JcrProcessThread.java

   1 /*
   2  * Copyright (C) 2007-2012 Mathieu Baudier
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *         http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.argeo.slc.jcr.execution;
  17
  18 import java.util.HashMap;
  19 import java.util.Map;
  20
  21 import javax.jcr.Node;
  22 import javax.jcr.NodeIterator;
  23 import javax.jcr.Property;
  24 import javax.jcr.RepositoryException;
  25
  26 import org.argeo.ArgeoException;
  27 import org.argeo.slc.SlcException;
  28 import org.argeo.slc.core.execution.DefaultExecutionSpec;
  29 import org.argeo.slc.core.execution.PrimitiveSpecAttribute;
  30 import org.argeo.slc.core.execution.PrimitiveUtils;
  31 import org.argeo.slc.core.execution.ProcessThread;
  32 import org.argeo.slc.core.execution.RefSpecAttribute;
  33 import org.argeo.slc.execution.ExecutionFlowDescriptor;
  34 import org.argeo.slc.execution.ExecutionModulesManager;
  35 import org.argeo.slc.execution.ExecutionProcess;
  36 import org.argeo.slc.execution.ExecutionSpecAttribute;
  37 import org.argeo.slc.jcr.SlcJcrUtils;
  38 import org.argeo.slc.jcr.SlcNames;
  39 import org.argeo.slc.jcr.SlcTypes;
  40 import org.argeo.slc.process.RealizedFlow;
  41
  42 /** Where the actual execution takes place */
  43 public class JcrProcessThread extends ProcessThread implements SlcNames {
  44
  45         public JcrProcessThread(ThreadGroup processesThreadGroup,
  46                         ExecutionModulesManager executionModulesManager,
  47                         JcrExecutionProcess process) {
  48                 super(processesThreadGroup, executionModulesManager, process);
  49         }
  50
  51         @Override
  52         protected void process() throws InterruptedException {
  53                 try {
  54                         Node rootRealizedFlowNode = getNode().getNode(SLC_FLOW);
  55                         // we just manage one level for the time being
  56                         NodeIterator nit = rootRealizedFlowNode.getNodes(SLC_FLOW);
  57                         while (nit.hasNext()) {
  58                                 Node realizedFlowNode = nit.nextNode();
  59
  60                                 // set status on realized flow
  61                                 realizedFlowNode.setProperty(SLC_STATUS,
  62                                                 ExecutionProcess.RUNNING);
  63                                 realizedFlowNode.getSession().save();
  64                                 try {
  65                                         execute(realizedFlowNode);
  66
  67                                         // set status on realized flow
  68                                         realizedFlowNode.setProperty(SLC_STATUS,
  69                                                         ExecutionProcess.COMPLETED);
  70                                         realizedFlowNode.getSession().save();
  71                                 } catch (RepositoryException e) {
  72                                         throw e;
  73                                 } catch (InterruptedException e) {
  74                                         // set status on realized flow
  75                                         realizedFlowNode.setProperty(SLC_STATUS,
  76                                                         ExecutionProcess.KILLED);
  77                                         realizedFlowNode.getSession().save();
  78                                         throw e;
  79                                 } catch (RuntimeException e) {
  80                                         // set status on realized flow
  81                                         realizedFlowNode.setProperty(SLC_STATUS,
  82                                                         ExecutionProcess.ERROR);
  83                                         realizedFlowNode.getSession().save();
  84                                         throw e;
  85                                 }
  86                         }
  87                 } catch (RepositoryException e) {
  88                         throw new ArgeoException("Cannot process " + getNode(), e);
  89                 }
  90         }
  91
  92         /** Configure the realized flows */
  93         protected void execute(Node realizedFlowNode) throws RepositoryException,
  94                         InterruptedException {
  95                 if (realizedFlowNode.hasNode(SLC_ADDRESS)) {
  96                         String flowPath = realizedFlowNode.getNode(SLC_ADDRESS)
  97                                         .getProperty(Property.JCR_PATH).getString();
  98                         // TODO: convert to local path if remote
  99
 100                         Node flowNode = realizedFlowNode.getSession().getNode(flowPath);
 101                         String flowName = flowNode.getProperty(SLC_NAME).getString();
 102
 103                         String executionModuleName = SlcJcrUtils
 104                                         .flowExecutionModuleName(flowPath);
 105                         String executionModuleVersion = SlcJcrUtils
 106                                         .flowExecutionModuleVersion(flowPath);
 107
 108                         RealizedFlow realizedFlow = new RealizedFlow();
 109                         realizedFlow.setModuleName(executionModuleName);
 110                         realizedFlow.setModuleVersion(executionModuleVersion);
 111
 112                         // retrieve execution spec
 113                         DefaultExecutionSpec executionSpec = new DefaultExecutionSpec();
 114                         Map<String, ExecutionSpecAttribute> attrs = readExecutionSpecAttributes(realizedFlowNode);
 115                         executionSpec.setAttributes(attrs);
 116
 117                         // set execution spec name
 118                         if (flowNode.hasProperty(SlcNames.SLC_SPEC)) {
 119                                 Node executionSpecNode = flowNode.getProperty(SLC_SPEC)
 120                                                 .getNode();
 121                                 executionSpec.setBeanName(executionSpecNode.getProperty(
 122                                                 SLC_NAME).getString());
 123                         }
 124
 125                         // explicitly retrieve values
 126                         Map<String, Object> values = new HashMap<String, Object>();
 127                         for (String attrName : attrs.keySet()) {
 128                                 ExecutionSpecAttribute attr = attrs.get(attrName);
 129                                 Object value = attr.getValue();
 130                                 values.put(attrName, value);
 131                         }
 132
 133                         ExecutionFlowDescriptor efd = new ExecutionFlowDescriptor(flowName,
 134                                         values, executionSpec);
 135                         realizedFlow.setFlowDescriptor(efd);
 136
 137                         //
 138                         // EXECUTE THE FLOW
 139                         //
 140                         execute(realizedFlow, true);
 141                         //
 142                 }
 143         }
 144
 145         protected Map<String, ExecutionSpecAttribute> readExecutionSpecAttributes(
 146                         Node node) {
 147                 try {
 148                         Map<String, ExecutionSpecAttribute> attrs = new HashMap<String, ExecutionSpecAttribute>();
 149                         for (NodeIterator nit = node.getNodes(); nit.hasNext();) {
 150                                 Node specAttrNode = nit.nextNode();
 151                                 if (specAttrNode
 152                                                 .isNodeType(SlcTypes.SLC_PRIMITIVE_SPEC_ATTRIBUTE)) {
 153                                         String type = specAttrNode.getProperty(SLC_TYPE)
 154                                                         .getString();
 155                                         Object value = null;
 156                                         if (specAttrNode.hasProperty(SLC_VALUE)) {
 157                                                 String valueStr = specAttrNode.getProperty(SLC_VALUE)
 158                                                                 .getString();
 159                                                 value = PrimitiveUtils.convert(type, valueStr);
 160                                         }
 161                                         PrimitiveSpecAttribute specAttr = new PrimitiveSpecAttribute(
 162                                                         type, value);
 163                                         attrs.put(specAttrNode.getName(), specAttr);
 164                                 } else if (specAttrNode
 165                                                 .isNodeType(SlcTypes.SLC_REF_SPEC_ATTRIBUTE)) {
 166                                         if (!specAttrNode.hasProperty(SLC_VALUE)) {
 167                                                 continue;
 168                                         }
 169                                         Integer value = (int) specAttrNode.getProperty(SLC_VALUE)
 170                                                         .getLong();
 171                                         RefSpecAttribute specAttr = new RefSpecAttribute();
 172                                         NodeIterator children = specAttrNode.getNodes();
 173                                         int index = 0;
 174                                         String id = null;
 175                                         while (children.hasNext()) {
 176                                                 Node child = children.nextNode();
 177                                                 if (index == value)
 178                                                         id = child.getName();
 179                                                 index++;
 180                                         }
 181                                         specAttr.setValue(id);
 182                                         attrs.put(specAttrNode.getName(), specAttr);
 183                                 }
 184                                 // throw new SlcException("Unsupported spec attribute "
 185                                 // + specAttrNode);
 186                         }
 187                         return attrs;
 188                 } catch (RepositoryException e) {
 189                         throw new SlcException("Cannot read spec attributes from " + node,
 190                                         e);
 191                 }
 192         }
 193
 194         protected Node getNode() {
 195                 return ((JcrExecutionProcess) getProcess()).getNode();
 196         }
 197 }