You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by ra...@apache.org on 2012/06/26 19:57:15 UTC
svn commit: r1354136 - in
/incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor:
DoWhileHandler.java WorkflowInterpreter.java
Author: raminder
Date: Tue Jun 26 17:57:14 2012
New Revision: 1354136
URL: http://svn.apache.org/viewvc?rev=1354136&view=rev
Log:
added do while handler and changed workflow interpreter to call the handler.
Added:
incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/DoWhileHandler.java
Modified:
incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/WorkflowInterpreter.java
Added: incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/DoWhileHandler.java
URL: http://svn.apache.org/viewvc/incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/DoWhileHandler.java?rev=1354136&view=auto
==============================================================================
--- incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/DoWhileHandler.java (added)
+++ incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/DoWhileHandler.java Tue Jun 26 17:57:14 2012
@@ -0,0 +1,251 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.airavata.xbaya.interpretor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+
+import javax.xml.xpath.XPath;
+import javax.xml.xpath.XPathConstants;
+import javax.xml.xpath.XPathExpressionException;
+import javax.xml.xpath.XPathFactory;
+
+import org.apache.airavata.workflow.model.component.Component;
+import org.apache.airavata.workflow.model.component.ws.WSComponent;
+import org.apache.airavata.workflow.model.exceptions.WorkflowException;
+import org.apache.airavata.workflow.model.exceptions.WorkflowRuntimeException;
+import org.apache.airavata.workflow.model.graph.ControlPort;
+import org.apache.airavata.workflow.model.graph.DataPort;
+import org.apache.airavata.workflow.model.graph.Node;
+import org.apache.airavata.workflow.model.graph.impl.EdgeImpl;
+import org.apache.airavata.workflow.model.graph.system.DoWhileNode;
+import org.apache.airavata.workflow.model.graph.system.EndDoWhileNode;
+import org.apache.airavata.xbaya.graph.controller.NodeController;
+import org.apache.airavata.xbaya.invoker.Invoker;
+import org.apache.airavata.xbaya.ui.monitor.MonitorEventHandler.NodeState;
+import org.apache.airavata.xbaya.util.InterpreterUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DoWhileHandler implements Callable<Boolean> {
+ private static Logger log = LoggerFactory.getLogger(DoWhileHandler.class);
+ private DoWhileNode dowhilenode;
+ private Map<Node, Invoker> invokerMap;
+ private ArrayList<Node> waitingNode;
+ private ArrayList<Node> finishedNodes;
+ private WorkflowInterpreter interpreter;
+ private ExecutorService threadExecutor;
+
+ /**
+ *
+ * Constructs a DoWhileHandler.
+ *
+ * @param node
+ * @param invokerMap
+ * @param waitingNode
+ * @param finishedNodes
+ * @param interpreter
+ */
+
+ public DoWhileHandler(DoWhileNode node, Map<Node, Invoker> invokerMap, ArrayList<Node> waitingNode, ArrayList<Node> finishedNodes,
+ WorkflowInterpreter interpreter, ExecutorService threadExecutor) {
+ this.dowhilenode = node;
+ this.invokerMap = invokerMap;
+ this.waitingNode = waitingNode;
+ this.finishedNodes = finishedNodes;
+ this.interpreter = interpreter;
+ this.threadExecutor = threadExecutor;
+ }
+
+ /**
+ * To evaluate dowhile condition with the input values
+ *
+ * @param doWhileNode
+ * @param inputPorts
+ * @param invokerMap
+ * @return boolean value
+ * @throws WorkFlowInterpreterException
+ * @throws XBayaException
+ */
+ private boolean evaluate(DoWhileNode doWhileNode, List<DataPort> inputPorts, Map<Node, Invoker> invokerMap) throws WorkFlowInterpreterException,
+ WorkflowException {
+ String booleanExpression = doWhileNode.getXpath();
+ if (booleanExpression == null) {
+ throw new WorkFlowInterpreterException("XPath for if cannot be null");
+ }
+
+ int i = 0;
+ for (DataPort port : inputPorts) {
+ Object inputVal1 = InterpreterUtil.findInputFromPort(port, invokerMap);
+ if (null == inputVal1) {
+ throw new WorkFlowInterpreterException("Unable to find inputs for the node:" + doWhileNode.getID());
+ }
+ booleanExpression = booleanExpression.replaceAll("\\$" + i, "'" + inputVal1 + "'");
+ i++;
+ }
+ Boolean result = new Boolean(false);
+ // Now the XPath expression
+ try {
+ XPathFactory xpathFact = XPathFactory.newInstance();
+ XPath xpath = xpathFact.newXPath();
+ result = (Boolean) xpath.evaluate(booleanExpression, booleanExpression, XPathConstants.BOOLEAN);
+ } catch (XPathExpressionException e) {
+ throw new WorkFlowInterpreterException("Cannot evaluate XPath in If Condition: " + booleanExpression);
+ }
+ return result.booleanValue();
+ }
+
+ /**
+ * To get only web service components attached to dowhile
+ *
+ * @param waitingNode
+ * @return list
+ */
+ private ArrayList<Node> handleDowhile(ArrayList<Node> waitingNode, ArrayList<Node> finishedNodes) {
+ ArrayList<Node> list = new ArrayList<Node>();
+ for (Node node : waitingNode) {
+ Component component = node.getComponent();
+ if (component instanceof WSComponent) {
+ ControlPort control = node.getControlInPort();
+ boolean controlDone = true;
+ if (control != null) {
+ for (EdgeImpl edge : control.getEdges()) {
+ controlDone = controlDone && (finishedNodes.contains(edge.getFromPort().getNode())
+ || ((ControlPort) edge.getFromPort()).isConditionMet());
+ }
+ }
+
+ /*
+ * Check for input ports
+ */
+ List<DataPort> inputPorts = node.getInputPorts();
+ boolean inputsDone = true;
+ for (DataPort dataPort : inputPorts) {
+ inputsDone = inputsDone && finishedNodes.contains(dataPort.getFromNode());
+ }
+ if (inputsDone && controlDone) {
+ list.add(node);
+ }
+ }
+ }
+
+ return list;
+ }
+
+ /**
+ * @see java.util.concurrent.Callable#call()
+ */
+ @Override
+ public Boolean call() throws Exception {
+ log.info("Invoked Dowhile node");
+ SystemComponentInvoker dowhileinvoker = new SystemComponentInvoker();
+ // TODO check for multiple input case
+ Object inputVal1 = InterpreterUtil.findInputFromPort(this.dowhilenode.getInputPort(0), this.invokerMap);
+ dowhileinvoker.addOutput(this.dowhilenode.getOutputPort(0).getID(), inputVal1);
+ this.invokerMap.put(this.dowhilenode, dowhileinvoker);
+ this.finishedNodes.add(this.dowhilenode);
+
+ ArrayList<Node> readyNodes = this.handleDowhile(this.waitingNode, this.finishedNodes);
+
+ // When you are starting 1st time its getting input from 1st node and
+ // invoking all the webservice components
+ if (readyNodes.size() != 1) {
+ throw new WorkflowRuntimeException("More than one dowhile execution not supported");
+ }
+ Node donode = readyNodes.get(0);
+ this.interpreter.handleWSComponent(donode);
+ log.info("Invoked service " + donode.getName());
+
+ List<DataPort> inputPorts = this.dowhilenode.getInputPorts();
+ boolean runflag = true;
+ while (runflag) {
+ while (true) {
+ if (NodeController.isRunning(donode) || NodeController.isWaiting(donode)) {
+ Thread.sleep(500);
+ log.info("Service " + donode.getName() + " waiting");
+ } else if (NodeController.isFinished(donode)) {
+ log.info("Service " + donode.getName() + " Finished");
+ List<DataPort> ports = this.dowhilenode.getOutputPorts();
+ for (int outputPortIndex = 0, inputPortIndex = 1; outputPortIndex < ports.size(); outputPortIndex++) {
+ Object inputValue = InterpreterUtil.findInputFromPort(this.dowhilenode.getInputPort(inputPortIndex), this.invokerMap);
+ dowhileinvoker.addOutput(this.dowhilenode.getOutputPort(outputPortIndex).getID(), inputValue);
+ }
+ break;
+ } else if (NodeController.isFailed(donode)) {
+ log.info("Service " + donode.getName() + " Failed");
+ runflag = false;
+ NodeController.getGUI(dowhilenode).setBodyColor(NodeState.FAILED.color);
+ this.threadExecutor.shutdown();
+ return false;
+ } else if (donode.isBreak()) {
+ log.info("Service " + donode.getName() + " set to break");
+ runflag = false;
+ break;
+ } else {
+ log.error("Service " + donode.getName() + " have unknow status");
+ throw new WorkFlowInterpreterException("Unknow status of the node");
+ }
+ }
+
+ this.invokerMap.put(this.dowhilenode, dowhileinvoker);
+ log.info("Going to evaluate do while expression for " + donode.getName());
+ if (!evaluate(this.dowhilenode, inputPorts, this.invokerMap)) {
+ log.info("Expression evaluation is false so calling EndDoWhile");
+ runflag = false;
+ } else {
+ if (readyNodes.size() != 1) {
+ throw new WorkFlowInterpreterException("More than one dowhile execution not supported");
+ }
+
+ Node whileNode = readyNodes.get(0);
+ // this.dowhilenode.getGUI().setBodyColor(NodeState.STARTING.color);
+ log.info("Expression evaluation is true so invoking service again " + whileNode.getName());
+
+ this.interpreter.handleWSComponent(whileNode);
+ }
+ }
+ // WS node should be done
+ NodeController.getGUI(dowhilenode).setBodyColor(NodeState.FINISHED.color);
+ EndDoWhileNode endDoWhileNode = this.dowhilenode.getEndDoWhileNode();
+
+ // /////////////////////////////////////////////////////////
+ // // Do WHile finished execution thus we can set the //////
+ // //inputs to the EndDOWHile and resume the executions/////
+ SystemComponentInvoker invoker = new SystemComponentInvoker();
+
+ List<DataPort> inputports = endDoWhileNode.getInputPorts();
+
+ for (int inputPortIndex = 0; inputPortIndex < inputports.size(); inputPortIndex++) {
+ Object inputVal = dowhileinvoker.getOutput(inputports.get(inputPortIndex).getFromPort().getID());
+ invoker.addOutput(endDoWhileNode.getOutputPort(inputPortIndex).getID(), inputVal);
+ }
+
+ this.invokerMap.put(endDoWhileNode, invoker);
+ // TODO send mail once the iterations have converged
+
+ NodeController.getGUI(endDoWhileNode).setBodyColor(NodeState.FINISHED.color);
+ this.threadExecutor.shutdown();
+ return true;
+ }
+}
Modified: incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/WorkflowInterpreter.java
URL: http://svn.apache.org/viewvc/incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/WorkflowInterpreter.java?rev=1354136&r1=1354135&r2=1354136&view=diff
==============================================================================
--- incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/WorkflowInterpreter.java (original)
+++ incubator/airavata/trunk/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/interpretor/WorkflowInterpreter.java Tue Jun 26 17:57:14 2012
@@ -21,6 +21,7 @@
package org.apache.airavata.xbaya.interpretor;
+import java.awt.Color;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
@@ -32,6 +33,8 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import javax.xml.namespace.QName;
@@ -52,6 +55,8 @@ import org.apache.airavata.workflow.mode
import org.apache.airavata.workflow.model.component.dynamic.DynamicComponent;
import org.apache.airavata.workflow.model.component.system.ConstantComponent;
import org.apache.airavata.workflow.model.component.system.DifferedInputComponent;
+import org.apache.airavata.workflow.model.component.system.DoWhileComponent;
+import org.apache.airavata.workflow.model.component.system.EndDoWhileComponent;
import org.apache.airavata.workflow.model.component.system.EndForEachComponent;
import org.apache.airavata.workflow.model.component.system.EndifComponent;
import org.apache.airavata.workflow.model.component.system.ForEachComponent;
@@ -74,6 +79,7 @@ import org.apache.airavata.workflow.mode
import org.apache.airavata.workflow.model.graph.impl.NodeImpl;
import org.apache.airavata.workflow.model.graph.subworkflow.SubWorkflowNode;
import org.apache.airavata.workflow.model.graph.system.ConstantNode;
+import org.apache.airavata.workflow.model.graph.system.DoWhileNode;
import org.apache.airavata.workflow.model.graph.system.EndForEachNode;
import org.apache.airavata.workflow.model.graph.system.EndifNode;
import org.apache.airavata.workflow.model.graph.system.ForEachNode;
@@ -117,7 +123,7 @@ public class WorkflowInterpreter {
public static final String WORKFLOW_FINISHED = "Workflow Finished";
private WorkflowInterpreterConfiguration config;
-
+
private Map<Node, Invoker> invokerMap = new HashMap<Node, Invoker>();
private LeadResourceMapping resourceMapping;
@@ -127,9 +133,9 @@ public class WorkflowInterpreter {
private WorkflowInterpreterInteractor interactor;
/**
- *
+ *
* Constructs a WorkflowInterpreter.
- *
+ *
* @param engine
* @param topic
* @param workflow
@@ -138,7 +144,7 @@ public class WorkflowInterpreter {
public WorkflowInterpreter(WorkflowInterpreterConfiguration config, WorkflowInterpreterInteractor interactor) {
this.setConfig(config);
config.validateNotifier();
- this.interactor=interactor;
+ this.interactor = interactor;
config.setActOnProvenance(config.getConfiguration().isCollectProvenance());
config.setSubWorkflow(false);
}
@@ -147,34 +153,28 @@ public class WorkflowInterpreter {
this.resourceMapping = resourceMapping;
}
- private void notifyViaInteractor(WorkflowExecutionMessage messageType, Object data){
+ private void notifyViaInteractor(WorkflowExecutionMessage messageType, Object data) {
interactor.notify(messageType, config, data);
}
-
- private Object getInputViaInteractor(WorkflowExecutionMessage messageType, Object data) throws Exception{
+
+ private Object getInputViaInteractor(WorkflowExecutionMessage messageType, Object data) throws Exception {
return interactor.retrieveData(messageType, config, data);
}
-
-
+
/**
* @throws WorkflowException
*/
public void scheduleDynamically() throws WorkflowException {
try {
- if (!this.config.isSubWorkflow()
- && this.getWorkflow().getExecutionState() != WorkflowExecutionState.NONE) {
- throw new WorkFlowInterpreterException(
- "XBaya is already running a workflow");
+ if (!this.config.isSubWorkflow() && this.getWorkflow().getExecutionState() != WorkflowExecutionState.NONE) {
+ throw new WorkFlowInterpreterException("XBaya is already running a workflow");
}
this.getWorkflow().setExecutionState(WorkflowExecutionState.RUNNING);
if (this.config.isActOnProvenance()) {
try {
- this.getConfig().getConfiguration()
- .getJcrComponentRegistry()
- .getRegistry()
- .saveWorkflowExecutionStatus(this.config.getTopic(),
- ExecutionStatus.STARTED);
+ this.getConfig().getConfiguration().getJcrComponentRegistry().getRegistry()
+ .saveWorkflowExecutionStatus(this.config.getTopic(), ExecutionStatus.STARTED);
} catch (RegistryException e) {
throw new WorkflowException(e);
}
@@ -193,12 +193,13 @@ public class WorkflowInterpreter {
while (this.getWorkflow().getExecutionState() != WorkflowExecutionState.STOPPED) {
if (getRemainNodesDynamically() == 0) {
notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_STATE_CHANGED, WorkflowExecutionState.PAUSED);
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// this.notifyPause();
-// } else {
-// this.getWorkflow().setExecutionState(
-// WorkflowExecutionState.STOPPED);
-// }
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // this.notifyPause();
+ // } else {
+ // this.getWorkflow().setExecutionState(
+ // WorkflowExecutionState.STOPPED);
+ // }
}
// ok we have paused sleep
while (this.getWorkflow().getExecutionState() == WorkflowExecutionState.PAUSED) {
@@ -229,12 +230,11 @@ public class WorkflowInterpreter {
if (!nodeOutputLoadedFromProvenance) {
executeDynamically(node);
}
- }else{
+ } else {
executeDynamically(node);
}
if (this.getWorkflow().getExecutionState() == WorkflowExecutionState.STEP) {
- this.getWorkflow().setExecutionState(
- WorkflowExecutionState.PAUSED);
+ this.getWorkflow().setExecutionState(WorkflowExecutionState.PAUSED);
break;
}
}
@@ -248,8 +248,7 @@ public class WorkflowInterpreter {
// so we should pause the execution
if (InterpreterUtil.getRunningNodeCountDynamically(this.getGraph()) == 0
&& InterpreterUtil.getFailedNodeCountDynamically(this.getGraph()) != 0) {
- this.getWorkflow().setExecutionState(
- WorkflowExecutionState.PAUSED);
+ this.getWorkflow().setExecutionState(WorkflowExecutionState.PAUSED);
}
try {
@@ -264,11 +263,8 @@ public class WorkflowInterpreter {
if (this.config.isActOnProvenance()) {
try {
try {
- this.getConfig().getConfiguration()
- .getJcrComponentRegistry()
- .getRegistry()
- .saveWorkflowExecutionStatus(this.config.getTopic(),
- ExecutionStatus.FINISHED);
+ this.getConfig().getConfiguration().getJcrComponentRegistry().getRegistry()
+ .saveWorkflowExecutionStatus(this.config.getTopic(), ExecutionStatus.FINISHED);
} catch (Exception e) {
throw new WorkflowException(e);
}
@@ -280,47 +276,46 @@ public class WorkflowInterpreter {
} else {
if (this.config.isActOnProvenance()) {
try {
- this.getConfig().getConfiguration()
- .getJcrComponentRegistry()
- .getRegistry()
- .saveWorkflowExecutionStatus(this.config.getTopic(),
- ExecutionStatus.FAILED);
+ this.getConfig().getConfiguration().getJcrComponentRegistry().getRegistry()
+ .saveWorkflowExecutionStatus(this.config.getTopic(), ExecutionStatus.FAILED);
} catch (RegistryException e) {
throw new WorkflowException(e);
}
}
}
this.config.getNotifier().workflowTerminated();
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// final WaitDialog waitDialog = new WaitDialog(new Cancelable() {
-// @Override
-// public void cancel() {
-// // Do nothing
-// }
-// }, "Stop Workflow", "Cleaning up resources for Workflow",
-// this.config.getGUI());
-// new Thread(new Runnable() {
-// @Override
-// public void run() {
-// waitDialog.show();
-// }
-// }).start();
-// // Send Notification for output values
-// finish();
-// // Sleep to provide for notification delay
-// try {
-// Thread.sleep(1000);
-// } catch (InterruptedException e) {
-// e.printStackTrace();
-// }
-// cleanup();
-// this.config.getNotifier().cleanup();
-// waitDialog.hide();
-// } else {
-// finish();
-// }
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // final WaitDialog waitDialog = new WaitDialog(new Cancelable() {
+ // @Override
+ // public void cancel() {
+ // // Do nothing
+ // }
+ // }, "Stop Workflow", "Cleaning up resources for Workflow",
+ // this.config.getGUI());
+ // new Thread(new Runnable() {
+ // @Override
+ // public void run() {
+ // waitDialog.show();
+ // }
+ // }).start();
+ // // Send Notification for output values
+ // finish();
+ // // Sleep to provide for notification delay
+ // try {
+ // Thread.sleep(1000);
+ // } catch (InterruptedException e) {
+ // e.printStackTrace();
+ // }
+ // cleanup();
+ // this.config.getNotifier().cleanup();
+ // waitDialog.hide();
+ // } else {
+ // finish();
+ // }
UUID uuid = UUID.randomUUID();
- notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_TASK_START, new WorkflowInterpreterInteractor.TaskNotification("Stop Workflow", "Cleaning up resources for Workflow",uuid.toString()));
+ notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_TASK_START, new WorkflowInterpreterInteractor.TaskNotification("Stop Workflow",
+ "Cleaning up resources for Workflow", uuid.toString()));
// Send Notification for output values
finish();
// Sleep to provide for notification delay
@@ -330,10 +325,11 @@ public class WorkflowInterpreter {
e.printStackTrace();
}
cleanup();
- if (config.getNotifier()!=null){
+ if (config.getNotifier() != null) {
this.config.getNotifier().cleanup();
}
- notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_TASK_END, new WorkflowInterpreterInteractor.TaskNotification("Stop Workflow", "Cleaning up resources for Workflow",uuid.toString()));
+ notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_TASK_END, new WorkflowInterpreterInteractor.TaskNotification("Stop Workflow",
+ "Cleaning up resources for Workflow", uuid.toString()));
this.getWorkflow().setExecutionState(WorkflowExecutionState.NONE);
} catch (RuntimeException e) {
@@ -359,16 +355,12 @@ public class WorkflowInterpreter {
for (int i = 0; i < inputPorts.size(); ++i) {
String inputTagname = inputPorts.get(i).getName();
// cordinate return
- inputs[i] = new Pair<String, String>(inputTagname, "<"
- + inputTagname
- + ">"
- + InterpreterUtil.findInputFromPort(inputPorts.get(i),
- this.invokerMap).toString() + "</"
- + inputTagname + ">");
+ inputs[i] = new Pair<String, String>(inputTagname, "<" + inputTagname + ">"
+ + InterpreterUtil.findInputFromPort(inputPorts.get(i), this.invokerMap).toString() + "</" + inputTagname + ">");
}
- String output = ((String)new ProvenanceReader(node,this.config.getTopic(),this.config.getRegistry()).read());
- if(output != null){
+ String output = ((String) new ProvenanceReader(node, this.config.getTopic(), this.config.getRegistry()).read());
+ if (output != null) {
XmlElement result = XMLUtil.stringToXmlElement(output);
SystemComponentInvoker invoker = new SystemComponentInvoker();
List<DataPort> outPorts = node.getOutputPorts();
@@ -377,11 +369,8 @@ public class WorkflowInterpreter {
Iterable itr = result.children();
for (Object outputElem : itr) {
if (outputElem instanceof XmlElement) {
- if (((XmlElement) outputElem).getName().equals(
- dataPort.getName())) {
- invoker.addOutput(dataPort.getName(),
- ((XmlElement) outputElem).children()
- .iterator().next());
+ if (((XmlElement) outputElem).getName().equals(dataPort.getName())) {
+ invoker.addOutput(dataPort.getName(), ((XmlElement) outputElem).children().iterator().next());
}
}
}
@@ -390,7 +379,7 @@ public class WorkflowInterpreter {
this.invokerMap.put(node, invoker);
NodeController.getGUI(node).setBodyColor(NodeState.FINISHED.color);
return true;
- }else{
+ } else {
writeProvenanceLater(node);
}
} catch (Exception e) {
@@ -412,9 +401,8 @@ public class WorkflowInterpreter {
if (this.provenanceWriter == null) {
this.provenanceWriter = new PredicatedTaskRunner(1);
}
- this.provenanceWriter.scedule(new ProvenanceWrite(node, this
- .getWorkflow().getName(), invokerMap, this.config.getTopic(),
- this.getConfig().getConfiguration().getJcrComponentRegistry().getRegistry()));
+ this.provenanceWriter.scedule(new ProvenanceWrite(node, this.getWorkflow().getName(), invokerMap, this.config.getTopic(), this.getConfig()
+ .getConfiguration().getJcrComponentRegistry().getRegistry()));
}
/**
@@ -422,33 +410,36 @@ public class WorkflowInterpreter {
*/
public void raiseException(Throwable e) {
notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_ERROR, e);
-//// throw new RuntimeException(e);
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// this.config.getGUI().getErrorWindow().error(e);
-// } else {
-// throw new RuntimeException(e);
-// }
+ // // throw new RuntimeException(e);
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // this.config.getGUI().getErrorWindow().error(e);
+ // } else {
+ // throw new RuntimeException(e);
+ // }
}
/**
*
*/
private void notifyPause() {
- if (this.getWorkflow().getExecutionState() != WorkflowExecutionState.RUNNING
- && this.getWorkflow().getExecutionState() != WorkflowExecutionState.STEP) {
+ if (this.getWorkflow().getExecutionState() != WorkflowExecutionState.RUNNING && this.getWorkflow().getExecutionState() != WorkflowExecutionState.STEP) {
throw new WorkflowRuntimeException("Cannot pause when not running");
}
notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_STATE_CHANGED, WorkflowExecutionState.PAUSED);
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-//
-// if (this.getWorkflow().getExecutionState() == WorkflowExecutionState.RUNNING
-// || this.getWorkflow().getExecutionState() == WorkflowExecutionState.STEP) {
-// this.config.getGUI().getToolbar().getPlayAction()
-// .actionPerformed(null);
-// } else {
-// throw new WorkflowRuntimeException("Cannot pause when not running");
-// }
-// }
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ //
+ // if (this.getWorkflow().getExecutionState() ==
+ // WorkflowExecutionState.RUNNING
+ // || this.getWorkflow().getExecutionState() ==
+ // WorkflowExecutionState.STEP) {
+ // this.config.getGUI().getToolbar().getPlayAction()
+ // .actionPerformed(null);
+ // } else {
+ // throw new WorkflowRuntimeException("Cannot pause when not running");
+ // }
+ // }
}
/**
@@ -457,14 +448,15 @@ public class WorkflowInterpreter {
public void cleanup() throws MonitorException {
this.getWorkflow().setExecutionState(WorkflowExecutionState.STOPPED);
notifyViaInteractor(WorkflowExecutionMessage.EXECUTION_CLEANUP, null);
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// this.engine.resetWorkflowInterpreter();
-// try {
-// this.engine.getMonitor().stop();
-// } finally {
-// this.engine.getMonitor().reset();
-// }
-// }
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // this.engine.resetWorkflowInterpreter();
+ // try {
+ // this.engine.getMonitor().stop();
+ // } finally {
+ // this.engine.getMonitor().reset();
+ // }
+ // }
}
private void sendOutputsDynamically() throws WorkflowException {
@@ -482,19 +474,14 @@ public class WorkflowInterpreter {
List<DataPort> inputPorts = node.getInputPorts();
for (DataPort dataPort : inputPorts) {
- Object val = InterpreterUtil.findInputFromPort(dataPort,
- this.invokerMap);
+ Object val = InterpreterUtil.findInputFromPort(dataPort, this.invokerMap);
if (null == val) {
- throw new WorkFlowInterpreterException(
- "Unable to find output for the node:"
- + node.getID());
+ throw new WorkFlowInterpreterException("Unable to find output for the node:" + node.getID());
}
// This is ok because the outputnodes always got only one
// input
if (val instanceof org.xmlpull.v1.builder.XmlElement) {
- ((OutputNode) node)
- .setDescription(XMLUtil
- .xmlElementToString((org.xmlpull.v1.builder.XmlElement) val));
+ ((OutputNode) node).setDescription(XMLUtil.xmlElementToString((org.xmlpull.v1.builder.XmlElement) val));
} else {
((OutputNode) node).setDescription(val.toString());
}
@@ -502,19 +489,14 @@ public class WorkflowInterpreter {
if (this.config.isActOnProvenance()) {
try {
if (val instanceof String) {
- this.getConfig().getConfiguration()
- .getJcrComponentRegistry()
- .getRegistry()
- .saveWorkflowExecutionOutput(
- this.config.getTopic(), node.getName(),
- val.toString());
+ this.getConfig().getConfiguration().getJcrComponentRegistry().getRegistry()
+ .saveWorkflowExecutionOutput(this.config.getTopic(), node.getName(), val.toString());
} else if (val instanceof org.xmlpull.v1.builder.XmlElement) {
- this.getConfig().getConfiguration()
+ this.getConfig()
+ .getConfiguration()
.getJcrComponentRegistry()
.getRegistry()
- .saveWorkflowExecutionOutput(
- this.config.getTopic(),
- node.getName(),
+ .saveWorkflowExecutionOutput(this.config.getTopic(), node.getName(),
XMLUtil.xmlElementToString((org.xmlpull.v1.builder.XmlElement) val));
}
} catch (RegistryException e) {
@@ -528,8 +510,7 @@ public class WorkflowInterpreter {
}
System.out.println("Looping");
}
- this.config.getNotifier().sendingPartialResults(outputValues.toArray(),
- outputKeywords.toArray(new String[outputKeywords.size()]));
+ this.config.getNotifier().sendingPartialResults(outputValues.toArray(), outputKeywords.toArray(new String[outputKeywords.size()]));
}
}
@@ -554,33 +535,25 @@ public class WorkflowInterpreter {
OutputNode node = (OutputNode) outputNode;
List<DataPort> inputPorts = node.getInputPorts();
for (DataPort dataPort : inputPorts) {
- Object val = InterpreterUtil.findInputFromPort(dataPort,
- this.invokerMap);
+ Object val = InterpreterUtil.findInputFromPort(dataPort, this.invokerMap);
;
if (null == val) {
- throw new WorkFlowInterpreterException(
- "Unable to find output for the node:"
- + node.getID());
+ throw new WorkFlowInterpreterException("Unable to find output for the node:" + node.getID());
}
// Some node not yet updated
if (NodeController.getGUI(node).getBodyColor() != NodeState.FINISHED.color) {
if (this.config.isActOnProvenance()) {
try {
if (val instanceof String) {
- this.getConfig().getConfiguration()
- .getJcrComponentRegistry()
- .getRegistry()
- .saveWorkflowExecutionOutput(
- this.config.getTopic(), node.getName(),
- val.toString());
+ this.getConfig().getConfiguration().getJcrComponentRegistry().getRegistry()
+ .saveWorkflowExecutionOutput(this.config.getTopic(), node.getName(), val.toString());
} else if (val instanceof org.xmlpull.v1.builder.XmlElement) {
- this.getConfig().getConfiguration()
+ this.getConfig()
+ .getConfiguration()
.getJcrComponentRegistry()
.getRegistry()
- .saveWorkflowExecutionOutput(
- this.config.getTopic(),
- node.getName(),
+ .saveWorkflowExecutionOutput(this.config.getTopic(), node.getName(),
XMLUtil.xmlElementToString((org.xmlpull.v1.builder.XmlElement) val));
}
@@ -592,8 +565,7 @@ public class WorkflowInterpreter {
}
}
if (val instanceof XmlElement) {
- ((OutputNode) node).setDescription(XMLUtil
- .xmlElementToString((XmlElement) val));
+ ((OutputNode) node).setDescription(XMLUtil.xmlElementToString((XmlElement) val));
} else {
((OutputNode) node).setDescription(val.toString());
}
@@ -602,8 +574,7 @@ public class WorkflowInterpreter {
}
}
- this.config.getNotifier().sendingPartialResults(outputValues.toArray(),
- outputKeywords.toArray(new String[outputKeywords.size()]));
+ this.config.getNotifier().sendingPartialResults(outputValues.toArray(), outputKeywords.toArray(new String[outputKeywords.size()]));
cleanup();
this.config.getNotifier().cleanup();
@@ -624,19 +595,22 @@ public class WorkflowInterpreter {
handleIf(node);
} else if (component instanceof EndifComponent) {
handleEndIf(node);
+ } else if (component instanceof DoWhileComponent) {
+ // Executor thread is shutdown inside as thread gets killed when you
+ // shutdown
+ ExecutorService threadExecutor = Executors.newSingleThreadExecutor();
+ DoWhileHandler doWhileHandler = new DoWhileHandler((DoWhileNode) node, this.invokerMap, getWaitingNodesDynamically(),
+ getFinishedNodesDynamically(), this, threadExecutor);
+ threadExecutor.submit(doWhileHandler);
} else if (component instanceof InstanceComponent) {
- if (AmazonCredential.getInstance().getAwsAccessKeyId().isEmpty()
- || AmazonCredential.getInstance().getAwsSecretAccessKey()
- .isEmpty()) {
- throw new WorkFlowInterpreterException(
- "Please set Amazon Credential before using EC2 Instance Component");
+ if (AmazonCredential.getInstance().getAwsAccessKeyId().isEmpty() || AmazonCredential.getInstance().getAwsSecretAccessKey().isEmpty()) {
+ throw new WorkFlowInterpreterException("Please set Amazon Credential before using EC2 Instance Component");
}
for (ControlPort ports : node.getControlOutPorts()) {
ports.setConditionMet(true);
}
} else if (component instanceof TerminateInstanceComponent) {
- Object inputVal = InterpreterUtil.findInputFromPort(
- node.getInputPort(0), this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(node.getInputPort(0), this.invokerMap);
String instanceId = inputVal.toString();
/*
* Terminate instance
@@ -646,17 +620,19 @@ public class WorkflowInterpreter {
// set color to done
NodeController.getGUI(node).setBodyColor(NodeState.FINISHED.color);
} else {
- throw new WorkFlowInterpreterException(
- "Encountered Node that cannot be executed:" + node);
+ throw new WorkFlowInterpreterException("Encountered Node that cannot be executed:" + node);
}
}
private void handleSubWorkComponent(Node node) throws WorkflowException {
notifyViaInteractor(WorkflowExecutionMessage.OPEN_SUBWORKFLOW, node);
-// if ((this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) && (node instanceof SubWorkflowNodeGUI)) {
-// ((SubWorkflowNodeGUI) NodeController.getGUI(node)).openWorkflowTab(this.config.getGUI());
-// }
+ // if ((this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) && (node instanceof
+ // SubWorkflowNodeGUI)) {
+ // ((SubWorkflowNodeGUI)
+ // NodeController.getGUI(node)).openWorkflowTab(this.config.getGUI());
+ // }
// setting the inputs
Workflow subWorkflow = ((SubWorkflowNode) node).getWorkflow();
// List<WSComponentPort> subWorkflowdInputs = new ODEClient()
@@ -665,49 +641,47 @@ public class WorkflowInterpreter {
List<DataPort> inputPorts = node.getInputPorts();
for (DataPort port : inputPorts) {
- Object inputVal = InterpreterUtil.findInputFromPort(port,
- this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(port, this.invokerMap);
if (null == inputVal) {
- throw new WorkFlowInterpreterException(
- "Unable to find inputs for the subworkflow node node:"
- + node.getID());
+ throw new WorkFlowInterpreterException("Unable to find inputs for the subworkflow node node:" + node.getID());
}
- for (Iterator<Node> iterator = subWorkflowInputNodes.iterator(); iterator
- .hasNext();) {
+ for (Iterator<Node> iterator = subWorkflowInputNodes.iterator(); iterator.hasNext();) {
InputNode inputNode = (InputNode) iterator.next();
if (inputNode.getName().equals(port.getName())) {
inputNode.setDefaultValue(inputVal);
}
}
}
- for (Iterator<Node> iterator = subWorkflowInputNodes.iterator(); iterator
- .hasNext();) {
+ for (Iterator<Node> iterator = subWorkflowInputNodes.iterator(); iterator.hasNext();) {
InputNode inputNode = (InputNode) iterator.next();
if (inputNode.getDefaultValue() == null) {
- throw new WorkFlowInterpreterException("Input not set for :"
- + inputNode.getID());
+ throw new WorkFlowInterpreterException("Input not set for :" + inputNode.getID());
}
}
try {
- WorkflowInterpreter subworkflowInterpreter = (WorkflowInterpreter)getInputViaInteractor(WorkflowExecutionMessage.INPUT_WORKFLOWINTERPRETER_FOR_WORKFLOW, subWorkflow);
+ WorkflowInterpreter subworkflowInterpreter = (WorkflowInterpreter) getInputViaInteractor(
+ WorkflowExecutionMessage.INPUT_WORKFLOWINTERPRETER_FOR_WORKFLOW, subWorkflow);
subworkflowInterpreter.scheduleDynamically();
} catch (Exception e) {
throw new WorkflowException(e);
}
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// new WorkflowInterpreter(getConfig(), this.topic, subWorkflow, true,
-// false, new GUIWorkflowInterpreterInteractorImpl(engine, workflow)).scheduleDynamically();
-// } else {
-// new WorkflowInterpreter(this.getConfig(), this.topic,
-// subWorkflow, this.getUsername(), this.getPassword(),new SSWorkflowInterpreterInteractorImpl(workflow))
-// .scheduleDynamically();
-// }
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // new WorkflowInterpreter(getConfig(), this.topic, subWorkflow, true,
+ // false, new GUIWorkflowInterpreterInteractorImpl(engine,
+ // workflow)).scheduleDynamically();
+ // } else {
+ // new WorkflowInterpreter(this.getConfig(), this.topic,
+ // subWorkflow, this.getUsername(), this.getPassword(),new
+ // SSWorkflowInterpreterInteractorImpl(workflow))
+ // .scheduleDynamically();
+ // }
}
- private void handleWSComponent(Node node) throws WorkflowException {
+ protected void handleWSComponent(Node node) throws WorkflowException {
WSComponent wsComponent = ((WSComponent) node.getComponent());
QName portTypeQName = wsComponent.getPortTypeQName();
Invoker invoker = this.invokerMap.get(node);
@@ -721,45 +695,45 @@ public class WorkflowInterpreter {
final String gfacURLString = this.getConfig().getConfiguration().getGFacURL().toString();
if (null == wsdlLocation) {
if (gfacURLString.startsWith("https")) {
- LeadContextHeader leadCtxHeader=null;
-// try {
-// leadCtxHeader = (LeadContextHeader)getInputViaInteractor(WorkflowExecutionMessage.INPUT_LEAD_CONTEXT_HEADER, wsNode);
-// } catch (Exception e1) {
-// throw new WorkflowException(e1);
-// }
+ LeadContextHeader leadCtxHeader = null;
+ // try {
+ // leadCtxHeader =
+ // (LeadContextHeader)getInputViaInteractor(WorkflowExecutionMessage.INPUT_LEAD_CONTEXT_HEADER,
+ // wsNode);
+ // } catch (Exception e1) {
+ // throw new WorkflowException(e1);
+ // }
try {
- leadCtxHeader = XBayaUtil.buildLeadContextHeader(
- this.getWorkflow(),
- this.getConfig().getConfiguration(),
- new MonitorConfiguration(this.getConfig().getConfiguration()
- .getBrokerURL(), this.config.getTopic(), true,
- this.getConfig().getConfiguration().getMessageBoxURL()),
- wsNode.getID(), null);
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// leadCtxHeader = XBayaUtil.buildLeadContextHeader(
-// this.getWorkflow(),
-// this.getConfig().getConfiguration(),
-// new MonitorConfiguration(this.getConfig().getConfiguration()
-// .getBrokerURL(), this.config.getTopic(), true,
-// this.getConfig().getConfiguration().getMessageBoxURL()),
-// wsNode.getID(), null);
-// } else {
-// leadCtxHeader = XBayaUtil.buildLeadContextHeader(
-// this.getWorkflow(),
-// this.getConfig().getConfiguration(),
-// new MonitorConfiguration(this.getConfig().getConfiguration()
-// .getBrokerURL(), this.config.getTopic(), true,
-// this.getConfig().getConfiguration().getMessageBoxURL()),
-// wsNode.getID(), null);
-// }
+ leadCtxHeader = XBayaUtil.buildLeadContextHeader(this.getWorkflow(), this.getConfig().getConfiguration(), new MonitorConfiguration(this
+ .getConfig().getConfiguration().getBrokerURL(), this.config.getTopic(), true, this.getConfig().getConfiguration()
+ .getMessageBoxURL()), wsNode.getID(), null);
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // leadCtxHeader = XBayaUtil.buildLeadContextHeader(
+ // this.getWorkflow(),
+ // this.getConfig().getConfiguration(),
+ // new
+ // MonitorConfiguration(this.getConfig().getConfiguration()
+ // .getBrokerURL(), this.config.getTopic(), true,
+ // this.getConfig().getConfiguration().getMessageBoxURL()),
+ // wsNode.getID(), null);
+ // } else {
+ // leadCtxHeader = XBayaUtil.buildLeadContextHeader(
+ // this.getWorkflow(),
+ // this.getConfig().getConfiguration(),
+ // new
+ // MonitorConfiguration(this.getConfig().getConfiguration()
+ // .getBrokerURL(), this.config.getTopic(), true,
+ // this.getConfig().getConfiguration().getMessageBoxURL()),
+ // wsNode.getID(), null);
+ // }
} catch (URISyntaxException e) {
throw new WorkflowException(e);
}
leadCtxHeader.setServiceId(node.getID());
try {
- leadCtxHeader.setWorkflowId(new URI(this.getWorkflow()
- .getName()));
+ leadCtxHeader.setWorkflowId(new URI(this.getWorkflow().getName()));
// We do this so that the wsdl resolver can is setup
// wsdlresolver.getInstance is static so once this is
@@ -786,81 +760,59 @@ public class WorkflowInterpreter {
for (Node n : wsNode.getControlInPort().getFromNodes()) {
if (n instanceof InstanceNode) {
// TODO make it as constant
- LeadResourceMapping x = new LeadResourceMapping(
- "AMAZON");
+ LeadResourceMapping x = new LeadResourceMapping("AMAZON");
- x.addAttribute("ACCESS_KEY", AmazonCredential
- .getInstance().getAwsAccessKeyId());
- x.addAttribute("SECRET_KEY", AmazonCredential
- .getInstance().getAwsSecretAccessKey());
+ x.addAttribute("ACCESS_KEY", AmazonCredential.getInstance().getAwsAccessKeyId());
+ x.addAttribute("SECRET_KEY", AmazonCredential.getInstance().getAwsSecretAccessKey());
if (((InstanceNode) n).isStartNewInstance()) {
- x.addAttribute("AMI_ID",
- ((InstanceNode) n).getIdAsValue());
- x.addAttribute("INS_TYPE",
- ((InstanceNode) n).getInstanceType());
+ x.addAttribute("AMI_ID", ((InstanceNode) n).getIdAsValue());
+ x.addAttribute("INS_TYPE", ((InstanceNode) n).getInstanceType());
} else {
- x.addAttribute("INS_ID",
- ((InstanceNode) n).getIdAsValue());
+ x.addAttribute("INS_ID", ((InstanceNode) n).getIdAsValue());
}
- x.addAttribute("USERNAME",
- ((InstanceNode) n).getUsername());
+ x.addAttribute("USERNAME", ((InstanceNode) n).getUsername());
// set to leadHeader
leadCtxHeader.setResourceMapping(x);
}
}
- invoker = new WorkflowInvokerWrapperForGFacInvoker(
- portTypeQName, gfacURLString, this.config.getMonitor()
- .getConfiguration().getMessageBoxURL()
- .toString(), leadCtxHeader,
- this.config.getNotifier().createServiceNotificationSender(node
- .getID()));
+ invoker = new WorkflowInvokerWrapperForGFacInvoker(portTypeQName, gfacURLString, this.config.getMonitor().getConfiguration().getMessageBoxURL()
+ .toString(), leadCtxHeader, this.config.getNotifier().createServiceNotificationSender(node.getID()));
} else {
-// invoker=(Invoker)getInputViaInteractor(WorkflowExecutionMessage.INPUT_GFAC_INVOKER, new GFacInvokerData(gfacEmbeddedMode,portTypeQName,
-// WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(wsNode
-// .getComponent().getWSDL()), node.getID(),
-// this.engine.getMonitor().getConfiguration()
-// .getMessageBoxURL().toASCIIString(),
-// this.engine.getMonitor().getConfiguration().getBrokerURL().toASCIIString(), this.config.getNotifier(), this.config.getTopic(),
-// this.engine.getConfiguration().getJcrComponentRegistry().getRegistry(),
-// portTypeQName.getLocalPart(),this.engine.getConfiguration()));
+ // invoker=(Invoker)getInputViaInteractor(WorkflowExecutionMessage.INPUT_GFAC_INVOKER,
+ // new GFacInvokerData(gfacEmbeddedMode,portTypeQName,
+ // WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(wsNode
+ // .getComponent().getWSDL()), node.getID(),
+ // this.engine.getMonitor().getConfiguration()
+ // .getMessageBoxURL().toASCIIString(),
+ // this.engine.getMonitor().getConfiguration().getBrokerURL().toASCIIString(),
+ // this.config.getNotifier(), this.config.getTopic(),
+ // this.engine.getConfiguration().getJcrComponentRegistry().getRegistry(),
+ // portTypeQName.getLocalPart(),this.engine.getConfiguration()));
if (this.config.isGfacEmbeddedMode()) {
- invoker = new EmbeddedGFacInvoker(portTypeQName,
- WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(wsNode.getComponent().getWSDL()),
- node.getID(),
- this.config.getMessageBoxURL().toASCIIString(),
- this.config.getMessageBrokerURL().toASCIIString(),
- this.config.getNotifier(),
- this.config.getTopic(),
- this.config.getRegistry(),
- portTypeQName.getLocalPart(),
- this.config.getConfiguration());
- } else {
- invoker = new GenericInvoker(portTypeQName,
- WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(wsNode.getComponent().getWSDL()),
- node.getID(),
- this.config.getMessageBoxURL().toASCIIString(),
- gfacURLString,
- this.config.getNotifier());
- }
-
+ invoker = new EmbeddedGFacInvoker(portTypeQName, WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(wsNode.getComponent().getWSDL()), node.getID(),
+ this.config.getMessageBoxURL().toASCIIString(), this.config.getMessageBrokerURL().toASCIIString(), this.config.getNotifier(),
+ this.config.getTopic(), this.config.getRegistry(), portTypeQName.getLocalPart(), this.config.getConfiguration());
+ } else {
+ invoker = new GenericInvoker(portTypeQName, WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(wsNode.getComponent().getWSDL()), node.getID(),
+ this.config.getMessageBoxURL().toASCIIString(), gfacURLString, this.config.getNotifier());
+ }
+
}
} else {
if (wsdlLocation.endsWith("/")) {
- wsdlLocation = wsdlLocation.substring(0,
- wsdlLocation.length() - 1);
+ wsdlLocation = wsdlLocation.substring(0, wsdlLocation.length() - 1);
}
if (!wsdlLocation.endsWith("?wsdl")) {
wsdlLocation += "?wsdl";
}
- invoker = new GenericInvoker(portTypeQName, wsdlLocation,
- node.getID(), this.getConfig().getConfiguration().getMessageBoxURL()
- .toString(), gfacURLString, this.config.getNotifier());
+ invoker = new GenericInvoker(portTypeQName, wsdlLocation, node.getID(), this.getConfig().getConfiguration().getMessageBoxURL().toString(),
+ gfacURLString, this.config.getNotifier());
}
invoker.setup();
this.invokerMap.put(node, invoker);
@@ -870,26 +822,20 @@ public class WorkflowInterpreter {
List<DataPort> inputPorts = node.getInputPorts();
ODEClient odeClient = new ODEClient();
for (DataPort port : inputPorts) {
- Object inputVal = InterpreterUtil.findInputFromPort(port,
- this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(port, this.invokerMap);
/*
* Need to override inputValue if it is odeClient
*/
if (port.getFromNode() instanceof InputNode) {
- inputVal = ODEClientUtil.parseValue(
- (WSComponentPort) port.getComponentPort(),
- (String) inputVal);
+ inputVal = ODEClientUtil.parseValue((WSComponentPort) port.getComponentPort(), (String) inputVal);
}
if (null == inputVal) {
- throw new WorkFlowInterpreterException(
- "Unable to find inputs for the node:" + node.getID());
+ throw new WorkFlowInterpreterException("Unable to find inputs for the node:" + node.getID());
}
if (port.getFromNode() instanceof EndForEachNode) {
- inputVal = ODEClientUtil.parseValue(
- (WSComponentPort) port.getComponentPort(),
- (String) inputVal);
+ inputVal = ODEClientUtil.parseValue((WSComponentPort) port.getComponentPort(), (String) inputVal);
// org.xmlpull.v1.builder.XmlElement inputElem = XMLUtil
// .stringToXmlElement3("<" + port.getName() + ">"
// + inputVal.toString() + "</" + port.getName()
@@ -902,8 +848,7 @@ public class WorkflowInterpreter {
}
private void handleDynamicComponent(Node node) throws WorkflowException {
- DynamicComponent dynamicComponent = (DynamicComponent) node
- .getComponent();
+ DynamicComponent dynamicComponent = (DynamicComponent) node.getComponent();
String className = dynamicComponent.getClassName();
String operationName = dynamicComponent.getOperationName();
URL implJarLocation = dynamicComponent.getImplJarLocation();
@@ -911,8 +856,7 @@ public class WorkflowInterpreter {
LinkedList<Object> inputs = new LinkedList<Object>();
List<DataPort> inputPorts = dynamicNode.getInputPorts();
for (DataPort dataPort : inputPorts) {
- Object inputVal = InterpreterUtil.findInputFromPort(dataPort,
- this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(dataPort, this.invokerMap);
/*
* Set type after get input value, and override inputValue if output
@@ -925,30 +869,23 @@ public class WorkflowInterpreter {
} else if (fromNode instanceof ConstantNode) {
type = ((ConstantNode) fromNode).getType();
} else if ((dataPort.getFromPort() instanceof WSPort)
- && BasicTypeMapping.isArrayType(((WSPort) dataPort
- .getFromPort()).getComponentPort().getElement())) {
+ && BasicTypeMapping.isArrayType(((WSPort) dataPort.getFromPort()).getComponentPort().getElement())) {
Invoker fromInvoker = this.invokerMap.get(fromNode);
- inputVal = BasicTypeMapping.getOutputArray(XmlConstants.BUILDER
- .parseFragmentFromString(fromInvoker.getOutputs()
- .toString()), dataPort.getFromPort().getName(),
- BasicTypeMapping
- .getSimpleTypeIndex(((DataPort) dataPort
- .getFromPort()).getType()));
+ inputVal = BasicTypeMapping.getOutputArray(XmlConstants.BUILDER.parseFragmentFromString(fromInvoker.getOutputs().toString()), dataPort
+ .getFromPort().getName(), BasicTypeMapping.getSimpleTypeIndex(((DataPort) dataPort.getFromPort()).getType()));
type = ((DataPort) dataPort.getFromPort()).getType();
} else {
type = ((DataPort) dataPort.getFromPort()).getType();
}
if (null == inputVal) {
- throw new WorkFlowInterpreterException(
- "Unable to find inputs for the node:" + node.getID());
+ throw new WorkFlowInterpreterException("Unable to find inputs for the node:" + node.getID());
}
inputs.add(BasicTypeMapping.getObjectOfType(type, inputVal));
}
- DynamicInvoker dynamicInvoker = new DynamicInvoker(className,
- implJarLocation, operationName, inputs.toArray());
+ DynamicInvoker dynamicInvoker = new DynamicInvoker(className, implJarLocation, operationName, inputs.toArray());
this.invokerMap.put(node, dynamicInvoker);
dynamicInvoker.setup();
dynamicInvoker.invoke();
@@ -961,8 +898,7 @@ public class WorkflowInterpreter {
Collection<Node> repeatNodes = node.getOutputPort(0).getToNodes();
// we will support only one for now
if (repeatNodes.size() != 1) {
- throw new WorkFlowInterpreterException(
- "Only one node allowed inside foreach");
+ throw new WorkFlowInterpreterException("Only one node allowed inside foreach");
}
Iterator<Node> iterator = repeatNodes.iterator();
if (iterator.hasNext()) {
@@ -971,15 +907,11 @@ public class WorkflowInterpreter {
// forEachNode should point to a WSNode and should have only one
// output
- if ((!(middleNode instanceof WSNode))
- && (!(middleNode instanceof SubWorkflowNode))) {
- throw new WorkFlowInterpreterException(
- "Encountered Node inside foreach that is not a WSNode"
- + middleNode);
+ if ((!(middleNode instanceof WSNode)) && (!(middleNode instanceof SubWorkflowNode))) {
+ throw new WorkFlowInterpreterException("Encountered Node inside foreach that is not a WSNode" + middleNode);
} else if (middleNode instanceof SubWorkflowNode) {
/* Get the EndforEach Node of the Subworkflow */
- Iterator<Node> subWorkflowOut = middleNode.getOutputPort(0)
- .getToNodes().iterator();
+ Iterator<Node> subWorkflowOut = middleNode.getOutputPort(0).getToNodes().iterator();
while (subWorkflowOut.hasNext()) {
Node node2 = subWorkflowOut.next();
if (node2 instanceof EndForEachNode) {
@@ -988,13 +920,9 @@ public class WorkflowInterpreter {
}
final LinkedList<String> listOfValues = new LinkedList<String>();
- InterpreterUtil.getInputsForForEachNode(forEachNode,
- listOfValues, this.invokerMap);
- final Integer[] inputNumbers = InterpreterUtil
- .getNumberOfInputsForForEachNode(forEachNode,
- this.invokerMap);
- Workflow workflow1 = ((SubWorkflowNode) middleNode)
- .getWorkflow();
+ InterpreterUtil.getInputsForForEachNode(forEachNode, listOfValues, this.invokerMap);
+ final Integer[] inputNumbers = InterpreterUtil.getNumberOfInputsForForEachNode(forEachNode, this.invokerMap);
+ Workflow workflow1 = ((SubWorkflowNode) middleNode).getWorkflow();
List<NodeImpl> nodes = workflow1.getGraph().getNodes();
List<Node> wsNodes = new ArrayList<Node>();
/* Take the List of WSNodes in the subworkflow */
@@ -1010,8 +938,7 @@ public class WorkflowInterpreter {
List<DataPort> outputPorts1 = node1.getOutputPorts();
List<Node> endForEachNodes = new ArrayList<Node>();
for (DataPort port : outputPorts1) {
- Iterator<Node> endForEachNodeItr1 = port.getToNodes()
- .iterator();
+ Iterator<Node> endForEachNodeItr1 = port.getToNodes().iterator();
while (endForEachNodeItr1.hasNext()) {
Node node2 = endForEachNodeItr1.next();
if (node2 instanceof EndForEachNode) {
@@ -1019,26 +946,21 @@ public class WorkflowInterpreter {
} else if (node2 instanceof OutputNode) {
// intentionally left noop
} else {
- throw new WorkFlowInterpreterException(
- "Found More than one node inside foreach");
+ throw new WorkFlowInterpreterException("Found More than one node inside foreach");
}
}
}
final List<Node> finalEndForEachNodes = endForEachNodes;
- Iterator<Node> endForEachNodeItr1 = node1.getOutputPort(0)
- .getToNodes().iterator();
+ Iterator<Node> endForEachNodeItr1 = node1.getOutputPort(0).getToNodes().iterator();
while (endForEachNodeItr1.hasNext()) {
Node node2 = endForEachNodeItr1.next();
// Start reading input came for foreach node
- int parallelRuns = listOfValues.size()
- * node1.getOutputPorts().size();
+ int parallelRuns = listOfValues.size() * node1.getOutputPorts().size();
if (listOfValues.size() > 0) {
- NodeController.getGUI(forEachNode).setBodyColor(
- NodeState.EXECUTING.color);
- NodeController.getGUI(node1).setBodyColor(
- NodeState.EXECUTING.color);
+ NodeController.getGUI(forEachNode).setBodyColor(NodeState.EXECUTING.color);
+ NodeController.getGUI(node1).setBodyColor(NodeState.EXECUTING.color);
List<DataPort> outputPorts = node1.getOutputPorts();
final AtomicInteger counter = new AtomicInteger();
for (Node endFor : endForEachNodes) {
@@ -1050,13 +972,10 @@ public class WorkflowInterpreter {
@Override
public void run() {
try {
- runInThread(listOfValues, forEachNode,
- node1, finalEndForEachNodes,
- finalMap, counter, inputNumbers);
+ runInThread(listOfValues, forEachNode, node1, finalEndForEachNodes, finalMap, counter, inputNumbers);
} catch (WorkflowException e) {
- WorkflowInterpreter.this.config
- .getGUI().getErrorWindow().error(e);
+ WorkflowInterpreter.this.config.getGUI().getErrorWindow().error(e);
}
}
@@ -1072,21 +991,13 @@ public class WorkflowInterpreter {
}
if (!(node2 instanceof OutputNode)) {
listOfValues.removeAll(listOfValues);
- String output = (String) systemInvoker
- .getOutput(node1.getOutputPort(0)
- .getName());
- XmlElement xmlElement = XMLUtil
- .stringToXmlElement("<result>" + output
- + "</result>");
- Iterator iterator1 = xmlElement.children()
- .iterator();
+ String output = (String) systemInvoker.getOutput(node1.getOutputPort(0).getName());
+ XmlElement xmlElement = XMLUtil.stringToXmlElement("<result>" + output + "</result>");
+ Iterator iterator1 = xmlElement.children().iterator();
while (iterator1.hasNext()) {
Object next1 = iterator1.next();
if (next1 instanceof XmlElement) {
- listOfValues
- .add((String) ((XmlElement) next1)
- .children().iterator()
- .next());
+ listOfValues.add((String) ((XmlElement) next1).children().iterator().next());
}
}
}
@@ -1105,8 +1016,7 @@ public class WorkflowInterpreter {
List<DataPort> outputPorts1 = middleNode.getOutputPorts();
List<Node> endForEachNodes = new ArrayList<Node>();
for (DataPort port : outputPorts1) {
- Iterator<Node> endForEachNodeItr1 = port.getToNodes()
- .iterator();
+ Iterator<Node> endForEachNodeItr1 = port.getToNodes().iterator();
while (endForEachNodeItr1.hasNext()) {
Node node2 = endForEachNodeItr1.next();
if (node2 instanceof EndForEachNode) {
@@ -1114,8 +1024,7 @@ public class WorkflowInterpreter {
} else if (node2 instanceof OutputNode) {
// intentionally left noop
} else {
- throw new WorkFlowInterpreterException(
- "Found More than one node inside foreach");
+ throw new WorkFlowInterpreterException("Found More than one node inside foreach");
}
}
@@ -1125,20 +1034,14 @@ public class WorkflowInterpreter {
final LinkedList<String> listOfValues = new LinkedList<String>();
// Start reading input came for foreach node
- InterpreterUtil.getInputsForForEachNode(forEachNode,
- listOfValues, this.invokerMap);
- final Integer[] inputNumbers = InterpreterUtil
- .getNumberOfInputsForForEachNode(forEachNode,
- this.invokerMap);
+ InterpreterUtil.getInputsForForEachNode(forEachNode, listOfValues, this.invokerMap);
+ final Integer[] inputNumbers = InterpreterUtil.getNumberOfInputsForForEachNode(forEachNode, this.invokerMap);
- int parallelRuns = createInputValues(listOfValues, inputNumbers)
- .size() * outputPorts1.size();
+ int parallelRuns = createInputValues(listOfValues, inputNumbers).size() * outputPorts1.size();
if (listOfValues.size() > 0) {
- NodeController.getGUI(forEachNode)
- .setBodyColor(NodeState.EXECUTING.color);
- NodeController.getGUI(foreachWSNode).setBodyColor(
- NodeState.EXECUTING.color);
+ NodeController.getGUI(forEachNode).setBodyColor(NodeState.EXECUTING.color);
+ NodeController.getGUI(foreachWSNode).setBodyColor(NodeState.EXECUTING.color);
List<DataPort> outputPorts = middleNode.getOutputPorts();
final AtomicInteger counter = new AtomicInteger();
for (Node endFor : endForEachNodes) {
@@ -1151,12 +1054,9 @@ public class WorkflowInterpreter {
@Override
public void run() {
try {
- runInThread(listOfValues, forEachNode,
- foreachWSNode, finalEndForEachNodes,
- finalInvokerMap, counter, inputNumbers);
+ runInThread(listOfValues, forEachNode, foreachWSNode, finalEndForEachNodes, finalInvokerMap, counter, inputNumbers);
} catch (WorkflowException e) {
- WorkflowInterpreter.this.config
- .getGUI().getErrorWindow().error(e);
+ WorkflowInterpreter.this.config.getGUI().getErrorWindow().error(e);
}
}
@@ -1173,12 +1073,10 @@ public class WorkflowInterpreter {
// todo this has to be done in a separate thread
NodeController.getGUI(middleNode).setBodyColor(NodeState.FINISHED.color);
for (Node endForEach : endForEachNodes) {
- NodeController.getGUI(endForEach).setBodyColor(
- NodeState.FINISHED.color);
+ NodeController.getGUI(endForEach).setBodyColor(NodeState.FINISHED.color);
}
} else {
- throw new WorkFlowInterpreterException(
- "No array values found for foreach");
+ throw new WorkFlowInterpreterException("No array values found for foreach");
}
}
@@ -1193,31 +1091,26 @@ public class WorkflowInterpreter {
*/
String booleanExpression = ifNode.getXPath();
if (booleanExpression == null) {
- throw new WorkFlowInterpreterException(
- "XPath for if cannot be null");
+ throw new WorkFlowInterpreterException("XPath for if cannot be null");
}
List<DataPort> inputPorts = node.getInputPorts();
int i = 0;
for (DataPort port : inputPorts) {
- Object inputVal = InterpreterUtil.findInputFromPort(port,
- this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(port, this.invokerMap);
if (null == inputVal) {
- throw new WorkFlowInterpreterException(
- "Unable to find inputs for the node:" + node.getID());
+ throw new WorkFlowInterpreterException("Unable to find inputs for the node:" + node.getID());
}
- booleanExpression = booleanExpression.replaceAll("\\$" + i, "'"
- + inputVal + "'");
+ booleanExpression = booleanExpression.replaceAll("\\$" + i, "'" + inputVal + "'");
}
// Now the XPath expression
try {
XPathFactory xpathFact = XPathFactory.newInstance();
XPath xpath = xpathFact.newXPath();
- Boolean result = (Boolean) xpath.evaluate(booleanExpression,
- booleanExpression, XPathConstants.BOOLEAN);
+ Boolean result = (Boolean) xpath.evaluate(booleanExpression, booleanExpression, XPathConstants.BOOLEAN);
/*
* Set control port to make execution flow continue according to
@@ -1226,8 +1119,7 @@ public class WorkflowInterpreter {
for (ControlPort controlPort : node.getControlOutPorts()) {
if (controlPort.getName().equals(IfComponent.TRUE_PORT_NAME)) {
controlPort.setConditionMet(result.booleanValue());
- } else if (controlPort.getName().equals(
- IfComponent.FALSE_PORT_NAME)) {
+ } else if (controlPort.getName().equals(IfComponent.FALSE_PORT_NAME)) {
controlPort.setConditionMet(!result.booleanValue());
}
}
@@ -1235,9 +1127,7 @@ public class WorkflowInterpreter {
NodeController.getGUI(node).setBodyColor(NodeState.FINISHED.color);
} catch (XPathExpressionException e) {
- throw new WorkFlowInterpreterException(
- "Cannot evaluate XPath in If Condition: "
- + booleanExpression);
+ throw new WorkFlowInterpreterException("Cannot evaluate XPath in If Condition: " + booleanExpression);
}
}
@@ -1246,28 +1136,18 @@ public class WorkflowInterpreter {
SystemComponentInvoker invoker = new SystemComponentInvoker();
List<DataPort> ports = endIfNode.getOutputPorts();
- for (int outputPortIndex = 0, inputPortIndex = 0; outputPortIndex < ports
- .size(); outputPortIndex++, inputPortIndex = inputPortIndex + 2) {
+ for (int outputPortIndex = 0, inputPortIndex = 0; outputPortIndex < ports.size(); outputPortIndex++, inputPortIndex = inputPortIndex + 2) {
- Object inputVal = InterpreterUtil.findInputFromPort(
- endIfNode.getInputPort(inputPortIndex), this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(endIfNode.getInputPort(inputPortIndex), this.invokerMap);
- Object inputVal2 = InterpreterUtil
- .findInputFromPort(
- endIfNode.getInputPort(inputPortIndex + 1),
- this.invokerMap);
-
- if ((inputVal == null && inputVal2 == null)
- || (inputVal != null && inputVal2 != null)) {
- throw new WorkFlowInterpreterException(
- "EndIf gets wrong input number" + "Port:"
- + inputPortIndex + " and "
- + (inputPortIndex + 1));
+ Object inputVal2 = InterpreterUtil.findInputFromPort(endIfNode.getInputPort(inputPortIndex + 1), this.invokerMap);
+
+ if ((inputVal == null && inputVal2 == null) || (inputVal != null && inputVal2 != null)) {
+ throw new WorkFlowInterpreterException("EndIf gets wrong input number" + "Port:" + inputPortIndex + " and " + (inputPortIndex + 1));
}
Object value = inputVal != null ? inputVal : inputVal2;
- invoker.addOutput(endIfNode.getOutputPort(outputPortIndex).getID(),
- value);
+ invoker.addOutput(endIfNode.getOutputPort(outputPortIndex).getID(), value);
}
this.invokerMap.put(node, invoker);
@@ -1275,9 +1155,7 @@ public class WorkflowInterpreter {
NodeController.getGUI(node).setBodyColor(NodeState.FINISHED.color);
}
- private Invoker createInvokerForEachSingleWSNode(Node foreachWSNode,
- String gfacURLString, WSComponent wsComponent)
- throws WorkflowException {
+ private Invoker createInvokerForEachSingleWSNode(Node foreachWSNode, String gfacURLString, WSComponent wsComponent) throws WorkflowException {
Invoker invoker;
String wsdlLocation = InterpreterUtil.getEPR((WSNode) foreachWSNode);
QName portTypeQName = wsComponent.getPortTypeQName();
@@ -1285,82 +1163,64 @@ public class WorkflowInterpreter {
if (gfacURLString.startsWith("https")) {
LeadContextHeader leadCtxHeader = null;
try {
- leadCtxHeader = XBayaUtil.buildLeadContextHeader(
- this.getWorkflow(),
- this.getConfig().getConfiguration(),
- new MonitorConfiguration(this.getConfig().getConfiguration()
- .getBrokerURL(), this.config.getTopic(), true,
- this.getConfig().getConfiguration().getMessageBoxURL()),
- foreachWSNode.getID(), null);
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// leadCtxHeader = XBayaUtil.buildLeadContextHeader(
-// this.getWorkflow(),
-// this.getConfig().getConfiguration(),
-// new MonitorConfiguration(this.getConfig().getConfiguration()
-// .getBrokerURL(), this.config.getTopic(), true,
-// this.getConfig().getConfiguration().getMessageBoxURL()),
-// foreachWSNode.getID(), null);
-// } else {
-// leadCtxHeader = XBayaUtil.buildLeadContextHeader(
-// this.getWorkflow(),
-// this.getConfig().getConfiguration(),
-// new MonitorConfiguration(this.getConfig().getConfiguration()
-// .getBrokerURL(), this.config.getTopic(), true,
-// this.getConfig().getConfiguration().getMessageBoxURL()),
-// foreachWSNode.getID(), null);
-// }
+ leadCtxHeader = XBayaUtil.buildLeadContextHeader(this.getWorkflow(), this.getConfig().getConfiguration(), new MonitorConfiguration(this
+ .getConfig().getConfiguration().getBrokerURL(), this.config.getTopic(), true, this.getConfig().getConfiguration()
+ .getMessageBoxURL()), foreachWSNode.getID(), null);
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // leadCtxHeader = XBayaUtil.buildLeadContextHeader(
+ // this.getWorkflow(),
+ // this.getConfig().getConfiguration(),
+ // new
+ // MonitorConfiguration(this.getConfig().getConfiguration()
+ // .getBrokerURL(), this.config.getTopic(), true,
+ // this.getConfig().getConfiguration().getMessageBoxURL()),
+ // foreachWSNode.getID(), null);
+ // } else {
+ // leadCtxHeader = XBayaUtil.buildLeadContextHeader(
+ // this.getWorkflow(),
+ // this.getConfig().getConfiguration(),
+ // new
+ // MonitorConfiguration(this.getConfig().getConfiguration()
+ // .getBrokerURL(), this.config.getTopic(), true,
+ // this.getConfig().getConfiguration().getMessageBoxURL()),
+ // foreachWSNode.getID(), null);
+ // }
} catch (URISyntaxException e) {
throw new WorkflowException(e);
}
- invoker = new WorkflowInvokerWrapperForGFacInvoker(
- portTypeQName, gfacURLString, this.getConfig().getConfiguration()
- .getMessageBoxURL().toString(), leadCtxHeader,
- this.config.getNotifier()
- .createServiceNotificationSender(foreachWSNode
- .getID()));
+ invoker = new WorkflowInvokerWrapperForGFacInvoker(portTypeQName, gfacURLString, this.getConfig().getConfiguration().getMessageBoxURL()
+ .toString(), leadCtxHeader, this.config.getNotifier().createServiceNotificationSender(foreachWSNode.getID()));
} else {
- invoker = new GenericInvoker(portTypeQName, wsdlLocation,
- foreachWSNode.getID(), this.getConfig().getConfiguration()
- .getMessageBoxURL().toString(), gfacURLString,
- this.config.getNotifier());
+ invoker = new GenericInvoker(portTypeQName, wsdlLocation, foreachWSNode.getID(), this.getConfig().getConfiguration().getMessageBoxURL()
+ .toString(), gfacURLString, this.config.getNotifier());
}
} else {
if (wsdlLocation.endsWith("/")) {
- wsdlLocation = wsdlLocation.substring(0,
- wsdlLocation.length() - 1);
+ wsdlLocation = wsdlLocation.substring(0, wsdlLocation.length() - 1);
}
if (!wsdlLocation.endsWith("?wsdl")) {
wsdlLocation += "?wsdl";
}
- invoker = new GenericInvoker(portTypeQName, wsdlLocation,
- foreachWSNode.getID(), this.getConfig().getConfiguration()
- .getMessageBoxURL().toString(), gfacURLString,
- this.config.getNotifier());
+ invoker = new GenericInvoker(portTypeQName, wsdlLocation, foreachWSNode.getID(), this.getConfig().getConfiguration().getMessageBoxURL().toString(),
+ gfacURLString, this.config.getNotifier());
}
return invoker;
}
- private void runInThread(final LinkedList<String> listOfValues,
- ForEachNode forEachNode, final Node middleNode,
- List<Node> endForEachNodes, Map<Node, Invoker> tempInvoker,
- AtomicInteger counter, final Integer[] inputNumber)
- throws WorkflowException {
+ private void runInThread(final LinkedList<String> listOfValues, ForEachNode forEachNode, final Node middleNode, List<Node> endForEachNodes,
+ Map<Node, Invoker> tempInvoker, AtomicInteger counter, final Integer[] inputNumber) throws WorkflowException {
final LinkedList<Invoker> invokerList = new LinkedList<Invoker>();
if (inputNumber.length > 1) {
- List<String> inputValues = createInputValues(listOfValues,
- inputNumber);
- for (final Iterator<String> iterator = inputValues.iterator(); iterator
- .hasNext();) {
- final String gfacURLString = this.getConfig().getConfiguration().getGFacURL()
- .toString();
+ List<String> inputValues = createInputValues(listOfValues, inputNumber);
+ for (final Iterator<String> iterator = inputValues.iterator(); iterator.hasNext();) {
+ final String gfacURLString = this.getConfig().getConfiguration().getGFacURL().toString();
final String input = iterator.next();
- WSComponent wsComponent = (WSComponent) middleNode
- .getComponent();
- final Invoker invoker2 = createInvokerForEachSingleWSNode(
- middleNode, gfacURLString, wsComponent);
+ WSComponent wsComponent = (WSComponent) middleNode.getComponent();
+ final Invoker invoker2 = createInvokerForEachSingleWSNode(middleNode, gfacURLString, wsComponent);
invokerList.add(invoker2);
new Thread() {
@@ -1368,12 +1228,10 @@ public class WorkflowInterpreter {
public void run() {
try {
getInvoker(middleNode, invoker2);
- invokeGFacService(listOfValues, middleNode,
- inputNumber, input, invoker2);
+ invokeGFacService(listOfValues, middleNode, inputNumber, input, invoker2);
} catch (WorkflowException e) {
- WorkflowInterpreter.this.config.getGUI().getErrorWindow()
- .error(e);
+ WorkflowInterpreter.this.config.getGUI().getErrorWindow().error(e);
}
}
@@ -1387,38 +1245,30 @@ public class WorkflowInterpreter {
}
} else {
Invoker invoker = null;
- for (Iterator<String> iterator = listOfValues.iterator(); iterator
- .hasNext();) {
+ for (Iterator<String> iterator = listOfValues.iterator(); iterator.hasNext();) {
String input = iterator.next();
- final String gfacURLString = this.getConfig().getConfiguration().getGFacURL()
- .toString();
+ final String gfacURLString = this.getConfig().getConfiguration().getGFacURL().toString();
- WSComponent wsComponent = (WSComponent) middleNode
- .getComponent();
- invoker = createInvokerForEachSingleWSNode(middleNode,
- gfacURLString, wsComponent);
+ WSComponent wsComponent = (WSComponent) middleNode.getComponent();
+ invoker = createInvokerForEachSingleWSNode(middleNode, gfacURLString, wsComponent);
invokerList.add(invoker);
getInvoker(middleNode, invoker);
// find inputs
List<DataPort> inputPorts = middleNode.getInputPorts();
for (DataPort port : inputPorts) {
- Object inputVal = InterpreterUtil.findInputFromPort(port,
- this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(port, this.invokerMap);
/*
* Handle ForEachNode
*/
Node fromNode = port.getFromNode();
// if (fromNode instanceof ForEachNode) {
- inputVal = ODEClientUtil.parseValue(
- (WSComponentPort) port.getComponentPort(), input);
+ inputVal = ODEClientUtil.parseValue((WSComponentPort) port.getComponentPort(), input);
// }
if (null == inputVal) {
- throw new WorkFlowInterpreterException(
- "Unable to find inputs for the node:"
- + middleNode.getID());
+ throw new WorkFlowInterpreterException("Unable to find inputs for the node:" + middleNode.getID());
}
invoker.setInput(port.getName(), inputVal);
}
@@ -1437,15 +1287,14 @@ public class WorkflowInterpreter {
int i = 0;
for (DataPort port : middleNode.getOutputPorts()) {
String outputString = "";
- for (Iterator<Invoker> iterator = invokerList.iterator(); iterator
- .hasNext();) {
+ for (Iterator<Invoker> iterator = invokerList.iterator(); iterator.hasNext();) {
Invoker workflowInvoker = iterator.next();
// /
Object output = workflowInvoker.getOutput(port.getName());
if (output instanceof org.xmlpull.v1.builder.XmlElement) {
- org.xmlpull.v1.builder.XmlElement element = (org.xmlpull.v1.builder.XmlElement) ((org.xmlpull.v1.builder.XmlElement) output)
- .children().next();
+ org.xmlpull.v1.builder.XmlElement element = (org.xmlpull.v1.builder.XmlElement) ((org.xmlpull.v1.builder.XmlElement) output).children()
+ .next();
outputString += "\n" + XMLUtil.xmlElementToString(element);
} else {
outputString += "\n<value>" + output + "</value>";
@@ -1463,9 +1312,7 @@ public class WorkflowInterpreter {
for (Node endForEachNode : endForEachNodes) {
if (tempInvoker.get(endForEachNode) != null) {
if (!(endForEachNode instanceof OutputNode)) {
- ((SystemComponentInvoker) tempInvoker
- .get(endForEachNode)).addOutput(port.getName(),
- outputStr[i]);
+ ((SystemComponentInvoker) tempInvoker.get(endForEachNode)).addOutput(port.getName(), outputStr[i]);
}
}
outputPortIndex++;
@@ -1475,9 +1322,8 @@ public class WorkflowInterpreter {
NodeController.getGUI(forEachNode).setBodyColor(NodeState.FINISHED.color);
}
- private void invokeGFacService(LinkedList<String> listOfValues,
- Node middleNode, Integer[] inputNumber, String input,
- Invoker invoker) throws WorkflowException {
+ private void invokeGFacService(LinkedList<String> listOfValues, Node middleNode, Integer[] inputNumber, String input, Invoker invoker)
+ throws WorkflowException {
// find inputs
List<DataPort> inputPorts = middleNode.getInputPorts();
@@ -1489,8 +1335,7 @@ public class WorkflowInterpreter {
}
int index = 0;
for (DataPort port : inputPorts) {
- Object inputVal = InterpreterUtil.findInputFromPort(port,
- this.invokerMap);
+ Object inputVal = InterpreterUtil.findInputFromPort(port, this.invokerMap);
/*
* Handle ForEachNode
*/
@@ -1500,9 +1345,7 @@ public class WorkflowInterpreter {
}
if (null == inputVal) {
- throw new WorkFlowInterpreterException(
- "Unable to find inputs for the node:"
- + middleNode.getID());
+ throw new WorkFlowInterpreterException("Unable to find inputs for the node:" + middleNode.getID());
}
invoker.setInput(port.getName(), inputVal);
}
@@ -1510,8 +1353,7 @@ public class WorkflowInterpreter {
}
- private Invoker getInvoker(Node middleNode, Invoker invoker)
- throws WorkflowException {
+ private Invoker getInvoker(Node middleNode, Invoker invoker) throws WorkflowException {
if (middleNode instanceof WSNode) {
WSComponent wsComponent = (WSComponent) middleNode.getComponent();
invoker.setup();
@@ -1522,17 +1364,12 @@ public class WorkflowInterpreter {
// TODO : Need to create a invoker!
// new WorkflowInterpreter()
} else {
- throw new WorkflowRuntimeException(
- "Only Web services and subworkflows are supported for For-Each : Found : "
- + middleNode);
+ throw new WorkflowRuntimeException("Only Web services and subworkflows are supported for For-Each : Found : " + middleNode);
}
return invoker;
}
-
-
- private List<String> createInputValues(LinkedList<String> listOfValues,
- Integer[] inputNumbers) throws WorkflowException {
+ private List<String> createInputValues(LinkedList<String> listOfValues, Integer[] inputNumbers) throws WorkflowException {
List<String> inputValues = null;
try {
inputValues = new ArrayList<String>();
@@ -1542,8 +1379,7 @@ public class WorkflowInterpreter {
if (this.config.isRunWithCrossProduct()) {
for (int i = 0; i < inputNumbers[0]; i++) {
for (int j = 0; j < inputNumbers[1]; j++) {
- inputValues.add(listOfValues.get(i) + ","
- + listOfValues.get(inputNumbers[0] + j));
+ inputValues.add(listOfValues.get(i) + "," + listOfValues.get(inputNumbers[0] + j));
}
}
@@ -1595,10 +1431,8 @@ public class WorkflowInterpreter {
ArrayList<Node> list = new ArrayList<Node>();
List<NodeImpl> nodes = this.getGraph().getNodes();
for (Node node : nodes) {
- if (node instanceof OutputNode
- && NodeController.getGUI(node).getBodyColor() == NodeGUI.DEFAULT_BODY_COLOR
- && NodeController.getGUI(node.getInputPort(0).getFromNode())
- .getBodyColor() == NodeState.FINISHED.color) {
+ if (node instanceof OutputNode && NodeController.getGUI(node).getBodyColor() == NodeGUI.DEFAULT_BODY_COLOR
+ && NodeController.getGUI(node.getInputPort(0).getFromNode()).getBodyColor() == NodeState.FINISHED.color) {
list.add(node);
}
@@ -1607,8 +1441,7 @@ public class WorkflowInterpreter {
}
private int getRemainNodesDynamically() {
- return InterpreterUtil.getWaitingNodeCountDynamically(this.getGraph())
- + InterpreterUtil.getRunningNodeCountDynamically(this.getGraph()) ;
+ return InterpreterUtil.getWaitingNodeCountDynamically(this.getGraph()) + InterpreterUtil.getRunningNodeCountDynamically(this.getGraph());
}
private ArrayList<Node> getInputNodesDynamically() {
@@ -1620,9 +1453,7 @@ public class WorkflowInterpreter {
List<NodeImpl> nodes = wf.getGraph().getNodes();
for (Node node : nodes) {
String name = node.getComponent().getName();
- if (InputComponent.NAME.equals(name)
- || ConstantComponent.NAME.equals(name)
- || S3InputComponent.NAME.equals(name)) {
+ if (InputComponent.NAME.equals(name) || ConstantComponent.NAME.equals(name) || S3InputComponent.NAME.equals(name)) {
list.add(node);
}
}
@@ -1635,12 +1466,8 @@ public class WorkflowInterpreter {
ArrayList<Node> finishedNodes = InterpreterUtil.getFinishedNodesDynamically(this.getGraph());
for (Node node : waiting) {
Component component = node.getComponent();
- if (component instanceof WSComponent
- || component instanceof DynamicComponent
- || component instanceof SubWorkflowComponent
- || component instanceof ForEachComponent
- || component instanceof EndForEachComponent
- || component instanceof IfComponent
+ if (component instanceof WSComponent || component instanceof DynamicComponent || component instanceof SubWorkflowComponent
+ || component instanceof ForEachComponent || component instanceof EndForEachComponent || component instanceof IfComponent
|| component instanceof InstanceComponent) {
/*
@@ -1650,13 +1477,10 @@ public class WorkflowInterpreter {
boolean controlDone = true;
if (control != null) {
for (EdgeImpl edge : control.getEdges()) {
- controlDone = controlDone
- && (finishedNodes.contains(edge.getFromPort()
- .getNode())
- // amazon component use condition met to check
- // whether the control port is done
- || ((ControlPort) edge.getFromPort())
- .isConditionMet());
+ controlDone = controlDone && (finishedNodes.contains(edge.getFromPort().getNode())
+ // amazon component use condition met to check
+ // whether the control port is done
+ || ((ControlPort) edge.getFromPort()).isConditionMet());
}
}
@@ -1666,8 +1490,7 @@ public class WorkflowInterpreter {
List<DataPort> inputPorts = node.getInputPorts();
boolean inputsDone = true;
for (DataPort dataPort : inputPorts) {
- inputsDone = inputsDone
- && finishedNodes.contains(dataPort.getFromNode());
+ inputsDone = inputsDone && finishedNodes.contains(dataPort.getFromNode());
}
if (inputsDone && controlDone) {
list.add(node);
@@ -1696,9 +1519,7 @@ public class WorkflowInterpreter {
boolean controlDone = true;
if (control != null) {
for (EdgeImpl edge : control.getEdges()) {
- controlDone = controlDone
- && finishedNodes.contains(edge.getFromPort()
- .getFromNode());
+ controlDone = controlDone && finishedNodes.contains(edge.getFromPort().getFromNode());
}
}
@@ -1708,33 +1529,43 @@ public class WorkflowInterpreter {
List<DataPort> inputPorts = node.getInputPorts();
boolean inputsDone = true;
for (DataPort dataPort : inputPorts) {
- inputsDone = inputsDone
- && finishedNodes.contains(dataPort.getFromNode());
+ inputsDone = inputsDone && finishedNodes.contains(dataPort.getFromNode());
}
if (inputsDone && controlDone) {
list.add(node);
}
- } else if (InputComponent.NAME.equals(component.getName())
- || DifferedInputComponent.NAME.equals(component.getName())
- || S3InputComponent.NAME.equals(component.getName())
- || OutputComponent.NAME.equals(component.getName())
- || MemoComponent.NAME.equals(component.getName())) {
+ } else if (InputComponent.NAME.equals(component.getName()) || DifferedInputComponent.NAME.equals(component.getName())
+ || S3InputComponent.NAME.equals(component.getName()) || OutputComponent.NAME.equals(component.getName())
+ || MemoComponent.NAME.equals(component.getName()) || component instanceof EndDoWhileComponent) {
// no op
+ } else if (component instanceof DoWhileComponent) {
+ ControlPort control = node.getControlInPort();
+ boolean controlDone = true;
+ if (control != null) {
+ for (EdgeImpl edge : control.getEdges()) {
+ controlDone = controlDone && finishedNodes.contains(edge.getFromPort().getFromNode());
+ }
+ }
+
+ if (controlDone) {
+ list.add(node);
+ }
} else {
- throw new WorkFlowInterpreterException(
- "Component Not handled :" + component.getName());
+ throw new WorkFlowInterpreterException("Component Not handled :" + component.getName());
}
}
notifyViaInteractor(WorkflowExecutionMessage.HANDLE_DEPENDENT_NODES_DIFFERED_INPUTS, this.getGraph());
-// if (this.config.getMode() == WorkflowInterpreterConfiguration.GUI_MODE) {
-// ArrayList<Node> waitingNodes = InterpreterUtil.getWaitingNodesDynamically(this.getGraph());
-// for (Node readyNode : waitingNodes) {
-// DifferedInputHandler.handleDifferredInputsofDependentNodes(
-// readyNode, config.getGUI());
-// }
-// }
+ // if (this.config.getMode() ==
+ // WorkflowInterpreterConfiguration.GUI_MODE) {
+ // ArrayList<Node> waitingNodes =
+ // InterpreterUtil.getWaitingNodesDynamically(this.getGraph());
+ // for (Node readyNode : waitingNodes) {
+ // DifferedInputHandler.handleDifferredInputsofDependentNodes(
+ // readyNode, config.getGUI());
+ // }
+ // }
return list;
@@ -1759,4 +1590,24 @@ public class WorkflowInterpreter {
private WSGraph getGraph() {
return this.getWorkflow().getGraph();
}
+
+ private ArrayList<Node> getFinishedNodesDynamically() {
+ return this.getNodesWithBodyColor(NodeState.FINISHED.color);
+ }
+
+ private ArrayList<Node> getWaitingNodesDynamically() {
+ return this.getNodesWithBodyColor(NodeState.DEFAULT.color);
+ }
+
+ private ArrayList<Node> getNodesWithBodyColor(Color color) {
+ ArrayList<Node> list = new ArrayList<Node>();
+ List<NodeImpl> nodes = getGraph().getNodes();
+ for (Node node : nodes) {
+ if (NodeController.getGUI(node).getBodyColor() == color) {
+ list.add(node);
+ }
+ }
+ return list;
+ }
+
}
\ No newline at end of file