runtime/org.argeo.slc.support.osgi/src/main/java/org/argeo/slc/osgi/OsgiExecutionModulesManager.java

   1 package org.argeo.slc.osgi;
   2
   3 import java.util.ArrayList;
   4 import java.util.Arrays;
   5 import java.util.HashMap;
   6 import java.util.HashSet;
   7 import java.util.Iterator;
   8 import java.util.List;
   9 import java.util.Map;
  10 import java.util.Properties;
  11 import java.util.Set;
  12
  13 import org.apache.commons.logging.Log;
  14 import org.apache.commons.logging.LogFactory;
  15 import org.argeo.slc.SlcException;
  16 import org.argeo.slc.build.BasicNameVersion;
  17 import org.argeo.slc.core.execution.AbstractExecutionModulesManager;
  18 import org.argeo.slc.core.execution.DefaultExecutionFlowDescriptorConverter;
  19 import org.argeo.slc.deploy.ModuleDescriptor;
  20 import org.argeo.slc.execution.ExecutionContext;
  21 import org.argeo.slc.execution.ExecutionFlow;
  22 import org.argeo.slc.execution.ExecutionFlowDescriptor;
  23 import org.argeo.slc.execution.ExecutionFlowDescriptorConverter;
  24 import org.argeo.slc.execution.ExecutionModuleDescriptor;
  25 import org.argeo.slc.execution.ExecutionModulesListener;
  26 import org.argeo.slc.process.RealizedFlow;
  27 import org.osgi.framework.Bundle;
  28 import org.osgi.framework.Constants;
  29 import org.osgi.framework.InvalidSyntaxException;
  30 import org.osgi.framework.ServiceReference;
  31 import org.osgi.util.tracker.ServiceTracker;
  32 import org.springframework.beans.factory.DisposableBean;
  33 import org.springframework.beans.factory.InitializingBean;
  34 import org.springframework.osgi.service.importer.OsgiServiceLifecycleListener;
  35
  36 public class OsgiExecutionModulesManager extends
  37                 AbstractExecutionModulesManager implements InitializingBean,
  38                 DisposableBean, OsgiServiceLifecycleListener {
  39
  40         static {
  41                 // Force usage of vanilla Xalan when in OSGi
  42                 // We would like to do it in a cleaner way
  43                 // but the integration of Xalan and Xerces in the JRE
  44                 // makes it very difficult
  45                 // Suggestions welcome!
  46                 Properties systemProperties = System.getProperties();
  47                 // if (!systemProperties
  48                 // .containsKey("javax.xml.parsers.DocumentBuilderFactory"))
  49                 // System.setProperty("javax.xml.parsers.DocumentBuilderFactory",
  50                 // "org.apache.xerces.jaxp.DocumentBuilderFactoryImpl");
  51                 // if
  52                 // (!systemProperties.containsKey("javax.xml.parsers.SAXParserFactory"))
  53                 // System.setProperty("javax.xml.parsers.SAXParserFactory",
  54                 // "org.apache.xerces.jaxp.SAXParserFactoryImpl");
  55                 if (!systemProperties
  56                                 .containsKey("javax.xml.transform.TransformerFactory"))
  57                         System.setProperty("javax.xml.transform.TransformerFactory",
  58                                         "org.apache.xalan.processor.TransformerFactoryImpl");
  59         }
  60
  61         private final static String PROPERTY_CACHE_SERVICES = "slc.osgi.execution.cacheServices";
  62
  63         private final static Log log = LogFactory
  64                         .getLog(OsgiExecutionModulesManager.class);
  65
  66         private BundlesManager bundlesManager;
  67         private ServiceTracker executionContextsTracker;
  68         private Map<OsgiBundle, ExecutionContext> executionContexts = new HashMap<OsgiBundle, ExecutionContext>();
  69         private Map<OsgiBundle, ExecutionFlowDescriptorConverter> executionFlowDescriptorConverters = new HashMap<OsgiBundle, ExecutionFlowDescriptorConverter>();
  70         private Map<OsgiBundle, Set<ExecutionFlow>> executionFlows = new HashMap<OsgiBundle, Set<ExecutionFlow>>();
  71         private ExecutionFlowDescriptorConverter defaultDescriptorConverter = new DefaultExecutionFlowDescriptorConverter();
  72
  73         private Boolean useCachedServices = Boolean.parseBoolean(System
  74                         .getProperty(PROPERTY_CACHE_SERVICES, "true"));
  75
  76         public synchronized ExecutionModuleDescriptor getExecutionModuleDescriptor(
  77                         String moduleName, String version) {
  78                 ExecutionModuleDescriptor md = new ExecutionModuleDescriptor();
  79                 if (useCachedServices) {
  80                         OsgiBundle osgiBundle = null;
  81                         BasicNameVersion nameVersion = new BasicNameVersion(moduleName,
  82                                         version);
  83                         bundles: for (Iterator<OsgiBundle> iterator = executionContexts
  84                                         .keySet().iterator(); iterator.hasNext();) {
  85                                 OsgiBundle ob = iterator.next();
  86                                 if (ob.equals(nameVersion)) {
  87                                         osgiBundle = ob;
  88                                         break bundles;
  89                                 }
  90                         }
  91                         if (osgiBundle == null)
  92                                 throw new SlcException("No execution module registered for "
  93                                                 + nameVersion);
  94                         md.setName(osgiBundle.getName());
  95                         md.setVersion(osgiBundle.getVersion());
  96                         md.setLabel(osgiBundle.getLabel());
  97                         md.setDescription(osgiBundle.getDescription());
  98                 } else {
  99                         md.setName(moduleName);
 100                         md.setVersion(version);
 101                         setMetadataFromBundle(md, null);
 102                 }
 103                 ExecutionFlowDescriptorConverter executionFlowDescriptorConverter = getExecutionFlowDescriptorConverter(
 104                                 moduleName, version);
 105                 if (executionFlowDescriptorConverter == null)
 106                         throw new SlcException("No flow converter found.");
 107                 executionFlowDescriptorConverter.addFlowsToDescriptor(md, listFlows(
 108                                 moduleName, version));
 109                 return md;
 110         }
 111
 112         public synchronized List<ExecutionModuleDescriptor> listExecutionModules() {
 113                 List<ExecutionModuleDescriptor> descriptors = new ArrayList<ExecutionModuleDescriptor>();
 114
 115                 if (useCachedServices) {
 116                         for (Iterator<OsgiBundle> iterator = executionContexts.keySet()
 117                                         .iterator(); iterator.hasNext();) {
 118                                 OsgiBundle osgiBundle = iterator.next();
 119                                 ExecutionModuleDescriptor md = new ExecutionModuleDescriptor();
 120                                 setMetadataFromBundle(md, bundlesManager
 121                                                 .findRelatedBundle(osgiBundle));
 122                                 descriptors.add(md);
 123                         }
 124                 } else {
 125                         ServiceReference[] arr = executionContextsTracker
 126                                         .getServiceReferences();
 127                         if (arr == null) {
 128                                 log.error("Tracker returned null.");
 129                                 return descriptors;
 130                         }
 131
 132                         List<ServiceReference> srs = Arrays.asList(arr);
 133                         // ServiceReference[] srs =
 134                         // executionContexts.getServiceReferences();
 135                         for (ServiceReference sr : srs) {
 136                                 ExecutionModuleDescriptor md = new ExecutionModuleDescriptor();
 137                                 setMetadataFromBundle(md, sr.getBundle());
 138                                 descriptors.add(md);
 139                         }
 140                 }
 141                 return descriptors;
 142         }
 143
 144         protected synchronized Map<String, ExecutionFlow> listFlows(
 145                         String moduleName, String moduleVersion) {
 146
 147                 Map<String, ExecutionFlow> flows = new HashMap<String, ExecutionFlow>();
 148                 if (useCachedServices) {
 149                         Set<ExecutionFlow> flowsT = executionFlows.get(new OsgiBundle(
 150                                         moduleName, moduleVersion));
 151                         for (ExecutionFlow flow : flowsT)
 152                                 flows.put(flow.getName(), flow);
 153                 } else {
 154
 155                         // TODO: use service trackers?
 156                         // String filter = OsgiFilterUtils.unifyFilter(ExecutionFlow.class,
 157                         // null);
 158
 159                         String filter = "(Bundle-SymbolicName=" + moduleName + ")";
 160                         ServiceReference[] sfs;
 161                         try {
 162                                 sfs = bundlesManager.getBundleContext().getServiceReferences(
 163                                                 ExecutionFlow.class.getName(), filter);
 164                         } catch (InvalidSyntaxException e) {
 165                                 throw new SlcException(
 166                                                 "Cannot retrieve service reference for flow " + filter,
 167                                                 e);
 168                         }
 169
 170                         for (ServiceReference sf : sfs) {
 171                                 ExecutionFlow flow = (ExecutionFlow) bundlesManager
 172                                                 .getBundleContext().getService(sf);
 173                                 flows.put(flow.getName(), flow);
 174                         }
 175                 }
 176                 return flows;
 177         }
 178
 179         protected ExecutionFlow findExecutionFlow(String moduleName,
 180                         String moduleVersion, String flowName) {
 181                 String filter = "(&(Bundle-SymbolicName=" + moduleName
 182                                 + ")(org.springframework.osgi.bean.name=" + flowName + "))";
 183                 return bundlesManager.getSingleServiceStrict(ExecutionFlow.class,
 184                                 filter);
 185         }
 186
 187         protected ExecutionContext findExecutionContext(String moduleName,
 188                         String moduleVersion) {
 189                 String filter = "(&(Bundle-SymbolicName=" + moduleName
 190                                 + ")(Bundle-Version=" + moduleVersion + "))";
 191                 return bundlesManager.getSingleServiceStrict(ExecutionContext.class,
 192                                 filter);
 193         }
 194
 195         protected ExecutionFlowDescriptorConverter findExecutionFlowDescriptorConverter(
 196                         String moduleName, String moduleVersion) {
 197
 198                 String filter = "(&(Bundle-SymbolicName=" + moduleName
 199                                 + ")(Bundle-Version=" + moduleVersion + "))";
 200                 return bundlesManager.getSingleService(
 201                                 ExecutionFlowDescriptorConverter.class, filter);
 202         }
 203
 204         public void setBundlesManager(BundlesManager bundlesManager) {
 205                 this.bundlesManager = bundlesManager;
 206         }
 207
 208         public void afterPropertiesSet() throws Exception {
 209                 if (!useCachedServices)
 210                         executionContextsTracker = bundlesManager
 211                                         .newTracker(ExecutionContext.class);
 212         }
 213
 214         public void destroy() throws Exception {
 215                 if (executionContextsTracker != null)
 216                         executionContextsTracker.close();
 217         }
 218
 219         /**
 220          * Builds a minimal realized flow, based on the provided information
 221          * (typically from the command line).
 222          *
 223          * @param module
 224          *            a bundle id, or a pattern contained in a bundle symbolic name
 225          * @param module
 226          *            the execution flow name
 227          * @return a minimal realized flow, to be used in an execution
 228          */
 229         public RealizedFlow findRealizedFlow(String module, String executionName) {
 230                 // First check whether we have a bundleId
 231                 Long bundleId = null;
 232                 try {
 233                         bundleId = Long.parseLong(module);
 234                 } catch (NumberFormatException e) {
 235                         // silent
 236                 }
 237
 238                 // Look for bundle names containing pattern
 239                 OsgiBundle bundle = null;
 240                 if (bundleId != null) {
 241                         bundle = bundlesManager.getBundle(bundleId);
 242                 } else {
 243                         bundle = bundlesManager.findFromPattern(module);
 244                 }
 245
 246                 if (bundle != null) {
 247                         RealizedFlow launch = new RealizedFlow();
 248                         launch.setModuleName(bundle.getName());
 249                         launch.setModuleVersion(bundle.getVersion());
 250                         ExecutionFlowDescriptor descriptor = new ExecutionFlowDescriptor();
 251                         descriptor.setName(executionName);
 252                         launch.setFlowDescriptor(descriptor);
 253                         return launch;
 254                 } else {
 255                         log
 256                                         .warn("Could not find any execution module matching these requirements.");
 257                         return null;
 258                 }
 259         }
 260
 261         public void updateAndExecute(RealizedFlow realizedFlow) {
 262                 OsgiBundle osgiBundle = new OsgiBundle(realizedFlow);
 263                 bundlesManager.upgradeSynchronous(osgiBundle);
 264                 execute(realizedFlow);
 265         }
 266
 267         protected synchronized ExecutionFlowDescriptorConverter getExecutionFlowDescriptorConverter(
 268                         String moduleName, String moduleVersion) {
 269                 if (useCachedServices) {
 270                         OsgiBundle osgiBundle = new OsgiBundle(moduleName, moduleVersion);
 271                         if (executionFlowDescriptorConverters.containsKey(osgiBundle))
 272                                 return executionFlowDescriptorConverters.get(osgiBundle);
 273                         else
 274                                 return defaultDescriptorConverter;
 275                 } else {
 276                         // Check whether a descriptor converter is published by this module
 277                         ExecutionFlowDescriptorConverter descriptorConverter = findExecutionFlowDescriptorConverter(
 278                                         moduleName, moduleVersion);
 279                         if (descriptorConverter == null)
 280                                 return defaultDescriptorConverter;
 281                         else
 282                                 return descriptorConverter;
 283                 }
 284         }
 285
 286         public void execute(RealizedFlow realizedFlow) {
 287                 if (log.isTraceEnabled())
 288                         log.trace("Executing " + realizedFlow);
 289
 290                 String moduleName = realizedFlow.getModuleName();
 291                 String moduleVersion = realizedFlow.getModuleVersion();
 292
 293                 Map<? extends String, ? extends Object> variablesToAdd = getExecutionFlowDescriptorConverter(
 294                                 moduleName, moduleVersion).convertValues(
 295                                 realizedFlow.getFlowDescriptor());
 296                 ExecutionContext executionContext = findExecutionContext(moduleName,
 297                                 moduleVersion);
 298                 for (String key : variablesToAdd.keySet())
 299                         executionContext.setVariable(key, variablesToAdd.get(key));
 300
 301                 ExecutionFlow flow = findExecutionFlow(moduleName, moduleVersion,
 302                                 realizedFlow.getFlowDescriptor().getName());
 303
 304                 //
 305                 // Actually runs the flow, IN THIS THREAD
 306                 //
 307                 flow.run();
 308                 //
 309                 //
 310                 //
 311         }
 312
 313         public ModuleDescriptor getModuleDescriptor(String moduleName,
 314                         String version) {
 315                 return getExecutionModuleDescriptor(moduleName, version);
 316         }
 317
 318         public List<ModuleDescriptor> listModules() {
 319                 Bundle[] bundles = bundlesManager.getBundleContext().getBundles();
 320                 List<ModuleDescriptor> lst = new ArrayList<ModuleDescriptor>();
 321                 for (Bundle bundle : bundles) {
 322                         ModuleDescriptor moduleDescriptor = new ModuleDescriptor();
 323                         setMetadataFromBundle(moduleDescriptor, bundle);
 324                         lst.add(moduleDescriptor);
 325                 }
 326                 return lst;
 327         }
 328
 329         protected void setMetadataFromBundle(ModuleDescriptor md, Bundle bundle) {
 330                 Bundle bdl = bundle;
 331                 if (bdl == null) {
 332                         if (md.getName() == null || md.getVersion() == null)
 333                                 throw new SlcException("Name and version not available.");
 334
 335                         Bundle[] bundles = bundlesManager.getBundleContext().getBundles();
 336                         for (Bundle b : bundles) {
 337                                 if (b.getSymbolicName().equals(md.getName())
 338                                                 && md.getVersion().equals(
 339                                                                 getHeaderSafe(b, Constants.BUNDLE_VERSION))) {
 340                                         bdl = b;
 341                                         break;
 342                                 }
 343                         }
 344
 345                 }
 346
 347                 if (bdl == null)
 348                         throw new SlcException("Cannot find bundle.");
 349
 350                 md.setName(bdl.getSymbolicName());
 351                 md.setVersion(getHeaderSafe(bdl, Constants.BUNDLE_VERSION));
 352                 md.setLabel(getHeaderSafe(bdl, Constants.BUNDLE_NAME));
 353                 md.setDescription(getHeaderSafe(bdl, Constants.BUNDLE_DESCRIPTION));
 354         }
 355
 356         private String getHeaderSafe(Bundle bundle, Object key) {
 357                 Object obj = bundle.getHeaders().get(key);
 358                 if (obj == null)
 359                         return null;
 360                 else
 361                         return obj.toString();
 362         }
 363
 364         @SuppressWarnings("unchecked")
 365         public synchronized void bind(Object service, Map properties)
 366                         throws Exception {
 367                 if (service instanceof ExecutionContext) {
 368                         ExecutionContext executionContext = (ExecutionContext) service;
 369                         OsgiBundle osgiBundle = asOsgiBundle(properties);
 370                         Bundle bundle = bundlesManager.findRelatedBundle(osgiBundle);
 371                         osgiBundle.setLabel(getHeaderSafe(bundle, Constants.BUNDLE_NAME));
 372                         osgiBundle.setDescription(getHeaderSafe(bundle,
 373                                         Constants.BUNDLE_DESCRIPTION));
 374                         executionContexts.put(osgiBundle, executionContext);
 375                         if (log.isTraceEnabled())
 376                                 log.debug("Registered execution context from " + osgiBundle);
 377                         // Notify
 378                         for (ExecutionModulesListener listener : getExecutionModulesListeners())
 379                                 listener.executionModuleAdded(osgiBundle, executionContext);
 380
 381                 } else if (service instanceof ExecutionFlow) {
 382                         ExecutionFlow executionFlow = (ExecutionFlow) service;
 383                         OsgiBundle osgiBundle = asOsgiBundle(properties);
 384                         if (!executionFlows.containsKey(osgiBundle)) {
 385                                 executionFlows.put(osgiBundle, new HashSet());
 386                         }
 387                         executionFlows.get(osgiBundle).add(executionFlow);
 388                         if (log.isTraceEnabled())
 389                                 log
 390                                                 .debug("Registered " + executionFlow + " from "
 391                                                                 + osgiBundle);
 392                         for (ExecutionModulesListener listener : getExecutionModulesListeners())
 393                                 listener.executionFlowAdded(osgiBundle, executionFlow);
 394
 395                 } else if (service instanceof ExecutionFlowDescriptorConverter) {
 396                         ExecutionFlowDescriptorConverter executionFlowDescriptorConverter = (ExecutionFlowDescriptorConverter) service;
 397                         OsgiBundle osgiBundle = asOsgiBundle(properties);
 398                         executionFlowDescriptorConverters.put(osgiBundle,
 399                                         executionFlowDescriptorConverter);
 400                         if (log.isTraceEnabled())
 401                                 log
 402                                                 .debug("Registered execution flow descriptor converter from "
 403                                                                 + osgiBundle);
 404                 } else {
 405                         // ignore
 406                 }
 407         }
 408
 409         @SuppressWarnings("unchecked")
 410         public synchronized void unbind(Object service, Map properties)
 411                         throws Exception {
 412                 if (service instanceof ExecutionContext) {
 413                         OsgiBundle osgiBundle = asOsgiBundle(properties);
 414                         if (executionContexts.containsKey(osgiBundle)) {
 415                                 ExecutionContext executionContext = executionContexts
 416                                                 .remove(osgiBundle);
 417                                 if (log.isTraceEnabled())
 418                                         log.debug("Removed execution context from " + osgiBundle);
 419                                 // Notify
 420                                 for (ExecutionModulesListener listener : getExecutionModulesListeners())
 421                                         listener.executionModuleRemoved(osgiBundle,
 422                                                         executionContext);
 423                         }
 424                 } else if (service instanceof ExecutionFlow) {
 425                         ExecutionFlow executionFlow = (ExecutionFlow) service;
 426                         OsgiBundle osgiBundle = asOsgiBundle(properties);
 427                         if (executionFlows.containsKey(osgiBundle)) {
 428                                 Set flows = executionFlows.get(osgiBundle);
 429                                 flows.remove(executionFlow);
 430                                 if (log.isTraceEnabled())
 431                                         log.debug("Removed " + executionFlow + " from "
 432                                                         + osgiBundle);
 433                                 if (flows.size() == 0) {
 434                                         executionFlows.remove(osgiBundle);
 435                                         if (log.isTraceEnabled())
 436                                                 log.debug("Removed flows set from " + osgiBundle);
 437                                 }
 438                                 for (ExecutionModulesListener listener : getExecutionModulesListeners())
 439                                         listener.executionFlowRemoved(osgiBundle, executionFlow);
 440                         }
 441                 } else if (service instanceof ExecutionFlowDescriptorConverter) {
 442                         OsgiBundle osgiBundle = asOsgiBundle(properties);
 443                         if (executionFlowDescriptorConverters.containsKey(osgiBundle)) {
 444                                 executionFlowDescriptorConverters.remove(osgiBundle);
 445                                 if (log.isTraceEnabled())
 446                                         log
 447                                                         .debug("Removed execution flow descriptor converter from "
 448                                                                         + osgiBundle);
 449                         }
 450                 } else {
 451                         // ignore
 452                 }
 453         }
 454
 455         @SuppressWarnings("unchecked")
 456         private OsgiBundle asOsgiBundle(Map properties) {
 457                 String bundleSymbolicName = checkAndGet(Constants.BUNDLE_SYMBOLICNAME,
 458                                 properties);
 459                 String bundleVersion = checkAndGet(Constants.BUNDLE_VERSION, properties);
 460                 return new OsgiBundle(bundleSymbolicName, bundleVersion);
 461         }
 462
 463         @SuppressWarnings("unchecked")
 464         private String checkAndGet(Object key, Map properties) {
 465                 if (!properties.containsKey(key) || properties.get(key) == null)
 466                         throw new SlcException(key + " not set in " + properties);
 467                 else
 468                         return properties.get(key).toString();
 469         }
 470
 471         public void setDefaultDescriptorConverter(
 472                         ExecutionFlowDescriptorConverter defaultDescriptorConverter) {
 473                 this.defaultDescriptorConverter = defaultDescriptorConverter;
 474         }
 475 }