runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/AbstractExecutionModulesManager.java

   1 package org.argeo.slc.core.execution;
   2
   3 import java.util.ArrayList;
   4 import java.util.List;
   5
   6 import org.argeo.slc.core.execution.internal.ProcessThread;
   7 import org.argeo.slc.execution.ExecutionModulesManager;
   8 import org.argeo.slc.process.SlcExecution;
   9 import org.argeo.slc.process.SlcExecutionNotifier;
  10
  11 public abstract class AbstractExecutionModulesManager implements
  12                 ExecutionModulesManager {
  13         private List<SlcExecutionNotifier> slcExecutionNotifiers = new ArrayList<SlcExecutionNotifier>();
  14         private ThreadGroup processesThreadGroup = new ThreadGroup("Processes");
  15
  16         public void process(SlcExecution slcExecution) {
  17                 new ProcessThread(this, slcExecution).start();
  18         }
  19 /*
  20         protected void dispatchUpdateStatus(SlcExecution slcExecution,
  21                         String oldStatus, String newStatus) {
  22                 for (Iterator<SlcExecutionNotifier> it = slcExecutionNotifiers
  23                                 .iterator(); it.hasNext();) {
  24                         it.next().updateStatus(slcExecution, oldStatus, newStatus);
  25                 }
  26         }
  27
  28         protected synchronized void dispatchAddStep(SlcExecution slcExecution,
  29                         SlcExecutionStep step) {
  30                 slcExecution.getSteps().add(step);
  31                 List<SlcExecutionStep> steps = new ArrayList<SlcExecutionStep>();
  32                 steps.add(step);
  33                 for (Iterator<SlcExecutionNotifier> it = slcExecutionNotifiers
  34                                 .iterator(); it.hasNext();) {
  35                         it.next().addSteps(slcExecution, steps);
  36                 }
  37         }*/
  38
  39         public void setSlcExecutionNotifiers(
  40                         List<SlcExecutionNotifier> slcExecutionNotifiers) {
  41                 this.slcExecutionNotifiers = slcExecutionNotifiers;
  42         }
  43 /*
  44         protected static void addFlowsToDescriptor(ExecutionModuleDescriptor md,
  45                         Map<String, ExecutionFlow> executionFlows) {
  46                 // TODO: put this in a separate configurable object
  47                 for (String name : executionFlows.keySet()) {
  48                         ExecutionFlow executionFlow = executionFlows.get(name);
  49
  50                         Assert.notNull(executionFlow.getName());
  51                         Assert.state(name.equals(executionFlow.getName()));
  52
  53                         ExecutionSpec executionSpec = executionFlow.getExecutionSpec();
  54                         Assert.notNull(executionSpec);
  55                         Assert.notNull(executionSpec.getName());
  56
  57                         Map<String, Object> values = new TreeMap<String, Object>();
  58                         for (String key : executionSpec.getAttributes().keySet()) {
  59                                 ExecutionSpecAttribute attribute = executionSpec
  60                                                 .getAttributes().get(key);
  61
  62                                 if (executionFlow.isSetAsParameter(key)) {
  63                                         Object value = executionFlow.getParameter(key);
  64                                         if (attribute instanceof PrimitiveSpecAttribute) {
  65                                                 PrimitiveValue primitiveValue = new PrimitiveValue();
  66                                                 primitiveValue
  67                                                                 .setType(((PrimitiveSpecAttribute) attribute)
  68                                                                                 .getType());
  69                                                 primitiveValue.setValue(value);
  70                                                 values.put(key, primitiveValue);
  71                                         } else if (attribute instanceof RefSpecAttribute) {
  72                                                 RefValue refValue = new RefValue();
  73                                                 if (value instanceof ScopedObject) {
  74                                                         refValue.setLabel("RUNTIME "
  75                                                                         + value.getClass().getName());
  76                                                 } else {
  77                                                         refValue.setLabel("STATIC "
  78                                                                         + value.getClass().getName());
  79                                                 }
  80                                                 values.put(key, refValue);
  81                                         } else {
  82                                                 throw new SlcException("Unkown spec attribute type "
  83                                                                 + attribute.getClass());
  84                                         }
  85                                 }
  86
  87                         }
  88
  89                         ExecutionFlowDescriptor efd = new ExecutionFlowDescriptor(name,
  90                                         values, executionSpec);
  91                         if (executionFlow.getPath() != null)
  92                                 efd.setPath(executionFlow.getPath());
  93
  94                         // Add execution spec if necessary
  95                         if (!md.getExecutionSpecs().contains(executionSpec))
  96                                 md.getExecutionSpecs().add(executionSpec);
  97
  98                         // Add execution flow
  99                         md.getExecutionFlows().add(efd);
 100                 }
 101         }
 102 */
 103         /**
 104          * Thread of the SLC Process, starting the sub executions. private class
 105          * ProcessThread extends Thread { private final SlcExecution slcProcess;
 106          * private final ThreadGroup processThreadGroup; private final
 107          * List<RealizedFlow> flowsToProcess = new ArrayList<RealizedFlow>();
 108          *
 109          * public ProcessThread(ThreadGroup processesThreadGroup, SlcExecution
 110          * slcExecution) { super(processesThreadGroup, "SLC Process #" +
 111          * slcExecution.getUuid()); this.slcProcess = slcExecution;
 112          * processThreadGroup = new ThreadGroup("SLC Process #" +
 113          * slcExecution.getUuid() + " thread group"); }
 114          *
 115          * public void run() { log.info("\n##\n## Process SLC Execution " +
 116          * slcProcess + "\n##\n");
 117          *
 118          * slcProcess.setStatus(SlcExecution.STATUS_RUNNING);
 119          * dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_SCHEDULED,
 120          * SlcExecution.STATUS_RUNNING);
 121          *
 122          * flowsToProcess.addAll(slcProcess.getRealizedFlows());
 123          *
 124          * while (flowsToProcess.size() > 0) { RealizedFlow flow =
 125          * flowsToProcess.remove(0); ExecutionThread thread = new
 126          * ExecutionThread(this, flow); thread.start();
 127          *
 128          * synchronized (this) { try { wait(); } catch (InterruptedException e) { //
 129          * silent } } }
 130          *
 131          * slcProcess.setStatus(SlcExecution.STATUS_FINISHED);
 132          * dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_RUNNING,
 133          * SlcExecution.STATUS_FINISHED); }
 134          *
 135          * public synchronized void flowCompleted() { notifyAll(); }
 136          *
 137          * public SlcExecution getSlcProcess() { return slcProcess; }
 138          *
 139          * public ThreadGroup getProcessThreadGroup() { return processThreadGroup; }
 140          * }
 141          */
 142
 143         /**
 144          * Thread of a single execution private class ExecutionThread extends Thread
 145          * { private final RealizedFlow realizedFlow; private final ProcessThread
 146          * processThread;
 147          *
 148          * public ExecutionThread(ProcessThread processThread, RealizedFlow
 149          * realizedFlow) { super(processThread.getProcessThreadGroup(), "Flow " +
 150          * realizedFlow.getFlowDescriptor().getName()); this.realizedFlow =
 151          * realizedFlow; this.processThread = processThread; }
 152          *
 153          * public void run() { ExecutionFlowDescriptor executionFlowDescriptor =
 154          * realizedFlow .getFlowDescriptor(); String flowName =
 155          * executionFlowDescriptor.getName();
 156          *
 157          * dispatchAddStep(processThread.getSlcProcess(), new
 158          * SlcExecutionStep(SlcExecutionStep.TYPE_PHASE_START, "Flow " + flowName));
 159          *
 160          * try { execute(realizedFlow); } catch (Exception e) { // TODO: re-throw
 161          * exception ? String msg = "Execution of flow " + flowName + " failed.";
 162          * log.error(msg, e); dispatchAddStep(processThread.getSlcProcess(), new
 163          * SlcExecutionStep(msg + " " + e.getMessage())); } finally {
 164          * processThread.flowCompleted();
 165          * dispatchAddStep(processThread.getSlcProcess(), new
 166          * SlcExecutionStep(SlcExecutionStep.TYPE_PHASE_END, "Flow " + flowName)); }
 167          * } }
 168          */
 169
 170         public List<SlcExecutionNotifier> getSlcExecutionNotifiers() {
 171                 return slcExecutionNotifiers;
 172         }
 173
 174         public ThreadGroup getProcessesThreadGroup() {
 175                 return processesThreadGroup;
 176         }
 177
 178 }