X-Git-Url: http://git.argeo.org/?a=blobdiff_plain;f=runtime%2Forg.argeo.slc.core%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fslc%2Fcore%2Fexecution%2FExecutionThread.java;h=6a1c9728465aeeed4a4ff60cdc3a77ef4ea28184;hb=a17824ea2422474f25e2ea0eeae310f4dd9e6361;hp=5e288f264f0de75447afbbbfbeecff7fa82c7392;hpb=24d560ee846fda5d7954d44f83cb22ab449dbe61;p=gpl%2Fargeo-slc.git diff --git a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionThread.java b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionThread.java index 5e288f264..6a1c97284 100644 --- a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionThread.java +++ b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionThread.java @@ -1,5 +1,5 @@ /* - * Copyright (C) 2010 Mathieu Baudier + * Copyright (C) 2007-2012 Argeo GmbH * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,14 +13,20 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.argeo.slc.core.execution; +import java.util.ArrayList; +import java.util.List; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.argeo.slc.SlcException; import org.argeo.slc.execution.ExecutionFlowDescriptor; +import org.argeo.slc.execution.ExecutionModulesManager; import org.argeo.slc.execution.ExecutionStep; -import org.argeo.slc.process.RealizedFlow; +import org.argeo.slc.execution.RealizedFlow; +import org.springframework.security.Authentication; +import org.springframework.security.context.SecurityContextHolder; /** Thread of a single execution */ public class ExecutionThread extends Thread { @@ -28,58 +34,97 @@ public class ExecutionThread extends Thread { private final static Log log = LogFactory.getLog(ExecutionThread.class); + private ExecutionModulesManager executionModulesManager; private final RealizedFlow realizedFlow; - private final ProcessThread processThread; + // private final ProcessThread processThread; + + private List destructionCallbacks = new ArrayList(); - public ExecutionThread(ProcessThread processThread, + public ExecutionThread(ProcessThreadGroup processThreadGroup, + ExecutionModulesManager executionModulesManager, RealizedFlow realizedFlow) { - super(processThread.getProcessThreadGroup(), "Flow " + super(processThreadGroup, "Flow " + realizedFlow.getFlowDescriptor().getName()); this.realizedFlow = realizedFlow; - this.processThread = processThread; + this.executionModulesManager = executionModulesManager; + // this.processThread = processThread; } public void run() { - if (getContextClassLoader() != null) { - if (log.isTraceEnabled()) - log.trace("Context class loader set to " - + getContextClassLoader()); - } + // authenticate thread + Authentication authentication = getProcessThreadGroup() + .getAuthentication(); + if (authentication == null) + throw new SlcException("Can only execute authenticated threads"); + SecurityContextHolder.getContext().setAuthentication(authentication); // Retrieve execution flow descriptor ExecutionFlowDescriptor executionFlowDescriptor = realizedFlow .getFlowDescriptor(); String flowName = executionFlowDescriptor.getName(); - dispatchAddStep(new ExecutionStep(ExecutionStep.PHASE_START, "Flow " - + flowName)); + getProcessThreadGroup().dispatchAddStep( + new ExecutionStep(realizedFlow.getModuleName(), + ExecutionStep.PHASE_START, "Flow " + flowName)); try { String autoUpgrade = System .getProperty(SYSPROP_EXECUTION_AUTO_UPGRADE); if (autoUpgrade != null && autoUpgrade.equals("true")) - processThread.getExecutionModulesManager().upgrade( - realizedFlow.getModuleNameVersion()); + executionModulesManager.upgrade(realizedFlow + .getModuleNameVersion()); + executionModulesManager.start(realizedFlow.getModuleNameVersion()); // START FLOW - processThread.getExecutionModulesManager().execute(realizedFlow); + executionModulesManager.execute(realizedFlow); // END FLOW } catch (Exception e) { // TODO: re-throw exception ? String msg = "Execution of flow " + flowName + " failed."; log.error(msg, e); - dispatchAddStep(new ExecutionStep(ExecutionStep.ERROR, msg + " " - + e.getMessage())); - processThread.notifyError(); + getProcessThreadGroup().dispatchAddStep( + new ExecutionStep(realizedFlow.getModuleName(), + ExecutionStep.ERROR, msg + " " + e.getMessage())); + // processThread.notifyError(); } finally { - processThread.flowCompleted(); - dispatchAddStep(new ExecutionStep(ExecutionStep.PHASE_END, "Flow " - + flowName)); + // processThread.flowCompleted(); + getProcessThreadGroup().dispatchAddStep( + new ExecutionStep(realizedFlow.getModuleName(), + ExecutionStep.PHASE_END, "Flow " + flowName)); + processDestructionCallbacks(); } } - private void dispatchAddStep(ExecutionStep step) { - processThread.getProcessThreadGroup().dispatchAddStep(step); + // private void dispatchAddStep(ExecutionStep step) { + // getProcessThreadGroup().dispatchAddStep(step); + // } + + private synchronized void processDestructionCallbacks() { + for (int i = destructionCallbacks.size() - 1; i >= 0; i--) { + try { + destructionCallbacks.get(i).run(); + } catch (Exception e) { + log.warn("Could not process destruction callback " + i + + " in thread " + getName(), e); + } + } } + /** + * Gather object destruction callback to be called in reverse order at the + * end of the thread + */ + synchronized void registerDestructionCallback(String name, Runnable callback) { + destructionCallbacks.add(callback); + } + + protected ProcessThreadGroup getProcessThreadGroup() { + return (ProcessThreadGroup) getThreadGroup(); + // return processThread.getProcessThreadGroup(); + } + + // public RealizedFlow getRealizedFlow() { + // return realizedFlow; + // } + }