[jbpm-commits] JBoss JBPM SVN: r3090 - in projects/spec/trunk/modules: api/src/main/java/org/jbpm/api/service and 4 other directories.

Author: thomas.diesler(a)jboss.com Date: 2008-11-26 03:04:34 -0500 (Wed, 26 Nov 2008) New Revision: 3090 Added: projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ThreadingService.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ThreadingServiceImpl.java Removed: projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ExecutionService.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ExecutionServiceImpl.java Modified: projects/spec/trunk/modules/api/.classpath projects/spec/trunk/modules/api/src/main/java/org/jbpm/api/service/ProcessService.java projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/incubator/startevent/StartEventSignalTest.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/ProcessImpl.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/StartEventImpl.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessDefinitionServiceImpl.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessServiceImpl.java Log: Add getProcess by procDef Modified: projects/spec/trunk/modules/api/.classpath =================================================================== --- projects/spec/trunk/modules/api/.classpath 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/api/.classpath 2008-11-26 08:04:34 UTC (rev 3090) @@ -1,7 +1,7 @@ <?xml version="1.0" encoding="UTF-8"?> <classpath> - <classpathentry kind="src" output="target/classes" path="src/main/java"/> - <classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/J2SE-1.5"/> + <classpathentry kind="src" path="src/main/java"/> <classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/> + <classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/> <classpathentry kind="output" path="target/classes"/> </classpath> Modified: projects/spec/trunk/modules/api/src/main/java/org/jbpm/api/service/ProcessService.java =================================================================== --- projects/spec/trunk/modules/api/src/main/java/org/jbpm/api/service/ProcessService.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/api/src/main/java/org/jbpm/api/service/ProcessService.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -75,11 +75,11 @@ public abstract Set<ObjectName> getProcesses(); /** - * Find the set of Processes for a given name + * Get the set of Processes for a given ProcessDefinition * - * @param name The process name + * @param procDefID The process definition id * @param status The optional process status * @return An empty set if the process cannot be found */ - public abstract Set<ObjectName> getProcesses(String name, ProcessStatus status); + public abstract Set<ObjectName> getProcesses(ObjectName procDefID, ProcessStatus status); } \ No newline at end of file Deleted: projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ExecutionService.java =================================================================== --- projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ExecutionService.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ExecutionService.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -1,79 +0,0 @@ -/* - * JBoss, Home of Professional Open Source - * Copyright 2005, JBoss Inc., and individual contributors as indicated - * by the @authors tag. See the copyright.txt in the distribution for a - * full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.jbpm.incubator.service; - -// $Id$ - -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - -import org.jbpm.api.model.Process; -import org.jbpm.api.model.StartEvent; -import org.jbpm.api.model.Process.ProcessStatus; -import org.jbpm.api.runtime.Attachments; -import org.jbpm.api.service.AbstractService; - -/** - * The ExecutionService executes processes - * - * @author thomas.diesler(a)jboss.com - * @since 18-Jun-2008 - */ -public abstract class ExecutionService extends AbstractService -{ - private ExecutorService procExecutor = Executors.newCachedThreadPool(); - - // Hide public constructor - protected ExecutionService() - { - } - - /** - * Get the process executor service - */ - public ExecutorService getProcessExecutor() - { - return procExecutor; - } - - /** - * Start the Process from a given start event - * - * @param start The StartEvent that triggers the process - * @param att The Attachments in the ExecutionContext - */ - public abstract void startProcess(Process proc, StartEvent start, Attachments att); - - /** - * All Tokens that are generated at the Start Event for that Process must eventually arrive at an End Event. The - * Process will be in a running state until all Tokens are consumed. <p/> This method until the process ends without - * timeout. - */ - public abstract ProcessStatus waitForEnd(Process proc); - - /** - * All Tokens that are generated at the Start Event for that Process must eventually arrive at an End Event. The - * Process will be in a running state until all Tokens are consumed. <p/> This method until the process ends with a - * given timeout. - */ - public abstract ProcessStatus waitForEnd(Process proc, long timeout); -} Copied: projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ThreadingService.java (from rev 3087, projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ExecutionService.java) =================================================================== --- projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ThreadingService.java (rev 0) +++ projects/spec/trunk/modules/api/src/main/java/org/jbpm/incubator/service/ThreadingService.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -0,0 +1,79 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2005, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jbpm.incubator.service; + +// $Id$ + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import org.jbpm.api.model.Process; +import org.jbpm.api.model.StartEvent; +import org.jbpm.api.model.Process.ProcessStatus; +import org.jbpm.api.runtime.Attachments; +import org.jbpm.api.service.AbstractService; + +/** + * The ThreadingService executes processes + * + * @author thomas.diesler(a)jboss.com + * @since 18-Jun-2008 + */ +public abstract class ThreadingService extends AbstractService +{ + private ExecutorService procExecutor = Executors.newCachedThreadPool(); + + // Hide public constructor + protected ThreadingService() + { + } + + /** + * Get the process executor service + */ + public ExecutorService getProcessExecutor() + { + return procExecutor; + } + + /** + * Start the Process from a given start event + * + * @param start The StartEvent that triggers the process + * @param att The Attachments in the ExecutionContext + */ + public abstract void startProcess(Process proc, StartEvent start, Attachments att); + + /** + * All Tokens that are generated at the Start Event for that Process must eventually arrive at an End Event. The + * Process will be in a running state until all Tokens are consumed. <p/> This method until the process ends without + * timeout. + */ + public abstract ProcessStatus waitForEnd(Process proc); + + /** + * All Tokens that are generated at the Start Event for that Process must eventually arrive at an End Event. The + * Process will be in a running state until all Tokens are consumed. <p/> This method until the process ends with a + * given timeout. + */ + public abstract ProcessStatus waitForEnd(Process proc, long timeout); +} Modified: projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/incubator/startevent/StartEventSignalTest.java =================================================================== --- projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/incubator/startevent/StartEventSignalTest.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/incubator/startevent/StartEventSignalTest.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -62,6 +62,7 @@ // You need to register the process definition to activate the start trigger ProcessDefinitionService procDefService = ProcessDefinitionService.locateProcessDefinitionService(); procDefService.registerProcessDefinition(procDef); + ObjectName procDefID = procDef.getKey(); ProcessExt proc = (ProcessExt)procDef.newInstance(); try @@ -80,7 +81,7 @@ // Get the just started process ProcessService procService = ProcessService.locateProcessService(); - Set<ObjectName> procIDs = procService.getProcesses("StartEventSignal", null); + Set<ObjectName> procIDs = procService.getProcesses(procDefID, null); proc = (ProcessExt) procService.getProcess(procIDs.iterator().next()); // Wait for the process to end Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/ProcessImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/ProcessImpl.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/ProcessImpl.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -59,7 +59,7 @@ import org.jbpm.incubator.model.ProcessStructureExt; import org.jbpm.incubator.model.Property; import org.jbpm.incubator.model.StartEventExt; -import org.jbpm.incubator.service.ExecutionService; +import org.jbpm.incubator.service.ThreadingService; import org.jbpm.ri.runtime.TokenImpl; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -322,7 +322,7 @@ StartEvent start = prepareProcessStart(); ProcessEngine engine = getProcessEngine(); - ExecutionService exService = engine.getService(ExecutionService.class); + ThreadingService exService = engine.getService(ThreadingService.class); try { exService.startProcess(this, start, att); @@ -390,7 +390,7 @@ private ProcessStatus waitForEndInternal(long timeout) { ProcessEngine engine = getProcessEngine(); - ExecutionService exService = engine.getService(ExecutionService.class); + ThreadingService exService = engine.getService(ThreadingService.class); return exService.waitForEnd(this, timeout); } Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/StartEventImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/StartEventImpl.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/StartEventImpl.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -43,7 +43,7 @@ import org.jbpm.incubator.model.StartEventExt; import org.jbpm.incubator.model.Signal.SignalType; import org.jbpm.incubator.runtime.SignalHandler; -import org.jbpm.incubator.service.ExecutionService; +import org.jbpm.incubator.service.ThreadingService; import org.jbpm.incubator.service.SignalService; import org.jbpm.ri.model.builder.SingleOutFlowSupport; import org.slf4j.Logger; @@ -194,7 +194,7 @@ Process proc = procDef.newInstance(); StartEvent startEvent = proc.getNode(StartEvent.class, startEventName); - ExecutionService execService = engine.getService(ExecutionService.class); + ThreadingService execService = engine.getService(ThreadingService.class); execService.startProcess(proc, startEvent, null); } Deleted: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ExecutionServiceImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ExecutionServiceImpl.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ExecutionServiceImpl.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -1,329 +0,0 @@ -/* - * JBoss, Home of Professional Open Source - * Copyright 2005, JBoss Inc., and individual contributors as indicated - * by the @authors tag. See the copyright.txt in the distribution for a - * full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.jbpm.ri.service; - -// $Id$ - -import java.util.HashMap; -import java.util.Map; - -import javax.management.ObjectName; - -import org.jbpm.api.BPMException; -import org.jbpm.api.ProcessTimeoutException; -import org.jbpm.api.client.ProcessEngine; -import org.jbpm.api.client.Token; -import org.jbpm.api.model.Expression; -import org.jbpm.api.model.Process; -import org.jbpm.api.model.StartEvent; -import org.jbpm.api.model.Process.ProcessStatus; -import org.jbpm.api.runtime.Attachments; -import org.jbpm.api.service.ProcessService; -import org.jbpm.incubator.client.ProcessExt; -import org.jbpm.incubator.model.Assignment; -import org.jbpm.incubator.model.Assignment.AssignTime; -import org.jbpm.incubator.model.Signal.SignalType; -import org.jbpm.incubator.runtime.TokenExecutor; -import org.jbpm.incubator.service.ExecutionService; -import org.jbpm.incubator.service.SignalService; -import org.jbpm.ri.model.ProcessImpl; -import org.jbpm.ri.model.SequenceFlowImpl; -import org.jbpm.ri.model.SignalImpl; -import org.jbpm.ri.runtime.DelegatingToken; -import org.jbpm.ri.runtime.ExpressionEvaluator; -import org.jbpm.ri.runtime.MutableToken; -import org.jbpm.ri.runtime.RuntimeProcess; -import org.jbpm.ri.runtime.RuntimeProcessImpl; -import org.jbpm.ri.runtime.ThreadingTokenExecutor; -import org.jbpm.ri.runtime.TokenImpl; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * The process manager is the entry point to create, find and otherwise manage processes. - * - * @author thomas.diesler(a)jboss.com - * @since 18-Jun-2008 - */ -public class ExecutionServiceImpl extends ExecutionService implements MutableService -{ - // provide logging - final static Logger log = LoggerFactory.getLogger(ExecutionServiceImpl.class); - - // The map of active runtime processes - private Map<ObjectName, RuntimeProcess> runtimeProcesses = new HashMap<ObjectName, RuntimeProcess>(); - - @Override - public void setProcessEngine(ProcessEngine engine) - { - super.setProcessEngine(engine); - } - - @Override - public void startProcess(Process proc, StartEvent start, Attachments att) - { - startProcessInternal(proc, start, att); - } - - private synchronized void startProcessInternal(Process proc, StartEvent start, Attachments att) - { - @SuppressWarnings("serial") - class InitialFlow extends SequenceFlowImpl - { - InitialFlow(StartEvent start) - { - super(start.getName()); - } - } - - RuntimeProcess rtProc = getRuntimeProcess(proc, false); - boolean startProcessThread = (rtProc == null); - - // Create initial Token - TokenImpl initialToken = new TokenImpl(proc, att); - InitialFlow initialFlow = new InitialFlow(start); - initialToken.setLastFlow(initialFlow); - - // Register the initial Token - rtProc = getRuntimeProcess(proc, true); - TokenExecutor tokenExecutor = rtProc.getTokenExecutor(); - tokenExecutor.create(initialToken, initialFlow); - - // Start a new process thread - if (startProcessThread) - { - RunnableProcess runnable = new RunnableProcess(rtProc); - getProcessExecutor().execute(runnable); - synchronized (proc) - { - while (proc.getProcessStatus() != ProcessStatus.Active) - { - try - { - proc.wait(); - } - catch (InterruptedException ex) - { - log.error("Process interrupted", ex); - } - } - } - } - - // Do the start time assignments - startTimeAssignments((ProcessExt)proc, initialToken); - - // Start the initial token - tokenExecutor.start(initialToken); - } - - // Evaluate the Start time assignments - private void startTimeAssignments(ProcessExt proc, Token token) - { - DelegatingToken delegatingToken = new DelegatingToken((MutableToken)token); - Attachments atts = token.getAttachments(); - for (Assignment ass : proc.getAssignments()) - { - if (ass.getAssignTime() == AssignTime.Start) - { - Expression expr = ass.getFrom(); - ExpressionEvaluator exprEvaluator = new ExpressionEvaluator(expr); - Object result = exprEvaluator.evaluateExpression(delegatingToken); - String propName = ass.getTo().getName(); - atts.addAttachment(propName, result); - } - } - } - - public ProcessStatus waitForEnd(Process proc) - { - return waitForEndInternal(proc, 0); - } - - public ProcessStatus waitForEnd(Process proc, long timeout) - { - return waitForEndInternal(proc, timeout); - } - - /** - * Wait for the Process to end. All Tokens that are generated at the Start Event for that Process must eventually - * arrive at an End Event. The Process will be in a running state until all Tokens are consumed. If the process was - * aborted this method throws the causing RuntimeException if avaialable. - */ - private ProcessStatus waitForEndInternal(Process proc, long timeout) - { - ProcessImpl procImpl = (ProcessImpl)proc; - - ProcessStatus status = proc.getProcessStatus(); - if (status == ProcessStatus.None) - throw new IllegalStateException("Cannot wait for process in state: " + status); - - // Wait a little for the process to end - boolean forever = (timeout < 1); - long now = System.currentTimeMillis(); - long until = now + timeout; - try - { - while (forever || now < until) - { - synchronized (proc) - { - if (isProcessTerminated(proc)) - { - if (procImpl.getRuntimeException() != null) - { - throw new BPMException("Process aborted", procImpl.getRuntimeException()); - } - else - { - break; - } - } - - // Start waiting to get notified - long waitTimeout = forever ? 0 : until - now; - proc.wait(waitTimeout); - } - now = System.currentTimeMillis(); - } - - // Throw timeout exception if it took too long - if (isProcessTerminated(proc) == false) - { - RuntimeException rte = new ProcessTimeoutException("Process timeout after " + timeout + "ms for: " + proc.getKey()); - procImpl.setRuntimeException(rte); - log.error(rte.getMessage()); - throw rte; - } - } - catch (InterruptedException ex) - { - log.warn("Process interrupted", ex); - } - finally - { - // Unregister the process if not done already - // this could happen when the Process never received a start signal - // and then we get here because of a ProcessTimeoutException - ProcessEngine engine = getProcessEngine(); - ProcessService procService = engine.getService(ProcessService.class); - if (procService.getProcess(proc.getKey()) != null) - procService.unregisterProcess(proc.getKey()); - } - - status = proc.getProcessStatus(); - return status; - } - - private boolean isProcessTerminated(Process proc) - { - ProcessStatus status = proc.getProcessStatus(); - return status == ProcessStatus.Cancelled || status == ProcessStatus.Completed || status == ProcessStatus.Aborted; - } - - private RuntimeProcess getRuntimeProcess(Process proc, boolean createNew) - { - RuntimeProcess rtProcess; - synchronized (runtimeProcesses) - { - rtProcess = runtimeProcesses.get(proc.getKey()); - if (rtProcess == null && createNew) - { - rtProcess = new RuntimeProcessImpl(proc); - runtimeProcesses.put(proc.getKey(), rtProcess); - } - } - return rtProcess; - } - - /*************************************************************** - * The runnable Process - */ - class RunnableProcess implements Runnable - { - private RuntimeProcess rtProc; - - public RunnableProcess(RuntimeProcess rtProc) - { - this.rtProc = rtProc; - } - - public void run() - { - ThreadingTokenExecutor tokenExecutor = rtProc.getTokenExecutor(); - ProcessImpl procImpl = (ProcessImpl)rtProc.getProcess(); - Process proc = rtProc.getProcess(); - - ProcessEngine engine = getProcessEngine(); - SignalService sigService = engine.getService(SignalService.class); - - ObjectName procID = proc.getKey(); - String procName = proc.getName(); - try - { - synchronized (proc) - { - procImpl.setProcessStatus(ProcessStatus.Active); - sigService.throwSignal(new SignalImpl(SignalType.SYSTEM_PROCESS_ENTER, procID)); - - // Notify that the process is now Active - proc.notifyAll(); - } - - synchronized (rtProc) - { - // Wait until there are no more runnable tokens - while (tokenExecutor.hasRunnableTokens()) - { - try - { - rtProc.wait(); - } - catch (InterruptedException ex) - { - log.error("Process interrupted", ex); - } - } - - log.debug("End execution thread [proc=" + procName + ",status=" + proc.getProcessStatus() + "]"); - - if (proc.getProcessStatus() == ProcessStatus.Active) - procImpl.setProcessStatus(ProcessStatus.Completed); - } - } - finally - { - sigService.throwSignal(new SignalImpl(SignalType.SYSTEM_PROCESS_EXIT, procID)); - - synchronized (proc) - { - ProcessService procService = engine.getService(ProcessService.class); - procService.unregisterProcess(procID); - runtimeProcesses.remove(procID); - - // Notify that the process has now ended - proc.notifyAll(); - } - } - } - } -} \ No newline at end of file Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessDefinitionServiceImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessDefinitionServiceImpl.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessDefinitionServiceImpl.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -107,7 +107,7 @@ // Unregister the associated process instances ProcessService procService = engine.getService(ProcessService.class); - for (ObjectName procID : procService.getProcesses(procDef.getName(), null)) + for (ObjectName procID : procService.getProcesses(procDefID, null)) procService.unregisterProcess(procID); // Delete the ProcessDefinition through the PersistenceService Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessServiceImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessServiceImpl.java 2008-11-26 07:33:48 UTC (rev 3089) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ProcessServiceImpl.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -101,17 +101,18 @@ /** * Find the set of Processes for a given name * - * @param name The process name + * @param procDefID The process name * @param status The optional process status * @return An empty set if the process cannot be found */ @Override - public Set<ObjectName> getProcesses(String name, ProcessStatus status) + public Set<ObjectName> getProcesses(ObjectName procDefID, ProcessStatus status) { Set<ObjectName> procSet = new HashSet<ObjectName>(); for (Process auxProc : registeredProcs.values()) { - if (auxProc.getName().equals(name)) + ObjectName auxProcDefID = auxProc.getProcessDefinition().getKey(); + if (auxProcDefID.equals(procDefID)) { if (status == null || auxProc.getProcessStatus() == status) procSet.add(auxProc.getKey()); Copied: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ThreadingServiceImpl.java (from rev 3087, projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ExecutionServiceImpl.java) =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ThreadingServiceImpl.java (rev 0) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/ThreadingServiceImpl.java 2008-11-26 08:04:34 UTC (rev 3090) @@ -0,0 +1,327 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2005, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jbpm.ri.service; + +// $Id$ + +import java.util.HashMap; +import java.util.Map; + +import javax.management.ObjectName; + +import org.jbpm.api.BPMException; +import org.jbpm.api.ProcessTimeoutException; +import org.jbpm.api.client.ProcessEngine; +import org.jbpm.api.client.Token; +import org.jbpm.api.model.Expression; +import org.jbpm.api.model.Process; +import org.jbpm.api.model.StartEvent; +import org.jbpm.api.model.Process.ProcessStatus; +import org.jbpm.api.runtime.Attachments; +import org.jbpm.api.service.ProcessService; +import org.jbpm.incubator.client.ProcessExt; +import org.jbpm.incubator.model.Assignment; +import org.jbpm.incubator.model.Assignment.AssignTime; +import org.jbpm.incubator.model.Signal.SignalType; +import org.jbpm.incubator.runtime.TokenExecutor; +import org.jbpm.incubator.service.ThreadingService; +import org.jbpm.incubator.service.SignalService; +import org.jbpm.ri.model.ProcessImpl; +import org.jbpm.ri.model.SequenceFlowImpl; +import org.jbpm.ri.model.SignalImpl; +import org.jbpm.ri.runtime.DelegatingToken; +import org.jbpm.ri.runtime.ExpressionEvaluator; +import org.jbpm.ri.runtime.MutableToken; +import org.jbpm.ri.runtime.RuntimeProcess; +import org.jbpm.ri.runtime.RuntimeProcessImpl; +import org.jbpm.ri.runtime.ThreadingTokenExecutor; +import org.jbpm.ri.runtime.TokenImpl; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author thomas.diesler(a)jboss.com + * @since 18-Jun-2008 + */ +public class ThreadingServiceImpl extends ThreadingService implements MutableService +{ + // provide logging + final static Logger log = LoggerFactory.getLogger(ThreadingServiceImpl.class); + + // The map of active runtime processes + private Map<ObjectName, RuntimeProcess> runtimeProcesses = new HashMap<ObjectName, RuntimeProcess>(); + + @Override + public void setProcessEngine(ProcessEngine engine) + { + super.setProcessEngine(engine); + } + + @Override + public void startProcess(Process proc, StartEvent start, Attachments att) + { + startProcessInternal(proc, start, att); + } + + private synchronized void startProcessInternal(Process proc, StartEvent start, Attachments att) + { + @SuppressWarnings("serial") + class InitialFlow extends SequenceFlowImpl + { + InitialFlow(StartEvent start) + { + super(start.getName()); + } + } + + RuntimeProcess rtProc = getRuntimeProcess(proc, false); + boolean startProcessThread = (rtProc == null); + + // Create initial Token + TokenImpl initialToken = new TokenImpl(proc, att); + InitialFlow initialFlow = new InitialFlow(start); + initialToken.setLastFlow(initialFlow); + + // Register the initial Token + rtProc = getRuntimeProcess(proc, true); + TokenExecutor tokenExecutor = rtProc.getTokenExecutor(); + tokenExecutor.create(initialToken, initialFlow); + + // Start a new process thread + if (startProcessThread) + { + RunnableProcess runnable = new RunnableProcess(rtProc); + getProcessExecutor().execute(runnable); + synchronized (proc) + { + while (proc.getProcessStatus() != ProcessStatus.Active) + { + try + { + proc.wait(); + } + catch (InterruptedException ex) + { + log.error("Process interrupted", ex); + } + } + } + } + + // Do the start time assignments + startTimeAssignments((ProcessExt)proc, initialToken); + + // Start the initial token + tokenExecutor.start(initialToken); + } + + // Evaluate the Start time assignments + private void startTimeAssignments(ProcessExt proc, Token token) + { + DelegatingToken delegatingToken = new DelegatingToken((MutableToken)token); + Attachments atts = token.getAttachments(); + for (Assignment ass : proc.getAssignments()) + { + if (ass.getAssignTime() == AssignTime.Start) + { + Expression expr = ass.getFrom(); + ExpressionEvaluator exprEvaluator = new ExpressionEvaluator(expr); + Object result = exprEvaluator.evaluateExpression(delegatingToken); + String propName = ass.getTo().getName(); + atts.addAttachment(propName, result); + } + } + } + + public ProcessStatus waitForEnd(Process proc) + { + return waitForEndInternal(proc, 0); + } + + public ProcessStatus waitForEnd(Process proc, long timeout) + { + return waitForEndInternal(proc, timeout); + } + + /** + * Wait for the Process to end. All Tokens that are generated at the Start Event for that Process must eventually + * arrive at an End Event. The Process will be in a running state until all Tokens are consumed. If the process was + * aborted this method throws the causing RuntimeException if avaialable. + */ + private ProcessStatus waitForEndInternal(Process proc, long timeout) + { + ProcessImpl procImpl = (ProcessImpl)proc; + + ProcessStatus status = proc.getProcessStatus(); + if (status == ProcessStatus.None) + throw new IllegalStateException("Cannot wait for process in state: " + status); + + // Wait a little for the process to end + boolean forever = (timeout < 1); + long now = System.currentTimeMillis(); + long until = now + timeout; + try + { + while (forever || now < until) + { + synchronized (proc) + { + if (isProcessTerminated(proc)) + { + if (procImpl.getRuntimeException() != null) + { + throw new BPMException("Process aborted", procImpl.getRuntimeException()); + } + else + { + break; + } + } + + // Start waiting to get notified + long waitTimeout = forever ? 0 : until - now; + proc.wait(waitTimeout); + } + now = System.currentTimeMillis(); + } + + // Throw timeout exception if it took too long + if (isProcessTerminated(proc) == false) + { + RuntimeException rte = new ProcessTimeoutException("Process timeout after " + timeout + "ms for: " + proc.getKey()); + procImpl.setRuntimeException(rte); + log.error(rte.getMessage()); + throw rte; + } + } + catch (InterruptedException ex) + { + log.warn("Process interrupted", ex); + } + finally + { + // Unregister the process if not done already + // this could happen when the Process never received a start signal + // and then we get here because of a ProcessTimeoutException + ProcessEngine engine = getProcessEngine(); + ProcessService procService = engine.getService(ProcessService.class); + if (procService.getProcess(proc.getKey()) != null) + procService.unregisterProcess(proc.getKey()); + } + + status = proc.getProcessStatus(); + return status; + } + + private boolean isProcessTerminated(Process proc) + { + ProcessStatus status = proc.getProcessStatus(); + return status == ProcessStatus.Cancelled || status == ProcessStatus.Completed || status == ProcessStatus.Aborted; + } + + private RuntimeProcess getRuntimeProcess(Process proc, boolean createNew) + { + RuntimeProcess rtProcess; + synchronized (runtimeProcesses) + { + rtProcess = runtimeProcesses.get(proc.getKey()); + if (rtProcess == null && createNew) + { + rtProcess = new RuntimeProcessImpl(proc); + runtimeProcesses.put(proc.getKey(), rtProcess); + } + } + return rtProcess; + } + + /*************************************************************** + * The runnable Process + */ + class RunnableProcess implements Runnable + { + private RuntimeProcess rtProc; + + public RunnableProcess(RuntimeProcess rtProc) + { + this.rtProc = rtProc; + } + + public void run() + { + ThreadingTokenExecutor tokenExecutor = rtProc.getTokenExecutor(); + ProcessImpl procImpl = (ProcessImpl)rtProc.getProcess(); + Process proc = rtProc.getProcess(); + + ProcessEngine engine = getProcessEngine(); + SignalService sigService = engine.getService(SignalService.class); + + ObjectName procID = proc.getKey(); + String procName = proc.getName(); + try + { + synchronized (proc) + { + procImpl.setProcessStatus(ProcessStatus.Active); + sigService.throwSignal(new SignalImpl(SignalType.SYSTEM_PROCESS_ENTER, procID)); + + // Notify that the process is now Active + proc.notifyAll(); + } + + synchronized (rtProc) + { + // Wait until there are no more runnable tokens + while (tokenExecutor.hasRunnableTokens()) + { + try + { + rtProc.wait(); + } + catch (InterruptedException ex) + { + log.error("Process interrupted", ex); + } + } + + log.debug("End execution thread [proc=" + procName + ",status=" + proc.getProcessStatus() + "]"); + + if (proc.getProcessStatus() == ProcessStatus.Active) + procImpl.setProcessStatus(ProcessStatus.Completed); + } + } + finally + { + sigService.throwSignal(new SignalImpl(SignalType.SYSTEM_PROCESS_EXIT, procID)); + + synchronized (proc) + { + ProcessService procService = engine.getService(ProcessService.class); + procService.unregisterProcess(procID); + runtimeProcesses.remove(procID); + + // Notify that the process has now ended + proc.notifyAll(); + } + } + } + } +} \ No newline at end of file