X-Git-Url: http://git.argeo.org/?a=blobdiff_plain;f=runtime%2Forg.argeo.slc.support.simple%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fslc%2Fcore%2Fexecution%2FDefaultModulesManager.java;h=05a0e8daef263a5936a22b2d2ad43da7bfac2699;hb=8536d871354d83c032dfd85f763c7fdf965c02b7;hp=fdeb864a5b6b19be2b8c2346655ad692e524be5e;hpb=693cc53b426d088e49d746585bf7c5e197fc3998;p=gpl%2Fargeo-slc.git diff --git a/runtime/org.argeo.slc.support.simple/src/main/java/org/argeo/slc/core/execution/DefaultModulesManager.java b/runtime/org.argeo.slc.support.simple/src/main/java/org/argeo/slc/core/execution/DefaultModulesManager.java index fdeb864a5..05a0e8dae 100644 --- a/runtime/org.argeo.slc.support.simple/src/main/java/org/argeo/slc/core/execution/DefaultModulesManager.java +++ b/runtime/org.argeo.slc.support.simple/src/main/java/org/argeo/slc/core/execution/DefaultModulesManager.java @@ -1,15 +1,12 @@ package org.argeo.slc.core.execution; import java.util.ArrayList; -import java.util.HashMap; import java.util.Iterator; import java.util.List; -import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.argeo.slc.SlcException; -import org.argeo.slc.execution.ExecutionContext; import org.argeo.slc.execution.ExecutionFlowDescriptor; import org.argeo.slc.execution.ExecutionModule; import org.argeo.slc.execution.ExecutionModuleDescriptor; @@ -17,7 +14,6 @@ import org.argeo.slc.execution.ExecutionModulesManager; import org.argeo.slc.process.RealizedFlow; import org.argeo.slc.process.SlcExecution; import org.argeo.slc.process.SlcExecutionNotifier; -import org.dbunit.operation.UpdateOperation; import org.springframework.util.Assert; public class DefaultModulesManager implements ExecutionModulesManager { @@ -43,7 +39,8 @@ public class DefaultModulesManager implements ExecutionModulesManager { String moduleName, String version) { ExecutionModule module = getExecutionModule(moduleName, version); - Assert.notNull(module); + if(module==null) + throw new SlcException("Module "+moduleName+" ("+version+") not found"); return module.getDescriptor(); } @@ -56,50 +53,8 @@ public class DefaultModulesManager implements ExecutionModulesManager { this.executionModules = executionModules; } - protected Map convertValues( - ExecutionFlowDescriptor executionFlowDescriptor) { - // convert the values of flow.getFlowDescriptor() - Map values = executionFlowDescriptor.getValues(); - - Map convertedValues = new HashMap(); - - for (String key : values.keySet()) { - Object value = values.get(key); - if (value instanceof PrimitiveValue) { - PrimitiveValue primitiveValue = (PrimitiveValue) value; - - // TODO: check that the class of the the primitiveValue.value - // matches - // the primitiveValue.type - convertedValues.put(key, primitiveValue.getValue()); - } else if (value instanceof RefValue) { - RefValue refValue = (RefValue) value; - convertedValues.put(key, refValue.getLabel()); - } - } - return convertedValues; - } - public void process(SlcExecution slcExecution) { - log.info("\n##\n## Process SLC Execution " + slcExecution + "\n##\n"); - - for (RealizedFlow flow : slcExecution.getRealizedFlows()) { - ExecutionModule module = getExecutionModule(flow.getModuleName(), - flow.getModuleVersion()); - if (module != null) { - ExecutionThread thread = new ExecutionThread(flow - .getFlowDescriptor(), module); - thread.start(); - } else { - throw new SlcException("ExecutionModule " - + flow.getModuleName() + ", version " - + flow.getModuleVersion() + " not found."); - } - } - - slcExecution.setStatus(SlcExecution.STATUS_RUNNING); - dispatchUpdateStatus(slcExecution, SlcExecution.STATUS_SCHEDULED, - SlcExecution.STATUS_RUNNING); + new ProcessThread(slcExecution).start(); } protected void dispatchUpdateStatus(SlcExecution slcExecution, @@ -115,28 +70,96 @@ public class DefaultModulesManager implements ExecutionModulesManager { this.slcExecutionNotifiers = slcExecutionNotifiers; } + private class ProcessThread extends Thread { + private final SlcExecution slcExecution; + private final List flowsToProcess = new ArrayList(); + + public ProcessThread(SlcExecution slcExecution) { + this.slcExecution = slcExecution; + } + + public void run() { + log.info("\n##\n## Process SLC Execution " + slcExecution + + "\n##\n"); + + // FIXME: hack to let the SlcExecution be registered on server + try { + Thread.sleep(500); + } catch (InterruptedException e1) { + // silent + } + + slcExecution.setStatus(SlcExecution.STATUS_RUNNING); + dispatchUpdateStatus(slcExecution, SlcExecution.STATUS_SCHEDULED, + SlcExecution.STATUS_RUNNING); + + flowsToProcess.addAll(slcExecution.getRealizedFlows()); + + while (flowsToProcess.size() > 0) { + RealizedFlow flow = flowsToProcess.remove(0); + ExecutionModule module = getExecutionModule(flow + .getModuleName(), flow.getModuleVersion()); + if (module != null) { + ExecutionThread thread = new ExecutionThread(this, flow + .getFlowDescriptor(), module); + thread.start(); + } else { + throw new SlcException("ExecutionModule " + + flow.getModuleName() + ", version " + + flow.getModuleVersion() + " not found."); + } + + synchronized (this) { + try { + wait(); + } catch (InterruptedException e) { + // silent + } + } + } + + slcExecution.setStatus(SlcExecution.STATUS_RUNNING); + dispatchUpdateStatus(slcExecution, SlcExecution.STATUS_RUNNING, + SlcExecution.STATUS_FINISHED); + /* + * for (RealizedFlow flow : slcExecution.getRealizedFlows()) { + * ExecutionModule module = getExecutionModule(flow.getModuleName(), + * flow.getModuleVersion()); if (module != null) { ExecutionThread + * thread = new ExecutionThread(flow .getFlowDescriptor(), module); + * thread.start(); } else { throw new + * SlcException("ExecutionModule " + flow.getModuleName() + + * ", version " + flow.getModuleVersion() + " not found."); } } + */ + } + + public synchronized void flowCompleted() { + notifyAll(); + } + } + private class ExecutionThread extends Thread { private final ExecutionFlowDescriptor executionFlowDescriptor; private final ExecutionModule executionModule; + private final ProcessThread processThread; - public ExecutionThread(ExecutionFlowDescriptor executionFlowDescriptor, + public ExecutionThread(ProcessThread processThread, + ExecutionFlowDescriptor executionFlowDescriptor, ExecutionModule executionModule) { super("SLC Execution #" /* + executionContext.getUuid() */); this.executionFlowDescriptor = executionFlowDescriptor; this.executionModule = executionModule; + this.processThread = processThread; } public void run() { - ExecutionContext executionContext = executionModule - .getExecutionContext(); - executionContext - .addVariables(convertValues(executionFlowDescriptor)); try { executionModule.execute(executionFlowDescriptor); } catch (Exception e) { // TODO: re-throw exception ? - log.error("Execution " + executionContext.getUuid() + log.error("Execution "/* + executionContext.getUuid() */ + " failed.", e); + } finally { + processThread.flowCompleted(); } } }