[jbpm-commits] JBoss JBPM SVN: r2134 - in jbpm3/trunk/modules: enterprise/scripts and 17 other directories.

Author: alex.guizar(a)jboss.com Date: 2008-09-05 20:28:41 -0400 (Fri, 05 Sep 2008) New Revision: 2134 Added: jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/test/EventCallback.java Modified: jbpm3/trunk/modules/enterprise/pom.xml jbpm3/trunk/modules/enterprise/scripts/assembly-test-dependencies.xml jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/LocalTimerEntity.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/TimerEntityBean.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerService.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EntitySchedulerService.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/ExecuteTimerCommand.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/LocalTimerService.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/TimerServiceBean.java jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/ejb-jar.xml jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/jbosscmp-jdbc.xml jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerTest.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/db/JobSession.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/graph/node/Join.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/Timer.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutor.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/jpdl/el/parser/ELParserTokenManager.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/msg/db/DbMessageService.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/SchedulerService.java jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/db/DbSchedulerService.java jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/mail/TaskMailTest.java jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/scheduler/exe/SchedulerTest.java jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/svc/ServicesTest.java Log: [JBPM-1708] rewrote EjbSchedulerTest in terms of monitors instead of sleeping ensured that timer cancelations would not result in stale state or entity not found exceptions made improvements to the timer beans and the job objects while trying to figure out why they failed, until realizing the problem described in JBPM-1717 Modified: jbpm3/trunk/modules/enterprise/pom.xml =================================================================== --- jbpm3/trunk/modules/enterprise/pom.xml 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/pom.xml 2008-09-06 00:28:41 UTC (rev 2134) @@ -169,9 +169,7 @@ <artifactId>maven-surefire-plugin</artifactId> <configuration> <excludes> - <!-- https://jira.jboss.org/jira/browse/JBPM-1708 --> - <exclude>org/jbpm/scheduler/ejbtimer/EjbSchedulerTest.java</exclude> - <!-- https://jira.jboss.org/jira/browse/JBPM-1709 --> + <!-- https://jira.jboss.org/jira/browse/JBPM-1717 --> <exclude>org/jbpm/msg/jms/AsyncProcessingTest.java</exclude> </excludes> </configuration> Modified: jbpm3/trunk/modules/enterprise/scripts/assembly-test-dependencies.xml =================================================================== --- jbpm3/trunk/modules/enterprise/scripts/assembly-test-dependencies.xml 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/scripts/assembly-test-dependencies.xml 2008-09-06 00:28:41 UTC (rev 2134) @@ -13,12 +13,12 @@ <outputDirectory>/</outputDirectory> <useStrictFiltering>true</useStrictFiltering> <includes> - <include>*:aspectjrt:jar</include> - <include>*:cactus.integration.shared.api:jar</include> - <include>*:cactus.core.framework.wrapper.javaEE.14</include> - <include>*:cactus.core.framework.javaEE.13-14</include> - <include>*:cactus.core.framework.javaEE.12-13-14</include> - <include>*:junit:jar</include> + <include>aspectj:aspectjrt:jar</include> + <include>org.apache.cactus:cactus.integration.shared.api</include> + <include>org.apache.cactus:cactus.core.framework.wrapper.javaEE.14</include> + <include>org.apache.cactus:cactus.core.framework.javaEE.13-14</include> + <include>org.apache.cactus:cactus.core.framework.javaEE.12-13-14</include> + <include>junit:junit</include> </includes> <unpack>false</unpack> <scope>test</scope> Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/LocalTimerEntity.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/LocalTimerEntity.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/LocalTimerEntity.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -4,15 +4,16 @@ import org.jbpm.graph.exe.ProcessInstance; import org.jbpm.graph.exe.Token; +import org.jbpm.job.Timer; public interface LocalTimerEntity extends EJBLocalObject { - public void cancelTimer(); + public void createTimer(Timer timer); - public void createTimer(org.jbpm.job.Timer timer); + public void cancelTimer(Timer timer); public void cancelTimersByName(String timerName, Token token); - public void deleteTimersForProcessInstance(ProcessInstance processInstance); + public void cancelTimersForProcessInstance(ProcessInstance processInstance); } \ No newline at end of file Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -1,5 +1,8 @@ package org.jbpm.ejb.impl; +import java.io.PrintWriter; +import java.io.StringWriter; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.jbpm.JbpmContext; @@ -9,10 +12,10 @@ public class ExecuteJobCommand implements Command { + private final long jobId; + private static final long serialVersionUID = 1L; - - long jobId; - + public ExecuteJobCommand(long jobId) { this.jobId = jobId; } @@ -20,11 +23,19 @@ public Object execute(JbpmContext jbpmContext) throws Exception { JobSession jobSession = jbpmContext.getJobSession(); Job job = jobSession.loadJob(jobId); - log.debug("executing job "+jobId); - if (job.execute(jbpmContext)) { - log.debug("deleting job "+jobId); - jobSession.deleteJob(job); + job.setLockOwner(getClass().getName()); // prevent others from removing job + log.debug("executing " + job); + try { + if (job.execute(jbpmContext)) { + jobSession.deleteJob(job); + } } + catch (Exception e) { + log.debug("exception while executing " + job, e); + StringWriter memoryWriter = new StringWriter(); + e.printStackTrace(new PrintWriter(memoryWriter)); + job.setException(memoryWriter.toString()); + } return job; } Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/TimerEntityBean.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/TimerEntityBean.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/TimerEntityBean.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -74,6 +74,10 @@ public abstract void setProcessInstanceId(Long processInstanceId); + public abstract char getDiscriminator(); + + public abstract void setDiscriminator(char discriminator); + public void ejbActivate() { try { Context initial = new InitialContext(); @@ -122,56 +126,70 @@ public void ejbPostCreate() {} public void ejbTimeout(javax.ejb.Timer ejbTimer) { - log.debug("ejb timer " + ejbTimer + " fires"); + log.debug(ejbTimer + " fired"); TimerInfo timerInfo = (TimerInfo) ejbTimer.getInfo(); Timer timer = (Timer) commandService.execute(new ExecuteTimerCommand(timerInfo.getTimerId())); // if the timer has repeat - if(timer != null && timer.getRepeat() != null) { + if (timer.getRepeat() != null) { // create a new timer - log.debug("scheduling timer for repeat at " + timer.getDueDate()); + log.debug("scheduling timer for repeat on " + timer.getDueDate()); createTimer(timer); } } public void createTimer(org.jbpm.job.Timer timer) { - log.debug("Creating timer " + timer + " in the ejb timer service"); TimerService timerService = entityContext.getTimerService(); - timerService.createTimer(timer.getDueDate(), new TimerInfo(timer)); + javax.ejb.Timer ejbTimer = timerService.createTimer(timer.getDueDate(), new TimerInfo(timer)); + log.debug("created " + ejbTimer); } - public void cancelTimer() { - log.debug("Cancelling timer: " + this.getName()); + public void cancelTimer(org.jbpm.job.Timer timer) { + long timerId = timer.getId(); Collection timers = entityContext.getTimerService().getTimers(); + log.debug("retrieved " + timers.size() + " ejb timer(s) by id " + timerId); + + int count = 0; for (Iterator i = timers.iterator(); i.hasNext();) { javax.ejb.Timer ejbTimer = (javax.ejb.Timer) i.next(); - ejbTimer.cancel(); + TimerInfo timerInfo = (TimerInfo) ejbTimer.getInfo(); + if (timerInfo.getTimerId() == timerId) { + ejbTimer.cancel(); + ++count; + } } + log.debug("canceled " + count + " ejb timer(s) by id " + timerId); } public void cancelTimersByName(String timerName, Token token) { - log.debug("cancelling timers with name " + timerName - + " from the ejb timer service"); Collection timers = entityContext.getTimerService().getTimers(); + log.debug("retrieved " + timers.size() + " ejb timer(s) by name '" + timerName + "' for " + token); + + int count = 0; for (Iterator i = timers.iterator(); i.hasNext();) { javax.ejb.Timer ejbTimer = (javax.ejb.Timer) i.next(); TimerInfo timerInfo = (TimerInfo) ejbTimer.getInfo(); - if(timerInfo.matchesName(timerName, token)) { + if (timerInfo.matchesName(timerName, token)) { ejbTimer.cancel(); + ++count; } } + log.debug("canceled " + count + " ejb timer(s) by name '" + timerName + "' for " + token); } - public void deleteTimersForProcessInstance(ProcessInstance processInstance) { - log.debug("deleting timers for process instance " + processInstance - + " from the ejb timer service"); + public void cancelTimersForProcessInstance(ProcessInstance processInstance) { Collection timers = entityContext.getTimerService().getTimers(); + log.debug("retrieved " + timers.size() + " timer(s) for " + processInstance); + + int count = 0; for (Iterator i = timers.iterator(); i.hasNext();) { javax.ejb.Timer ejbTimer = (javax.ejb.Timer) i.next(); TimerInfo timerInfo = (TimerInfo) ejbTimer.getInfo(); - if(timerInfo.matchesProcessInstance(processInstance)) { + if (timerInfo.matchesProcessInstance(processInstance)) { ejbTimer.cancel(); + ++count; } } + log.debug("canceled " + count + " ejb timer(s) for " + processInstance); } } Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerService.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerService.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerService.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -42,22 +42,28 @@ } public void createTimer(Timer timer) { - log.debug("creating timer "+timer); + log.debug("creating " + timer); jobSession.saveJob(timer); session.flush(); timerService.createTimer(timer); } + public void deleteTimer(Timer timer) { + log.debug("deleting " + timer); + timerService.cancelTimer(timer); + jobSession.deleteJob(timer); + } + public void deleteTimersByName(String timerName, Token token) { - log.debug("deleting timers by name "+timerName); - jobSession.cancelTimersByName(timerName, token); + log.debug("deleting timers by name '" + timerName + "' for " + token); timerService.cancelTimersByName(timerName, token); + jobSession.deleteTimersByName(timerName, token); } public void deleteTimersByProcessInstance(ProcessInstance processInstance) { - log.debug("deleting timers for process instance "+processInstance); + log.debug("deleting timers for " + processInstance); + timerService.cancelTimersForProcessInstance(processInstance); jobSession.deleteJobsForProcessInstance(processInstance); - timerService.deleteTimersForProcessInstance(processInstance); } public void close() { Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EntitySchedulerService.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EntitySchedulerService.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/EntitySchedulerService.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -38,7 +38,7 @@ } public void createTimer(Timer timer) { - log.debug("creating timer " + timer); + log.debug("creating " + timer); jobSession.saveJob(timer); session.flush(); try { @@ -46,42 +46,54 @@ timerEntity.createTimer(timer); } catch (FinderException e) { - log.error("failed to retrieve entity for timer " + timer, e); + log.error("failed to retrieve entity for " + timer, e); } } - public void deleteTimersByName(String timerName, Token token) { - log.debug("deleting timers by name " + timerName); - Collection timerEntities = null; + public void deleteTimer(Timer timer) { + log.debug("deleting " + timer); + try { + LocalTimerEntity timerEntity = timerEntityHome.findByPrimaryKey(new Long(timer.getId())); + timerEntity.cancelTimer(timer); + } + catch (FinderException e) { + log.error("failed to retrieve entity for " + timer, e); + } + jobSession.deleteJob(timer); + } + + public void deleteTimersByName(String timerName, Token token) { try { + Collection timerEntities; if(timerName == null || timerName.equals("")) { timerEntities = timerEntityHome.findByTokenId(new Long(token.getId())); } else { timerEntities = timerEntityHome.findByTokenIdAndName(new Long(token.getId()), timerName); } + log.debug("found " + timerEntities.size() + " timer entities by name '" + timerName + "' for " + token); for (Iterator i = timerEntities.iterator(); i.hasNext();) { LocalTimerEntity timerEntity = (LocalTimerEntity) i.next(); timerEntity.cancelTimersByName(timerName, token); } } catch (FinderException e) { - log.error("failed to retrieve timer entities for name " + timerName + " and token " + token, e); + log.error("failed to retrieve timer entities by name '" + timerName + "' for " + token, e); } - jobSession.cancelTimersByName(timerName, token); + jobSession.deleteTimersByName(timerName, token); } public void deleteTimersByProcessInstance(ProcessInstance processInstance) { - log.debug("deleting timers for process instance " + processInstance); try { Collection timerEntities = timerEntityHome.findByProcessInstanceId(new Long(processInstance.getId())); + log.debug("found " + timerEntities.size() + " timer entities for " + processInstance); for (Iterator i = timerEntities.iterator(); i.hasNext();) { LocalTimerEntity timerEntity = (LocalTimerEntity) i.next(); - timerEntity.deleteTimersForProcessInstance(processInstance); + timerEntity.cancelTimersForProcessInstance(processInstance); } } catch (FinderException e) { - log.error("failed to retrieve timer entities for process instance " + processInstance, e); + log.error("failed to retrieve timer entities for " + processInstance, e); } jobSession.deleteJobsForProcessInstance(processInstance); } Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/ExecuteTimerCommand.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/ExecuteTimerCommand.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/ExecuteTimerCommand.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -1,17 +1,20 @@ package org.jbpm.scheduler.ejbtimer; +import java.io.PrintWriter; +import java.io.StringWriter; + 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; +import org.jbpm.job.Timer; public class ExecuteTimerCommand implements Command { + private final long timerId; + private static final long serialVersionUID = 1L; - - final long timerId; public ExecuteTimerCommand(long timerId) { this.timerId = timerId; @@ -19,18 +22,22 @@ public Object execute(JbpmContext jbpmContext) throws Exception { JobSession jobSession = jbpmContext.getJobSession(); - Job timer = jobSession.getJob(timerId); - if (timer!=null) { - log.debug("executing timer "+timerId); + Timer timer = jobSession.loadTimer(timerId); + timer.setLockOwner(getClass().getName()); // prevent others from removing timer + log.debug("executing " + timer); + try { if (timer.execute(jbpmContext)) { - log.debug("deleting timer "+timerId); - jobSession.deleteJob(timer); + jbpmContext.getServices().getSchedulerService().deleteTimer(timer); } - } else { - log.info("timer "+timerId+" was deleted, cannot execute it"); } + catch (Exception e) { + log.debug("exception while executing " + timer, e); + StringWriter memoryWriter = new StringWriter(); + e.printStackTrace(new PrintWriter(memoryWriter)); + timer.setException(memoryWriter.toString()); + } return timer; } - private static Log log = LogFactory.getLog(ExecuteTimerCommand.class); + private static final Log log = LogFactory.getLog(ExecuteTimerCommand.class); } Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/LocalTimerService.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/LocalTimerService.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/LocalTimerService.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -32,7 +32,8 @@ public interface LocalTimerService extends javax.ejb.EJBLocalObject { void createTimer(Timer timer); + void cancelTimer(Timer timer); void cancelTimersByName(String timerName, Token token); - void deleteTimersForProcessInstance(ProcessInstance processInstance); + void cancelTimersForProcessInstance(ProcessInstance processInstance); } Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/TimerServiceBean.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/TimerServiceBean.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/scheduler/ejbtimer/TimerServiceBean.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -23,6 +23,7 @@ import java.io.Serializable; import java.rmi.RemoteException; +import java.util.Collection; import java.util.Iterator; import javax.ejb.EJBException; @@ -58,54 +59,78 @@ public void ejbCreate() { } - public void createTimer(Timer timer) { + public void createTimer(org.jbpm.job.Timer timer) { TimerService timerService = sessionContext.getTimerService(); - log.debug("creating timer "+timer+" in the ejb timer service"); - timerService.createTimer(timer.getDueDate(), new TimerInfo(timer)); + javax.ejb.Timer ejbTimer = timerService.createTimer(timer.getDueDate(), new TimerInfo(timer)); + log.debug("created " + ejbTimer); } + public void cancelTimer(org.jbpm.job.Timer timer) { + // TODO make the scanning of timers for cancellation optional by only deleting the timerjobs in the db. + // of course, the corresponding ejb timer notifications have to be ignored. + + long timerId = timer.getId(); + Collection timers = sessionContext.getTimerService().getTimers(); + log.debug("examining " + timers.size() + " ejb timer(s) by id " + timerId); + + int count = 0; + for (Iterator i = timers.iterator(); i.hasNext();) { + javax.ejb.Timer ejbTimer = (javax.ejb.Timer) i.next(); + if (ejbTimer.getInfo() instanceof TimerInfo) { + TimerInfo timerInfo = (TimerInfo) ejbTimer.getInfo(); + if (timerId == timerInfo.getTimerId()) { + ejbTimer.cancel(); + ++count; + } + } + } + log.debug("canceled " + count + " ejb timer(s) by id " + timerId); + } + public void cancelTimersByName(String timerName, Token token) { - // TODO make the scanning of timers for cancellation optional by only deleting the timerjobs in the db. // of course, the corresponding ejb timer notifications have to be ignored. - log.debug("cancelling timers with name "+timerName+" from the ejb timer service"); + Collection timers = sessionContext.getTimerService().getTimers(); + log.debug("examining " + timers.size() + " ejb timer(s) by name '" + timerName + "' for " + token); - TimerService timerService = sessionContext.getTimerService(); - Iterator iter = timerService.getTimers().iterator(); - while (iter.hasNext()) { - javax.ejb.Timer ejbTimer = (javax.ejb.Timer) iter.next(); + int count = 0; + for (Iterator i = timers.iterator(); i.hasNext();) { + javax.ejb.Timer ejbTimer = (javax.ejb.Timer) i.next(); if (ejbTimer.getInfo() instanceof TimerInfo) { TimerInfo timerInfo = (TimerInfo) ejbTimer.getInfo(); if (timerInfo.matchesName(timerName, token)) { ejbTimer.cancel(); + ++count; } } } + log.debug("canceled " + count + " ejb timer(s) by name '" + timerName + "' for " + token); } - public void deleteTimersForProcessInstance(ProcessInstance processInstance) { - + public void cancelTimersForProcessInstance(ProcessInstance processInstance) { // TODO make the scanning of timers for cancellation optional by only deleting the timerjobs in the db. // of course, the corresponding ejb timer notifications have to be ignored. + + Collection timers = sessionContext.getTimerService().getTimers(); + log.debug("examining " + timers.size() + " timer(s) for " + processInstance); - log.debug("deleting timers for process instance "+processInstance+" from the ejb timer service"); - - TimerService timerService = sessionContext.getTimerService(); - Iterator iter = timerService.getTimers().iterator(); - while (iter.hasNext()) { - javax.ejb.Timer ejbTimer = (javax.ejb.Timer) iter.next(); + int count = 0; + for (Iterator i = timers.iterator(); i.hasNext();) { + javax.ejb.Timer ejbTimer = (javax.ejb.Timer) i.next(); if (ejbTimer.getInfo() instanceof TimerInfo) { TimerInfo timerInfo = (TimerInfo) ejbTimer.getInfo(); if (timerInfo.matchesProcessInstance(processInstance)) { ejbTimer.cancel(); + ++count; } } } + log.debug("canceled " + count + " ejb timer(s) for " + processInstance); } public void ejbTimeout(javax.ejb.Timer ejbTimer) { - log.debug("ejb timer "+ejbTimer+" fires"); + log.debug(ejbTimer + " fired"); String localCommandServiceJndiName = "java:comp/env/ejb/LocalCommandServiceBean"; try { Context initial = new InitialContext(); @@ -146,6 +171,6 @@ } public void ejbRemove() throws EJBException, RemoteException { } - + private static Log log = LogFactory.getLog(TimerServiceBean.class); } Modified: jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/ejb-jar.xml =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/ejb-jar.xml 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/ejb-jar.xml 2008-09-06 00:28:41 UTC (rev 2134) @@ -213,6 +213,9 @@ <cmp-field> <field-name>processInstanceId</field-name> </cmp-field> + <cmp-field> + <field-name>discriminator</field-name> + </cmp-field> <primkey-field>timerId</primkey-field> <ejb-local-ref> <description> @@ -235,7 +238,7 @@ <ejb-ql><![CDATA[ select object(obj) from TimerEntityBean obj - where obj.tokenId = ?1 + where obj.tokenId = ?1 and obj.discriminator = 84 ]]></ejb-ql> </query> <query> @@ -250,7 +253,7 @@ <ejb-ql><![CDATA[ select object(obj) from TimerEntityBean obj - where obj.tokenId = ?1 and obj.name = ?2 + where obj.tokenId = ?1 and obj.name = ?2 and obj.discriminator = 84 ]]></ejb-ql> </query> <query> @@ -264,7 +267,7 @@ <ejb-ql><![CDATA[ select object(obj) from TimerEntityBean obj - where obj.processInstanceId = ?1 + where obj.processInstanceId = ?1 and obj.discriminator = 84 ]]></ejb-ql> </query> </entity> Modified: jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/jbosscmp-jdbc.xml =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/jbosscmp-jdbc.xml 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/main/resources/META-INF/jbosscmp-jdbc.xml 2008-09-06 00:28:41 UTC (rev 2134) @@ -31,6 +31,10 @@ <field-name>processInstanceId</field-name> <column-name>PROCESSINSTANCE_</column-name> </cmp-field> + <cmp-field> + <field-name>discriminator</field-name> + <column-name>CLASS_</column-name> + </cmp-field> </entity> </enterprise-beans> </jbosscmp-jdbc> \ No newline at end of file Added: jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/test/EventCallback.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/test/EventCallback.java (rev 0) +++ jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/test/EventCallback.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -0,0 +1,111 @@ +/* + * 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.enterprise.test; + +import java.io.Serializable; + +import javax.transaction.Status; +import javax.transaction.Synchronization; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.jbpm.JbpmContext; +import org.jbpm.graph.def.Event; + +public class EventCallback implements Serializable { + + private static final long serialVersionUID = 1L; + private static final Log log = LogFactory.getLog(EventCallback.class); + + public void processStart() { + registerNotification(Event.EVENTTYPE_PROCESS_START); + } + + public void processEnd() { + registerNotification(Event.EVENTTYPE_PROCESS_END); + } + + public void nodeEnter() { + registerNotification(Event.EVENTTYPE_NODE_ENTER); + } + + public void nodeLeave() { + registerNotification(Event.EVENTTYPE_NODE_LEAVE); + } + + public void taskCreate() { + registerNotification(Event.EVENTTYPE_TASK_CREATE); + } + + public void taskEnd() { + registerNotification(Event.EVENTTYPE_TASK_END); + } + + public void timerCreate() { + registerNotification(Event.EVENTTYPE_TIMER_CREATE); + } + + public void timer() { + registerNotification(Event.EVENTTYPE_TIMER); + } + + private static void registerNotification(final String event) { + Synchronization notification = new Synchronization() { + + public void beforeCompletion() {} + + public void afterCompletion(int status) { + if (status == Status.STATUS_COMMITTED) { + log.info("notifying " + event); + synchronized (event) { + event.notify(); + } + } + } + + }; + JbpmContext.getCurrentJbpmContext() + .getSession() + .getTransaction() + .registerSynchronization(notification); + } + + public static void waitForEvent(String event) { + waitForEvent(event, 10000); + } + + public static void waitForEvent(String event, long timeout) { + synchronized (event) { + try { + long startTime = System.currentTimeMillis(); + event.wait(timeout); + if (System.currentTimeMillis() - startTime >= timeout) { + log.warn("event '" + event + "' took longer than " + timeout + " ms to occur"); + } + } + catch (InterruptedException e) { + // reassert interruption + Thread.currentThread().interrupt(); + } + } + } +} \ No newline at end of file Modified: jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerTest.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerTest.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/scheduler/ejbtimer/EjbSchedulerTest.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -21,11 +21,17 @@ */ package org.jbpm.scheduler.ejbtimer; +import java.util.Collections; + +import javax.naming.Context; import javax.naming.InitialContext; import junit.framework.Test; import org.apache.cactus.ServletTestCase; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.jbpm.IntegrationTestSetup; import org.jbpm.JbpmContext; import org.jbpm.command.Command; @@ -33,252 +39,277 @@ import org.jbpm.command.StartProcessInstanceCommand; import org.jbpm.ejb.LocalCommandService; import org.jbpm.ejb.LocalCommandServiceHome; -import org.jbpm.enterprise.test.ActionRecorder; +import org.jbpm.enterprise.test.EventCallback; import org.jbpm.enterprise.test.GetCurrentTime; -import org.jbpm.enterprise.test.Recorder; +import org.jbpm.graph.def.Event; import org.jbpm.graph.def.ProcessDefinition; import org.jbpm.graph.exe.ProcessInstance; import org.jbpm.graph.exe.Token; public class EjbSchedulerTest extends ServletTestCase { - + private LocalCommandService commandService; private static LocalCommandServiceHome commandServiceHome; + private static final Log log = LogFactory.getLog(EjbSchedulerTest.class); + public static Test suite() throws Exception { - return new IntegrationTestSetup(EjbSchedulerTest.class, "enterprise-test.war, enterprise-test.jar"); + return new IntegrationTestSetup(EjbSchedulerTest.class, + "enterprise-test.war, enterprise-test.jar"); } - + protected void setUp() throws Exception { if (commandServiceHome == null) { - InitialContext initialContext = new InitialContext(); - commandServiceHome = (LocalCommandServiceHome) initialContext.lookup("java:comp/env/ejb/CommandServiceBean"); + Context initialContext = new InitialContext(); + try { + commandServiceHome = (LocalCommandServiceHome) initialContext.lookup("java:comp/env/ejb/CommandServiceBean"); + } + finally { + initialContext.close(); + } } commandService = commandServiceHome.create(); + log.info("### " + getName() + " started ###"); } protected void tearDown() throws Exception { + log.info("### " + getName() + " done ###"); commandService = null; } public void testScheduleFuture() throws Exception { - Recorder.resetCollections(); - deployProcess( - "<process-definition name='timer included'>" + - " <start-state>" + - " <transition to='a' />" + - " </start-state>" + - " <state name='a'>" + - " <timer duedate='1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - "</process-definition>"); - launchProcess("timer included"); - assertEquals(0, Recorder.executions); - Thread.sleep(500); - assertEquals(0, Recorder.executions); - Thread.sleep(1000); - assertEquals(1, Recorder.executions); + deployProcess("<process-definition name='future'>" + + " <event type='process-end'>" + + " <action expression='#{eventCallback.processEnd}'/>" + + " </event>" + + " <start-state name='start'>" + + " <transition to='a' />" + + " </start-state>" + + " <state name='a'>" + + " <timer duedate='1 second' transition='timeout' />" + + " <transition name='timeout' to='end' />" + + " </state>" + + " <end-state name='end' />" + + "</process-definition>"); + long processId = launchProcess("future").getId(); + EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END); + assertTrue(isProcessFinished(processId)); } public void testSchedulePast() throws Exception { - Recorder.resetCollections(); - deployProcess( - "<process-definition name='timer included'>" + - " <start-state>" + - " <transition to='a'>" + - " <action class='" + GetCurrentTime.class.getName() + "'>" + - " <variable>now</variable>" + - " </action>" + - " </transition>" + - " </start-state>" + - " <state name='a'>" + - " <timer duedate='#{now} - 1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - "</process-definition>"); - launchProcess("timer included"); - assertEquals(0, Recorder.executions); - Thread.sleep(500); - assertEquals(1, Recorder.executions); + deployProcess("<process-definition name='past'>" + + " <event type='process-end'>" + + " <action expression='#{eventCallback.processEnd}'/>" + + " </event>" + + " <start-state name='start'>" + + " <transition to='a'>" + + " <action class='" + + GetCurrentTime.class.getName() + + "'>" + + " <variable>now</variable>" + + " </action>" + + " </transition>" + + " </start-state>" + + " <state name='a'>" + + " <timer duedate='#{now} - 10 seconds' transition='timeout' />" + + " <transition name='timeout' to='end' />" + + " </state>" + + " <end-state name='end' />" + + "</process-definition>"); + long processId = launchProcess("past").getId(); + EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END); + assertTrue(isProcessFinished(processId)); } public void testScheduleRepeat() throws Exception { - Recorder.resetCollections(); - deployProcess( - "<process-definition name='repeat included'>" + - " <start-state>" + - " <transition to='a' />" + - " </start-state>" + - " <state name='a'>" + - " <timer duedate='1 second' repeat='2 seconds'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - "</process-definition>"); - long tokenId = launchProcess("repeat included").getRootToken().getId(); - assertEquals(0, Recorder.executions); - Thread.sleep(500); - assertEquals(0, Recorder.executions); - Thread.sleep(1000); - assertEquals(1, Recorder.executions); - Thread.sleep(1000); - assertEquals(1, Recorder.executions); - Thread.sleep(1000); - assertEquals(2, Recorder.executions); - cancelTimer("a", tokenId); + deployProcess("<process-definition name='repeat'>" + + " <event type='timer'>" + + " <action expression='#{eventCallback.timer}'/>" + + " </event>" + + " <start-state name='start'>" + + " <transition to='a' />" + + " </start-state>" + + " <state name='a'>" + + " <timer duedate='1 second' repeat='1 second' />" + + " <transition to='end' />" + + " </state>" + + " <end-state name='end' />" + + "</process-definition>"); + long processId = launchProcess("repeat").getId(); + for (int i = 0; i < 3; i++) { + EventCallback.waitForEvent(Event.EVENTTYPE_TIMER); + assertEquals("a", getProcessState(processId)); + } + signalProcess(processId); + assertTrue(isProcessFinished(processId)); } public void testCancel() throws Exception { - Recorder.resetCollections(); - deployProcess( - "<process-definition name='repeat included'>" + - " <start-state>" + - " <transition to='a' />" + - " </start-state>" + - " <state name='a'>" + - " <timer duedate='1 second' repeat='1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - "</process-definition>"); - long tokenId = launchProcess("repeat included").getRootToken().getId(); - assertEquals(0, Recorder.executions); - Thread.sleep(500); - assertEquals(0, Recorder.executions); - Thread.sleep(1000); - assertEquals(1, Recorder.executions); - cancelTimer("a", tokenId); - Thread.sleep(1000); - assertEquals(1, Recorder.executions); - Thread.sleep(1000); - assertEquals(1, Recorder.executions); + deployProcess("<process-definition name='cancel'>" + + " <event type='timer'>" + + " <action expression='#{eventCallback.timer}'/>" + + " </event>" + + " <start-state name='start'>" + + " <transition to='a' />" + + " </start-state>" + + " <state name='a'>" + + " <timer duedate='1 second' repeat='1 second' />" + + " <transition to='end' />" + + " </state>" + + " <end-state name='end' />" + + "</process-definition>"); + ProcessInstance process = launchProcess("cancel"); + long processId = process.getId(); + // first expiration + EventCallback.waitForEvent(Event.EVENTTYPE_TIMER); + assertEquals("a", getProcessState(processId)); + // repeated expiration + EventCallback.waitForEvent(Event.EVENTTYPE_TIMER); + assertEquals("a", getProcessState(processId)); + cancelTimer("a", process.getRootToken().getId()); + // no more expirations + EventCallback.waitForEvent(Event.EVENTTYPE_TIMER, 2000); + signalProcess(processId); + assertTrue(isProcessFinished(processId)); } public void testScheduleMultiple() throws Exception { - Recorder.resetCollections(); - deployProcess( - "<process-definition name='timers included'>" + - " <start-state>" + - " <transition to='a' />" + - " </start-state>" + - " <state name='a'>" + - " <timer duedate='500 milliseconds' transition='next'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " <transition name='next' to='b' />" + - " </state>" + - " <state name='b'>" + - " <timer duedate='500 milliseconds' transition='next'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " <transition name='next' to='c' />" + - " </state>" + - " <state name='c'>" + - " <timer duedate='500 milliseconds' transition='next'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " <transition name='next' to='d' />" + - " </state>" + - " <state name='d'>" + - " <timer duedate='500 milliseconds' transition='next'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " <transition name='next' to='e' />" + - " </state>" + - " <state name='e'>" + - " <timer duedate='500 milliseconds'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - "</process-definition>"); - launchProcess("timers included"); - assertEquals(0, Recorder.executions); - Thread.sleep(250); - assertEquals(0, Recorder.executions); - for (int i = 1; i <= 5; i++) { - Thread.sleep(500); - assertEquals(i, Recorder.executions); + deployProcess("<process-definition name='multiple'>" + + " <event type='process-end'>" + + " <action expression='#{eventCallback.processEnd}'/>" + + " </event>" + + " <event type='timer'>" + + " <action expression='#{eventCallback.timer}'/>" + + " </event>" + + " <start-state>" + + " <transition to='a' />" + + " </start-state>" + + " <state name='a'>" + + " <timer duedate='500 milliseconds' transition='next' />" + + " <transition name='next' to='b' />" + + " </state>" + + " <state name='b'>" + + " <timer duedate='500 milliseconds' transition='next' />" + + " <transition name='next' to='c' />" + + " </state>" + + " <state name='c'>" + + " <timer duedate='500 milliseconds' transition='next' />" + + " <transition name='next' to='d' />" + + " </state>" + + " <state name='d'>" + + " <timer duedate='500 milliseconds' transition='next' />" + + " <transition name='next' to='e' />" + + " </state>" + + " <state name='e'>" + + " <timer duedate='500 milliseconds' transition='next' />" + + " <transition name='next' to='end' />" + + " </state>" + + " <end-state name='end' />" + + "</process-definition>"); + long processId = launchProcess("multiple").getId(); + char state = 'b'; + for (int i = 0; i < 4; i++) { + EventCallback.waitForEvent(Event.EVENTTYPE_TIMER); + assertEquals(Character.toString(state++), getProcessState(processId)); } + EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END); + assertTrue(isProcessFinished(processId)); } public void testScheduleConcurrent() throws InterruptedException { - Recorder.resetCollections(); - deployProcess( - "<process-definition name='timers included'>" + - " <start-state>" + - " <transition to='f' />" + - " </start-state>" + - " <fork name='f'>" + - " <transition name='a' to='a' />" + - " <transition name='b' to='b' />" + - " <transition name='c' to='c' />" + - " <transition name='d' to='d' />" + - " <transition name='e' to='e' />" + - " </fork>" + - " <state name='a'>" + - " <timer duedate='1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - " <state name='b'>" + - " <timer duedate='1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - " <state name='c'>" + - " <timer duedate='1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - " <state name='d'>" + - " <timer duedate='1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - " <state name='e'>" + - " <timer duedate='1 second'>" + - " <action class='" + ActionRecorder.class.getName() + "' />" + - " </timer>" + - " </state>" + - "</process-definition>"); - launchProcess("timers included"); - assertEquals(0, Recorder.executions); - Thread.sleep(500); - assertEquals(0, Recorder.executions); - Thread.sleep(1000); - assertEquals(5, Recorder.executions); + deployProcess("<process-definition name='concurrent'>" + + " <event type='process-end'>" + + " <action expression='#{eventCallback.processEnd}'/>" + + " </event>" + + " <start-state>" + + " <transition to='f' />" + + " </start-state>" + + " <fork name='f'>" + + " <transition name='a' to='a' />" + + " <transition name='b' to='b' />" + + " <transition name='c' to='c' />" + + " <transition name='d' to='d' />" + + " <transition name='e' to='e' />" + + " </fork>" + + " <state name='a'>" + + " <timer duedate='0.25 seconds' transition='timeout' />" + + " <transition name='timeout' to='j' />" + + " </state>" + + " <state name='b'>" + + " <timer duedate='0.5 seconds' transition='timeout' />" + + " <transition name='timeout' to='j' />" + + " </state>" + + " <state name='c'>" + + " <timer duedate='0.75 seconds' transition='timeout' />" + + " <transition name='timeout' to='j' />" + + " </state>" + + " <state name='d'>" + + " <timer duedate='1 second' transition='timeout' />" + + " <transition name='timeout' to='j' />" + + " </state>" + + " <state name='e'>" + + " <timer duedate='1.25 second' transition='timeout' />" + + " <transition name='timeout' to='j' />" + + " </state>" + + " <join name='j'>" + + " <transition to='end' />" + + " </join>" + + " <end-state name='end' />" + + "</process-definition>"); + long processId = launchProcess("concurrent").getId(); + EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END); + assertTrue(isProcessFinished(processId)); } protected ProcessDefinition deployProcess(String xml) { return (ProcessDefinition) commandService.execute(new DeployProcessCommand(xml)); } - + protected ProcessInstance launchProcess(final String processName) { StartProcessInstanceCommand command = new StartProcessInstanceCommand(); command.setProcessName(processName); + command.setVariables(Collections.singletonMap("eventCallback", new EventCallback())); return (ProcessInstance) commandService.execute(command); } - protected void cancelTimer(String timerName, long tokenId) { - commandService.execute(new CancelTimerCommand(timerName, tokenId)); + protected void signalProcess(final long processId) { + commandService.execute(new Command() { + private static final long serialVersionUID = 1L; + public Object execute(JbpmContext jbpmContext) throws Exception { + jbpmContext.loadProcessInstanceForUpdate(processId).signal(); + return null; + } + }); } - protected static class CancelTimerCommand implements Command { - final String timerName; - final long tokenId; - private static final long serialVersionUID = 1L; - public CancelTimerCommand(String timerName, long tokenId) { - this.timerName = timerName; - this.tokenId = tokenId; - } - public Object execute(JbpmContext jbpmContext) throws Exception { - Token token = jbpmContext.loadToken(tokenId); - jbpmContext.getJobSession().cancelTimersByName(timerName, token); - return null; - } + protected String getProcessState(final long processId) { + return (String) commandService.execute(new Command() { + private static final long serialVersionUID = 1L; + public Object execute(JbpmContext jbpmContext) throws Exception { + return jbpmContext.loadProcessInstance(processId).getRootToken().getNode().getName(); + } + }); } + + protected boolean isProcessFinished(final long processId) { + Boolean isFinished = (Boolean) commandService.execute(new Command() { + private static final long serialVersionUID = 1L; + public Object execute(JbpmContext jbpmContext) throws Exception { + return jbpmContext.loadProcessInstance(processId).hasEnded(); + } + }); + return isFinished.booleanValue(); + } + + protected void cancelTimer(final String timerName, final long tokenId) { + commandService.execute(new Command() { + private static final long serialVersionUID = 1L; + public Object execute(JbpmContext jbpmContext) throws Exception { + Token token = jbpmContext.loadToken(tokenId); + jbpmContext.getServices().getSchedulerService().deleteTimersByName(timerName, token); + return null; + } + }); + } } Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/db/JobSession.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/db/JobSession.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/db/JobSession.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -2,12 +2,10 @@ import java.util.Collection; import java.util.Date; -import java.util.Iterator; import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.hibernate.LockMode; import org.hibernate.Query; import org.hibernate.Session; import org.jbpm.JbpmContext; @@ -90,21 +88,14 @@ if (job instanceof Timer) { Timer timer = (Timer) job; Action action = timer.getAction(); - if ( (action!=null) - && (! session.contains(action)) - ) { - log.debug("cascading timer save to timer action"); + if (action != null && !session.contains(action)) { + log.debug("cascading timer save to action"); session.save(action); } } } - public void reattachUnmodifiedJob(Job job) { - session.lock(job, LockMode.NONE); - } - public void deleteJob(Job job) { - log.debug("deleting "+job); session.delete(job); } @@ -117,6 +108,14 @@ } } + public Timer loadTimer(long timerId) { + try { + return (Timer) session.load(Timer.class, new Long(timerId)); + } catch (Exception e) { + log.error(e); + throw new JbpmException("couldn't load timer " + timerId, e); + } + } public Job getJob(long jobId) { try { @@ -151,17 +150,17 @@ } } - public void cancelTimersByName(String name, Token token) { + public void deleteTimersByName(String name, Token token) { try { - log.debug("canceling timers by name " + name + " for " + token); + log.debug("deleting timers by name '" + name + "' for " + token); Query query = session.getNamedQuery("JobSession.deleteTimersByName"); query.setString("name", name); query.setParameter("token", token); int entityCount = query.executeUpdate(); - log.debug(entityCount+" timers by name " + name + " for " + token + " were deleted"); + log.debug(entityCount + " timers by name '" + name + "' for " + token + " were deleted"); } catch (Exception e) { log.error(e); - throw new JbpmException("couldn't cancel timers by name '"+name+"' for '"+token+"'", e); + throw new JbpmException("couldn't delete timers by name '" + name + "' for " + token, e); } } Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/graph/node/Join.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/graph/node/Join.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/graph/node/Join.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -112,7 +112,8 @@ lockMode = LockMode.parse(parentLockMode); } log.debug("forcing version increment on parent token "+parentToken); - session.flush(); + // TODO why is a flush needed?! + // session.flush(); session.lock(parentToken, lockMode); } Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/Timer.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/Timer.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/Timer.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -103,10 +103,8 @@ } public String toString() { - StringBuffer buffer = new StringBuffer(); - buffer.append("timer"); - if ( (name!=null) || (dueDate!=null)) { - + StringBuffer buffer = new StringBuffer("Timer"); + if (name != null || dueDate != null) { buffer.append("("); if (name!=null) { buffer.append(name).append(","); Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutor.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutor.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutor.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -2,6 +2,7 @@ import java.io.Serializable; import java.net.InetAddress; +import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -35,19 +36,20 @@ Map monitoredJobIds = Collections.synchronizedMap(new HashMap()); boolean isStarted = false; - + + private static String hostName; + public synchronized void start() { if (! isStarted) { log.debug("starting thread group '"+name+"'..."); for (int i=0; i<nbrOfThreads; i++) { startThread(); } + lockMonitorThread = new LockMonitorThread(jbpmConfiguration, lockMonitorInterval, maxLockTime, lockBufferTime); isStarted = true; } else { log.debug("ignoring start: thread group '"+name+"' is already started'"); } - - lockMonitorThread = new LockMonitorThread(jbpmConfiguration, lockMonitorInterval, maxLockTime, lockBufferTime); } /** @@ -65,6 +67,7 @@ for (int i=0; i<nbrOfThreads; i++) { stoppedThreads.add(stopThread()); } + lockMonitorThread.deactivate(); isStarted = false; } else { log.debug("ignoring stop: thread group '"+name+"' not started"); @@ -78,6 +81,7 @@ Thread thread = (Thread) iter.next(); thread.join(); } + lockMonitorThread.join(); } protected synchronized void startThread() { @@ -99,12 +103,15 @@ return name + ":" + getHostName() + ":" + index; } - private String getHostName() { - try { - return InetAddress.getLocalHost().getHostAddress(); - } catch (Exception e) { - return "unknown"; + private static String getHostName() { + if (hostName == null) { + try { + hostName = InetAddress.getLocalHost().getHostAddress(); + } catch (UnknownHostException e) { + hostName = "127.0.0.1"; + } } + return hostName; } protected synchronized Thread stopThread() { Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -13,6 +13,7 @@ import org.jbpm.JbpmConfiguration; import org.jbpm.JbpmContext; import org.jbpm.db.JobSession; +import org.jbpm.graph.exe.ProcessInstance; import org.jbpm.job.Job; import org.jbpm.persistence.JbpmPersistenceException; import org.jbpm.svc.Services; @@ -97,7 +98,7 @@ Collection acquiredJobs; synchronized (jobExecutor) { log.debug("acquiring jobs for execution..."); - Collection jobsToLock = Collections.EMPTY_LIST; + List jobsToLock = Collections.EMPTY_LIST; JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); try { JobSession jobSession = jbpmContext.getJobSession(); @@ -106,12 +107,13 @@ Job job = jobSession.getFirstAcquirableJob(lockOwner); if (job!=null) { if (job.isExclusive()) { - log.debug("exclusive acquirable job found ("+job+"). querying for other exclusive jobs to lock them all in one tx..."); - List otherExclusiveJobs = jobSession.findExclusiveJobs(lockOwner, job.getProcessInstance()); - jobsToLock = otherExclusiveJobs; - log.debug("trying to obtain a process-instance exclusive locks for '"+otherExclusiveJobs+"'"); + log.debug("found exclusive " + job); + ProcessInstance processInstance = job.getProcessInstance(); + log.debug("finding other exclusive jobs for " + processInstance); + jobsToLock = jobSession.findExclusiveJobs(lockOwner, processInstance); + log.debug("trying to obtain exclusive locks on " + jobsToLock + " for " + processInstance); } else { - log.debug("trying to obtain a lock for '"+job+"'"); + log.debug("trying to obtain lock on " + job); jobsToLock = Collections.singletonList(job); } @@ -124,7 +126,7 @@ } // HACKY HACK : this is a workaround for a hibernate problem that is fixed in hibernate 3.2.1 - // TODO is this still needed? + // TODO remove this hack already? // if (job instanceof Timer) { // Hibernate.initialize(((Timer)job).getGraphElement()); // } @@ -158,15 +160,15 @@ job = jobSession.loadJob(job.getId()); try { - log.debug("executing job "+job); + log.debug("executing " + job); if (job.execute(jbpmContext)) { jobSession.deleteJob(job); } } catch (Exception e) { - log.debug("exception while executing '"+job+"'", e); - StringWriter sw = new StringWriter(); - e.printStackTrace(new PrintWriter(sw)); - job.setException(sw.toString()); + log.debug("exception while executing " + job, e); + StringWriter memoryWriter = new StringWriter(); + e.printStackTrace(new PrintWriter(memoryWriter)); + job.setException(memoryWriter.toString()); job.setRetries(job.getRetries()-1); } @@ -230,7 +232,8 @@ } /** - * Signals this thread to stop running. Execution should cease shortly afterwards. + * Indicates that this thread should stop running. + * Execution will cease shortly afterwards. */ public void deactivate() { if (isActive) { Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -83,9 +83,24 @@ } } + /** + * @deprecated As of jBPM 3.2.3, replaced by {@link #deactivate()} + */ public void setActive(boolean isActive) { - this.isActive = isActive; + if (isActive == false) + deactivate(); } + /** + * Indicates that this thread should stop running. + * Execution will cease shortly afterwards. + */ + public void deactivate() { + if (isActive) { + isActive = false; + interrupt(); + } + } + private static Log log = LogFactory.getLog(LockMonitorThread.class); } Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/jpdl/el/parser/ELParserTokenManager.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/jpdl/el/parser/ELParserTokenManager.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/jpdl/el/parser/ELParserTokenManager.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -121,7 +121,7 @@ if (curChar < 64) { long l = 1L << curChar; - MatchLoop: do + do { switch(jjstateSet[--i]) { @@ -160,7 +160,7 @@ else if (curChar < 128) { long l = 1L << (curChar & 077); - MatchLoop: do + do { switch(jjstateSet[--i]) { @@ -188,7 +188,7 @@ long l1 = 1L << (hiByte & 077); int i2 = (curChar & 0xff) >> 6; long l2 = 1L << (curChar & 077); - MatchLoop: do + do { switch(jjstateSet[--i]) { @@ -541,7 +541,7 @@ if (curChar < 64) { long l = 1L << curChar; - MatchLoop: do + do { switch(jjstateSet[--i]) { @@ -710,7 +710,7 @@ else if (curChar < 128) { long l = 1L << (curChar & 077); - MatchLoop: do + do { switch(jjstateSet[--i]) { @@ -790,7 +790,7 @@ long l1 = 1L << (hiByte & 077); int i2 = (curChar & 0xff) >> 6; long l2 = 1L << (curChar & 077); - MatchLoop: do + do { switch(jjstateSet[--i]) { Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/msg/db/DbMessageService.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/msg/db/DbMessageService.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/msg/db/DbMessageService.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -21,11 +21,8 @@ */ package org.jbpm.msg.db; -import java.util.Collection; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.jbpm.JbpmConfiguration; import org.jbpm.JbpmContext; import org.jbpm.JbpmException; import org.jbpm.db.JobSession; Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/SchedulerService.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/SchedulerService.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/SchedulerService.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -21,17 +21,15 @@ */ package org.jbpm.scheduler; -import java.io.Serializable; - import org.jbpm.graph.exe.ProcessInstance; import org.jbpm.graph.exe.Token; import org.jbpm.job.Timer; import org.jbpm.svc.Service; -public interface SchedulerService extends Service, Serializable { +public interface SchedulerService extends Service { void createTimer(Timer timer); + void deleteTimer(Timer timer); void deleteTimersByName(String timerName, Token token); void deleteTimersByProcessInstance(ProcessInstance processInstance); - void close(); } Modified: jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/db/DbSchedulerService.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/db/DbSchedulerService.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/main/java/org/jbpm/scheduler/db/DbSchedulerService.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -56,14 +56,15 @@ hasProducedJobs = true; } + public void deleteTimer(Timer timer) { + jobSession.deleteJob(timer); + } + public void deleteTimersByName(String timerName, Token token) { - jobSession.cancelTimersByName(timerName, token); + jobSession.deleteTimersByName(timerName, token); } public void deleteTimersByProcessInstance(ProcessInstance processInstance) { - if (processInstance==null) { - throw new JbpmException("couldn't cancel timers for null process instance"); - } jobSession.deleteJobsForProcessInstance(processInstance); } Modified: jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/mail/TaskMailTest.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/mail/TaskMailTest.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/mail/TaskMailTest.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -84,7 +84,7 @@ assertTrue(-1!=email.getBody().indexOf("http://localhost:8080/jbpm/home?taskId=0")); } - public static class TestSchedulerService implements SchedulerService, Service, ServiceFactory { + public static class TestSchedulerService implements SchedulerService, ServiceFactory { private static final long serialVersionUID = 1L; List createdTimers = new ArrayList(); @@ -93,11 +93,12 @@ public void createTimer(Timer timer) { createdTimers.add(timer); } - + public void deleteTimer(Timer timer) { + cancelledTimers.add(timer.getName()); + } public void deleteTimersByName(String timerName, Token token) { cancelledTimers.add(timerName); } - public void deleteTimersByProcessInstance(ProcessInstance processInstance) { } public Service openService() { @@ -105,6 +106,7 @@ } public void close() { } + } public void testTaskInstanceReminder() throws Exception { Modified: jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/scheduler/exe/SchedulerTest.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/scheduler/exe/SchedulerTest.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/scheduler/exe/SchedulerTest.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -59,6 +59,9 @@ public void createTimer(Timer timer) { createdTimers.add(timer); } + public void deleteTimer(Timer timer) { + cancelledTimersByName.add(new Object[]{timer.getName(), timer.getToken()}); + } public void deleteTimersByName(String timerName, Token token) { cancelledTimersByName.add(new Object[]{timerName, token}); } Modified: jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/svc/ServicesTest.java =================================================================== --- jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/svc/ServicesTest.java 2008-09-05 16:51:08 UTC (rev 2133) +++ jbpm3/trunk/modules/jpdl/core/src/test/java/org/jbpm/svc/ServicesTest.java 2008-09-06 00:28:41 UTC (rev 2134) @@ -20,6 +20,7 @@ * 02110-1301 USA, or see the FSF site: http://www.fsf.org. */ package org.jbpm.svc; + import java.security.AccessControlException; import java.security.Permission; import java.util.HashMap; @@ -27,7 +28,6 @@ import junit.framework.TestCase; -import org.hibernate.SessionFactory; import org.jbpm.db.ContextSession; import org.jbpm.db.GraphSession; import org.jbpm.db.JobSession; @@ -160,7 +160,6 @@ public void setLoggingSession(LoggingSession loggingSession) {} public void setJobSession(JobSession jobSession) {} public void setTaskMgmtSession(TaskMgmtSession taskMgmtSession) {} - public void setSessionFactory(SessionFactory sessionFactory) {} }; serviceFactories.put("persistence", new TestGivenServiceFactory(service)); Services services = new Services(serviceFactories); @@ -171,6 +170,7 @@ SchedulerService schedulerService = new SchedulerService() { private static final long serialVersionUID = 1L; public void createTimer(Timer timer) {} + public void deleteTimer(Timer timer) {} public void deleteTimersByName(String timerName, Token token) {} public void deleteTimersByProcessInstance(ProcessInstance processInstance) {} public void close() {}