runtime/org.argeo.slc.support.jcr/src/main/java/org/argeo/slc/jcr/execution/JcrProcessThread.java

   1 package org.argeo.slc.jcr.execution;
   2
   3 import java.util.HashMap;
   4 import java.util.Map;
   5
   6 import javax.jcr.Node;
   7 import javax.jcr.NodeIterator;
   8 import javax.jcr.Property;
   9 import javax.jcr.RepositoryException;
  10
  11 import org.argeo.ArgeoException;
  12 import org.argeo.slc.SlcException;
  13 import org.argeo.slc.core.execution.DefaultExecutionSpec;
  14 import org.argeo.slc.core.execution.PrimitiveSpecAttribute;
  15 import org.argeo.slc.core.execution.PrimitiveUtils;
  16 import org.argeo.slc.core.execution.ProcessThread;
  17 import org.argeo.slc.execution.ExecutionFlowDescriptor;
  18 import org.argeo.slc.execution.ExecutionModulesManager;
  19 import org.argeo.slc.execution.ExecutionProcess;
  20 import org.argeo.slc.execution.ExecutionSpecAttribute;
  21 import org.argeo.slc.jcr.SlcJcrUtils;
  22 import org.argeo.slc.jcr.SlcNames;
  23 import org.argeo.slc.jcr.SlcTypes;
  24 import org.argeo.slc.process.RealizedFlow;
  25
  26 /** Where the actual execution takes place */
  27 public class JcrProcessThread extends ProcessThread implements SlcNames {
  28
  29         public JcrProcessThread(ThreadGroup processesThreadGroup,
  30                         ExecutionModulesManager executionModulesManager,
  31                         JcrExecutionProcess process) {
  32                 super(processesThreadGroup, executionModulesManager, process);
  33         }
  34
  35         @Override
  36         protected void process() throws InterruptedException {
  37                 try {
  38                         Node rootRealizedFlowNode = getNode().getNode(SLC_FLOW);
  39                         // we just manage one level for the time being
  40                         NodeIterator nit = rootRealizedFlowNode.getNodes(SLC_FLOW);
  41                         while (nit.hasNext()) {
  42                                 Node realizedFlowNode = nit.nextNode();
  43
  44                                 // set status on realized flow
  45                                 realizedFlowNode.setProperty(SLC_STATUS,
  46                                                 ExecutionProcess.RUNNING);
  47                                 realizedFlowNode.getSession().save();
  48                                 try {
  49                                         execute(realizedFlowNode);
  50
  51                                         // set status on realized flow
  52                                         realizedFlowNode.setProperty(SLC_STATUS,
  53                                                         ExecutionProcess.COMPLETED);
  54                                         realizedFlowNode.getSession().save();
  55                                 } catch (RepositoryException e) {
  56                                         throw e;
  57                                 } catch (InterruptedException e) {
  58                                         // set status on realized flow
  59                                         realizedFlowNode.setProperty(SLC_STATUS,
  60                                                         ExecutionProcess.KILLED);
  61                                         realizedFlowNode.getSession().save();
  62                                         throw e;
  63                                 } catch (RuntimeException e) {
  64                                         // set status on realized flow
  65                                         realizedFlowNode.setProperty(SLC_STATUS,
  66                                                         ExecutionProcess.ERROR);
  67                                         realizedFlowNode.getSession().save();
  68                                         throw e;
  69                                 }
  70                         }
  71                 } catch (RepositoryException e) {
  72                         throw new ArgeoException("Cannot process " + getNode(), e);
  73                 }
  74         }
  75
  76         /** Configure the realized flows */
  77         protected void execute(Node realizedFlowNode) throws RepositoryException,
  78                         InterruptedException {
  79                 if (realizedFlowNode.hasNode(SLC_ADDRESS)) {
  80                         String flowPath = realizedFlowNode.getNode(SLC_ADDRESS)
  81                                         .getProperty(Property.JCR_PATH).getString();
  82                         // TODO: convert to local path if remote
  83
  84                         Node flowNode = realizedFlowNode.getSession().getNode(flowPath);
  85                         String flowName = flowNode.getProperty(SLC_NAME).getString();
  86
  87                         String executionModuleName = SlcJcrUtils
  88                                         .flowExecutionModuleName(flowPath);
  89                         String executionModuleVersion = SlcJcrUtils
  90                                         .flowExecutionModuleVersion(flowPath);
  91
  92                         RealizedFlow realizedFlow = new RealizedFlow();
  93                         realizedFlow.setModuleName(executionModuleName);
  94                         realizedFlow.setModuleVersion(executionModuleVersion);
  95
  96                         // retrieve execution spec
  97                         DefaultExecutionSpec executionSpec = null;
  98                         if (flowNode.hasProperty(SlcNames.SLC_SPEC)) {
  99                                 Node executionSpecNode = flowNode.getProperty(SLC_SPEC)
 100                                                 .getNode();
 101                                 executionSpec = new DefaultExecutionSpec();
 102                                 executionSpec.setBeanName(executionSpecNode.getProperty(
 103                                                 SLC_NAME).getString());
 104                                 executionSpec
 105                                                 .setAttributes(readExecutionSpecAttributes(executionSpecNode));
 106                         }
 107                         // TODO: will with original attr
 108                         Map<String, ExecutionSpecAttribute> attrs = readExecutionSpecAttributes(realizedFlowNode);
 109                         Map<String, Object> values = new HashMap<String, Object>();
 110                         for (String attrName : attrs.keySet()) {
 111 //                              if (flowNode.hasNode(attrName)) {
 112 //                                      // we assume this is a primitive
 113 //                                      // since ref are not yet implemented
 114 //                                      Node valueNode = flowNode.getNode(attrName);
 115 //                                      String type = valueNode.getProperty(SLC_TYPE).getString();
 116 //                                      String valueStr = valueNode.getProperty(SLC_VALUE)
 117 //                                                      .getString();
 118 //                                      Object value = PrimitiveUtils.convert(type, valueStr);
 119 //                                      values.put(attrName, value);
 120 //                              } else {
 121                                         ExecutionSpecAttribute attr = attrs.get(attrName);
 122                                         Object value = attr.getValue();
 123                                         values.put(attrName, value);
 124 //                              }
 125                         }
 126
 127                         // if(executionSpec!=null)
 128                         // executionSpec.setAttributes(attrs);
 129                         ExecutionFlowDescriptor efd = new ExecutionFlowDescriptor(flowName,
 130                                         values, executionSpec);
 131                         realizedFlow.setFlowDescriptor(efd);
 132
 133                         //
 134                         // EXECUTE THE FLOW
 135                         //
 136                         execute(realizedFlow, true);
 137                         //
 138                 }
 139         }
 140
 141         protected Map<String, ExecutionSpecAttribute> readExecutionSpecAttributes(
 142                         Node node) {
 143                 try {
 144                         Map<String, ExecutionSpecAttribute> attrs = new HashMap<String, ExecutionSpecAttribute>();
 145                         for (NodeIterator nit = node.getNodes(); nit.hasNext();) {
 146                                 Node specAttrNode = nit.nextNode();
 147                                 if (specAttrNode
 148                                                 .isNodeType(SlcTypes.SLC_PRIMITIVE_SPEC_ATTRIBUTE)) {
 149                                         String type = specAttrNode.getProperty(SLC_TYPE)
 150                                                         .getString();
 151                                         Object value = null;
 152                                         if (specAttrNode.hasProperty(SLC_VALUE)) {
 153                                                 String valueStr = specAttrNode.getProperty(SLC_VALUE)
 154                                                                 .getString();
 155                                                 value = PrimitiveUtils.convert(type, valueStr);
 156                                         }
 157                                         PrimitiveSpecAttribute specAttr = new PrimitiveSpecAttribute(
 158                                                         type, value);
 159                                         attrs.put(specAttrNode.getName(), specAttr);
 160                                 }
 161
 162                         }
 163                         return attrs;
 164                 } catch (RepositoryException e) {
 165                         throw new SlcException("Cannot read spec attributes from " + node,
 166                                         e);
 167                 }
 168         }
 169
 170         protected Node getNode() {
 171                 return ((JcrExecutionProcess) getProcess()).getNode();
 172         }
 173 }