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