You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oodt.apache.org by ma...@apache.org on 2012/09/13 23:19:16 UTC
svn commit: r1384539 - in /oodt/trunk/workflow/src:
main/java/org/apache/oodt/cas/workflow/engine/
main/java/org/apache/oodt/cas/workflow/engine/processor/
main/java/org/apache/oodt/cas/workflow/engine/runner/
main/java/org/apache/oodt/cas/workflow/ins...
Author: mattmann
Date: Thu Sep 13 21:19:15 2012
New Revision: 1384539
URL: http://svn.apache.org/viewvc?rev=1384539&view=rev
Log:
- OODT-495: WIP
- start to flesh out WorkflowProcesor nextState function (1/4 implemented; need pre/post cond status eval; move status checkoff to EngineRunner per OODT-496)
- lifecycle status messages make sense
- fix unit tests to pass with changes to workflow Insts and Priorities, etc.
- in-mem cacheWPs (TODO: delete WPs when they are done likely per OODT-496)
Modified:
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/PrioritizedQueueBasedWorkflowEngine.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/TaskQuerier.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngine.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngineFactory.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ConditionProcessor.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ParallelProcessor.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/SequentialProcessor.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/TaskProcessor.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessor.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorBuilder.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorHelper.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorQueue.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/runner/AsynchronousLocalEngineRunner.java
oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/instrepo/LuceneWorkflowInstanceRepository.java
oodt/trunk/workflow/src/main/resources/examples/wengine/wengine-lifecycle.xml
oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/QuerierAndRunnerUtils.java
oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/SimpleTester.java
oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/TestAsynchronousLocalEngineRunner.java
oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestFILOPrioritySorter.java
oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestFIFOPrioritySorter.java
oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestPrioritySorter.java
oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/system/TestXmlRpcWorkflowManager.java
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/PrioritizedQueueBasedWorkflowEngine.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/PrioritizedQueueBasedWorkflowEngine.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/PrioritizedQueueBasedWorkflowEngine.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/PrioritizedQueueBasedWorkflowEngine.java Thu Sep 13 21:19:15 2012
@@ -114,23 +114,13 @@ public class PrioritizedQueueBasedWorkfl
(ParentChildWorkflow)workflow:new ParentChildWorkflow(workflow));
inst.setStartDate(Calendar.getInstance().getTime());
inst.setCurrentTaskId(workflow.getTasks().get(0).getTaskId());
- inst.setId(UUID.randomUUID().toString());
+ inst.setId(UUID.randomUUID().toString()); //TODO: decide whether or not this ID setting makes sense
inst.setSharedContext(metadata);
- inst.setPriority(Priority.getDefault());
- WorkflowLifecycle cycle =
- lifecycle.getLifecycleForWorkflow(workflow) != null ?
- lifecycle.getLifecycleForWorkflow(workflow):
- lifecycle.getDefaultLifecycle();
- WorkflowState state = cycle.getStateByName("Queued");
- state.setMessage("Workflow started and Queued.");
+ inst.setPriority(Priority.getDefault()); //FIXME: this should be sensed or passed in
+ WorkflowLifecycle cycle = getLifecycleForWorkflow(workflow);
+ WorkflowState state = cycle.createState("Null", "initial", "Workflow created by Engine.");
inst.setState(state);
- try {
- this.repo.addWorkflowInstance(inst);
- } catch (InstanceRepositoryException e) {
- e.printStackTrace();
- throw new EngineException(e.getMessage());
- }
-
+ persist(inst);
return inst;
}
@@ -247,7 +237,20 @@ public class PrioritizedQueueBasedWorkfl
// TODO Auto-generated method stub
return null;
}
+
+ private void persist(WorkflowInstance inst) throws EngineException{
+ try {
+ this.repo.addWorkflowInstance(inst);
+ } catch (InstanceRepositoryException e) {
+ e.printStackTrace();
+ throw new EngineException(e.getMessage());
+ }
+ }
- // FIXME: add in methods from WEngine
+ private WorkflowLifecycle getLifecycleForWorkflow(Workflow workflow){
+ return lifecycle.getLifecycleForWorkflow(workflow) != null ?
+ lifecycle.getLifecycleForWorkflow(workflow):
+ lifecycle.getDefaultLifecycle();
+ }
}
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/TaskQuerier.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/TaskQuerier.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/TaskQuerier.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/TaskQuerier.java Thu Sep 13 21:19:15 2012
@@ -26,11 +26,13 @@ import java.util.logging.Logger;
//OODT imports
import org.apache.oodt.cas.workflow.engine.processor.TaskProcessor;
import org.apache.oodt.cas.workflow.engine.processor.WorkflowProcessor;
+import org.apache.oodt.cas.workflow.engine.processor.WorkflowProcessorHelper;
import org.apache.oodt.cas.workflow.engine.processor.WorkflowProcessorQueue;
import org.apache.oodt.cas.workflow.instrepo.WorkflowInstanceRepository;
import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycle;
import org.apache.oodt.cas.workflow.lifecycle.WorkflowState;
import org.apache.oodt.cas.workflow.structs.PrioritySorter;
+import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
/**
*
@@ -98,31 +100,32 @@ public class TaskQuerier implements Runn
List<WorkflowProcessor> processors = processorQueue.getProcessors();
List<WorkflowProcessor> processorsToRun = new Vector<WorkflowProcessor>();
+ System.out.println("HERE HERE HERE!");
for (WorkflowProcessor processor : processors) {
// OK now get its lifecycle
- WorkflowLifecycle lifecycle = getLifecycleForProcessor(processor);
- if (!(processor.getWorkflowInstance().getState().getCategory().getName().equals("done") || processor
- .getWorkflowInstance().getState().getCategory().getName().equals("holding"))) {
+ WorkflowProcessorHelper helper = new WorkflowProcessorHelper(
+ processor.getLifecycleManager());
+ WorkflowLifecycle lifecycle = helper
+ .getLifecycleForProcessor(processor);
+ System.out.println("TESTING STATE!");
+ if (!(processor.getWorkflowInstance().getState().getCategory()
+ .getName().equals("done") || processor.getWorkflowInstance()
+ .getState().getCategory().getName().equals("holding")) &&
+ !processor.getWorkflowInstance().getState().getName().equals("Executing")) {
+ System.out.println("Checking for RUNNABLE TASK PROCESSORS");
for (TaskProcessor tp : processor.getRunnableWorkflowProcessors()) {
WorkflowState state = lifecycle.createState("Executing", "running",
"Added to Runnable queue");
tp.getWorkflowInstance().setState(state);
- if (this.repo != null) {
- try {
- this.repo.updateWorkflowInstance(tp.getWorkflowInstance());
- } catch (Exception e) {
- e.printStackTrace();
- LOG.log(Level.WARNING, "Unable to update workflow instance: ["
- + tp.getWorkflowInstance().getId()
- + "] status to Executing. Message: " + e.getMessage());
- }
- }
- LOG.log(Level.INFO,
- "Added processor with priority: [" + tp.getWorkflowInstance().getPriority() + "]");
+ persist(tp.getWorkflowInstance());
+ LOG.log(Level.INFO, "Added processor with priority: ["
+ + tp.getWorkflowInstance().getPriority() + "]");
processorsToRun.add(tp);
}
- prioritizer.sort(processorsToRun);
+ if(processorsToRun != null && processorsToRun.size() > 1){
+ prioritizer.sort(processorsToRun);
+ }
synchronized (runnableProcessors) {
if (running)
@@ -130,7 +133,12 @@ public class TaskQuerier implements Runn
}
} else {
- continue;
+ // simply call nextState and persist it
+ LOG.log(Level.FINE, "Processor for workflow instance: ["
+ + processor.getWorkflowInstance().getId()
+ + "] not ready to Execute or already Executing: advancing it to next state.");
+ processor.nextState();
+ persist(processor.getWorkflowInstance());
}
}
}
@@ -172,13 +180,17 @@ public class TaskQuerier implements Runn
return (TaskProcessor) getRunnableProcessors().remove(0);
}
- private WorkflowLifecycle getLifecycleForProcessor(WorkflowProcessor processor) {
- if (processor.getWorkflowInstance() != null
- && processor.getWorkflowInstance().getParentChildWorkflow() != null) {
- return processor.getLifecycleManager().getLifecycleForWorkflow(
- processor.getWorkflowInstance().getParentChildWorkflow());
- } else
- return processor.getLifecycleManager().getDefaultLifecycle();
+ private void persist(WorkflowInstance instance) {
+ if (this.repo != null) {
+ try {
+ this.repo.updateWorkflowInstance(instance);
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOG.log(Level.WARNING, "Unable to update workflow instance: ["
+ + instance.getId() + "] status to " + instance.getState().getName()
+ + "]. Message: " + e.getMessage());
+ }
+ }
}
}
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngine.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngine.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngine.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngine.java Thu Sep 13 21:19:15 2012
@@ -19,7 +19,6 @@ package org.apache.oodt.cas.workflow.eng
//OODT imports
import org.apache.oodt.cas.metadata.Metadata;
-import org.apache.oodt.cas.resource.structs.Job;
import org.apache.oodt.cas.workflow.engine.processor.ConditionProcessor;
import org.apache.oodt.cas.workflow.engine.processor.SequentialProcessor;
import org.apache.oodt.cas.workflow.engine.runner.EngineRunner;
@@ -28,7 +27,6 @@ import org.apache.oodt.cas.workflow.life
import org.apache.oodt.cas.workflow.structs.Workflow;
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.structs.WorkflowStatus;
-import org.apache.oodt.cas.workflow.structs.WorkflowTask;
import org.apache.oodt.cas.workflow.structs.exceptions.EngineException;
import org.apache.oodt.cas.workflow.structs.exceptions.InstanceRepositoryException;
import org.apache.oodt.commons.util.DateConvert;
@@ -46,7 +44,6 @@ import EDU.oswego.cs.dl.util.concurrent.
import EDU.oswego.cs.dl.util.concurrent.Channel;
import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
-import EDU.oswego.cs.dl.util.concurrent.ThreadedExecutor;
/**
*
@@ -106,7 +103,7 @@ public class ThreadPoolWorkflowEngine im
*/
public ThreadPoolWorkflowEngine(WorkflowInstanceRepository instRep,
int queueSize, int maxPoolSize, int minPoolSize,
- long threadKeepAliveTime, boolean unlimitedQueue) {
+ long threadKeepAliveTime, boolean unlimitedQueue, WorkflowLifecycleManager lifecycleManager) {
this.instRep = instRep;
@@ -120,6 +117,7 @@ public class ThreadPoolWorkflowEngine im
pool = new PooledExecutor(c, maxPoolSize);
pool.setMinimumPoolSize(minPoolSize);
pool.setKeepAliveTime(1000 * 60 * threadKeepAliveTime);
+ this.lifecycleManager = lifecycleManager;
workerMap = new HashMap();
@@ -127,7 +125,7 @@ public class ThreadPoolWorkflowEngine im
"org.apache.oodt.cas.workflow.engine.preConditionWaitTime", 10)
.longValue();
- this.condProcessor = new ConditionProcessor(lifecycleManager);
+ this.condProcessor = new ConditionProcessor(lifecycleManager, new WorkflowInstance());
}
@Override
@@ -159,7 +157,7 @@ public class ThreadPoolWorkflowEngine im
wInst.setStatus(CREATED);
persistWorkflowInstance(wInst);
- SequentialProcessor worker = new SequentialProcessor(this.lifecycleManager);
+ SequentialProcessor worker = new SequentialProcessor(this.lifecycleManager, wInst);
worker.setWorkflowInstance(wInst);
workerMap.put(wInst.getId(), worker);
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngineFactory.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngineFactory.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngineFactory.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/ThreadPoolWorkflowEngineFactory.java Thu Sep 13 21:19:15 2012
@@ -22,62 +22,72 @@ import static java.lang.Integer.getInteg
import static java.lang.Long.getLong;
//OODT imports
+import org.apache.oodt.cas.metadata.util.PathUtils;
import org.apache.oodt.cas.workflow.instrepo.WorkflowInstanceRepository;
+import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycleManager;
import org.apache.oodt.cas.workflow.util.GenericWorkflowObjectFactory;
/**
* A Factory class for creating {@link ThreadPoolWorkflowEngine}s.
- *
+ *
* @author mattmann (Chris Mattmann)
* @author bfoster (Brian Foster)
*/
public class ThreadPoolWorkflowEngineFactory implements WorkflowEngineFactory {
- private static final String INSTANCE_REPO_FACTORY_PROPERTY = "workflow.engine.instanceRep.factory";
- private static final String QUEUE_SIZE_PROPERTY = "org.apache.oodt.cas.workflow.engine.queueSize";
- private static final String MAX_POOL_SIZE_PROPERTY = "org.apache.oodt.cas.workflow.engine.maxPoolSize";
- private static final String MIN_POOL_SIZE_PROPERTY = "org.apache.oodt.cas.workflow.engine.minPoolSize";
- private static final String THREAD_KEEP_ALIVE_PROPERTY = "org.apache.oodt.cas.workflow.engine.threadKeepAlive.minutes";
- private static final String UNLIMITED_QUEUE_PROPERTY = "org.apache.oodt.cas.workflow.engine.unlimitedQueue";
-
- private static final int DEFAULT_QUEUE_SIZE = 10;
- private static final int DEFAULT_MAX_POOL_SIZE = 10;
- private static final int DEFAULT_MIN_POOL_SIZE = 4;
- private static final long DEFAULT_THREAD_KEEP_ALIVE_MINS = 5;
-
- @Override
- public WorkflowEngine createWorkflowEngine() {
- return new ThreadPoolWorkflowEngine(getWorkflowInstanceRepository(),
- getQueueSize(),
- getMaxPoolSize(),
- getMinPoolSize(),
- getThreadKeepAliveMinutes(),
- isUnlimitedQueue());
- }
-
- protected WorkflowInstanceRepository getWorkflowInstanceRepository() {
- return GenericWorkflowObjectFactory
- .getWorkflowInstanceRepositoryFromClassName(System
- .getProperty(INSTANCE_REPO_FACTORY_PROPERTY));
- }
-
- protected int getQueueSize() {
- return getInteger(QUEUE_SIZE_PROPERTY, DEFAULT_QUEUE_SIZE);
- }
-
- protected int getMaxPoolSize() {
- return getInteger(MAX_POOL_SIZE_PROPERTY, DEFAULT_MAX_POOL_SIZE);
- }
-
- protected int getMinPoolSize() {
- return getInteger(MIN_POOL_SIZE_PROPERTY, DEFAULT_MIN_POOL_SIZE);
- }
-
- protected long getThreadKeepAliveMinutes() {
- return getLong(THREAD_KEEP_ALIVE_PROPERTY, DEFAULT_THREAD_KEEP_ALIVE_MINS);
- }
-
- protected boolean isUnlimitedQueue() {
- return getBoolean(UNLIMITED_QUEUE_PROPERTY);
- }
+ private static final String INSTANCE_REPO_FACTORY_PROPERTY = "workflow.engine.instanceRep.factory";
+ private static final String QUEUE_SIZE_PROPERTY = "org.apache.oodt.cas.workflow.engine.queueSize";
+ private static final String MAX_POOL_SIZE_PROPERTY = "org.apache.oodt.cas.workflow.engine.maxPoolSize";
+ private static final String MIN_POOL_SIZE_PROPERTY = "org.apache.oodt.cas.workflow.engine.minPoolSize";
+ private static final String THREAD_KEEP_ALIVE_PROPERTY = "org.apache.oodt.cas.workflow.engine.threadKeepAlive.minutes";
+ private static final String UNLIMITED_QUEUE_PROPERTY = "org.apache.oodt.cas.workflow.engine.unlimitedQueue";
+ private static final String LIFECYCLES_FILE_PATH_PROPERTY = "org.apache.oodt.cas.workflow.lifecycle.filePath";
+
+ private static final int DEFAULT_QUEUE_SIZE = 10;
+ private static final int DEFAULT_MAX_POOL_SIZE = 10;
+ private static final int DEFAULT_MIN_POOL_SIZE = 4;
+ private static final long DEFAULT_THREAD_KEEP_ALIVE_MINS = 5;
+
+ @Override
+ public WorkflowEngine createWorkflowEngine() {
+ return new ThreadPoolWorkflowEngine(getWorkflowInstanceRepository(),
+ getQueueSize(), getMaxPoolSize(), getMinPoolSize(),
+ getThreadKeepAliveMinutes(), isUnlimitedQueue(), getWorkflowLifecycle());
+ }
+
+ protected WorkflowInstanceRepository getWorkflowInstanceRepository() {
+ return GenericWorkflowObjectFactory
+ .getWorkflowInstanceRepositoryFromClassName(System
+ .getProperty(INSTANCE_REPO_FACTORY_PROPERTY));
+ }
+
+ protected WorkflowLifecycleManager getWorkflowLifecycle() {
+ try {
+ return new WorkflowLifecycleManager(PathUtils.replaceEnvVariables(System
+ .getProperty(LIFECYCLES_FILE_PATH_PROPERTY)));
+ } catch (Exception e) {
+ e.printStackTrace();
+ return null;
+ }
+ }
+
+ protected int getQueueSize() {
+ return getInteger(QUEUE_SIZE_PROPERTY, DEFAULT_QUEUE_SIZE);
+ }
+
+ protected int getMaxPoolSize() {
+ return getInteger(MAX_POOL_SIZE_PROPERTY, DEFAULT_MAX_POOL_SIZE);
+ }
+
+ protected int getMinPoolSize() {
+ return getInteger(MIN_POOL_SIZE_PROPERTY, DEFAULT_MIN_POOL_SIZE);
+ }
+
+ protected long getThreadKeepAliveMinutes() {
+ return getLong(THREAD_KEEP_ALIVE_PROPERTY, DEFAULT_THREAD_KEEP_ALIVE_MINS);
+ }
+
+ protected boolean isUnlimitedQueue() {
+ return getBoolean(UNLIMITED_QUEUE_PROPERTY);
+ }
}
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ConditionProcessor.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ConditionProcessor.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ConditionProcessor.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ConditionProcessor.java Thu Sep 13 21:19:15 2012
@@ -32,8 +32,8 @@ import org.apache.oodt.cas.workflow.stru
*/
public class ConditionProcessor extends TaskProcessor {
- public ConditionProcessor(WorkflowLifecycleManager lifecycleManager) {
- super(lifecycleManager);
+ public ConditionProcessor(WorkflowLifecycleManager lifecycleManager, WorkflowInstance inst) {
+ super(lifecycleManager, inst);
}
@Override
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ParallelProcessor.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ParallelProcessor.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ParallelProcessor.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/ParallelProcessor.java Thu Sep 13 21:19:15 2012
@@ -21,6 +21,7 @@ import java.util.List;
//OODT imports
import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycleManager;
+import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
/**
*
@@ -33,12 +34,8 @@ import org.apache.oodt.cas.workflow.life
*/
public class ParallelProcessor extends WorkflowProcessor {
- public ParallelProcessor() {
- this(null);
- }
-
- public ParallelProcessor(WorkflowLifecycleManager lifecycleMgr) {
- super(lifecycleMgr);
+ public ParallelProcessor(WorkflowLifecycleManager lifecycleMgr, WorkflowInstance instance) {
+ super(lifecycleMgr, instance);
}
public List<WorkflowProcessor> getRunnableSubProcessors() {
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/SequentialProcessor.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/SequentialProcessor.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/SequentialProcessor.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/SequentialProcessor.java Thu Sep 13 21:19:15 2012
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Vector;
import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycleManager;
+import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
/**
*
@@ -33,12 +34,8 @@ import org.apache.oodt.cas.workflow.life
*/
public class SequentialProcessor extends WorkflowProcessor {
- public SequentialProcessor() {
- this(null);
- }
-
- public SequentialProcessor(WorkflowLifecycleManager lifecycleManager) {
- super(lifecycleManager);
+ public SequentialProcessor(WorkflowLifecycleManager lifecycleManager, WorkflowInstance instance) {
+ super(lifecycleManager, instance);
}
@Override
@@ -58,7 +55,7 @@ public class SequentialProcessor extends
private WorkflowProcessor getNext() {
for (WorkflowProcessor wp : this.getSubProcessors())
if (!wp.getWorkflowInstance().getState().getCategory().getName()
- .equals("done"))
+ .equals("done") && !wp.getWorkflowInstance().getState().getName().equals("Executing"))
return wp;
return null;
}
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/TaskProcessor.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/TaskProcessor.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/TaskProcessor.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/TaskProcessor.java Thu Sep 13 21:19:15 2012
@@ -41,13 +41,9 @@ public class TaskProcessor extends Workf
private Class<? extends WorkflowTaskInstance> instanceClass;
private String jobId;
-
- public TaskProcessor() {
- this(null);
- }
-
- public TaskProcessor(WorkflowLifecycleManager lifecycleManager) {
- super(lifecycleManager);
+
+ public TaskProcessor(WorkflowLifecycleManager lifecycleManager, WorkflowInstance instance) {
+ super(lifecycleManager, instance);
}
public Class<? extends WorkflowTaskInstance> getInstanceClass() {
@@ -95,9 +91,11 @@ public class TaskProcessor extends Workf
.getTimeInMillis()) / 1000) / 60;
if (elapsedTime >= requiredBlockTimeElapse)
tps.add(this);
- } else if (this.getWorkflowInstance().getState().getName()
- .equals("Queued")
- && this.passedPreConditions()
+ } else if (((this.getWorkflowInstance().getState().getName()
+ .equals("Queued") || this.getWorkflowInstance().getState()
+ .getName().equals("Loaded") || this.getWorkflowInstance().getState()
+ .getName().equals("Executing"))
+ && this.passedPreConditions())
|| this.getWorkflowInstance().getState().getName()
.equals("PreConditionSuccess")) {
tps.add(this);
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessor.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessor.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessor.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessor.java Thu Sep 13 21:19:15 2012
@@ -56,18 +56,19 @@ public abstract class WorkflowProcessor
private int minReqSuccessfulSubProcessors; //FIXME: read this in PackagedRepo: flow through instance
protected WorkflowLifecycleManager lifecycleManager;
protected WorkflowProcessorHelper helper;
-
- public WorkflowProcessor(WorkflowLifecycleManager lifecycleManager) {
+
+ public WorkflowProcessor(WorkflowLifecycleManager lifecycleManager, WorkflowInstance workflowInstance){
this.subProcessors = new Vector<WorkflowProcessor>();
this.listeners = new Vector<WorkflowProcessorListener>();
this.excusedSubProcessorIds = new Vector<String>();
this.minReqSuccessfulSubProcessors = -1;
- this.workflowInstance = new WorkflowInstance();
this.lifecycleManager = lifecycleManager;
+ this.workflowInstance = workflowInstance;
this.helper = new WorkflowProcessorHelper(lifecycleManager);
+ WorkflowState initState = helper.getLifecycleForProcessor(this).createState("Null", "initial", "Instance created by workflow processor.");
+ this.workflowInstance.setState(initState);
}
-
/**
* @return the workflowInstance
*/
@@ -308,6 +309,52 @@ public abstract class WorkflowProcessor
"All sub-processors for Workflow Processor handling workflow id: ["
+ workflowInstance.getId() + "] are " + "not complete");
}
+
+
+ public void nextState(){
+ if(this.workflowInstance != null && this.workflowInstance.getState() != null){
+ WorkflowState currState = this.workflowInstance.getState();
+ WorkflowState nextState = null;
+ if(currState.getName().equals("Null")){
+ nextState = this.helper.getLifecycleForProcessor(this)
+ .createState("Loaded", "initial", "Workflow Processor: nextState: " +
+ "loading workflow instance: ["+this.workflowInstance.getId()+"]");
+ }
+ else if(currState.getName().equals("Loaded")){
+ nextState = this.helper.getLifecycleForProcessor(this)
+ .createState("Queued", "initial", "Workflow Processor: nextState: " +
+ "queueing instance: ["+this.workflowInstance.getId()+"]");
+ }
+ else if(currState.getName().equals("Queued")){
+ if(!this.passedPreConditions()){
+ nextState = this.helper.getLifecycleForProcessor(this)
+ .createState("PreConditionEval", "running", "Workflow Processor: nextState: " +
+ "running preconditiosn for workflow instance: ["+this.workflowInstance.getId()+"]");
+ }
+ }
+ else if(currState.getName().equals("Executing")){
+ nextState = this.helper.getLifecycleForProcessor(this)
+ .createState("Success", "done", "Workflow Processor: nextState: " +
+ "workflow instance: ["+this.workflowInstance.getId()+"] completed successfully");
+ }
+
+ if(nextState != null){
+ this.workflowInstance.setState(nextState);
+ }
+ else{
+ nextState = this.helper.getLifecycleForProcessor(this)
+ .createState("Unknown", "holding", "Unable to determine next state " +
+ "for workflow processor for instance: ["+this.workflowInstance.getId()+"]: " +
+ "current state: ["+currState.getName()+"]");
+ }
+ this.workflowInstance.setState(nextState);
+ }
+ else{
+ this.workflowInstance.setState(helper.getLifecycleForProcessor(this).createState("Unknown",
+ "holding", "The Workflow Processor for instance : ["+this.getWorkflowInstance().getId()+"] " +
+ "had a null state"));
+ }
+ }
/**
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorBuilder.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorBuilder.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorBuilder.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorBuilder.java Thu Sep 13 21:19:15 2012
@@ -18,11 +18,14 @@
package org.apache.oodt.cas.workflow.engine.processor;
//JDK imorts
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
import java.util.List;
//OODT imports
import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycleManager;
import org.apache.oodt.cas.workflow.structs.Priority;
+import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
//Google imports
import com.google.common.collect.Lists;
@@ -42,12 +45,14 @@ public class WorkflowProcessorBuilder {
private double priority;
private List<WorkflowProcessor> subProcessors;
private WorkflowLifecycleManager lifecycleManager;
+ private WorkflowInstance workflowInstance;
private WorkflowProcessorBuilder() {
subProcessors = Lists.newArrayList();
this.id = null;
this.priority = -1;
this.lifecycleManager = null;
+ this.workflowInstance = null;
}
public static WorkflowProcessorBuilder aWorkflowProcessor() {
@@ -70,20 +75,37 @@ public class WorkflowProcessorBuilder {
return this;
}
+ public WorkflowProcessorBuilder withInstance(WorkflowInstance workflowInstance) {
+ this.workflowInstance = workflowInstance;
+ return this;
+ }
+
public WorkflowProcessorBuilder with(WorkflowProcessorBuilder wpb,
Class<? extends WorkflowProcessor> clazz) throws InstantiationException,
- IllegalAccessException {
+ IllegalAccessException, IllegalArgumentException,
+ InvocationTargetException, SecurityException, NoSuchMethodException {
subProcessors.add(wpb.build(clazz));
return this;
}
public WorkflowProcessor build(Class<? extends WorkflowProcessor> clazz)
- throws InstantiationException, IllegalAccessException {
- WorkflowProcessor wp = clazz.newInstance();
- if(this.id != null) wp.getWorkflowInstance().setId(id);
- if(this.lifecycleManager != null) wp.setLifecycleManager(lifecycleManager);
- if(this.priority != -1) wp.getWorkflowInstance().setPriority(Priority.getPriority(priority));
- if(this.subProcessors != null) wp.setSubProcessors(subProcessors);
+ throws InstantiationException, IllegalAccessException,
+ IllegalArgumentException, InvocationTargetException, SecurityException,
+ NoSuchMethodException {
+ Constructor<? extends WorkflowProcessor> clazzConstructor = clazz
+ .getConstructor(new Class[] { WorkflowLifecycleManager.class,
+ WorkflowInstance.class });
+ WorkflowProcessor wp = clazzConstructor.newInstance(this.lifecycleManager,
+ this.workflowInstance);
+ if (this.id != null)
+ wp.getWorkflowInstance().setId(id);
+ if (this.priority != -1)
+ wp.getWorkflowInstance().setPriority(Priority.getPriority(priority));
+ if (this.subProcessors != null)
+ wp.setSubProcessors(subProcessors);
+ if (this.workflowInstance != null){
+ wp.setWorkflowInstance(workflowInstance);
+ }
return wp;
}
}
\ No newline at end of file
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorHelper.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorHelper.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorHelper.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorHelper.java Thu Sep 13 21:19:15 2012
@@ -48,6 +48,10 @@ import org.apache.commons.lang.StringUti
public class WorkflowProcessorHelper {
private WorkflowLifecycleManager lifecycle;
+
+ public WorkflowProcessorHelper(){
+ this(null);
+ }
public WorkflowProcessorHelper(WorkflowLifecycleManager lifecycle) {
this.lifecycle = lifecycle;
@@ -355,6 +359,15 @@ public class WorkflowProcessorHelper {
return "Unknown";
return host;
}
+
+ public WorkflowLifecycle getLifecycleForProcessor(WorkflowProcessor processor) {
+ if (processor.getWorkflowInstance() != null
+ && processor.getWorkflowInstance().getParentChildWorkflow() != null) {
+ return processor.getLifecycleManager().getLifecycleForWorkflow(
+ processor.getWorkflowInstance().getParentChildWorkflow());
+ } else
+ return processor.getLifecycleManager().getDefaultLifecycle();
+ }
private WorkflowLifecycle getLifecycle(ParentChildWorkflow model) {
return lifecycle.getLifecycleForWorkflow(model) != null ? lifecycle
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorQueue.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorQueue.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorQueue.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/processor/WorkflowProcessorQueue.java Thu Sep 13 21:19:15 2012
@@ -18,7 +18,9 @@
package org.apache.oodt.cas.workflow.engine.processor;
//JDK imports
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import java.util.Vector;
import java.util.logging.Level;
@@ -58,12 +60,15 @@ public class WorkflowProcessorQueue {
private WorkflowRepository modelRepo;
private WorkflowLifecycleManager lifecycle;
+
+ private Map<String, WorkflowProcessor> processorCache;
public WorkflowProcessorQueue(WorkflowInstanceRepository repo,
WorkflowLifecycleManager lifecycle, WorkflowRepository modelRepo) {
this.repo = repo;
this.lifecycle = lifecycle;
this.modelRepo = modelRepo;
+ this.processorCache = new HashMap<String, WorkflowProcessor>();
}
/**
@@ -74,7 +79,7 @@ public class WorkflowProcessorQueue {
public synchronized List<WorkflowProcessor> getProcessors() {
WorkflowInstancePage page = null;
try {
- page = repo.getPagedWorkflows(1, "Queued");
+ page = repo.getPagedWorkflows(1);
} catch (Exception e) {
e.printStackTrace();
LOG.log(Level.WARNING, "Unable to load workflow processors: Message: "
@@ -83,28 +88,12 @@ public class WorkflowProcessorQueue {
}
List<WorkflowProcessor> processors = new Vector<WorkflowProcessor>(
- page.getPageSize());
+ page.getPageWorkflows() != null ? page.getPageWorkflows().size():0);
for (WorkflowInstance inst : (List<WorkflowInstance>) (List<?>) page
.getPageWorkflows()) {
- if ((inst.getState() == null)
- || (inst.getState() != null && inst.getState().getCategory() == null)) {
- WorkflowLifecycle cycle = getLifecycle(inst.getWorkflow());
- WorkflowState state = cycle.getStateByName(inst.getStatus());
- state.setMessage("Queued by WorkflowProcessorQueue.");
- inst.setState(state);
- try {
- this.repo.updateWorkflowInstance(inst);
- } catch (Exception e) {
- e.printStackTrace();
- LOG.log(
- Level.WARNING,
- "Unable to update workflow instance: [" + inst.getId()
- + "] with status: [" + inst.getStatus() + "]: Message: "
- + e.getMessage());
- }
-
+ if(!inst.getState().getCategory().getName().equals("done")){
+ processors.add(fromWorkflowInstance(inst));
}
- processors.add(fromWorkflowInstance(inst));
}
return processors;
@@ -112,44 +101,82 @@ public class WorkflowProcessorQueue {
private WorkflowProcessor fromWorkflowInstance(WorkflowInstance inst) {
WorkflowProcessor processor = null;
- if (inst.getParentChildWorkflow().getTasks() != null
- && inst.getParentChildWorkflow().getTasks().size() > 1) {
- processor = new SequentialProcessor();
- processor.setWorkflowInstance(inst);
-
- for (WorkflowTask task : inst.getParentChildWorkflow().getTasks()) {
- WorkflowInstance instance = new WorkflowInstance();
- instance.setState(inst.getState());
- instance.setCurrentTaskId(task.getTaskId());
- ParentChildWorkflow workflow = new ParentChildWorkflow(new Graph());
- workflow.setId("task-workflow-"+UUID.randomUUID().toString());
- workflow.setName("Task Workflow-"+task.getTaskName());
- workflow.getTasks().add(task);
- workflow.getGraph().setTask(task);
- instance.setId(UUID.randomUUID().toString());
- instance.setParentChildWorkflow(workflow);
- if(modelRepo != null){
- try {
- modelRepo.addWorkflow(workflow);
- } catch (RepositoryException e) {
- e.printStackTrace();
- }
- }
-
- WorkflowProcessor subProcessor = fromWorkflowInstance(instance);
- processor.getSubProcessors().add(subProcessor);
- }
- processor.getWorkflowInstance().setState(inst.getState());
+ if(processorCache.containsKey(inst.getId())){
+ return processorCache.get(inst.getId());
}
else{
- processor = new TaskProcessor();
- processor.setWorkflowInstance(inst);
- processor.getWorkflowInstance().setState(inst.getState());
+ if (inst.getParentChildWorkflow().getTasks() != null
+ && inst.getParentChildWorkflow().getTasks().size() > 1) {
+ processor = new SequentialProcessor(lifecycle, inst);
+ WorkflowState seqProcessorState =
+ getLifecycle(inst.getParentChildWorkflow()).
+ createState("Loaded", "initial", "Sequential Workflow instance with id: ["
+ + inst.getId()+"] loaded by processor queue.");
+ inst.setState(seqProcessorState);
+ persist(inst);
+
+ for (WorkflowTask task : inst.getParentChildWorkflow().getTasks()) {
+ WorkflowInstance instance = new WorkflowInstance();
+ WorkflowState taskWorkflowState =
+ lifecycle.getDefaultLifecycle().createState("Null", "initial",
+ "Sub Task Workflow created by Workflow Processor Queue for workflow instance: " +
+ "["+inst.getId()+"]");
+ instance.setState(taskWorkflowState);
+ instance.setPriority(inst.getPriority());
+ instance.setCurrentTaskId(task.getTaskId());
+ ParentChildWorkflow workflow = new ParentChildWorkflow(new Graph());
+ String taskWorkflowId = UUID.randomUUID().toString();
+ workflow.setId("task-workflow-"+ taskWorkflowId);
+ workflow.setName("Task Workflow-"+task.getTaskName());
+ workflow.getTasks().add(task);
+ workflow.getGraph().setTask(task);
+ instance.setId(taskWorkflowId);
+ instance.setParentChildWorkflow(workflow);
+ this.addToModelRepo(workflow);
+ persist(inst);
+ WorkflowProcessor subProcessor = fromWorkflowInstance(instance);
+ processor.getSubProcessors().add(subProcessor);
+ }
+ }
+ else{
+ processor = new TaskProcessor(lifecycle, inst);
+ WorkflowState taskProcessorState =
+ getLifecycle(inst.getParentChildWorkflow()).
+ createState("Loaded", "initial", "Task Workflow instance with id: ["
+ + inst.getId()+"] loaded by processor queue.");
+ inst.setState(taskProcessorState);
+ persist(inst);
+ }
+
+ synchronized(processorCache){
+ processorCache.put(inst.getId(), processor);
+ }
+ return processor;
}
- processor.setLifecycleManager(lifecycle);
-
- return processor;
+ }
+
+ private void addToModelRepo(Workflow workflow){
+ if(modelRepo != null){
+ try {
+ modelRepo.addWorkflow(workflow);
+ } catch (RepositoryException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ private void persist(WorkflowInstance instance){
+ try {
+ this.repo.updateWorkflowInstance(instance);
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOG.log(
+ Level.WARNING,
+ "Unable to update workflow instance: [" + instance.getId()
+ + "] with status: [" + instance.getState().getName() + "]: Message: "
+ + e.getMessage());
+ }
}
private WorkflowLifecycle getLifecycle(Workflow workflow) {
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/runner/AsynchronousLocalEngineRunner.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/runner/AsynchronousLocalEngineRunner.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/runner/AsynchronousLocalEngineRunner.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/runner/AsynchronousLocalEngineRunner.java Thu Sep 13 21:19:15 2012
@@ -76,10 +76,10 @@ public class AsynchronousLocalEngineRunn
try {
inst.run(dynMetadata, workflowTask.getTaskConfig());
} catch (Exception e) {
+ e.printStackTrace();
LOG.log(Level.WARNING,
"Exception executing task: [" + workflowTask.getTaskName()
+ "]: Message: " + e.getMessage());
- e.printStackTrace();
}
}
Modified: oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/instrepo/LuceneWorkflowInstanceRepository.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/instrepo/LuceneWorkflowInstanceRepository.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/instrepo/LuceneWorkflowInstanceRepository.java (original)
+++ oodt/trunk/workflow/src/main/java/org/apache/oodt/cas/workflow/instrepo/LuceneWorkflowInstanceRepository.java Thu Sep 13 21:19:15 2012
@@ -507,7 +507,6 @@ public class LuceneWorkflowInstanceRepos
if(state.getCategory() != null && state.getCategory().getName() != null){
doc.add(new Field("workflow_inst_state_category",
state.getCategory().getName(), Field.Store.YES, Field.Index.UN_TOKENIZED));
- System.out.println("Indexing category: ["+state.getCategory().getName()+"]");
}
}
@@ -663,7 +662,6 @@ public class LuceneWorkflowInstanceRepos
if(doc.get("workflow_inst_state_category") != null){
WorkflowLifecycleStage category = new WorkflowLifecycleStage();
category.setName(doc.get("workflow_inst_state_category"));
- System.out.println("unserializing category: ["+category.getName()+"]");
state.setCategory(category);
}
Modified: oodt/trunk/workflow/src/main/resources/examples/wengine/wengine-lifecycle.xml
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/main/resources/examples/wengine/wengine-lifecycle.xml?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/main/resources/examples/wengine/wengine-lifecycle.xml (original)
+++ oodt/trunk/workflow/src/main/resources/examples/wengine/wengine-lifecycle.xml Thu Sep 13 21:19:15 2012
@@ -57,14 +57,6 @@ the License.
<description>Waiting for resource to execute</description>
</status>
</stage>
- <stage name="holding">
- <status name="Unknown">
- <description>State is Unknown</description>
- </status>
- <status name="Paused">
- <description>Has been manually paused</description>
- </status>
- </stage>
<stage name="transition">
<status name="PreConditionSuccess">
<description>All PreCondition Finished Successfully</description>
@@ -73,6 +65,14 @@ the License.
<description>Execution Completed Successfully</description>
</status>
</stage>
+ <stage name="holding">
+ <status name="Unknown">
+ <description>State is Unknown</description>
+ </status>
+ <status name="Paused">
+ <description>Has been manually paused</description>
+ </status>
+ </stage>
<stage name="running">
<status name="PreConditionEval">
<description>Executing PreConditions</description>
@@ -84,6 +84,17 @@ the License.
<description>Executing PostConditions</description>
</status>
</stage>
+ <stage name="results">
+ <status name="ResultsSuccess">
+ <description>Notification of success</description>
+ </status>
+ <status name="ResultsFailure">
+ <description>Notification of failure</description>
+ </status>
+ <status name="ResultsBail">
+ <description>Notification to bail workflow</description>
+ </status>
+ </stage>
<stage name="done">
<status name="Stopped">
<description>Force Killed</description>
@@ -97,19 +108,7 @@ the License.
<status name="Success">
<description>Successfully Completed</description>
</status>
- </stage>
- <stage name="results">
- <status name="ResultsSuccess">
- <description>Notification of success</description>
- </status>
- <status name="ResultsFailure">
- <description>Notification of failure</description>
- </status>
- <status name="ResultsBail">
- <description>Notification to bail workflow</description>
- </status>
- </stage>
-
+ </stage>
</default>
<!--
Lifecycles can also be specified at the
Modified: oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/QuerierAndRunnerUtils.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/QuerierAndRunnerUtils.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/QuerierAndRunnerUtils.java (original)
+++ oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/QuerierAndRunnerUtils.java Thu Sep 13 21:19:15 2012
@@ -20,6 +20,7 @@ package org.apache.oodt.cas.workflow.eng
//JDK imports
import java.io.File;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
import java.util.Collections;
import java.util.Date;
import java.util.List;
@@ -72,7 +73,7 @@ public class QuerierAndRunnerUtils {
public WorkflowProcessor getProcessor(double priority, String stateName,
String categoryName) throws InstantiationException,
- IllegalAccessException, IOException {
+ IllegalAccessException, IOException, IllegalArgumentException, SecurityException, InvocationTargetException, NoSuchMethodException {
WorkflowLifecycleManager lifecycleManager = new WorkflowLifecycleManager(
"./src/main/resources/examples/wengine/wengine-lifecycle.xml");
WorkflowInstance inst = new WorkflowInstance();
@@ -80,34 +81,21 @@ public class QuerierAndRunnerUtils {
sd.setTime(sd.getTime() + (this.dateGen * 5000));
this.dateGen++;
inst.setStartDate(sd);
- inst.setId("winst-" + priority);
+ inst.setId("task-winst-" + priority);
ParentChildWorkflow workflow = new ParentChildWorkflow(new Graph());
- workflow.setTasks(Collections.EMPTY_LIST);
- inst.setWorkflow(workflow);
+ workflow.getTasks().add(getTask(getTmpPath()));
+ inst.setParentChildWorkflow(workflow);
inst.setPriority(Priority.getPriority(priority));
+ inst.setCurrentTaskId(workflow.getTasks().get(0).getTaskId());
+ inst.setParentChildWorkflow(workflow);
WorkflowProcessorBuilder builder = WorkflowProcessorBuilder
.aWorkflowProcessor().withLifecycleManager(lifecycleManager)
- .withPriority(priority);
- SequentialProcessor processor = (SequentialProcessor) builder
- .build(SequentialProcessor.class);
- processor.setWorkflowInstance(inst);
- processor.getWorkflowInstance().setState(lifecycleManager.getDefaultLifecycle().createState(
- stateName, categoryName, ""));
- List<WorkflowProcessor> runnables = new Vector<WorkflowProcessor>();
+ .withPriority(priority).withInstance(inst);
TaskProcessor taskProcessor = (TaskProcessor) builder
.build(TaskProcessor.class);
- ParentChildWorkflow taskWorkflow = new ParentChildWorkflow(new Graph());
- taskWorkflow.getTasks().add(getTask(getTmpPath()));
- WorkflowInstance taskWorkflowInst = new WorkflowInstance();
- taskWorkflowInst.setPriority(Priority.getPriority(priority));
- taskWorkflowInst.setWorkflow(taskWorkflow);
- taskWorkflowInst.setCurrentTaskId(taskWorkflow.getTasks().get(0).getTaskId());
- taskProcessor.setWorkflowInstance(taskWorkflowInst);
taskProcessor.getWorkflowInstance().setState(lifecycleManager.getDefaultLifecycle().createState(
- "Queued", "waiting", ""));
- runnables.add(taskProcessor);
- processor.setSubProcessors(runnables);
- return processor;
+ stateName, categoryName, ""));
+ return taskProcessor;
}
private File getTmpPath() throws IOException{
Modified: oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/SimpleTester.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/SimpleTester.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/SimpleTester.java (original)
+++ oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/SimpleTester.java Thu Sep 13 21:19:15 2012
@@ -18,6 +18,7 @@
package org.apache.oodt.cas.workflow.engine;
//JDK imports
+import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintWriter;
@@ -58,6 +59,12 @@ public class SimpleTester implements Wor
try {
String jobFilePath = config.getProperty("TestDirPath") + "task-"
+ metadata.getMetadata("StartDateTime") + ".job";
+ int n=0;
+ while(new File(jobFilePath).exists()){
+ jobFilePath = config.getProperty("TestDirPath") + "task-"
+ + n + metadata.getMetadata("StartDateTime") + ".job";
+ n++;
+ }
LOG.log(Level.INFO, "Creating job file: [" + jobFilePath + "]");
pw = new PrintWriter(new FileOutputStream(jobFilePath));
pw.println("StartDateTime=" + metadata.getMetadata("StartDateTime"));
Modified: oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/TestAsynchronousLocalEngineRunner.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/TestAsynchronousLocalEngineRunner.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/TestAsynchronousLocalEngineRunner.java (original)
+++ oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/engine/TestAsynchronousLocalEngineRunner.java Thu Sep 13 21:19:15 2012
@@ -20,6 +20,7 @@ package org.apache.oodt.cas.workflow.eng
//JDK imports
import java.io.BufferedReader;
import java.io.File;
+import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
@@ -79,6 +80,7 @@ public class TestAsynchronousLocalEngine
try {
String line = FileUtils.readFileToString(f);
String[] toks = line.split(",");
+ assertEquals("Toks not equal to 2: toks=["+Arrays.asList(toks)+"]", 2, toks.length);
Date dateTime = DateConvert.isoParse(toks[1]);
Seconds seconds = Seconds.secondsBetween(new DateTime(dateTime),
new DateTime());
Modified: oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestFILOPrioritySorter.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestFILOPrioritySorter.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestFILOPrioritySorter.java (original)
+++ oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestFILOPrioritySorter.java Thu Sep 13 21:19:15 2012
@@ -80,8 +80,7 @@ public class TestFILOPrioritySorter exte
workflow.setTasks(Collections.EMPTY_LIST);
inst.setWorkflow(workflow);
inst.setPriority(Priority.getPriority(priority));
- SequentialProcessor processor = new SequentialProcessor(lifecycleManager);
- processor.setWorkflowInstance(inst);
+ SequentialProcessor processor = new SequentialProcessor(lifecycleManager, inst);
return processor;
}
Modified: oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestFIFOPrioritySorter.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestFIFOPrioritySorter.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestFIFOPrioritySorter.java (original)
+++ oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestFIFOPrioritySorter.java Thu Sep 13 21:19:15 2012
@@ -75,8 +75,7 @@ public class TestHighestFIFOPrioritySort
workflow.setTasks(Collections.EMPTY_LIST);
inst.setWorkflow(workflow);
inst.setPriority(Priority.getPriority(priority));
- SequentialProcessor processor = new SequentialProcessor(lifecycleManager);
- processor.setWorkflowInstance(inst);
+ SequentialProcessor processor = new SequentialProcessor(lifecycleManager, inst);
return processor;
}
Modified: oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestPrioritySorter.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestPrioritySorter.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestPrioritySorter.java (original)
+++ oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/structs/TestHighestPrioritySorter.java Thu Sep 13 21:19:15 2012
@@ -78,8 +78,7 @@ public class TestHighestPrioritySorter e
workflow.setTasks(Collections.EMPTY_LIST);
inst.setWorkflow(workflow);
inst.setPriority(Priority.getPriority(priority));
- SequentialProcessor processor = new SequentialProcessor(lifecycleManager);
- processor.setWorkflowInstance(inst);
+ SequentialProcessor processor = new SequentialProcessor(lifecycleManager, inst);
return processor;
}
Modified: oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/system/TestXmlRpcWorkflowManager.java
URL: http://svn.apache.org/viewvc/oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/system/TestXmlRpcWorkflowManager.java?rev=1384539&r1=1384538&r2=1384539&view=diff
==============================================================================
--- oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/system/TestXmlRpcWorkflowManager.java (original)
+++ oodt/trunk/workflow/src/test/org/apache/oodt/cas/workflow/system/TestXmlRpcWorkflowManager.java Thu Sep 13 21:19:15 2012
@@ -132,10 +132,13 @@ public class TestXmlRpcWorkflowManager e
System
.setProperty("org.apache.oodt.cas.workflow.instanceRep.lucene.idxPath",
luceneCatLoc);
+
try {
System.setProperty("org.apache.oodt.cas.workflow.repo.dirs", "file://"
+ new File("./src/main/resources/examples").getCanonicalPath());
+ System.setProperty("org.apache.oodt.cas.workflow.lifecycle.filePath",
+ new File("./src/main/resources/examples/workflow-lifecycle.xml").getCanonicalPath());
} catch (Exception e) {
fail(e.getMessage());
}
@@ -143,6 +146,7 @@ public class TestXmlRpcWorkflowManager e
try {
wmgr = new XmlRpcWorkflowManager(WM_PORT);
} catch (Exception e) {
+ e.printStackTrace();
fail(e.getMessage());
}