1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.argeo.slc.osgi;
17
18 import java.lang.management.ManagementFactory;
19 import java.util.ArrayList;
20 import java.util.Dictionary;
21 import java.util.Enumeration;
22 import java.util.HashMap;
23 import java.util.HashSet;
24 import java.util.Iterator;
25 import java.util.List;
26 import java.util.Map;
27 import java.util.Set;
28
29 import javax.management.MBeanServer;
30 import javax.management.ObjectName;
31 import javax.management.StandardMBean;
32
33 import org.apache.commons.logging.Log;
34 import org.apache.commons.logging.LogFactory;
35 import org.argeo.slc.DefaultNameVersion;
36 import org.argeo.slc.NameVersion;
37 import org.argeo.slc.SlcException;
38 import org.argeo.slc.core.execution.AbstractExecutionModulesManager;
39 import org.argeo.slc.core.execution.DefaultExecutionFlowDescriptorConverter;
40 import org.argeo.slc.deploy.Module;
41 import org.argeo.slc.deploy.ModuleDescriptor;
42 import org.argeo.slc.execution.ExecutionContext;
43 import org.argeo.slc.execution.ExecutionFlow;
44 import org.argeo.slc.execution.ExecutionFlowDescriptor;
45 import org.argeo.slc.execution.ExecutionFlowDescriptorConverter;
46 import org.argeo.slc.execution.ExecutionModuleDescriptor;
47 import org.argeo.slc.execution.ExecutionModulesListener;
48 import org.argeo.slc.execution.RealizedFlow;
49 import org.eclipse.gemini.blueprint.service.importer.OsgiServiceLifecycleListener;
50 import org.osgi.framework.Bundle;
51 import org.osgi.framework.BundleEvent;
52 import org.osgi.framework.BundleException;
53 import org.osgi.framework.BundleListener;
54 import org.osgi.framework.Constants;
55 import org.osgi.framework.launch.Framework;
56 import org.springframework.context.ApplicationContext;
57
58
59 public class OsgiExecutionModulesManager extends
60 AbstractExecutionModulesManager implements
61 OsgiServiceLifecycleListener, BundleListener {
62
63 private final static Log log = LogFactory
64 .getLog(OsgiExecutionModulesManager.class);
65
66 private BundlesManager bundlesManager;
67 private Map<OsgiBundle, ExecutionContext> executionContexts = new HashMap<OsgiBundle, ExecutionContext>();
68 private Map<OsgiBundle, ExecutionFlowDescriptorConverter> executionFlowDescriptorConverters = new HashMap<OsgiBundle, ExecutionFlowDescriptorConverter>();
69 private Map<OsgiBundle, Set<ExecutionFlow>> executionFlows = new HashMap<OsgiBundle, Set<ExecutionFlow>>();
70 private ExecutionFlowDescriptorConverter defaultDescriptorConverter = new DefaultExecutionFlowDescriptorConverter();
71
72 private List<ExecutionModulesListener> executionModulesListeners = new ArrayList<ExecutionModulesListener>();
73
74 private Boolean registerFlowsToJmx = false;
75
76 public void init() throws Exception {
77 bundlesManager.getBundleContext().addBundleListener(this);
78
79 final String module = System.getProperty(UNIQUE_LAUNCH_MODULE_PROPERTY);
80 final String flow = System.getProperty(UNIQUE_LAUNCH_FLOW_PROPERTY);
81 if (module != null) {
82
83 new Thread("Unique Flow") {
84 @Override
85 public void run() {
86 executeFlowAndExit(module, null, flow);
87 }
88 }.start();
89 }
90 }
91
92 public void destroy() {
93 bundlesManager.getBundleContext().removeBundleListener(this);
94 }
95
96
97 protected void executeFlowAndExit(final String module,
98 final String version, final String flow) {
99 if (log.isDebugEnabled())
100 log.debug("Launch unique flow " + flow + " from module " + module);
101 try {
102 OsgiBundle osgiBundle = bundlesManager.findFromPattern(module);
103 if (osgiBundle == null)
104 throw new SlcException("No OSGi bundle found for " + module);
105
106
107 start(osgiBundle);
108
109 RealizedFlow lastLaunch = findRealizedFlow(module, flow);
110 if (lastLaunch == null)
111 throw new SlcException("Cannot find launch for " + module + " "
112 + flow);
113 execute(lastLaunch);
114 } catch (Exception e) {
115 log.error(
116 "Error in unique flow " + flow + " from module " + module,
117 e);
118 } finally {
119 if (log.isDebugEnabled())
120 log.debug("Shutdown OSGi runtime...");
121 Framework framework = (Framework) bundlesManager.getBundleContext()
122 .getBundle(0);
123 try {
124
125 framework.stop();
126
127 framework.waitForStop(60 * 1000);
128
129 System.exit(0);
130 } catch (Exception e) {
131 e.printStackTrace();
132 System.exit(1);
133 }
134 }
135 }
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154 public synchronized ExecutionModuleDescriptor getExecutionModuleDescriptor(
155 String moduleName, String version) {
156 ExecutionModuleDescriptor md = new ExecutionModuleDescriptor();
157 OsgiBundle osgiBundle = null;
158 DefaultNameVersion nameVersion = new DefaultNameVersion(moduleName,
159 version);
160 bundles: for (Iterator<OsgiBundle> iterator = executionContexts
161 .keySet().iterator(); iterator.hasNext();) {
162 OsgiBundle ob = iterator.next();
163 if (nameVersion.getVersion() != null) {
164 if (ob.equals(nameVersion)) {
165 osgiBundle = ob;
166 break bundles;
167 }
168 } else {
169 if (ob.getName().equals(nameVersion.getName())) {
170 osgiBundle = ob;
171 break bundles;
172 }
173 }
174 }
175 if (osgiBundle == null)
176 throw new SlcException("No execution module registered for "
177 + nameVersion);
178 md.setName(osgiBundle.getName());
179 md.setVersion(osgiBundle.getVersion());
180 md.setTitle(osgiBundle.getTitle());
181 md.setDescription(osgiBundle.getDescription());
182
183 ExecutionFlowDescriptorConverter executionFlowDescriptorConverter = getExecutionFlowDescriptorConverter(
184 moduleName, version);
185 if (executionFlowDescriptorConverter == null)
186 throw new SlcException("No flow converter found.");
187 executionFlowDescriptorConverter.addFlowsToDescriptor(md,
188 listFlows(moduleName, version));
189 return md;
190 }
191
192 public synchronized List<ExecutionModuleDescriptor> listExecutionModules() {
193 List<ExecutionModuleDescriptor> descriptors = new ArrayList<ExecutionModuleDescriptor>();
194
195 for (Iterator<OsgiBundle> iterator = executionContexts.keySet()
196 .iterator(); iterator.hasNext();) {
197 OsgiBundle osgiBundle = iterator.next();
198 ExecutionModuleDescriptor md = new ExecutionModuleDescriptor();
199 setMetadataFromBundle(md,
200 bundlesManager.findRelatedBundle(osgiBundle));
201 descriptors.add(md);
202 }
203 return descriptors;
204 }
205
206 protected synchronized Map<String, ExecutionFlow> listFlows(
207 String moduleName, String moduleVersion) {
208
209 Map<String, ExecutionFlow> flows = new HashMap<String, ExecutionFlow>();
210 OsgiBundle key = bundlesManager.findRelatedBundle(moduleName,
211 moduleVersion);
212 if (!executionFlows.containsKey(key))
213 return flows;
214 Set<ExecutionFlow> flowsT = executionFlows.get(key);
215 for (ExecutionFlow flow : flowsT)
216 flows.put(flow.getName(), flow);
217 return flows;
218 }
219
220 protected ExecutionFlow findExecutionFlow(String moduleName,
221 String moduleVersion, String flowName) {
222 String filter = moduleVersion == null || moduleVersion.equals("0.0.0") ? "(&(Bundle-SymbolicName="
223 + moduleName
224 + ")(org.eclipse.gemini.blueprint.bean.name="
225 + flowName + "))"
226 : "(&(Bundle-SymbolicName=" + moduleName + ")(Bundle-Version="
227 + moduleVersion
228 + ")(org.eclipse.gemini.blueprint.bean.name="
229 + flowName + "))";
230 return bundlesManager.getSingleServiceStrict(ExecutionFlow.class,
231 filter, true);
232 }
233
234 protected ExecutionContext findExecutionContext(String moduleName,
235 String moduleVersion) {
236 String filter = moduleFilter(moduleName, moduleVersion);
237 return bundlesManager.getSingleServiceStrict(ExecutionContext.class,
238 filter, true);
239 }
240
241 protected ExecutionFlowDescriptorConverter findExecutionFlowDescriptorConverter(
242 String moduleName, String moduleVersion) {
243 String filter = moduleFilter(moduleName, moduleVersion);
244 return bundlesManager.getSingleService(
245 ExecutionFlowDescriptorConverter.class, filter, false);
246 }
247
248
249 protected String moduleFilter(String moduleName, String moduleVersion) {
250 return moduleVersion == null || moduleVersion.equals("0.0.0") ? "(Bundle-SymbolicName="
251 + moduleName + ")"
252 : "(&(Bundle-SymbolicName=" + moduleName + ")(Bundle-Version="
253 + moduleVersion + "))";
254
255 }
256
257
258
259
260
261
262
263
264
265
266
267 public RealizedFlow findRealizedFlow(String module, String executionName) {
268
269 Long bundleId = null;
270 try {
271 bundleId = Long.parseLong(module);
272 } catch (NumberFormatException e) {
273
274 }
275
276
277 OsgiBundle bundle = null;
278 if (bundleId != null) {
279 bundle = bundlesManager.getBundle(bundleId);
280 } else {
281 bundle = bundlesManager.findFromPattern(module);
282 }
283
284 if (bundle != null) {
285 RealizedFlow launch = new RealizedFlow();
286 launch.setModuleName(bundle.getName());
287 launch.setModuleVersion(bundle.getVersion());
288 ExecutionFlowDescriptor descriptor = new ExecutionFlowDescriptor();
289 descriptor.setName(executionName);
290 launch.setFlowDescriptor(descriptor);
291 return launch;
292 } else {
293 log.warn("Could not find any execution module matching these requirements.");
294 return null;
295 }
296 }
297
298 public void upgrade(NameVersion nameVersion) {
299 OsgiBundle osgiBundle = new OsgiBundle(nameVersion);
300 bundlesManager.upgradeSynchronous(osgiBundle);
301 }
302
303 protected synchronized ExecutionFlowDescriptorConverter getExecutionFlowDescriptorConverter(
304 String moduleName, String moduleVersion) {
305 return findExecutionFlowDescriptorConverter(moduleName, moduleVersion);
306
307
308 }
309
310 protected synchronized ExecutionFlowDescriptorConverter getExecutionFlowDescriptorConverter(
311 OsgiBundle osgiBundle) {
312 if (executionFlowDescriptorConverters.containsKey(osgiBundle))
313 return executionFlowDescriptorConverters.get(osgiBundle);
314 else
315 return defaultDescriptorConverter;
316 }
317
318 public ModuleDescriptor getModuleDescriptor(String moduleName,
319 String version) {
320 return getExecutionModuleDescriptor(moduleName, version);
321 }
322
323 public List<ModuleDescriptor> listModules() {
324 Bundle[] bundles = bundlesManager.getBundleContext().getBundles();
325 List<ModuleDescriptor> lst = new ArrayList<ModuleDescriptor>();
326 for (Bundle bundle : bundles) {
327 ModuleDescriptor moduleDescriptor = new ModuleDescriptor();
328 setMetadataFromBundle(moduleDescriptor, bundle);
329 lst.add(moduleDescriptor);
330 }
331 return lst;
332 }
333
334 public void start(NameVersion nameVersion) {
335 try {
336 Bundle bundle = bundlesManager.findRelatedBundle(new OsgiBundle(
337 nameVersion));
338 if (bundle == null)
339 throw new SlcException("Could not find bundle for "
340 + nameVersion);
341
342 bundlesManager.startSynchronous(bundle);
343 if (isSpringInstrumented(bundle)) {
344
345 String filter = "(Bundle-SymbolicName="
346 + bundle.getSymbolicName() + ")";
347 try {
348 bundlesManager.getServiceRefSynchronous(
349 ApplicationContext.class, filter);
350 } catch (Exception e) {
351
352 bundle.stop();
353 throw e;
354 }
355 }
356 } catch (Exception e) {
357 throw new SlcException("Cannot start " + nameVersion, e);
358 }
359 }
360
361
362 private Boolean isSpringInstrumented(Bundle bundle) {
363 Dictionary<?, ?> headers = bundle.getHeaders();
364 if (headers != null && headers.get("Spring-Context") != null)
365 return true;
366 Enumeration<?> springEntryPaths = bundle
367 .getEntryPaths("/META-INF/spring");
368 if (springEntryPaths != null && springEntryPaths.hasMoreElements())
369 return true;
370 return false;
371 }
372
373 public void stop(NameVersion nameVersion) {
374 try {
375 Bundle bundle = bundlesManager.findRelatedBundle(new OsgiBundle(
376 nameVersion));
377 bundlesManager.stopSynchronous(bundle);
378 } catch (BundleException e) {
379 throw new SlcException("Cannot stop " + nameVersion, e);
380 }
381 }
382
383 protected void setMetadataFromBundle(ModuleDescriptor md, Bundle bundle) {
384 Bundle bdl = bundle;
385 if (bdl == null) {
386 if (md.getName() == null || md.getVersion() == null)
387 throw new SlcException("Name and version not available.");
388
389 Bundle[] bundles = bundlesManager.getBundleContext().getBundles();
390 for (Bundle b : bundles) {
391 if (b.getSymbolicName().equals(md.getName())
392 && md.getVersion().equals(
393 getHeaderSafe(b, Constants.BUNDLE_VERSION))) {
394 bdl = b;
395 break;
396 }
397 }
398
399 }
400
401 if (bdl == null)
402 throw new SlcException("Cannot find bundle.");
403
404 md.setName(bdl.getSymbolicName());
405 md.setVersion(getHeaderSafe(bdl, Constants.BUNDLE_VERSION));
406 md.setTitle(getHeaderSafe(bdl, Constants.BUNDLE_NAME));
407 md.setDescription(getHeaderSafe(bdl, Constants.BUNDLE_DESCRIPTION));
408
409
410 Dictionary<?, ?> headers = bundle.getHeaders();
411 Enumeration<?> keys = headers.keys();
412 while (keys.hasMoreElements()) {
413 Object key = keys.nextElement();
414 Object value = headers.get(key);
415 if (value != null)
416 md.getMetadata().put(key.toString(), value.toString());
417 }
418
419
420 if (bundle.getState() == Bundle.ACTIVE
421 || bundle.getState() == Bundle.STARTING)
422 md.setStarted(true);
423 else
424 md.setStarted(false);
425 }
426
427 private String getHeaderSafe(Bundle bundle, Object key) {
428 Object obj = bundle.getHeaders().get(key);
429 if (obj == null)
430 return null;
431 else
432 return obj.toString();
433 }
434
435
436
437
438
439
440 public synchronized void register(ExecutionContext executionContext,
441 Map<String, String> properties) {
442 OsgiBundle osgiBundle = asOsgiBundle(properties);
443 Bundle bundle = bundlesManager.findRelatedBundle(osgiBundle);
444 osgiBundle.setTitle(getHeaderSafe(bundle, Constants.BUNDLE_NAME));
445 osgiBundle.setDescription(getHeaderSafe(bundle,
446 Constants.BUNDLE_DESCRIPTION));
447 executionContexts.put(osgiBundle, executionContext);
448 if (log.isTraceEnabled())
449 log.trace("Registered execution context from " + osgiBundle);
450
451 ModuleDescriptor md = osgiBundle.getModuleDescriptor();
452 md.setStarted(true);
453 for (ExecutionModulesListener listener : executionModulesListeners)
454 listener.executionModuleAdded(md);
455 }
456
457
458 public synchronized void unregister(ExecutionContext executionContext,
459 Map<String, String> properties) {
460
461 if (properties == null)
462 return;
463 OsgiBundle osgiBundle = asOsgiBundle(properties);
464 if (executionContexts.containsKey(osgiBundle)) {
465 executionContexts.remove(osgiBundle);
466 if (log.isTraceEnabled())
467 log.trace("Removed execution context from " + osgiBundle);
468
469 ModuleDescriptor md = osgiBundle.getModuleDescriptor();
470 md.setStarted(false);
471 for (ExecutionModulesListener listener : executionModulesListeners)
472 listener.executionModuleRemoved(md);
473 }
474 }
475
476
477 public synchronized void register(ExecutionFlow executionFlow,
478 Map<String, String> properties) {
479 OsgiBundle osgiBundle = asOsgiBundle(properties);
480 if (!executionFlows.containsKey(osgiBundle)) {
481 executionFlows.put(osgiBundle, new HashSet<ExecutionFlow>());
482 }
483 executionFlows.get(osgiBundle).add(executionFlow);
484 if (log.isTraceEnabled())
485 log.trace("Registered " + executionFlow + " from " + osgiBundle);
486
487
488 if (registerFlowsToJmx)
489 registerMBean(osgiBundle, executionFlow);
490 ExecutionFlowDescriptorConverter efdc = getExecutionFlowDescriptorConverter(osgiBundle);
491 for (ExecutionModulesListener listener : executionModulesListeners)
492 listener.executionFlowAdded(osgiBundle.getModuleDescriptor(),
493 efdc.getExecutionFlowDescriptor(executionFlow));
494 }
495
496
497 public synchronized void unregister(ExecutionFlow executionFlow,
498 Map<String, String> properties) {
499
500 if (properties == null)
501 return;
502 OsgiBundle osgiBundle = asOsgiBundle(properties);
503 if (executionFlows.containsKey(osgiBundle)) {
504 Set<ExecutionFlow> flows = executionFlows.get(osgiBundle);
505 flows.remove(executionFlow);
506 if (log.isTraceEnabled())
507 log.trace("Removed " + executionFlow + " from " + osgiBundle);
508 if (flows.size() == 0) {
509 executionFlows.remove(osgiBundle);
510 if (log.isTraceEnabled())
511 log.trace("Removed flows set from " + osgiBundle);
512 }
513
514
515 if (registerFlowsToJmx)
516 unregisterMBean(osgiBundle, executionFlow);
517 ExecutionFlowDescriptorConverter efdc = getExecutionFlowDescriptorConverter(osgiBundle);
518 for (ExecutionModulesListener listener : executionModulesListeners)
519 listener.executionFlowRemoved(osgiBundle.getModuleDescriptor(),
520 efdc.getExecutionFlowDescriptor(executionFlow));
521 }
522 }
523
524
525 public synchronized void register(
526 ExecutionModulesListener executionModulesListener,
527 Map<String, String> properties) {
528
529 for (OsgiBundle osgiBundle : executionContexts.keySet()) {
530 executionModulesListener.executionModuleAdded(osgiBundle
531 .getModuleDescriptor());
532 }
533 for (OsgiBundle osgiBundle : executionFlows.keySet()) {
534 ExecutionFlowDescriptorConverter efdc = getExecutionFlowDescriptorConverter(osgiBundle);
535 for (ExecutionFlow executionFlow : executionFlows.get(osgiBundle))
536 executionModulesListener.executionFlowAdded(
537 osgiBundle.getModuleDescriptor(),
538 efdc.getExecutionFlowDescriptor(executionFlow));
539 }
540 executionModulesListeners.add(executionModulesListener);
541 }
542
543
544 public synchronized void unregister(
545 ExecutionModulesListener executionModulesListener,
546 Map<String, String> properties) {
547 executionModulesListeners.remove(executionModulesListener);
548 }
549
550
551
552
553
554 public void bundleChanged(BundleEvent evt) {
555 Bundle bundle = evt.getBundle();
556 if (bundle.getHeaders().get(
557 ExecutionModuleDescriptor.SLC_EXECUTION_MODULE) != null) {
558 OsgiBundle osgiBundle = new OsgiBundle(bundle);
559 if (evt.getType() == BundleEvent.INSTALLED)
560 for (ExecutionModulesListener listener : executionModulesListeners)
561 listener.executionModuleAdded(osgiBundle
562 .getModuleDescriptor());
563 else if (evt.getType() == BundleEvent.UNINSTALLED)
564 for (ExecutionModulesListener listener : executionModulesListeners)
565 listener.executionModuleRemoved(osgiBundle
566 .getModuleDescriptor());
567 }
568
569 }
570
571 @SuppressWarnings({ "rawtypes" })
572 public synchronized void bind(Object service, Map properties)
573 throws Exception {
574 if (service instanceof ExecutionFlowDescriptorConverter) {
575 ExecutionFlowDescriptorConverter executionFlowDescriptorConverter = (ExecutionFlowDescriptorConverter) service;
576 OsgiBundle osgiBundle = asOsgiBundle(properties);
577 executionFlowDescriptorConverters.put(osgiBundle,
578 executionFlowDescriptorConverter);
579 if (log.isTraceEnabled())
580 log.debug("Registered execution flow descriptor converter from "
581 + osgiBundle);
582 } else {
583
584 }
585 }
586
587 @SuppressWarnings("rawtypes")
588 public synchronized void unbind(Object service, Map properties)
589 throws Exception {
590 if (service instanceof ExecutionFlowDescriptorConverter) {
591 OsgiBundle osgiBundle = asOsgiBundle(properties);
592 if (executionFlowDescriptorConverters.containsKey(osgiBundle)) {
593 executionFlowDescriptorConverters.remove(osgiBundle);
594 if (log.isTraceEnabled())
595 log.debug("Removed execution flow descriptor converter from "
596 + osgiBundle);
597 }
598 } else {
599
600 }
601 }
602
603
604
605
606 protected MBeanServer getMBeanServer() {
607 return ManagementFactory.getPlatformMBeanServer();
608 }
609
610 public void registerMBean(Module module, ExecutionFlow executionFlow) {
611 try {
612 StandardMBean mbean = new StandardMBean(executionFlow,
613 ExecutionFlow.class);
614 getMBeanServer().registerMBean(mbean,
615 flowMBeanName(module, executionFlow));
616 } catch (Exception e) {
617 String msg = "Cannot register execution flow " + executionFlow
618 + " as mbean";
619 throw new SlcException(msg, e);
620 }
621 }
622
623 public void unregisterMBean(Module module, ExecutionFlow executionFlow) {
624 try {
625 getMBeanServer().unregisterMBean(
626 flowMBeanName(module, executionFlow));
627 } catch (Exception e) {
628 String msg = "Cannot unregister execution flow " + executionFlow
629 + " as mbean";
630 throw new SlcException(msg, e);
631 }
632 }
633
634 protected ObjectName flowMBeanName(Module module,
635 ExecutionFlow executionFlow) {
636 String executionModulesPrefix = "SLCExecutionModules";
637
638 String name = executionFlow.getName();
639
640
641
642
643
644 StringBuffer buf = new StringBuffer(executionModulesPrefix + ":"
645 + "module=" + module.getName() + " [" + module.getVersion()
646 + "],");
647
648
649
650
651
652
653
654
655
656
657
658
659
660 buf.append("name=").append(name);
661 try {
662 return new ObjectName(buf.toString());
663 } catch (Exception e) {
664 throw new SlcException("Cannot generate object name based on "
665 + buf, e);
666 }
667 }
668
669
670
671
672 @SuppressWarnings("rawtypes")
673 private OsgiBundle asOsgiBundle(Map properties) {
674 String bundleSymbolicName = checkAndGet(Constants.BUNDLE_SYMBOLICNAME,
675 properties);
676 String bundleVersion = checkAndGet(Constants.BUNDLE_VERSION, properties);
677 return new OsgiBundle(bundleSymbolicName, bundleVersion);
678 }
679
680 @SuppressWarnings("rawtypes")
681 private String checkAndGet(Object key, Map properties) {
682 if (!properties.containsKey(key) || properties.get(key) == null)
683 throw new SlcException(key + " not set in " + properties);
684 else
685 return properties.get(key).toString();
686 }
687
688 public void setBundlesManager(BundlesManager bundlesManager) {
689 this.bundlesManager = bundlesManager;
690 }
691
692 public void setDefaultDescriptorConverter(
693 ExecutionFlowDescriptorConverter defaultDescriptorConverter) {
694 this.defaultDescriptorConverter = defaultDescriptorConverter;
695 }
696
697 public void setRegisterFlowsToJmx(Boolean registerFlowsToJmx) {
698 this.registerFlowsToJmx = registerFlowsToJmx;
699 }
700
701 }