add stop condition to process-flow engine that can be set dynamically by tasks

1 files changed, 71 insertions, 46 deletions

diff --git a/eaaf_core/src/main/java/at/gv/egiz/eaaf/core/impl/idp/process/ProcessEngineImpl.java b/eaaf_core/src/main/java/at/gv/egiz/eaaf/core/impl/idp/process/ProcessEngineImpl.java
index b6b42850..53f50e1f 100644
--- a/eaaf_core/src/main/java/at/gv/egiz/eaaf/core/impl/idp/process/ProcessEngineImpl.java
+++ b/eaaf_core/src/main/java/at/gv/egiz/eaaf/core/impl/idp/process/ProcessEngineImpl.java
@@ -64,14 +64,14 @@ import at.gv.egiz.eaaf.core.impl.idp.process.model.Transition;
  */
 public class ProcessEngineImpl implements ProcessEngine {
  
-	private Logger log = LoggerFactory.getLogger(getClass());
+	private final Logger log = LoggerFactory.getLogger(getClass());
 
 	@Autowired ProcessInstanceStoreDAO piStoreDao;
 	@Autowired ApplicationContext context;
 	
-	private ProcessDefinitionParser pdp = new ProcessDefinitionParser();
+	private final ProcessDefinitionParser pdp = new ProcessDefinitionParser();
 
-	private Map<String, ProcessDefinition> processDefinitions = new ConcurrentHashMap<String, ProcessDefinition>();
+	private final Map<String, ProcessDefinition> processDefinitions = new ConcurrentHashMap<String, ProcessDefinition>();
 
 	private final static String MDC_CTX_PI_NAME = "processInstanceId";
 	private final static String MDC_CTX_TASK_NAME = "taskId";
@@ -86,7 +86,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 
 	@Override
 	public String registerProcessDefinition(InputStream processDefinitionInputStream) throws ProcessDefinitionParserException{
-		ProcessDefinition pd = pdp.parse(processDefinitionInputStream);
+		final ProcessDefinition pd = pdp.parse(processDefinitionInputStream);
 		
 		postValidationOfProcessDefintion(pd);
 		
@@ -104,7 +104,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 	 */
 	public void setProcessDefinitions(Iterable<ProcessDefinition> processDefinitions) {
 		this.processDefinitions.clear();
-		for (ProcessDefinition pd : processDefinitions) {
+		for (final ProcessDefinition pd : processDefinitions) {
 			if (this.processDefinitions.containsKey(pd.getId())) {
 				throw new IllegalArgumentException("Duplicate process definition identifier '" + pd.getId() + "'.");
 			}
@@ -125,18 +125,18 @@ public class ProcessEngineImpl implements ProcessEngine {
 	@Override
 	public String createProcessInstance(String processDefinitionId, ExecutionContext executionContext) throws ProcessExecutionException {
 		// look for respective process definition
-		ProcessDefinition pd = processDefinitions.get(processDefinitionId);
+		final ProcessDefinition pd = processDefinitions.get(processDefinitionId);
 		if (pd == null) {
 			throw new ProcessExecutionException("Unable to find process definition for process '" + processDefinitionId + "'.");
 		}
 		// create and keep process instance
-		ProcessInstance pi = new ProcessInstance(pd, executionContext);
+		final ProcessInstance pi = new ProcessInstance(pd, executionContext);
 		log.info("Creating process instance from process definition '{}': {}", processDefinitionId, pi.getId());
 
 		try {
 			saveOrUpdateProcessInstance(pi);
 			
-		} catch (EAAFException e) {
+		} catch (final EAAFException e) {
 			throw new ProcessExecutionException("Unable to persist process instance.", e);
 		}
 
@@ -158,7 +158,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 					+ " includes NO 'ProcessInstanceId'");
 			}
 			
-			ProcessInstance pi = loadProcessInstance(pendingReq.getProcessInstanceId());
+			final ProcessInstance pi = loadProcessInstance(pendingReq.getProcessInstanceId());
 
 			if (pi == null ) {
 				throw new ProcessExecutionException("Process instance '" + pendingReq.getProcessInstanceId() + "' does not exist.");
@@ -179,7 +179,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 			if (!ProcessInstanceState.ENDED.equals(pi.getState()))
 				saveOrUpdateProcessInstance(pi);
 				
-		} catch (EAAFException e) {
+		} catch (final EAAFException e) {
 			throw new ProcessExecutionException("Unable to load/save process instance.", e);
 
 		} finally {
@@ -198,7 +198,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 					+ " includes NO 'ProcessInstanceId'");
 			}
 			
-			ProcessInstance pi = loadProcessInstance(pendingReq.getProcessInstanceId());
+			final ProcessInstance pi = loadProcessInstance(pendingReq.getProcessInstanceId());
 
 			if (pi == null ) {
 				throw new ProcessExecutionException("Process instance '" + pendingReq.getProcessInstanceId() + "' does not exist.");
@@ -223,7 +223,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 			if (!ProcessInstanceState.ENDED.equals(pi.getState()))
 				saveOrUpdateProcessInstance(pi);
 						
-		} catch (EAAFException e) {
+		} catch (final EAAFException e) {
 			throw new ProcessExecutionException("Unable to load/save process instance.", e);
 
 		} finally {
@@ -239,7 +239,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 	 * @throws ProcessExecutionException Thrown in case of error (when the referenced class does not implement {@link Task} for instance).
 	 */
 	private Task createTaskInstance(TaskInfo ti) throws ProcessExecutionException {
-		String clazz = StringUtils.trimToNull(ti.getTaskImplementingClass());
+		final String clazz = StringUtils.trimToNull(ti.getTaskImplementingClass());
 		Task task = null;
 		
 		if (clazz != null) {
@@ -248,7 +248,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 			try {
 				instanceClass = context.getBean(clazz);
 				
-			} catch (Exception e) {
+			} catch (final Exception e) {
 				throw new ProcessExecutionException("Unable to get class '" + clazz + "' associated with task '" + ti.getId() + "' .", e);
 				
 			}
@@ -259,7 +259,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 			try {
 				task = (Task) instanceClass;
 				
-			} catch (Exception e) {
+			} catch (final Exception e) {
 				throw new ProcessExecutionException("Unable to instantiate class '" + clazz + "' associated with task '" + ti.getId() + "' .", e);
 			}
 		}
@@ -277,19 +277,19 @@ public class ProcessEngineImpl implements ProcessEngine {
 		if (ProcessInstanceState.ENDED.equals(pi.getState())) {
 			throw new ProcessExecutionException("Process for instance '" + pi.getId() + "' has already been ended.");
 		}
-		ProcessDefinition pd = pi.getProcessDefinition();
-		ProcessNode processNode = pd.getProcessNode(pi.getNextId());
+		final ProcessDefinition pd = pi.getProcessDefinition();
+		final ProcessNode processNode = pd.getProcessNode(pi.getNextId());
 		log.debug("Processing node '{}'.", processNode.getId());
 		
 		// distinguish process node types StartEvent, TaskInfo and EndEvent
 		
 		if (processNode instanceof TaskInfo) {
 			// TaskInfo types need to be executed
-			TaskInfo ti = (TaskInfo) processNode;
+			final TaskInfo ti = (TaskInfo) processNode;
 			MDC.put(MDC_CTX_TASK_NAME, ti.getId());
 			try {
 				log.debug("Processing task '{}'.", ti.getId());
-				Task task = createTaskInstance(ti);
+				final Task task = createTaskInstance(ti);
 				if (task != null) {
 					try {
 						log.debug("Executing task implementation for task '{}'.", ti.getId());
@@ -297,35 +297,39 @@ public class ProcessEngineImpl implements ProcessEngine {
 						pendingReq = task.execute(pendingReq, pi.getExecutionContext());
 						log.debug("Returned from execution of task '{}'.", ti.getId());
 						log.trace("Execution context after task execution: {}", pi.getExecutionContext().keySet());
-					} catch (Throwable t) {
+												
+					} catch (final Throwable t) {
 						throw new ProcessExecutionException("Error executing task '" + ti.getId() + "'.", t);
+						
+					}
+					
+					//check if process was cancelled dynamically by task
+					if (pi.getExecutionContext().isProcessCancelled()) {
+						log.debug("Processing task '{}' was cancelled by Task: '{}'.", pi.getId(), ti.getId());
+						processFinishEvent(pi);
+						return;
+						
 					}
+					
 				} else {
 					log.debug("No task implementing class set.");
+					
 				}
 			} finally {
 				MDC.remove(MDC_CTX_TASK_NAME);
+				
 			}
 			
 		} else if (processNode instanceof EndEvent) {
-			log.info("Finishing process instance '{}'.", pi.getId());
-
-			try {
-				piStoreDao.remove(pi.getId());
-				
-			} catch (EAAFException e) {
-				throw new ProcessExecutionException("Unable to remove process instance.", e);
-				
-			}
-			pi.setState(ProcessInstanceState.ENDED);
-			log.debug("Final process context: {}", pi.getExecutionContext().keySet());
+			processFinishEvent(pi);
 			return;
+			
 		}
 		
 		final ExpressionEvaluationContext expressionContext = new ExpressionEvaluationContextImpl(pi);
 		
 		// traverse pointer
-		Transition t = IterableUtils.find(processNode.getOutgoingTransitions(), new Predicate<Transition>() {
+		final Transition t = IterableUtils.find(processNode.getOutgoingTransitions(), new Predicate<Transition>() {
 			@Override
 			public boolean evaluate(Transition transition) {
 				if (transitionConditionExpressionEvaluator != null && transition.getConditionExpression() != null) {
@@ -356,7 +360,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 			execute(pi, pendingReq);
 		}
 	}
-
+	
 	@Override
 	public ProcessInstance getProcessInstance(String processInstanceId) {
 
@@ -364,7 +368,7 @@ public class ProcessEngineImpl implements ProcessEngine {
 		try {
 			processInstance = loadProcessInstance(processInstanceId);
 
-		} catch (EAAFException e) {
+		} catch (final EAAFException e) {
 			throw new RuntimeException("The process instance '" + processInstanceId + "' could not be retrieved.", e);
 		}
 
@@ -381,12 +385,12 @@ public class ProcessEngineImpl implements ProcessEngine {
 	 * @throws MOADatabaseException Thrown if an error occurs while accessing the database.
 	 */
 	private void saveOrUpdateProcessInstance(ProcessInstance processInstance) throws EAAFException {
-		ProcessInstanceStore store = new ProcessInstanceStore();
+		final ProcessInstanceStore store = new ProcessInstanceStore();
 
-		ExecutionContext ctx = processInstance.getExecutionContext();
+		final ExecutionContext ctx = processInstance.getExecutionContext();
 
-		Map<String, Serializable> ctxData = new HashMap<String, Serializable>();
-		for (String key : ctx.keySet()) {
+		final Map<String, Serializable> ctxData = new HashMap<String, Serializable>();
+		for (final String key : ctx.keySet()) {
 			ctxData.put(key, ctx.get(key));
 		}
 		store.setExecutionContextData(ctxData);
@@ -408,20 +412,20 @@ public class ProcessEngineImpl implements ProcessEngine {
 	 */
 	private ProcessInstance loadProcessInstance(String processInstanceId) throws EAAFException {
 
-		ProcessInstanceStore piStore = piStoreDao.load(processInstanceId);
+		final ProcessInstanceStore piStore = piStoreDao.load(processInstanceId);
 
 		if (piStore == null) {
 			return null;
 		}
 
-		ExecutionContext executionContext = new ExecutionContextImpl(piStore.getProcessInstanceId());
+		final ExecutionContext executionContext = new ExecutionContextImpl(piStore.getProcessInstanceId());
 
-		Map<String, Serializable> executionContextData = piStore.getExecutionContextData();
-		for (String key : executionContextData.keySet()) {
+		final Map<String, Serializable> executionContextData = piStore.getExecutionContextData();
+		for (final String key : executionContextData.keySet()) {
 			executionContext.put(key, executionContextData.get(key));
 		}
 
-		ProcessInstance pi = new ProcessInstance(processDefinitions.get(piStore.getProcessDefinitionId()), executionContext);
+		final ProcessInstance pi = new ProcessInstance(processDefinitions.get(piStore.getProcessDefinitionId()), executionContext);
 		pi.setNextId(piStore.getNextTaskId());
 		pi.setState(piStore.getProcessState());
 
@@ -441,10 +445,31 @@ public class ProcessEngineImpl implements ProcessEngine {
 		try {
 			piStoreDao.remove(processInstanceId);
 			
-		} catch (EAAFException e) {
+		} catch (final EAAFException e) {
+			throw new ProcessExecutionException("Unable to remove process instance.", e);
+			
+		}
+		
+	}
+	
+	/**
+	 * Finish a process-flow and remove any process-flow related information
+	 * 
+	 * @param pi
+	 * @throws ProcessExecutionException
+	 */
+	private void processFinishEvent(ProcessInstance pi) throws ProcessExecutionException {
+		log.info("Finishing process instance '{}'.", pi.getId());
+
+		try {
+			piStoreDao.remove(pi.getId());
+			
+		} catch (final EAAFException e) {
 			throw new ProcessExecutionException("Unable to remove process instance.", e);
 			
 		}
+		pi.setState(ProcessInstanceState.ENDED);
+		log.debug("Final process context: {}", pi.getExecutionContext().keySet());
 		
 	}
 	
@@ -458,11 +483,11 @@ public class ProcessEngineImpl implements ProcessEngine {
 	 */
 	private void postValidationOfProcessDefintion(ProcessDefinition pd) throws ProcessDefinitionParserException{
 		try {
-			for(TaskInfo task : pd.getTaskInfos().values()) {
+			for(final TaskInfo task : pd.getTaskInfos().values()) {
 				createTaskInstance(task);		
 			}
 			
-		} catch (ProcessExecutionException e) {
+		} catch (final ProcessExecutionException e) {
 			log.error("Post-validation of process definition: {} find an error: {}", pd.getId(), e.getMessage());
 			throw new ProcessDefinitionParserException("Post-validation find an error in process definition:" + pd.getId(), e);