[jbpm-commits] JBoss JBPM SVN: r2621 - in jbpm3/trunk: modules/core/src/main/java/org/jbpm/db and 18 other directories.

Author: alex.guizar(a)jboss.com Date: 2008-10-26 03:28:32 -0400 (Sun, 26 Oct 2008) New Revision: 2621 Added: jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/CleanUpProcessJob.java jbpm3/trunk/modules/core/src/main/resources/org/jbpm/job/CleanUpProcessJob.hbm.xml Modified: jbpm3/trunk/.project jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/graph/exe/ProcessInstance.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java jbpm3/trunk/modules/core/src/main/resources/hibernate.mapping.default.xml jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1072/JBPM1072Test.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1755/JBPM1755Test.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm983/JBPM983Test.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/job/executor/JobExecutorDbTest.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/msg/command/AsyncExecutionDbTest.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/optimisticlocking/LockingTest.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/scheduler/exe/TimerDbTest.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/taskmgmt/exe/TaskTimerExecutionDbTest.java jbpm3/trunk/modules/distribution/pom.xml jbpm3/trunk/modules/enterprise/jar/pom.xml jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/CommandServiceBean.java jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/msg/jms/JmsMessageServiceFactory.java jbpm3/trunk/modules/enterprise/jar/src/test/java/org/jbpm/msg/jms/JmsMessageTest.java jbpm3/trunk/pom.xml Log: [JBPM-1709] introduced cleanup-process-job to decrease the likelihood of stale state exceptions while ending a process instance revised several tests in core module which hanged after the above change because they assumed no jobs remain once a process ends upgraded posgresql driver version to 8.3, for the xa datasource JmsMessageTest is good to go on hsql Modified: jbpm3/trunk/.project =================================================================== --- jbpm3/trunk/.project 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/.project 2008-10-26 07:28:32 UTC (rev 2621) @@ -5,7 +5,13 @@ <projects> </projects> <buildSpec> + <buildCommand> + <name>org.maven.ide.eclipse.maven2Builder</name> + <arguments> + </arguments> + </buildCommand> </buildSpec> <natures> + <nature>org.maven.ide.eclipse.maven2Nature</nature> </natures> </projectDescription> Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -63,14 +63,12 @@ protected void setUp() throws Exception { super.setUp(); - createJbpmContext(); - initializeMembers(); + beginSessionTransaction(); } protected void tearDown() throws Exception { - resetMembers(); - closeJbpmContext(); + commitAndCloseSession(); ensureCleanDatabase(); super.tearDown(); @@ -189,7 +187,10 @@ protected void closeJbpmContext() { if (jbpmContext != null) + { jbpmContext.close(); + jbpmContext = null; + } } protected void startJobExecutor() @@ -198,10 +199,8 @@ jobExecutor.start(); } - private void processAllJobs(final long maxWait, int maxJobs) + protected void processAllJobs(final long maxWait, int maxJobs) { - boolean jobsAvailable = true; - // install a timer that will interrupt if it takes too long // if that happens, it will lead to an interrupted exception and the test // will fail @@ -220,14 +219,11 @@ try { - while (jobsAvailable) + while (getNbrOfJobsAvailable() > maxJobs) { log.debug("going to sleep for 200 millis, waiting for the job executor to process more jobs"); Thread.sleep(200); - jobsAvailable = (getNbrOfJobsAvailable() > maxJobs); } - jobExecutor.stopAndJoin(); - } catch (InterruptedException e) { @@ -263,7 +259,7 @@ { int nbrOfJobsAvailable = 0; Number jobs = (Number)session.createQuery("select count(*) from org.jbpm.job.Job").uniqueResult(); - log.debug("there are '" + jobs + "' jobs currently in the job table"); + log.debug("there are " + jobs + " jobs in the database"); if (jobs != null) { nbrOfJobsAvailable = jobs.intValue(); @@ -271,9 +267,10 @@ return nbrOfJobsAvailable; } - protected boolean areJobsAvailable() - { - return (getNbrOfJobsAvailable() > 0); + protected int getTimerCount() { + Number timerCount = (Number)session.createQuery("select count(*) from org.jbpm.job.Timer").uniqueResult(); + log.debug("there are " + timerCount + " timers in the database"); + return timerCount.intValue(); } protected Job getJob() @@ -289,14 +286,6 @@ protected void processJobs(long maxWait, int maxJobs) { commitAndCloseSession(); - try - { - Thread.sleep(300); - } - catch (InterruptedException e) - { - e.printStackTrace(); - } startJobExecutor(); try { Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -1,3 +1,24 @@ +/* + * 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.db; import java.util.Collection; @@ -8,14 +29,13 @@ import org.apache.commons.logging.LogFactory; import org.hibernate.Query; import org.hibernate.Session; -import org.jbpm.JbpmContext; +import org.hibernate.criterion.Restrictions; import org.jbpm.JbpmException; import org.jbpm.graph.def.Action; import org.jbpm.graph.exe.ProcessInstance; import org.jbpm.graph.exe.Token; import org.jbpm.job.Job; import org.jbpm.job.Timer; -import org.jbpm.svc.save.SaveOperation; public class JobSession { @@ -117,6 +137,17 @@ } } + public List loadJobs(long[] jobIds) { + int jobCount = jobIds.length; + Long[] jobs = new Long[jobCount]; + for (int i = 0; i < jobCount; i++) { + jobs[i] = new Long(jobIds[i]); + } + return session.createCriteria(Job.class) + .add(Restrictions.in("id", jobs)) + .list(); + } + public Job getJob(long jobId) { try { return (Job) session.get(Job.class, new Long(jobId)); @@ -164,45 +195,21 @@ } } - private static class DeleteJobsOperation implements SaveOperation { + public void deleteJobsForProcessInstance(ProcessInstance processInstance) { + log.debug("deleting timers for "+processInstance); + Query query = session.getNamedQuery("JobSession.deleteTimersForProcessInstance"); + query.setParameter("processInstance", processInstance); + int entityCount = query.executeUpdate(); + log.debug(entityCount+" remaining timers for "+processInstance+" were deleted"); - private ProcessInstance targetProcessInstance; - - private static final long serialVersionUID = 1L; - - DeleteJobsOperation(ProcessInstance processInstance) { - targetProcessInstance = processInstance; - } - - public void save(ProcessInstance processInstance, JbpmContext jbpmContext) { - // avoid deleting jobs for process instances that did not request job deletion - if (!targetProcessInstance.equals(processInstance)) { - log.debug("forgiving " + processInstance + ", it is not the target of this operation"); - return; - } - - log.debug("deleting timers for "+processInstance); - Session session = jbpmContext.getSession(); - Query query = session.getNamedQuery("JobSession.deleteTimersForProcessInstance"); - query.setParameter("processInstance", processInstance); - int entityCount = query.executeUpdate(); - log.debug(entityCount+" remaining timers for "+processInstance+" were deleted"); - - log.debug("deleting execute-node-jobs for "+processInstance); - query = session.getNamedQuery("JobSession.deleteExecuteNodeJobsForProcessInstance"); - query.setParameter("processInstance", processInstance); - entityCount = query.executeUpdate(); - log.debug(entityCount+" remaining execute-node-jobs for "+processInstance+" were deleted"); - } - + log.debug("deleting execute-node-jobs for "+processInstance); + query = session.getNamedQuery("JobSession.deleteExecuteNodeJobsForProcessInstance"); + query.setParameter("processInstance", processInstance); + entityCount = query.executeUpdate(); + log.debug(entityCount+" remaining execute-node-jobs for "+processInstance+" were deleted"); } - public void deleteJobsForProcessInstance(ProcessInstance processInstance) { - SaveOperation operation = new DeleteJobsOperation(processInstance); - JbpmContext.getCurrentJbpmContext().getServices().addSaveOperation(operation); - } - public List findJobsWithOverdueLockTime(Date treshold) { Query query = session.getNamedQuery("JobSession.findJobsWithOverdueLockTime"); query.setDate("now", treshold); Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/graph/exe/ProcessInstance.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/graph/exe/ProcessInstance.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/graph/exe/ProcessInstance.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -38,11 +38,12 @@ import org.jbpm.graph.def.Transition; import org.jbpm.graph.log.ProcessInstanceCreateLog; import org.jbpm.graph.log.ProcessInstanceEndLog; +import org.jbpm.job.CleanUpProcessJob; import org.jbpm.logging.exe.LoggingInstance; import org.jbpm.logging.log.ProcessLog; import org.jbpm.module.def.ModuleDefinition; import org.jbpm.module.exe.ModuleInstance; -import org.jbpm.scheduler.SchedulerService; +import org.jbpm.msg.MessageService; import org.jbpm.svc.Services; import org.jbpm.taskmgmt.exe.TaskMgmtInstance; import org.jbpm.util.Clock; @@ -322,11 +323,14 @@ superProcessToken.signal(superExecutionContext); } - // make sure all the timers for this process instance are cancelled when the process end updates get saved in the database. - // TODO route this directly through the jobSession. just like the suspend and resume. - // NOTE Only timers should be deleted, messages-type of jobs should be kept. - SchedulerService schedulerService = (SchedulerService) Services.getCurrentService(Services.SERVICENAME_SCHEDULER, false); - if (schedulerService!=null) schedulerService.deleteTimersByProcessInstance(this); + // make sure all the timers for this process instance are canceled after the process end updates are posted to the database + // NOTE Only timers should be deleted, messages should be kept. + MessageService messageService = (MessageService) Services.getCurrentService(Services.SERVICENAME_MESSAGE, false); + if (messageService != null) { + CleanUpProcessJob job = new CleanUpProcessJob(this); + job.setDueDate(new Date()); + messageService.send(job); + } } } Added: jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/CleanUpProcessJob.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/CleanUpProcessJob.java (rev 0) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/CleanUpProcessJob.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -0,0 +1,51 @@ +/* + * 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.job; + +import org.jbpm.JbpmContext; +import org.jbpm.graph.exe.ProcessInstance; +import org.jbpm.scheduler.SchedulerService; + +/** + * @author Alejandro Guizar + */ +public class CleanUpProcessJob extends Job { + + private static final long serialVersionUID = 1L; + + public CleanUpProcessJob() { + // default constructor + } + + public CleanUpProcessJob(ProcessInstance processInstance) { + this.processInstance = processInstance; + } + + public boolean execute(JbpmContext jbpmContext) throws Exception { + SchedulerService schedulerService = jbpmContext.getServices().getSchedulerService(); + if (schedulerService!=null) { + schedulerService.deleteTimersByProcessInstance(processInstance); + } + return true; + } + +} Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -49,7 +49,7 @@ } protected boolean isTransactionActive() { - return isJtaTxCreated(); + return isJtaTransactionInProgress(); } public void close() { @@ -67,28 +67,28 @@ void beginJtaTransaction() { try { - log.debug("start user JTA transaction"); + log.debug("begin user transaction"); userTransaction = ((JtaDbPersistenceServiceFactory) persistenceServiceFactory).getUserTransaction(); userTransaction.begin(); } catch (Exception e) { - throw new JbpmException("couldn't start JTA transaction", e); + throw new JbpmException("couldn't begin user transaction", e); } } void endJtaTransaction() { if (isRollbackOnly() || JTAHelper.isRollback(getJtaTransactionStatus())) { - log.debug("end jta transation with ROLLBACK"); + log.debug("rolling back user transaction"); try { userTransaction.rollback(); } catch (Exception e) { - throw new JbpmException("couldn't rollback JTA transaction", e); + throw new JbpmException("couldn't rollback user transaction", e); } } else { - log.debug("end jta transation with COMMIT"); + log.debug("committing user transaction"); try { userTransaction.commit(); } catch (Exception e) { - throw new JbpmException("couldn't commit JTA transaction", e); + throw new JbpmException("couldn't commit user transaction", e); } } } Modified: jbpm3/trunk/modules/core/src/main/resources/hibernate.mapping.default.xml =================================================================== --- jbpm3/trunk/modules/core/src/main/resources/hibernate.mapping.default.xml 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/main/resources/hibernate.mapping.default.xml 2008-10-26 07:28:32 UTC (rev 2621) @@ -100,6 +100,7 @@ <mapping resource="org/jbpm/job/Timer.hbm.xml"/> <mapping resource="org/jbpm/job/ExecuteNodeJob.hbm.xml"/> <mapping resource="org/jbpm/job/ExecuteActionJob.hbm.xml"/> + <mapping resource="org/jbpm/job/CleanUpProcessJob.hbm.xml"/> <!-- taskmgmt.exe mapping files --> <mapping resource="org/jbpm/taskmgmt/exe/TaskMgmtInstance.hbm.xml"/> Added: jbpm3/trunk/modules/core/src/main/resources/org/jbpm/job/CleanUpProcessJob.hbm.xml =================================================================== --- jbpm3/trunk/modules/core/src/main/resources/org/jbpm/job/CleanUpProcessJob.hbm.xml (rev 0) +++ jbpm3/trunk/modules/core/src/main/resources/org/jbpm/job/CleanUpProcessJob.hbm.xml 2008-10-26 07:28:32 UTC (rev 2621) @@ -0,0 +1,13 @@ +<?xml version="1.0"?> + +<!DOCTYPE hibernate-mapping PUBLIC + "-//Hibernate/Hibernate Mapping DTD 3.0//EN" + "http://hibernate.sourceforge.net/hibernate-mapping-3.0.dtd"> + +<hibernate-mapping auto-import="false" default-access="field"> + + <subclass name="org.jbpm.job.CleanUpProcessJob" + discriminator-value="C" + extends="org.jbpm.job.Job" /> + +</hibernate-mapping> Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1072/JBPM1072Test.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1072/JBPM1072Test.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1072/JBPM1072Test.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -101,6 +101,7 @@ // wait for process end EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END); + assertEquals(2, Counter.getExecutionCount()); // stop job executors for (int i = jobExecutors.length - 1; i >= 0; i--) { @@ -113,7 +114,6 @@ } beginSessionTransaction(); - assertEquals(2, Counter.getExecutionCount()); } public static class Counter implements ActionHandler { Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1755/JBPM1755Test.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1755/JBPM1755Test.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm1755/JBPM1755Test.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -39,8 +39,9 @@ @Override protected void tearDown() throws Exception { EventCallback.clear(); + + stopJobExecutor(); graphSession.deleteProcessDefinition(processDefinitionId); - stopJobExecutor(); super.tearDown(); } @@ -76,7 +77,8 @@ waitForProcessInstanceEnd(processInstanceId); assertTrue(hasProcessInstanceEnded(processInstanceId)); } - beginSessionTransaction(); + + processJobs(maxWaitTime, 0); } private void waitForProcessInstanceEnd(long processInstanceId) { Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm983/JBPM983Test.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm983/JBPM983Test.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm983/JBPM983Test.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -5,7 +5,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.jbpm.JbpmContext; import org.jbpm.db.AbstractDbTestCase; import org.jbpm.graph.def.ActionHandler; import org.jbpm.graph.def.ProcessDefinition; @@ -18,29 +17,43 @@ * * https://jira.jboss.org/jira/browse/JBPM-983 * - * @author Tom Baeyens + * @author Tom Baeyens */ -public class JBPM983Test extends AbstractDbTestCase -{ +public class JBPM983Test extends AbstractDbTestCase { + private static Log log = LogFactory.getLog(JBPM983Test.class); + private long subProcessDefinitionId; + private long processDefinitionId; + @Override - protected String getJbpmTestConfig() - { + protected String getJbpmTestConfig() { return "org/jbpm/jbpm.test.cfg.xml"; } - protected void setUp() throws Exception - { + protected void setUp() throws Exception { super.setUp(); + + ProcessDefinition subProcessDefinition = ProcessDefinition.parseXmlString(SUBPROCESS_XML); + jbpmContext.deployProcessDefinition(subProcessDefinition); + subProcessDefinitionId = subProcessDefinition.getId(); + + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString(PROCESS_XML); + jbpmContext.deployProcessDefinition(processDefinition); + processDefinitionId = processDefinition.getId(); + JobExecutor jobExecutor = getJbpmConfiguration().getJobExecutor(); jobExecutor.setNbrOfThreads(5); jobExecutor.start(); } - protected void tearDown() throws Exception - { + protected void tearDown() throws Exception { getJbpmConfiguration().getJobExecutor().stop(); + + newTransaction(); + graphSession.deleteProcessDefinition(processDefinitionId); + graphSession.deleteProcessDefinition(subProcessDefinitionId); + super.tearDown(); } @@ -111,101 +124,91 @@ + "</end-state>" + "</process-definition>"; - public void testConcurrentJobs() throws Exception - { - assertTrue(getJbpmConfiguration().getJobExecutor().getNbrOfThreads() > 1); + public void testConcurrentJobs() throws Exception { + // create test properties + Map testVariables = new HashMap(); + testVariables.put("test", "true"); - ProcessDefinition subProcessDefinition = ProcessDefinition.parseXmlString(SUBPROCESS_XML); - jbpmContext.deployProcessDefinition(subProcessDefinition); + final int processCount = 10; + long[] processInstanceIds = new long[processCount]; + for (int i = 0; i < processCount; i++) { + newTransaction(); + ProcessInstance processInstance = jbpmContext.newProcessInstanceForUpdate("superprocess"); + processInstance.getContextInstance().addVariables(testVariables); + processInstance.signal(); + processInstanceIds[i] = processInstance.getId(); + } - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString(PROCESS_XML); - jbpmContext.deployProcessDefinition(processDefinition); - - newTransaction(); - - try { - // create test properties - Map testVariables = new HashMap(); - testVariables.put("test", "true"); + for (int i = 0; i < processCount; i++) { + long piId = processInstanceIds[i]; + waitFor(piId); - final int processCount = 10; - long[] piIds = new long[processCount]; - for (int i = 0; i < processCount; i++) - { - ProcessInstance pi = jbpmContext.newProcessInstanceForUpdate("superprocess"); - pi.getContextInstance().addVariables(testVariables); - pi.signal(); - piIds[i] = pi.getId(); - - newTransaction(); - } - - for (int i = 0; i < processCount; i++) - { - long piId = piIds[i]; - waitFor(piId); - - ProcessInstance pi = jbpmContext.loadProcessInstance(piId); - assertEquals("end-state-success", pi.getRootToken().getNode().getName()); - } - } finally { - newTransaction(); - graphSession.deleteProcessDefinition(processDefinition); - graphSession.deleteProcessDefinition(subProcessDefinition); + ProcessInstance pi = jbpmContext.loadProcessInstance(piId); + assertEquals("end-state-success", pi.getRootToken().getNode().getName()); } } - protected void waitFor(long piId) throws Exception - { + protected void waitFor(long piId) throws Exception { final int endTimeout = 30; long startTime = System.currentTimeMillis(); - + boolean processInstanceHasEnded = false; - - while (! processInstanceHasEnded) - { - if (System.currentTimeMillis() - startTime > endTimeout * 1000) - { + while (!processInstanceHasEnded) { + if (System.currentTimeMillis() - startTime > endTimeout * 1000) { fail("Aborting after " + endTimeout + " seconds."); return; } newTransaction(); - processInstanceHasEnded = jbpmContext.loadProcessInstance(piId).hasEnded(); log.info("waiting for workflow completion...."); - try - { + try { Thread.sleep(200); } - catch (InterruptedException e) - { + catch (InterruptedException e) { log.error("wait for workflow was interruputed", e); } } } - public static class TestAction implements ActionHandler - { + public static class TestAction implements ActionHandler { private static final long serialVersionUID = 1L; - public void execute(ExecutionContext executionContext) throws Exception - { - String processName = executionContext.getProcessDefinition().getName() + ":" + executionContext.getProcessInstance().getId(); + public void execute(ExecutionContext executionContext) throws Exception { + String processName = executionContext.getProcessDefinition().getName() + + ":" + + executionContext.getProcessInstance().getId(); String nodeName = executionContext.getToken().getNode().getName(); String tokenName = executionContext.getToken().toString(); - log.info("ACTION (process=" + processName + ",node=" + nodeName + ",token=" + tokenName + "): begin"); + log.info("ACTION (process=" + + processName + + ",node=" + + nodeName + + ",token=" + + tokenName + + "): begin"); - for (int i = 0; i < 5; i++) - { - log.info("ACTION (process=" + processName + ",node=" + nodeName + ",token=" + tokenName + "): working..."); + for (int i = 0; i < 5; i++) { + log.info("ACTION (process=" + + processName + + ",node=" + + nodeName + + ",token=" + + tokenName + + "): working..."); Thread.sleep(100); } - log.info("ACTION (process=" + processName + ",node=" + nodeName + ",token=" + tokenName + "): end"); + log.info("ACTION (process=" + + processName + + ",node=" + + nodeName + + ",token=" + + tokenName + + "): end"); executionContext.leaveNode(); } Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/job/executor/JobExecutorDbTest.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/job/executor/JobExecutorDbTest.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/job/executor/JobExecutorDbTest.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -53,7 +53,6 @@ JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); try { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( "<process-definition name='bulk messages'>" + " <start-state>" Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/msg/command/AsyncExecutionDbTest.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/msg/command/AsyncExecutionDbTest.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/msg/command/AsyncExecutionDbTest.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -33,35 +33,43 @@ import org.jbpm.graph.exe.ExecutionContext; import org.jbpm.graph.exe.ProcessInstance; -public class AsyncExecutionDbTest extends AbstractDbTestCase -{ +public class AsyncExecutionDbTest extends AbstractDbTestCase { static List recordedNodes = new ArrayList(); - public static class RecordNode implements ActionHandler - { + public static class RecordNode implements ActionHandler { + private static final long serialVersionUID = 1L; - public void execute(ExecutionContext executionContext) throws Exception - { + public void execute(ExecutionContext executionContext) throws Exception { Node node = executionContext.getNode(); recordedNodes.add(node.getName()); node.leave(executionContext); } } - public void testAsyncExecution() throws Exception - { + public void testAsyncExecution() throws Exception { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + " <start-state>" + " <transition to='one' />" - + " </start-state>" + " <node async='true' name='one'>" + " <action class='org.jbpm.msg.command.AsyncExecutionDbTest$RecordNode' />" - + " <transition to='two' />" + " </node>" + " <node async='exclusive' name='two'>" - + " <action class='org.jbpm.msg.command.AsyncExecutionDbTest$RecordNode' />" + " <transition to='three' />" + " </node>" - + " <node async='true' name='three'>" + " <action class='org.jbpm.msg.command.AsyncExecutionDbTest$RecordNode' />" + " <transition to='end' />" - + " </node>" + " <end-state name='end' />" + "</process-definition>"); + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='one' />" + + " </start-state>" + + " <node async='true' name='one'>" + + " <action class='org.jbpm.msg.command.AsyncExecutionDbTest$RecordNode' />" + + " <transition to='two' />" + + " </node>" + + " <node async='exclusive' name='two'>" + + " <action class='org.jbpm.msg.command.AsyncExecutionDbTest$RecordNode' />" + + " <transition to='three' />" + + " </node>" + + " <node async='true' name='three'>" + + " <action class='org.jbpm.msg.command.AsyncExecutionDbTest$RecordNode' />" + + " <transition to='end' />" + + " </node>" + + " <end-state name='end' />" + + "</process-definition>"); processDefinition = saveAndReload(processDefinition); - try - { + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); processInstance.signal(); jbpmContext.save(processInstance); @@ -85,8 +93,7 @@ assertTrue(processInstance.hasEnded()); assertEquals(processDefinition.getNode("end"), processInstance.getRootToken().getNode()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } @@ -94,20 +101,19 @@ static Set recordedActionNumbers = new HashSet(); - public static class RecordAction implements ActionHandler - { + public static class RecordAction implements ActionHandler { + private static final long serialVersionUID = 1L; String nbr; - public void execute(ExecutionContext executionContext) throws Exception - { + public void execute(ExecutionContext executionContext) throws Exception { recordedActionNumbers.add(nbr); } } - public void testAsyncAction() throws Exception - { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + " <event type='process-start'>" + public void testAsyncAction() throws Exception { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <event type='process-start'>" + " <action async='true' class='" + RecordAction.class.getName() + "'>" @@ -146,15 +152,17 @@ + " <nbr>6</nbr>" + " </action>" + " </event>" - + " <transition to='end' />" + " </node>" + " <end-state name='end' />" + "</process-definition>"); + + " <transition to='end' />" + + " </node>" + + " <end-state name='end' />" + + "</process-definition>"); processDefinition = saveAndReload(processDefinition); - try - { + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); processInstance.signal(); jbpmContext.save(processInstance); assertEquals(processDefinition.getNode("end"), processInstance.getRootToken().getNode()); - assertEquals(6, getNbrOfJobsAvailable()); + assertEquals(7, getNbrOfJobsAvailable()); assertEquals(0, recordedActionNumbers.size()); processJobs(5000); @@ -171,8 +179,7 @@ assertEquals(expected, recordedActionNumbers); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/optimisticlocking/LockingTest.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/optimisticlocking/LockingTest.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/optimisticlocking/LockingTest.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -5,7 +5,6 @@ import java.util.List; import org.hibernate.Query; -import org.hibernate.Session; import org.jbpm.JbpmConfiguration; import org.jbpm.JbpmContext; import org.jbpm.db.AbstractDbTestCase; @@ -18,216 +17,147 @@ * * https://jira.jboss.org/jira/browse/JBPM-1071 */ -public class LockingTest extends AbstractDbTestCase -{ +public class LockingTest extends AbstractDbTestCase { + + private long processDefinitionId; + static int nbrOfThreads = 5; static int nbrOfIterations = 20; - public void testLocking() - { - // the process will be executed in 2 separete transactions: + protected void setUp() throws Exception { + super.setUp(); + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition name='lockprocess'>" + + " <start-state name='start'>" + + " <transition to='end'/>" + + " </start-state>" + + " <end-state name='end' />" + + "</process-definition>"); + jbpmContext.deployProcessDefinition(processDefinition); + processDefinitionId = processDefinition.getId(); + } + + protected void tearDown() throws Exception { + graphSession.deleteProcessDefinition(processDefinitionId); + super.tearDown(); + } + + public void testLocking() { + // the process will be executed in 2 separate transactions: // Transaction 1 will create the process instance and position // the root token in the start state // Transaction 2 will signal the process instance while it is in the // start state, and that signal will bring the process to it's end state. - // It's the second transaction for which we'll set up multiple competing threads - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition name='lockprocess'>" + - " <start-state name='start'>" + - " <transition to='end'/>" + - " </start-state>" + - " <end-state name='end' />" + - "</process-definition>"); + // multiple competing threads will be set up for the second transaction + for (int i = 0; i < nbrOfIterations; i++) { + long processInstanceId = jbpmContext.newProcessInstanceForUpdate("lockprocess").getId(); + commitAndCloseSession(); - // deploy the process - JbpmContext jbpmContext = getJbpmConfiguration().createJbpmContext(); - long processDefinitionId = 0; - try - { - jbpmContext.deployProcessDefinition(processDefinition); - processDefinitionId = processDefinition.getId(); - } - finally - { - jbpmContext.close(); - } - - try - { - for (int i = 0; i < nbrOfIterations; i++) - { - long processInstanceId = launchProcessInstance(); + // create a bunch of threads that will all wait on the + // semaphore before they will try to signal the same process instance + Object semaphore = new Object(); + List threads = startThreads(semaphore, processInstanceId); - // create a bunch of threads that will all wait on the - // semaphore before they will try to signal the same process instance - Object semaphore = new Object(); - List threads = startThreads(semaphore, processInstanceId); + // release all the threads + synchronized (semaphore) { + semaphore.notifyAll(); + } - // release all the threads - synchronized (semaphore) - { - semaphore.notifyAll(); - } + // wait for all threads to finish + joinAllThreads(threads); - // wait for all threads to finish - joinAllThreads(threads); + // check that only 1 of those threads committed + beginSessionTransaction(); + Query query = session.createQuery("from org.jbpm.graph.exe.Comment"); + List results = query.list(); + assertEquals(results.toString(), 1, results.size()); - // check that only 1 of those threads committed - jbpmContext = getJbpmConfiguration().createJbpmContext(); - try - { - Session session = jbpmContext.getSession(); - Query query = session.createQuery("select c from org.jbpm.graph.exe.Comment as c"); - List results = query.list(); - // System.out.println("iteration "+i+": "+results); - assertEquals(results.toString(), 1, results.size()); + // delete the comment + session.delete(results.get(0)); - // delete the comment - session.delete(results.get(0)); - - } - finally - { - jbpmContext.close(); - } - - // check that the process instance has ended - jbpmContext = getJbpmConfiguration().createJbpmContext(); - try - { - ProcessInstance processInstance = jbpmContext.loadProcessInstance(processInstanceId); - assertTrue(processInstance.hasEnded()); - - } - finally - { - jbpmContext.close(); - } - } + // check that the process instance has ended + ProcessInstance processInstance = jbpmContext.loadProcessInstance(processInstanceId); + assertTrue(processInstance.hasEnded()); } - finally - { - jbpmContext = getJbpmConfiguration().createJbpmContext(); - try - { - jbpmContext.getGraphSession().deleteProcessDefinition(processDefinitionId); - } - finally - { - jbpmContext.close(); - } - } - } - private long launchProcessInstance() - { - JbpmContext jbpmContext = getJbpmConfiguration().createJbpmContext(); - try - { - return jbpmContext.newProcessInstance("lockprocess").getId(); - } - finally - { - jbpmContext.close(); - } - } - - private List startThreads(Object semaphore, long processInstanceId) - { + private List startThreads(Object semaphore, long processInstanceId) { List threads = new ArrayList(); - for (int i = 0; i < nbrOfThreads; i++) - { + for (int i = 0; i < nbrOfThreads; i++) { Thread thread = new LockThread(getJbpmConfiguration(), semaphore, processInstanceId); thread.start(); threads.add(thread); } - try - { + try { // giving the threads the opportunity to start and arrive in the wait Thread.sleep(200); } - catch (InterruptedException e) - { + catch (InterruptedException e) { // ignore } return threads; } - public static class LockThread extends Thread - { + static class LockThread extends Thread { + Object semaphore; long processInstanceId; JbpmConfiguration jbpmConfiguration; - public LockThread(JbpmConfiguration jbpmConfiguration, Object semaphore, long processInstanceId) - { + public LockThread(JbpmConfiguration jbpmConfiguration, Object semaphore, long processInstanceId) { this.semaphore = semaphore; this.processInstanceId = processInstanceId; this.jbpmConfiguration = jbpmConfiguration; } - public void run() - { - try - { + public void run() { + try { // first wait until the all threads are released at once in the // method testLocking - synchronized (semaphore) - { + synchronized (semaphore) { semaphore.wait(); } - // after a thread is released (=notified), it will try to load the process instance, + // after a thread is released (=notified), it will try to load the + // process instance, // signal it and then commit the transaction JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); - try - { + try { ProcessInstance processInstance = jbpmContext.loadProcessInstance(processInstanceId); processInstance.signal(); jbpmContext.save(processInstance); - // add a comment in the same transaction so that we can see which thread won + // add a comment in the same transaction so that we can see which + // thread won Comment comment = new Comment(getName() + " committed"); jbpmContext.getSession().save(comment); - } - catch (Exception e) - { + catch (Exception e) { jbpmContext.setRollbackOnly(); } - finally - { + finally { jbpmContext.close(); } - } - catch (InterruptedException e) - { + catch (InterruptedException e) { fail("semaphore waiting got interrupted"); } - catch (Throwable t) - { + catch (Throwable t) { // ignore } } } - private void joinAllThreads(List threads) - { + private void joinAllThreads(List threads) { Iterator iter = threads.iterator(); - while (iter.hasNext()) - { - Thread thread = (Thread)iter.next(); - try - { + while (iter.hasNext()) { + Thread thread = (Thread) iter.next(); + try { thread.join(); } - catch (InterruptedException e) - { + catch (InterruptedException e) { fail("join interrupted"); } } Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/scheduler/exe/TimerDbTest.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/scheduler/exe/TimerDbTest.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/scheduler/exe/TimerDbTest.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -24,13 +24,14 @@ import java.util.Calendar; import java.util.Date; import java.util.GregorianCalendar; -import java.util.Iterator; import java.util.List; +import org.hibernate.criterion.Restrictions; import org.jbpm.context.exe.ContextInstance; import org.jbpm.db.AbstractDbTestCase; -import org.jbpm.graph.def.Action; import org.jbpm.graph.def.ActionHandler; +import org.jbpm.graph.def.Event; +import org.jbpm.graph.def.Node; import org.jbpm.graph.def.ProcessDefinition; import org.jbpm.graph.exe.ExecutionContext; import org.jbpm.graph.exe.ProcessInstance; @@ -40,16 +41,14 @@ public class TimerDbTest extends AbstractDbTestCase { - static boolean isNoOpExecuted = false; + static boolean isNoOpExecuted = false; - protected void setUp() throws Exception - { + protected void setUp() throws Exception { super.setUp(); isNoOpExecuted = false; } - public void testSaveTimer() - { + public void testSaveTimer() { final Date now = Calendar.getInstance().getTime(); Timer timer = new Timer(); @@ -59,10 +58,9 @@ timer.setRepeat("repeat-duration"); session.save(timer); - try - { + try { newTransaction(); - timer = (Timer)session.load(Timer.class, new Long(timer.getId())); + timer = (Timer) session.load(Timer.class, new Long(timer.getId())); assertEquals("timer-name", timer.getName()); // we test for the same date in a simple format @@ -81,270 +79,238 @@ assertEquals(ncal.get(Calendar.HOUR_OF_DAY), tcal.get(Calendar.HOUR_OF_DAY)); assertEquals(ncal.get(Calendar.MINUTE), tcal.get(Calendar.MINUTE)); assertEquals(ncal.get(Calendar.SECOND), tcal.get(Calendar.SECOND)); - assertEquals(DateDbTestUtil.getInstance().convertDateToSeconds(now), DateDbTestUtil.getInstance().convertDateToSeconds(timer.getDueDate())); + assertEquals(DateDbTestUtil.getInstance().convertDateToSeconds(now), + DateDbTestUtil.getInstance().convertDateToSeconds(timer.getDueDate())); assertEquals("transition-name", timer.getTransitionName()); assertEquals("repeat-duration", timer.getRepeat()); } - finally - { + finally { session.delete(timer); } } - public void testTimerCreation() throws Exception - { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition>" + - " <start-state>" + - " <transition to='catch crooks' />" + - " </start-state>" + - " <state name='catch crooks'>" + - " <timer name='reminder' duedate='5 seconds' />" + - " <transition to='end'/>" + - " </state>" + - " <end-state name='end'/>" + - "</process-definition>"); + public void testTimerCreation() throws Exception { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='catch crooks' />" + + " </start-state>" + + " <state name='catch crooks'>" + + " <timer name='reminder' duedate='5 seconds' />" + + " <transition to='end'/>" + + " </state>" + + " <end-state name='end'/>" + + "</process-definition>"); - graphSession.saveProcessDefinition(processDefinition); - try - { + graphSession.saveProcessDefinition(processDefinition); + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); jbpmContext.save(processInstance); // long before = System.currentTimeMillis(); processInstance.signal(); // long after = System.currentTimeMillis(); jbpmContext.save(processInstance); + newTransaction(); - Timer timer = (Timer)session.createQuery("from org.jbpm.job.Timer").uniqueResult(); + Timer timer = (Timer) session.createQuery("from org.jbpm.job.Timer").uniqueResult(); assertNotNull("Timer is null", timer); assertEquals("reminder", timer.getName()); - // Commented out because of timer latency is changing between time required + // Commented out because of timer latency is changing between time + // required // to connect to the database // assertTrue((before + 5000) <= timer.getDueDate().getTime()); // assertTrue(timer.getDueDate().getTime() <= (after + 5000)); assertEquals("catch crooks", timer.getGraphElement().getName()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - public void testTimerCancellation() - { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition>" + - " <start-state>" + - " <transition to='catch crooks' />" + - " </start-state>" + - " <state name='catch crooks'>" + - " <timer name='reminder' duedate='5 seconds' />" + - " <transition to='end'/>" + - " </state>" + - " <end-state name='end'/>" + - "</process-definition>"); + public void testTimerCancellation() { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='catch crooks' />" + + " </start-state>" + + " <state name='catch crooks'>" + + " <timer name='reminder' duedate='5 seconds' />" + + " <transition to='end'/>" + + " </state>" + + " <end-state name='end'/>" + + "</process-definition>"); - graphSession.saveProcessDefinition(processDefinition); - try - { + graphSession.saveProcessDefinition(processDefinition); + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); processInstance.signal(); processInstance = saveAndReload(processInstance); - processInstance.signal(); newTransaction(); - - assertFalse(areJobsAvailable()); + assertEquals(0, getTimerCount()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - public void testTimerAction() - { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition name='process'>" + - " <start-state>" + - " <transition to='sometask' />" + - " </start-state>" + - " <task-node name='sometask'>" + - " <timer name='reminder'" + - " duedate='1 business minutes'" + - " repeat='1 business minutes'" + - " transition='time-out-transition' >" + - " <action class='my-action-handler-class-name' />" + - " </timer>" + - " <task name='do something'/>" + - " <transition name='time-out-transition' to='sometask' />" + - " </task-node>" + - "</process-definition>"); - - graphSession.saveProcessDefinition(processDefinition); - try - { - ProcessInstance processInstance = new ProcessInstance(processDefinition); + public void testTimerAction() { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition name='process'>" + + " <start-state>" + + " <transition to='sometask' />" + + " </start-state>" + + " <task-node name='sometask'>" + + " <timer name='reminder'" + + " duedate='1 business minutes'" + + " repeat='1 business minutes'" + + " transition='time-out-transition' >" + + " <action class='my-action-handler-class-name' />" + + " </timer>" + + " <task name='do something'/>" + + " <transition name='time-out-transition' to='sometask' />" + + " </task-node>" + + "</process-definition>"); - jbpmContext.save(processInstance); - newTransaction(); + graphSession.saveProcessDefinition(processDefinition); + try { + Node node = processDefinition.getNode("sometask"); + List actions = node.getEvent(Event.EVENTTYPE_NODE_ENTER).getActions(); + assertEquals(1, actions.size()); + actions = node.getEvent(Event.EVENTTYPE_NODE_LEAVE).getActions(); + assertEquals(1, actions.size()); - processInstance = graphSession.loadProcessInstance(processInstance.getId()); + ProcessInstance processInstance = new ProcessInstance(processDefinition); processInstance.signal(); - - processDefinition = processInstance.getProcessDefinition(); - Iterator iter = processDefinition.getNode("sometask").getEvent("node-enter").getActions().iterator(); - while (iter.hasNext()) - { - Action action = (Action)iter.next(); - action.getId(); - } - jbpmContext.save(processInstance); - newTransaction(); - assertTrue(areJobsAvailable()); + newTransaction(); + assertEquals(1, getTimerCount()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - public void testTaskTimerExecution() { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition>" + " <start-state>" + - " <transition to='timed task' />" + - " </start-state>" + - " <task-node name='timed task'>" + - " <task>" + - " <timer duedate='23 business seconds'>" + - " <action class='geftem-eu-shuppe-oender-ze-konte'/>" + - " </timer>" + - " </task>" + - " </task-node>" + - "</process-definition>"); - processDefinition = saveAndReload(processDefinition); - try - { + public void testTaskTimerExecution() { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='timed task' />" + + " </start-state>" + + " <task-node name='timed task'>" + + " <task>" + + " <timer duedate='23 business seconds'>" + + " <action class='geftem-eu-shuppe-oender-ze-konte'/>" + + " </timer>" + + " </task>" + + " </task-node>" + + "</process-definition>"); + processDefinition = saveAndReload(processDefinition); + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); processInstance.signal(); - processInstance = saveAndReload(processInstance); - - assertTrue(areJobsAvailable()); + newTransaction(); + assertEquals(1, getTimerCount()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - public void testTimerCancellationAtProcessEnd() { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition>" + - " <start-state>" + - " <transition to='s' />" + - " </start-state>" + - " <state name='s'>" + - " <event type='node-enter'>" + - " <create-timer duedate='26 business seconds'>" + - " <action class='claim.you.are.Innocent' />" + - " </create-timer>" + - " </event>" + - " <transition to='end' />" + - " </state>" + - " <end-state name='end' />" + - "</process-definition>"); - processDefinition = saveAndReload(processDefinition); - try - { + public void testTimerCancellationAtProcessEnd() { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='s' />" + + " </start-state>" + + " <state name='s'>" + + " <event type='node-enter'>" + + " <create-timer duedate='26 business seconds'>" + + " <action class='claim.you.are.Innocent' />" + + " </create-timer>" + + " </event>" + + " <transition to='end' />" + + " </state>" + + " <end-state name='end' />" + + "</process-definition>"); + processDefinition = saveAndReload(processDefinition); + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); processInstance.signal(); processInstance = saveAndReload(processInstance); - - assertTrue(areJobsAvailable()); + assertEquals(1, getTimerCount()); processInstance.signal(); - processInstance = saveAndReload(processInstance); - - assertFalse(areJobsAvailable()); + processJobs(5000, 0); + assertEquals(0, getTimerCount()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - public void testFindTimersByName() { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition>" + - " <start-state>" + - " <transition to='timed task' />" + - " </start-state>" + - " <task-node name='timed task'>" + - " <task name='find the hole in the market'>" + - " <timer name='reminder' duedate='23 business seconds'>" + - " <action class='geftem-eu-shuppe-oender-ze-konte'/>" + - " </timer>" + - " </task>" + - " </task-node>" + - "</process-definition>"); - processDefinition = saveAndReload(processDefinition); - try - { + public void testFindTimersByName() { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='timed task' />" + + " </start-state>" + + " <task-node name='timed task'>" + + " <task name='find the hole in the market'>" + + " <timer name='reminder' duedate='23 business seconds'>" + + " <action class='geftem-eu-shuppe-oender-ze-konte'/>" + + " </timer>" + + " </task>" + + " </task-node>" + + "</process-definition>"); + processDefinition = saveAndReload(processDefinition); + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); processInstance.signal(); processInstance = saveAndReload(processInstance); - List timersByName = session.createQuery("select timer from org.jbpm.job.Timer as timer where timer.name='reminder'").list(); + List timersByName = session.createCriteria(Timer.class) + .add(Restrictions.eq("name", "reminder")) + .list(); assertNotNull(timersByName); assertEquals(1, timersByName.size()); - Timer timer = (Timer)timersByName.get(0); - assertEquals("geftem-eu-shuppe-oender-ze-konte", timer.getAction().getActionDelegation().getClassName()); + Timer timer = (Timer) timersByName.get(0); + assertEquals("geftem-eu-shuppe-oender-ze-konte", timer.getAction() + .getActionDelegation() + .getClassName()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - - public void testTimerRepeat() { - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition>" + - " <start-state>" + - " <transition to='a' />" + - " </start-state>" + - " <state name='a'>" + - " <timer name='reminder' " + - " duedate='0 seconds'" + - " repeat='5 seconds' >" + - " <action class='org.jbpm.scheduler.exe.TimerDbTest$NoOp' />" + - " </timer>" + - " <transition to='b'/>" + - " <transition name='back' to='a'/>" + - " </state>" + - " <state name='b'/>" + - "</process-definition>"); - processDefinition = saveAndReload(processDefinition); - try - { + public void testTimerRepeat() { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='a' />" + + " </start-state>" + + " <state name='a'>" + + " <timer name='reminder' " + + " duedate='0 seconds'" + + " repeat='5 seconds' >" + + " <action class='org.jbpm.scheduler.exe.TimerDbTest$NoOp' />" + + " </timer>" + + " <transition to='b'/>" + + " <transition name='back' to='a'/>" + + " </state>" + + " <state name='b'/>" + + "</process-definition>"); + processDefinition = saveAndReload(processDefinition); + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); // long before = System.currentTimeMillis(); processInstance.signal(); // long after = System.currentTimeMillis(); - jbpmContext.save(processInstance); newTransaction(); - - Timer timer = (Timer)jobSession.getFirstAcquirableJob(null); + Timer timer = (Timer) jobSession.getFirstAcquirableJob(null); assertNotNull(timer); Date date = timer.getDueDate(); assertNotNull(date); @@ -354,7 +320,9 @@ processJobs(2000, 1); - timer = (Timer)session.createQuery("from org.jbpm.job.Timer").setMaxResults(1).uniqueResult(); + timer = (Timer) session.createQuery("from org.jbpm.job.Timer") + .setMaxResults(1) + .uniqueResult(); assertEquals(origDueDate + 5000, timer.getDueDate().getTime()); processInstance = jbpmContext.loadProcessInstance(processInstance.getId()); @@ -363,9 +331,9 @@ // after = System.currentTimeMillis(); jbpmContext.save(processInstance); - newTransaction(); - timer = (Timer)jobSession.getFirstAcquirableJob(null); + newTransaction(); + timer = (Timer) jobSession.getFirstAcquirableJob(null); assertNotNull(timer); date = timer.getDueDate(); assertNotNull(date); @@ -373,46 +341,41 @@ // assertTrue(date.getTime() <= after); newTransaction(); - processInstance = jbpmContext.loadProcessInstance(processInstance.getId()); processInstance.signal(); jbpmContext.save(processInstance); newTransaction(); - - assertFalse(areJobsAvailable()); + assertEquals(0, getTimerCount()); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - public void testTimerUpdatingProcessVariables() { + public void testTimerUpdatingProcessVariables() { // variable a will be a task instance local variable // variable b will be a process instance variable - ProcessDefinition processDefinition = ProcessDefinition.parseXmlString( - "<process-definition>" + - " <start-state>" + - " <transition to='a' />" + - " </start-state>" + - " <task-node name='a'>" + - " <task name='wait for var updates'>" + - " <controller>" + - " <variable name ='a' />" + - " </controller>" + - " <timer name='update variables' " + - " duedate='0 seconds'" + - " repeat='5 seconds' >" + - " <action class='org.jbpm.scheduler.exe.TimerDbTest$UpdateVariables' />" + - " </timer>" + - " </task>" + - " </task-node>" + - "</process-definition>"); - processDefinition = saveAndReload(processDefinition); - try - { + ProcessDefinition processDefinition = ProcessDefinition.parseXmlString("<process-definition>" + + " <start-state>" + + " <transition to='a' />" + + " </start-state>" + + " <task-node name='a'>" + + " <task name='wait for var updates'>" + + " <controller>" + + " <variable name ='a' />" + + " </controller>" + + " <timer name='update variables' " + + " duedate='0 seconds'" + + " repeat='5 seconds' >" + + " <action class='org.jbpm.scheduler.exe.TimerDbTest$UpdateVariables' />" + + " </timer>" + + " </task>" + + " </task-node>" + + "</process-definition>"); + processDefinition = saveAndReload(processDefinition); + try { ProcessInstance processInstance = new ProcessInstance(processDefinition); ContextInstance contextInstance = processInstance.getContextInstance(); contextInstance.setVariable("a", "value a"); @@ -428,43 +391,42 @@ assertEquals("value a", contextInstance.getVariable("a")); assertEquals("value b updated", contextInstance.getVariable("b")); - TaskInstance taskInstance = (TaskInstance)processInstance.getTaskMgmtInstance().getTaskInstances().iterator().next(); + TaskInstance taskInstance = (TaskInstance) processInstance.getTaskMgmtInstance() + .getTaskInstances() + .iterator() + .next(); assertEquals("value a updated", taskInstance.getVariable("a")); assertEquals("value b updated", taskInstance.getVariable("b")); } - finally - { + finally { jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } - } + } - public static class ConcurrentUpdateAction implements ActionHandler - { + public static class ConcurrentUpdateAction implements ActionHandler { + private static final long serialVersionUID = 1L; - public void execute(ExecutionContext executionContext) throws Exception - { + public void execute(ExecutionContext executionContext) throws Exception { executionContext.setVariable("a", "value a timer actioned updated"); } } - public static class NoOp implements ActionHandler - { + public static class NoOp implements ActionHandler { + private static final long serialVersionUID = 1L; - public void execute(ExecutionContext executionContext) throws Exception - { + public void execute(ExecutionContext executionContext) throws Exception { isNoOpExecuted = true; } } - public static class UpdateVariables implements ActionHandler - { + public static class UpdateVariables implements ActionHandler { + private static final long serialVersionUID = 1L; - public void execute(ExecutionContext executionContext) throws Exception - { + public void execute(ExecutionContext executionContext) throws Exception { executionContext.setVariable("a", "value a updated"); executionContext.setVariable("b", "value b updated"); } Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/taskmgmt/exe/TaskTimerExecutionDbTest.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/taskmgmt/exe/TaskTimerExecutionDbTest.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/taskmgmt/exe/TaskTimerExecutionDbTest.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -118,7 +118,7 @@ jbpmContext.save(taskInstance.getTaskMgmtInstance().getProcessInstance()); newTransaction(); - assertFalse(areJobsAvailable()); + assertEquals(0, getTimerCount()); } finally { Modified: jbpm3/trunk/modules/distribution/pom.xml =================================================================== --- jbpm3/trunk/modules/distribution/pom.xml 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/distribution/pom.xml 2008-10-26 07:28:32 UTC (rev 2621) @@ -157,7 +157,6 @@ <dependency> <groupId>postgresql</groupId> <artifactId>postgresql</artifactId> - <classifier>jdbc3</classifier> </dependency> </dependencies> Modified: jbpm3/trunk/modules/enterprise/jar/pom.xml =================================================================== --- jbpm3/trunk/modules/enterprise/jar/pom.xml 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/enterprise/jar/pom.xml 2008-10-26 07:28:32 UTC (rev 2621) @@ -243,8 +243,6 @@ <excludes> <!-- do not remove, ejb scheduler requires xa support --> <exclude>org/jbpm/scheduler/ejbtimer/EjbSchedulerTest.java</exclude> - <!-- https://jira.jboss.org/jira/browse/JBPM-1709 --> - <exclude>org/jbpm/msg/jms/JmsMessageTest.java</exclude> </excludes> </configuration> </plugin> Modified: jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/CommandServiceBean.java =================================================================== --- jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/CommandServiceBean.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/CommandServiceBean.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -37,6 +37,7 @@ import org.jbpm.command.CommandService; import org.jbpm.msg.jms.JmsMessageServiceFactory; import org.jbpm.persistence.jta.JtaDbPersistenceServiceFactory; +import org.jbpm.tx.TxService; /** * Stateless session bean that executes {@linkplain Command commands} by @@ -144,6 +145,11 @@ try { log.debug("executing " + command); result = command.execute(jbpmContext); + // check whether command requested a rollback + TxService txService = jbpmContext.getServices().getTxService(); + if (txService.isRollbackOnly()) { + sessionContext.setRollbackOnly(); + } } catch (RuntimeException e) { throw e; } catch (Exception e) { Modified: jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java =================================================================== --- jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -23,6 +23,7 @@ import java.io.PrintWriter; import java.io.StringWriter; +import java.util.Arrays; import java.util.List; import javax.jms.Destination; @@ -34,9 +35,9 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.hibernate.criterion.Restrictions; import org.jbpm.JbpmContext; import org.jbpm.command.Command; +import org.jbpm.db.JobSession; import org.jbpm.job.Job; import org.jbpm.msg.jms.JmsMessageService; import org.jbpm.msg.jms.JmsMessageServiceFactory; @@ -57,37 +58,38 @@ } public Object execute(JbpmContext jbpmContext) throws Exception { - Job job = (Job) jbpmContext.getSession().get(Job.class, jobId); + JobSession jobSession = jbpmContext.getJobSession(); + Job job = jobSession.getJob(jobId); if (job == null || job.getLockOwner() != null) { + log.debug("job " + jobId + " was deleted or is locked, abandoning execution"); jbpmContext.setRollbackOnly(); return null; } + String lockOwner = Long.toString(jobId); if (job.isExclusive()) { - // acquire exclusive jobs - List exclusiveJobs = jbpmContext.getSession().createCriteria(Job.class) - .add(Restrictions.isNull("lockMode")) - .add(Restrictions.eq("processInstance", job.getProcessInstance())) - .list(); - String lockOwner = Long.toString(jobId); - long[] exclusiveJobIds = new long[exclusiveJobs.size()]; - for (int i = 0; i < exclusiveJobIds.length; i++) { + List exclusiveJobs = jobSession.findExclusiveJobs(lockOwner, job.getProcessInstance()); + // lock exclusive jobs + int jobCount = exclusiveJobs.size(); + long[] exclusiveJobIds = new long[jobCount]; + for (int i = 0; i < jobCount; i++) { Job exclusiveJob = (Job) exclusiveJobs.get(i); exclusiveJob.setLockOwner(lockOwner); exclusiveJobIds[i] = exclusiveJob.getId(); } + log.debug("locked jobs " + Arrays.toString(exclusiveJobIds)); // execute exclusive jobs in separate transaction - executeJobs(jbpmContext, exclusiveJobIds); + postJobsExecution(jbpmContext, exclusiveJobIds); } else { + // lock job to prevent others from deleting it + job.setLockOwner(lockOwner); + log.debug("executing " + job); executeJob(job, jbpmContext); } return null; } static void executeJob(Job job, JbpmContext jbpmContext) { - // prevent others from removing job - job.setLockOwner(ExecuteJobCommand.class.getName()); - log.debug("executing " + job); try { if (job.execute(jbpmContext)) { jbpmContext.getJobSession().deleteJob(job); @@ -107,7 +109,7 @@ } } - private static void executeJobs(JbpmContext jbpmContext, long[] exclusiveJobIds) + private static void postJobsExecution(JbpmContext jbpmContext, long[] exclusiveJobIds) throws NamingException, JMSException { Services services = jbpmContext.getServices(); JmsMessageServiceFactory messageServiceFactory = (JmsMessageServiceFactory) services.getServiceFactory(Services.SERVICENAME_MESSAGE); Modified: jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java =================================================================== --- jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -21,9 +21,14 @@ */ package org.jbpm.ejb.impl; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.jbpm.JbpmContext; import org.jbpm.command.Command; -import org.jbpm.db.JobSession; import org.jbpm.job.Job; /** @@ -33,17 +38,20 @@ public class ExecuteJobsCommand implements Command { private long[] jobIds; + private static final long serialVersionUID = 1L; + private static Log log = LogFactory.getLog(ExecuteJobsCommand.class); public ExecuteJobsCommand(long[] jobIds) { this.jobIds = jobIds; } public Object execute(JbpmContext jbpmContext) throws Exception { - JobSession jobSession = jbpmContext.getJobSession(); - for (int i = 0; i < jobIds.length; i++) { - Job job = jobSession.loadJob(jobIds[i]); - ExecuteJobCommand.executeJob(job, jbpmContext); + log.debug("executing jobs " + Arrays.toString(jobIds)); + List jobs = jbpmContext.getJobSession().loadJobs(jobIds); + for (Iterator i = jobs.iterator(); i.hasNext();) { + Job job = (Job) i.next(); + job.execute(jbpmContext); } return null; } Modified: jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/msg/jms/JmsMessageServiceFactory.java =================================================================== --- jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/msg/jms/JmsMessageServiceFactory.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/enterprise/jar/src/main/java/org/jbpm/msg/jms/JmsMessageServiceFactory.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -85,13 +85,21 @@ destination = (Destination) lookup(destinationJndiName); } catch (NamingException e) { - throw new JbpmException("could not retrieve message destination", e); + throw new JbpmException("could not retrieve job destination", e); } } return destination; } public Destination getCommandDestination() { + if (commandDestination == null) { + try { + commandDestination = (Destination) lookup(commandDestinationJndiName); + } + catch (NamingException e) { + throw new JbpmException("could not retrieve command destination", e); + } + } return commandDestination; } Modified: jbpm3/trunk/modules/enterprise/jar/src/test/java/org/jbpm/msg/jms/JmsMessageTest.java =================================================================== --- jbpm3/trunk/modules/enterprise/jar/src/test/java/org/jbpm/msg/jms/JmsMessageTest.java 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/modules/enterprise/jar/src/test/java/org/jbpm/msg/jms/JmsMessageTest.java 2008-10-26 07:28:32 UTC (rev 2621) @@ -53,8 +53,8 @@ private static LocalCommandServiceHome commandServiceHome; - static final int processExecutionCount = 20; - static final int maxWaitTime = 10000; + static final int processExecutionCount = 5; + static final int maxWaitTime = 10 * 1000; public static Test suite() throws Exception { return new IntegrationTestSetup(JmsMessageTest.class, "enterprise-test.war"); Modified: jbpm3/trunk/pom.xml =================================================================== --- jbpm3/trunk/pom.xml 2008-10-25 14:35:37 UTC (rev 2620) +++ jbpm3/trunk/pom.xml 2008-10-26 07:28:32 UTC (rev 2621) @@ -80,7 +80,7 @@ <!-- Database Driver Versions --> <hsqldb.version>1.8.0.7</hsqldb.version> <mysql.connector.version>5.1.6</mysql.connector.version> - <postgresql.version>7.4</postgresql.version> + <postgresql.version>8.3-603.jdbc3</postgresql.version> </properties> <!-- DependencyManagement --> @@ -346,7 +346,6 @@ <dependency> <groupId>postgresql</groupId> <artifactId>postgresql</artifactId> - <classifier>jdbc3</classifier> <version>${postgresql.version}</version> </dependency> </dependencies> @@ -526,7 +525,6 @@ <dependency> <groupId>postgresql</groupId> <artifactId>postgresql</artifactId> - <classifier>jdbc3</classifier> <scope>test</scope> </dependency> </dependencies>