diff options
Diffstat (limited to 'id/server/idserverlib/src/main/java/at/gv/egovernment/moa/id/process/ProcessEngineImpl.java')
| -rw-r--r-- | id/server/idserverlib/src/main/java/at/gv/egovernment/moa/id/process/ProcessEngineImpl.java | 191 |
1 files changed, 125 insertions, 66 deletions
diff --git a/id/server/idserverlib/src/main/java/at/gv/egovernment/moa/id/process/ProcessEngineImpl.java b/id/server/idserverlib/src/main/java/at/gv/egovernment/moa/id/process/ProcessEngineImpl.java index 3ba8fb9ed..ea5a2684e 100644 --- a/id/server/idserverlib/src/main/java/at/gv/egovernment/moa/id/process/ProcessEngineImpl.java +++ b/id/server/idserverlib/src/main/java/at/gv/egovernment/moa/id/process/ProcessEngineImpl.java @@ -2,11 +2,8 @@ package at.gv.egovernment.moa.id.process; import java.io.InputStream; import java.io.Serializable; -import java.util.Date; -import java.util.Iterator; +import java.util.HashMap; import java.util.Map; -import java.util.Map.Entry; -import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import org.apache.commons.collections4.CollectionUtils; @@ -16,10 +13,14 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; +import at.gv.egovernment.moa.id.commons.db.ex.MOADatabaseException; import at.gv.egovernment.moa.id.process.api.ExecutionContext; import at.gv.egovernment.moa.id.process.api.ExpressionEvaluationContext; import at.gv.egovernment.moa.id.process.api.ExpressionEvaluator; import at.gv.egovernment.moa.id.process.api.Task; +import at.gv.egovernment.moa.id.process.dao.ProcessInstanceStore; +import at.gv.egovernment.moa.id.process.dao.ProcessInstanceStoreDAO; +import at.gv.egovernment.moa.id.process.dao.ProcessInstanceStoreDAOImpl; import at.gv.egovernment.moa.id.process.model.EndEvent; import at.gv.egovernment.moa.id.process.model.ProcessDefinition; import at.gv.egovernment.moa.id.process.model.ProcessNode; @@ -33,21 +34,20 @@ import at.gv.egovernment.moa.id.process.model.Transition; * */ public class ProcessEngineImpl implements ProcessEngine { - + private Logger log = LoggerFactory.getLogger(getClass()); - + private ProcessDefinitionParser pdp = new ProcessDefinitionParser(); + ProcessInstanceStoreDAO piStoreDao = ProcessInstanceStoreDAOImpl.getInstance(); + private Map<String, ProcessDefinition> processDefinitions = new ConcurrentHashMap<String, ProcessDefinition>(); - private Map<String, ProcessInstance> processInstances = new ConcurrentHashMap<String, ProcessInstance>(); - + private final static String MDC_CTX_PI_NAME = "processInstanceId"; private final static String MDC_CTX_TASK_NAME = "taskId"; - - private static final long DEFAULT_PROCESS_INSTANCE_MAX_AGE_SECONDS = 3600; - private long processInstanceIdleTimeSeconds = DEFAULT_PROCESS_INSTANCE_MAX_AGE_SECONDS; + private ExpressionEvaluator transitionConditionExpressionEvaluator; - + @Override public void registerProcessDefinition(ProcessDefinition processDefinition) { log.info("Registering process definition '{}'.", processDefinition.getId()); @@ -61,7 +61,7 @@ public class ProcessEngineImpl implements ProcessEngine { /** * Sets the process definitions. - * + * * @param processDefinitions * The process definitions. * @throws IllegalArgumentException @@ -76,19 +76,6 @@ public class ProcessEngineImpl implements ProcessEngine { registerProcessDefinition(pd); } } - - /** - * Defines the time frame in seconds an idle process instance will be managed by the process engine. A process - * instance with an idle time larger than the given time will be removed. - * <p/> - * Note that {@link #cleanup()} needs to be called in order to remove expired process instances. - * - * @param processInstanceMaxIdleTimeSeconds - * The maximum idle time in seconds. - */ - public void setProcessInstanceMaxIdleTimeSeconds(long processInstanceMaxIdleTimeSeconds) { - this.processInstanceIdleTimeSeconds = processInstanceMaxIdleTimeSeconds; - } /** * Sets an expression evaluator that should be used to process transition condition expressions. @@ -98,10 +85,10 @@ public class ProcessEngineImpl implements ProcessEngine { ExpressionEvaluator transitionConditionExpressionEvaluator) { this.transitionConditionExpressionEvaluator = transitionConditionExpressionEvaluator; } - + @Override - public ProcessInstance createProcessInstance(String processDefinitionId, ExecutionContext executionContext) throws ProcessExecutionException { + public String createProcessInstance(String processDefinitionId, ExecutionContext executionContext) throws ProcessExecutionException { // look for respective process definition ProcessDefinition pd = processDefinitions.get(processDefinitionId); if (pd == null) { @@ -110,19 +97,32 @@ public class ProcessEngineImpl implements ProcessEngine { // create and keep process instance ProcessInstance pi = new ProcessInstance(pd, executionContext); log.info("Creating process instance from process definition '{}': {}", processDefinitionId, pi.getId()); - processInstances.put(pi.getId(), pi); - return pi; + + try { + saveProcessInstance(pi); + } catch (MOADatabaseException e) { + throw new ProcessExecutionException("Unable to persist process instance.", e); + } + + return pi.getId(); } @Override - public ProcessInstance createProcessInstance(String processDefinitionId) throws ProcessExecutionException { + public String createProcessInstance(String processDefinitionId) throws ProcessExecutionException { return createProcessInstance(processDefinitionId, null); } @Override - public void start(ProcessInstance pi) throws ProcessExecutionException { - MDC.put(MDC_CTX_PI_NAME, pi.getId()); + public void start(String processInstanceId) throws ProcessExecutionException { + + boolean mdcEntryAdded = false; + try { + ProcessInstance pi = loadProcessInstance(processInstanceId); + + MDC.put(MDC_CTX_PI_NAME, pi.getId()); + mdcEntryAdded = true; + if (!ProcessInstanceState.NOT_STARTED.equals(pi.getState())) { throw new ProcessExecutionException("Process instance '" + pi.getId() + "' has already been started (current state is " + pi.getState() + ")."); } @@ -130,45 +130,49 @@ public class ProcessEngineImpl implements ProcessEngine { // execute process pi.setState(ProcessInstanceState.STARTED); execute(pi); + + saveProcessInstance(pi); + } catch (MOADatabaseException e) { + throw new ProcessExecutionException("Unable to load/save process instance.", e); + } finally { - MDC.remove(MDC_CTX_PI_NAME); + if (mdcEntryAdded) + MDC.remove(MDC_CTX_PI_NAME); } } @Override - public void signal(ProcessInstance pi) throws ProcessExecutionException { - MDC.put(MDC_CTX_PI_NAME, pi.getId()); + public void signal(String processInstanceId) throws ProcessExecutionException { + + boolean mdcEntryAdded = false; + try { + ProcessInstance pi = loadProcessInstance(processInstanceId); + + MDC.put(MDC_CTX_PI_NAME, pi.getId()); + mdcEntryAdded = true; + if (!ProcessInstanceState.SUSPENDED.equals(pi.getState())) { throw new ProcessExecutionException("Process instance '" + pi.getId() + "' has not been suspended (current state is " + pi.getState() + ")."); } + log.info("Waking up process instance '{}'.", pi.getId()); pi.setState(ProcessInstanceState.STARTED); execute(pi); + + saveProcessInstance(pi); + } catch (MOADatabaseException e) { + throw new ProcessExecutionException("Unable to load/save process instance.", e); + } finally { - MDC.remove(MDC_CTX_PI_NAME); + if (mdcEntryAdded) + MDC.remove(MDC_CTX_PI_NAME); } } - @Override - public synchronized void cleanup() { - log.trace("Cleanup job started."); - Iterator<Entry<String, ProcessInstance>> it = processInstances.entrySet().iterator(); - while (it.hasNext()) { - Entry<String, ProcessInstance> entry = it.next(); - ProcessInstance pi = entry.getValue(); - log.trace("Checking process instance {}.", pi); - long ageMillis = new Date().getTime() - pi.getLru().getTime(); - if (ageMillis > processInstanceIdleTimeSeconds * 1000) { - log.info("Removing process instance '{}'.", pi.getId()); - processInstances.remove(entry.getKey()); - } - } - log.trace("Cleanup job completed."); - } /** - * Instantates a task implementation given by a {@link TaskInfo}. + * Instantiates a task implementation given by a {@link TaskInfo}. * @param ti The task info. * @return A Task implementation or {@code null} if the task info does not reference any task implementing classes. * @throws ProcessExecutionException Thrown in case of error (when the referenced class does not implement {@link Task} for instance). @@ -194,7 +198,7 @@ public class ProcessEngineImpl implements ProcessEngine { throw new ProcessExecutionException("Unable to instantiate class '" + clazz + "' associated with task '" + ti.getId() + "' .", e); } } - + return task; } @@ -239,7 +243,12 @@ public class ProcessEngineImpl implements ProcessEngine { } else if (processNode instanceof EndEvent) { log.info("Finishing process instance '{}'.", pi.getId()); - processInstances.remove(pi.getId()); + + try { + piStoreDao.remove(pi.getId()); + } catch (MOADatabaseException e) { + throw new ProcessExecutionException("Unable to remove process instance.", e); + } pi.setState(ProcessInstanceState.ENDED); log.debug("Final process context: {}", pi.getExecutionContext().keySet()); return; @@ -282,23 +291,73 @@ public class ProcessEngineImpl implements ProcessEngine { @Override public ProcessInstance getProcessInstance(String processInstanceId) { - ProcessInstance processInstance = processInstances.get(processInstanceId); + + ProcessInstance processInstance; + try { + processInstance = loadProcessInstance(processInstanceId); + + } catch (MOADatabaseException e) { + throw new RuntimeException("The process instance '" + processInstanceId + "' could not be retrieved.", e); + } + if (processInstance == null) { throw new IllegalArgumentException("The process instance '" + processInstanceId + "' does not/no longer exist."); } + return processInstance; } - @Override - public ProcessInstance findProcessInstanceWith(String key, Serializable value) { - Iterator<ProcessInstance> it = processInstances.values().iterator(); - while (it.hasNext()) { - ProcessInstance pi = it.next(); - if (Objects.equals(pi.getExecutionContext().get(key), value)) { - return pi; - } + /** + * Persists a {@link ProcessInstance} to the database. + * @param processInstance The object to persist. + * @throws MOADatabaseException Thrown if an error occurs while accessing the database. + */ + private void saveProcessInstance(ProcessInstance processInstance) throws MOADatabaseException { + ProcessInstanceStore store = new ProcessInstanceStore(); + + ExecutionContext ctx = processInstance.getExecutionContext(); + + Map<String, Serializable> ctxData = new HashMap<String, Serializable>(); + for (String key : ctx.keySet()) { + ctxData.put(key, ctx.get(key)); + } + store.setExecutionContextData(ctxData); + + store.setNextTaskId(processInstance.getNextId()); + store.setProcessDefinitionId(processInstance.getProcessDefinition().getId()); + + store.setProcessInstanceId(processInstance.getId()); + store.setProcessState(processInstance.getState()); + + piStoreDao.save(store); + } + + /** + * Load a {@link ProcessInstance} with a certain id from the database. + * @param processInstanceId The process instance id + * @return The process instance corresponding to the id or {@code null} if no such object is found. + * @throws MOADatabaseException Thrown if an error occurs while accessing the database. + */ + private ProcessInstance loadProcessInstance(String processInstanceId) throws MOADatabaseException { + + ProcessInstanceStore piStore = piStoreDao.load(processInstanceId); + + if (piStore == null) { + return null; } - return null; + + ExecutionContext executionContext = new ExecutionContextImpl(piStore.getProcessInstanceId()); + + Map<String, Serializable> executionContextData = piStore.getExecutionContextData(); + for (String key : executionContextData.keySet()) { + executionContext.put(key, executionContextData.get(key)); + } + + ProcessInstance pi = new ProcessInstance(processDefinitions.get(piStore.getProcessDefinitionId()), executionContext); + pi.setNextId(piStore.getNextTaskId()); + pi.setState(piStore.getProcessState()); + + return pi; } } |