org.argeo.slc.runtime/src/org/argeo/slc/runtime/ExecutionThread.java

   1 package org.argeo.slc.runtime;
   2
   3 import java.security.AccessControlContext;
   4 import java.security.AccessController;
   5 import java.security.PrivilegedActionException;
   6 import java.security.PrivilegedExceptionAction;
   7 import java.util.ArrayList;
   8 import java.util.List;
   9
  10 import javax.security.auth.Subject;
  11
  12 import org.argeo.api.cms.CmsLog;
  13 import org.argeo.slc.execution.ExecutionFlowDescriptor;
  14 import org.argeo.slc.execution.ExecutionModulesManager;
  15 import org.argeo.slc.execution.ExecutionStep;
  16 import org.argeo.slc.execution.FlowConfigurationException;
  17 import org.argeo.slc.execution.RealizedFlow;
  18
  19 /** Thread of a single execution */
  20 public class ExecutionThread extends Thread {
  21         public final static String SYSPROP_EXECUTION_AUTO_UPGRADE = "slc.execution.autoupgrade";
  22         private final static CmsLog log = CmsLog.getLog(ExecutionThread.class);
  23
  24         private ExecutionModulesManager executionModulesManager;
  25         private final RealizedFlow realizedFlow;
  26         private final AccessControlContext accessControlContext;
  27
  28         private List<Runnable> destructionCallbacks = new ArrayList<Runnable>();
  29
  30         public ExecutionThread(ProcessThreadGroup processThreadGroup, ExecutionModulesManager executionModulesManager,
  31                         RealizedFlow realizedFlow) {
  32                 super(processThreadGroup, "Flow " + realizedFlow.getFlowDescriptor().getName());
  33                 this.realizedFlow = realizedFlow;
  34                 this.executionModulesManager = executionModulesManager;
  35                 accessControlContext = AccessController.getContext();
  36         }
  37
  38         public void run() {
  39                 // authenticate thread
  40                 // Authentication authentication = getProcessThreadGroup()
  41                 // .getAuthentication();
  42                 // if (authentication == null)
  43                 // throw new SlcException("Can only execute authenticated threads");
  44                 // SecurityContextHolder.getContext().setAuthentication(authentication);
  45
  46                 // Retrieve execution flow descriptor
  47                 ExecutionFlowDescriptor executionFlowDescriptor = realizedFlow.getFlowDescriptor();
  48                 String flowName = executionFlowDescriptor.getName();
  49
  50                 getProcessThreadGroup().dispatchAddStep(
  51                                 new ExecutionStep(realizedFlow.getModuleName(), ExecutionStep.PHASE_START, "Flow " + flowName));
  52
  53                 try {
  54                         Subject subject = Subject.getSubject(accessControlContext);
  55                         try {
  56                                 Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
  57
  58                                         @Override
  59                                         public Void run() throws Exception {
  60                                                 String autoUpgrade = System.getProperty(SYSPROP_EXECUTION_AUTO_UPGRADE);
  61                                                 if (autoUpgrade != null && autoUpgrade.equals("true"))
  62                                                         executionModulesManager.upgrade(realizedFlow.getModuleNameVersion());
  63                                                 executionModulesManager.start(realizedFlow.getModuleNameVersion());
  64                                                 //
  65                                                 // START FLOW
  66                                                 //
  67                                                 executionModulesManager.execute(realizedFlow);
  68                                                 // END FLOW
  69                                                 return null;
  70                                         }
  71
  72                                 });
  73                         } catch (PrivilegedActionException privilegedActionException) {
  74                                 throw (Exception) privilegedActionException.getCause();
  75                         }
  76                 } catch (FlowConfigurationException e) {
  77                         String msg = "Configuration problem with flow " + flowName + ":\n" + e.getMessage();
  78                         log.error(msg);
  79                         getProcessThreadGroup().dispatchAddStep(
  80                                         new ExecutionStep(realizedFlow.getModuleName(), ExecutionStep.ERROR, msg + " " + e.getMessage()));
  81                 } catch (Exception e) {
  82                         // TODO: re-throw exception ?
  83                         String msg = "Execution of flow " + flowName + " failed.";
  84                         log.error(msg, e);
  85                         getProcessThreadGroup().dispatchAddStep(
  86                                         new ExecutionStep(realizedFlow.getModuleName(), ExecutionStep.ERROR, msg + " " + e.getMessage()));
  87                 } finally {
  88                         getProcessThreadGroup().dispatchAddStep(
  89                                         new ExecutionStep(realizedFlow.getModuleName(), ExecutionStep.PHASE_END, "Flow " + flowName));
  90                         processDestructionCallbacks();
  91                 }
  92         }
  93
  94         private synchronized void processDestructionCallbacks() {
  95                 for (int i = destructionCallbacks.size() - 1; i >= 0; i--) {
  96                         try {
  97                                 destructionCallbacks.get(i).run();
  98                         } catch (Exception e) {
  99                                 log.warn("Could not process destruction callback " + i + " in thread " + getName(), e);
 100                         }
 101                 }
 102         }
 103
 104         /**
 105          * Gather object destruction callback to be called in reverse order at the
 106          * end of the thread
 107          */
 108         public synchronized void registerDestructionCallback(String name, Runnable callback) {
 109                 destructionCallbacks.add(callback);
 110         }
 111
 112         protected ProcessThreadGroup getProcessThreadGroup() {
 113                 return (ProcessThreadGroup) getThreadGroup();
 114         }
 115 }