runtime/org.argeo.slc.support.simple/src/main/java/org/argeo/slc/core/execution/DefaultModulesManager.java

   1 package org.argeo.slc.core.execution;
   2
   3 import java.util.ArrayList;
   4 import java.util.Iterator;
   5 import java.util.List;
   6
   7 import org.apache.commons.logging.Log;
   8 import org.apache.commons.logging.LogFactory;
   9 import org.argeo.slc.SlcException;
  10 import org.argeo.slc.execution.ExecutionFlowDescriptor;
  11 import org.argeo.slc.execution.ExecutionModule;
  12 import org.argeo.slc.execution.ExecutionModuleDescriptor;
  13 import org.argeo.slc.execution.ExecutionModulesManager;
  14 import org.argeo.slc.process.RealizedFlow;
  15 import org.argeo.slc.process.SlcExecution;
  16 import org.argeo.slc.process.SlcExecutionNotifier;
  17 import org.springframework.util.Assert;
  18
  19 public class DefaultModulesManager implements ExecutionModulesManager {
  20         private final static Log log = LogFactory
  21                         .getLog(DefaultModulesManager.class);
  22
  23         private List<ExecutionModule> executionModules = new ArrayList<ExecutionModule>();
  24         private List<SlcExecutionNotifier> slcExecutionNotifiers = new ArrayList<SlcExecutionNotifier>();
  25
  26         protected ExecutionModule getExecutionModule(String moduleName,
  27                         String version) {
  28                 for (ExecutionModule moduleT : executionModules) {
  29                         if (moduleT.getName().equals(moduleName)) {
  30                                 if (moduleT.getVersion().equals(version)) {
  31                                         return moduleT;
  32                                 }
  33                         }
  34                 }
  35                 return null;
  36         }
  37
  38         public ExecutionModuleDescriptor getExecutionModuleDescriptor(
  39                         String moduleName, String version) {
  40                 ExecutionModule module = getExecutionModule(moduleName, version);
  41
  42                 if(module==null)
  43                         throw new SlcException("Module "+moduleName+" ("+version+") not found");
  44
  45                 return module.getDescriptor();
  46         }
  47
  48         public List<ExecutionModule> listExecutionModules() {
  49                 return executionModules;
  50         }
  51
  52         public void setExecutionModules(List<ExecutionModule> executionModules) {
  53                 this.executionModules = executionModules;
  54         }
  55
  56         public void process(SlcExecution slcExecution) {
  57                 new ProcessThread(slcExecution).start();
  58         }
  59
  60         protected void dispatchUpdateStatus(SlcExecution slcExecution,
  61                         String oldStatus, String newStatus) {
  62                 for (Iterator<SlcExecutionNotifier> it = slcExecutionNotifiers
  63                                 .iterator(); it.hasNext();) {
  64                         it.next().updateStatus(slcExecution, oldStatus, newStatus);
  65                 }
  66         }
  67
  68         public void setSlcExecutionNotifiers(
  69                         List<SlcExecutionNotifier> slcExecutionNotifiers) {
  70                 this.slcExecutionNotifiers = slcExecutionNotifiers;
  71         }
  72
  73         private class ProcessThread extends Thread {
  74                 private final SlcExecution slcExecution;
  75                 private final List<RealizedFlow> flowsToProcess = new ArrayList<RealizedFlow>();
  76
  77                 public ProcessThread(SlcExecution slcExecution) {
  78                         this.slcExecution = slcExecution;
  79                 }
  80
  81                 public void run() {
  82                         log.info("\n##\n## Process SLC Execution " + slcExecution
  83                                         + "\n##\n");
  84
  85                         // FIXME: hack to let the SlcExecution be registered on server
  86                         try {
  87                                 Thread.sleep(500);
  88                         } catch (InterruptedException e1) {
  89                                 // silent
  90                         }
  91
  92                         slcExecution.setStatus(SlcExecution.STATUS_RUNNING);
  93                         dispatchUpdateStatus(slcExecution, SlcExecution.STATUS_SCHEDULED,
  94                                         SlcExecution.STATUS_RUNNING);
  95
  96                         flowsToProcess.addAll(slcExecution.getRealizedFlows());
  97
  98                         while (flowsToProcess.size() > 0) {
  99                                 RealizedFlow flow = flowsToProcess.remove(0);
 100                                 ExecutionModule module = getExecutionModule(flow
 101                                                 .getModuleName(), flow.getModuleVersion());
 102                                 if (module != null) {
 103                                         ExecutionThread thread = new ExecutionThread(this, flow
 104                                                         .getFlowDescriptor(), module);
 105                                         thread.start();
 106                                 } else {
 107                                         throw new SlcException("ExecutionModule "
 108                                                         + flow.getModuleName() + ", version "
 109                                                         + flow.getModuleVersion() + " not found.");
 110                                 }
 111
 112                                 synchronized (this) {
 113                                         try {
 114                                                 wait();
 115                                         } catch (InterruptedException e) {
 116                                                 // silent
 117                                         }
 118                                 }
 119                         }
 120
 121                         slcExecution.setStatus(SlcExecution.STATUS_RUNNING);
 122                         dispatchUpdateStatus(slcExecution, SlcExecution.STATUS_RUNNING,
 123                                         SlcExecution.STATUS_FINISHED);
 124                         /*
 125                          * for (RealizedFlow flow : slcExecution.getRealizedFlows()) {
 126                          * ExecutionModule module = getExecutionModule(flow.getModuleName(),
 127                          * flow.getModuleVersion()); if (module != null) { ExecutionThread
 128                          * thread = new ExecutionThread(flow .getFlowDescriptor(), module);
 129                          * thread.start(); } else { throw new
 130                          * SlcException("ExecutionModule " + flow.getModuleName() +
 131                          * ", version " + flow.getModuleVersion() + " not found."); } }
 132                          */
 133                 }
 134
 135                 public synchronized void flowCompleted() {
 136                         notifyAll();
 137                 }
 138         }
 139
 140         private class ExecutionThread extends Thread {
 141                 private final ExecutionFlowDescriptor executionFlowDescriptor;
 142                 private final ExecutionModule executionModule;
 143                 private final ProcessThread processThread;
 144
 145                 public ExecutionThread(ProcessThread processThread,
 146                                 ExecutionFlowDescriptor executionFlowDescriptor,
 147                                 ExecutionModule executionModule) {
 148                         super("SLC Execution #" /* + executionContext.getUuid() */);
 149                         this.executionFlowDescriptor = executionFlowDescriptor;
 150                         this.executionModule = executionModule;
 151                         this.processThread = processThread;
 152                 }
 153
 154                 public void run() {
 155                         try {
 156                                 executionModule.execute(executionFlowDescriptor);
 157                         } catch (Exception e) {
 158                                 // TODO: re-throw exception ?
 159                                 log.error("Execution "/* + executionContext.getUuid() */
 160                                                 + " failed.", e);
 161                         } finally {
 162                                 processThread.flowCompleted();
 163                         }
 164                 }
 165         }
 166 }