[jbpm-commits] JBoss JBPM SVN: r6472 - in jbpm4/trunk/modules: pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor and 10 other directories.

Author: rebody Date: 2010-07-08 01:28:43 -0400 (Thu, 08 Jul 2010) New Revision: 6472 Added: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/ jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java jbpm4/trunk/modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/ jbpm4/trunk/modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/job/MessageImpl.java jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/AcquireJobsCmd.java jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAddedNotification.java jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorMessageSession.java jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorTimerSession.java jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/model/op/ExecuteEventListenerMessage.java jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/MessageSessionBinding.java jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/TimerSessionBinding.java jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmCustomCfgTestCase.java jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmTestCase.java jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/idgenerator/DeploymentIdGenerationTest.java jbpm4/trunk/modules/test-db/src/test/java/org/jbpm/test/history/HistoryTaskDetailTest.java jbpm4/trunk/modules/test-load/src/test/java/org/jbpm/test/load/async/JobExecutorTestCase.java Log: JBPM-1453 merge job notification booleans Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/job/MessageImpl.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/job/MessageImpl.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/job/MessageImpl.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -36,10 +36,6 @@ public MessageImpl() { } - public String toString() { - return "message[" + dbid + "]"; - } - public MessageImpl(ExecutionImpl execution) { this.execution = execution; this.processInstance = execution.getProcessInstance(); Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/AcquireJobsCmd.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/AcquireJobsCmd.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/AcquireJobsCmd.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -25,8 +25,8 @@ import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.Date; -import java.util.List; import org.jbpm.api.cmd.Command; import org.jbpm.api.cmd.Environment; @@ -54,7 +54,7 @@ Collection<Long> acquiredJobDbids = new ArrayList<Long>(); try { - Collection<JobImpl> acquiredJobs = new ArrayList<JobImpl>(); + Collection<JobImpl> acquiredJobs; DbSession dbSession = environment.get(DbSession.class); if (log.isTraceEnabled()) log.trace("start querying first acquirable job..."); @@ -64,10 +64,9 @@ if (job!=null) { if (job.isExclusive()) { if (log.isTraceEnabled()) log.trace("exclusive acquirable job found ("+job+"). querying for other exclusive jobs to lock them all in one tx..."); - List<JobImpl> otherExclusiveJobs = dbSession.findExclusiveJobs(job.getProcessInstance()); - acquiredJobs.addAll(otherExclusiveJobs); + acquiredJobs = dbSession.findExclusiveJobs(job.getProcessInstance()); } else { - acquiredJobs.add(job); + acquiredJobs = Collections.singletonList(job); } for (JobImpl acquiredJob: acquiredJobs) { Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAddedNotification.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAddedNotification.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAddedNotification.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -21,6 +21,7 @@ */ package org.jbpm.pvm.internal.jobexecutor; +import javax.transaction.Status; import javax.transaction.Synchronization; import org.jbpm.internal.log.Log; @@ -41,9 +42,11 @@ this.jobExecutor = jobExecutor; } - public void afterCompletion(int arg0) { - log.trace("notifying job executor of added message"); - jobExecutor.jobWasAdded(); + public void afterCompletion(int status) { + if (status == Status.STATUS_COMMITTED) { + log.trace("notifying job executor of added message"); + jobExecutor.jobWasAdded(); + } } public void beforeCompletion() { Added: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java (rev 0) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -0,0 +1,65 @@ +/* + * 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.pvm.internal.jobexecutor; + +import org.jbpm.internal.log.Log; +import org.jbpm.pvm.internal.tx.Transaction; + +/** + * @author Huisheng Xu + */ +public class JobAdditionNotifier { + + private static final Log log = Log.getLog(JobAdditionNotifier.class.getName()); + + /* injected */ + Transaction transaction; + + /* injected */ + JobExecutor jobExecutor; + + private boolean notificationRegistered; + + public void registerNotification() { + if (jobExecutor == null) { + log.debug("cannot find jobExecutor"); + return; + } + + // a transaction is not required (can be null) + if (transaction == null) { + log.debug("cannot find transaction"); + return; + } + + if (notificationRegistered) + return; + + // notify the job executor after the transaction is completed + if (log.isTraceEnabled()) { + log.trace("registering job addition notifier with " + transaction); + } + transaction.registerSynchronization(new JobAddedNotification(jobExecutor)); + + notificationRegistered = true; + } +} Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorMessageSession.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorMessageSession.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorMessageSession.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -23,41 +23,32 @@ import org.jbpm.api.job.Message; import org.jbpm.internal.log.Log; -import org.jbpm.pvm.internal.env.EnvironmentImpl; import org.jbpm.pvm.internal.session.DbSession; import org.jbpm.pvm.internal.session.MessageSession; -import org.jbpm.pvm.internal.tx.Transaction; -import org.jbpm.pvm.internal.util.ReflectUtil; /** * @author Tom Baeyens + * @author Huisheng Xu */ public class JobExecutorMessageSession implements MessageSession { - + private static final Log log = Log.getLog(JobExecutorMessageSession.class.getName()); - + /* injected */ DbSession dbSession; - /* injected */ - Transaction transaction; - - boolean isNotificationAdded; + /* injected. */ + JobAdditionNotifier jobAdditionNotifier; public void send(Message message) { - log.debug("sending message "+ReflectUtil.getUnqualifiedClassName(message.getClass())); + if (log.isDebugEnabled()) { + log.debug("sending message " + message.getClass().getSimpleName()); + } dbSession.save(message); - if (!isNotificationAdded) { - isNotificationAdded = true; - - JobExecutor jobExecutor = EnvironmentImpl.getCurrent().get(JobExecutor.class); - if (jobExecutor!=null) { - // notify the job executor after the transaction is completed - log.trace("registering job executor notifier with "+transaction); - transaction.registerSynchronization(new JobAddedNotification(jobExecutor)); - } + if (jobAdditionNotifier != null) { + jobAdditionNotifier.registerNotification(); } } } Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorTimerSession.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorTimerSession.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorTimerSession.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -24,48 +24,36 @@ import org.jbpm.internal.log.Log; import org.jbpm.pvm.internal.job.TimerImpl; import org.jbpm.pvm.internal.session.TimerSession; -import org.jbpm.pvm.internal.tx.Transaction; import org.jbpm.pvm.internal.util.CollectionUtil; /** * Timers created with this service are committed at the end of the transaction, * so their execution will be late if the delay is shorter than the transaction. * In that case, they will be executed at the end of the transaction. - * + * * @author Tom Baeyens, Pascal Verdage + * @author Huisheng Xu */ public class JobExecutorTimerSession implements TimerSession { private static final Log log = Log.getLog(TimerSession.class.getName()); /* injected */ - Transaction transaction; - - /* injected */ - JobExecutor jobExecutor; - - /* injected */ Session session; - boolean jobExecutorNotificationScheduled = false; + /* injected. */ + JobAdditionNotifier jobAdditionNotifier; public void schedule(Timer timer) { if (timer == null) throw new JbpmException("null timer scheduled"); TimerImpl timerImpl = (TimerImpl) timer; timerImpl.validate(); - + log.debug("scheduling " + timer); session.save(timer); - - if ( (!jobExecutorNotificationScheduled) - && (jobExecutor!=null) - ) { - jobExecutorNotificationScheduled = true; - - //a transaction is not required (can be null) - if (transaction != null) { - transaction.registerSynchronization(new JobAddedNotification(jobExecutor)); - } + + if (jobAdditionNotifier != null) { + jobAdditionNotifier.registerNotification(); } } Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/model/op/ExecuteEventListenerMessage.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/model/op/ExecuteEventListenerMessage.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/model/op/ExecuteEventListenerMessage.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -152,4 +152,9 @@ execution.performAtomicOperationSync(AtomicOperation.EXECUTE_EVENT_LISTENER); } + + @Override + public String toString() { + return "ExecuteEventListenerMessage[" + dbid + "]"; + } } Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/MessageSessionBinding.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/MessageSessionBinding.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/MessageSessionBinding.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -21,6 +21,8 @@ */ package org.jbpm.pvm.internal.wire.binding; +import org.w3c.dom.Element; + import org.jbpm.pvm.internal.jms.JmsMessageSession; import org.jbpm.pvm.internal.jobexecutor.JobExecutorMessageSession; import org.jbpm.pvm.internal.session.DbSession; @@ -29,16 +31,15 @@ import org.jbpm.pvm.internal.wire.descriptor.ContextTypeRefDescriptor; import org.jbpm.pvm.internal.wire.descriptor.JndiDescriptor; import org.jbpm.pvm.internal.wire.descriptor.ObjectDescriptor; -import org.jbpm.pvm.internal.wire.descriptor.TransactionRefDescriptor; import org.jbpm.pvm.internal.xml.Parse; import org.jbpm.pvm.internal.xml.Parser; -import org.w3c.dom.Element; /** parses a descriptor for creating a {@link MessageSession}. * * See schema docs for more details. * * @author Tom Baeyens + * @author Huisheng Xu */ public class MessageSessionBinding extends WireDescriptorBinding { @@ -72,8 +73,9 @@ } else { objectDescriptor.setClassName(JobExecutorMessageSession.class.getName()); - objectDescriptor.addInjection("transaction", new TransactionRefDescriptor()); objectDescriptor.addInjection("dbSession", new ContextTypeRefDescriptor(DbSession.class)); + objectDescriptor.addInjection("jobAdditionNotifier", + TimerSessionBinding.getJobAdditionNotifierDescriptor(parse)); } return objectDescriptor; Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/TimerSessionBinding.java =================================================================== --- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/TimerSessionBinding.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/TimerSessionBinding.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -22,23 +22,27 @@ package org.jbpm.pvm.internal.wire.binding; import org.hibernate.Session; +import org.w3c.dom.Element; + +import org.jbpm.pvm.internal.jobexecutor.JobAdditionNotifier; import org.jbpm.pvm.internal.jobexecutor.JobExecutor; import org.jbpm.pvm.internal.jobexecutor.JobExecutorTimerSession; import org.jbpm.pvm.internal.session.TimerSession; import org.jbpm.pvm.internal.util.XmlUtil; +import org.jbpm.pvm.internal.wire.WireDefinition; import org.jbpm.pvm.internal.wire.descriptor.ContextTypeRefDescriptor; import org.jbpm.pvm.internal.wire.descriptor.EnvDescriptor; import org.jbpm.pvm.internal.wire.descriptor.ObjectDescriptor; import org.jbpm.pvm.internal.wire.descriptor.TransactionRefDescriptor; import org.jbpm.pvm.internal.xml.Parse; import org.jbpm.pvm.internal.xml.Parser; -import org.w3c.dom.Element; /** parses a descriptor for creating a {@link TimerSession}. * * See schema docs for more details. * * @author Tom Baeyens, Pascal Verdage + * @author Huisheng Xu */ public class TimerSessionBinding extends WireDescriptorBinding { @@ -59,12 +63,31 @@ objectDescriptor.setClassName(JobExecutorTimerSession.class.getName()); // inject fields - objectDescriptor.addInjection("transaction", new TransactionRefDescriptor()); - objectDescriptor.addInjection("jobExecutor", new EnvDescriptor(JobExecutor.class)); objectDescriptor.addInjection("session", new ContextTypeRefDescriptor(Session.class)); + objectDescriptor.addInjection("jobAdditionNotifier", + getJobAdditionNotifierDescriptor(parse)); } return objectDescriptor; } + + static ObjectDescriptor getJobAdditionNotifierDescriptor(Parse parse) { + String jobAdditionNotifierName; + ObjectDescriptor jobAdditionNotifierDescriptor; + + WireDefinition wireDefinition = parse.contextStackFind(WireDefinition.class); + if (wireDefinition != null + && (jobAdditionNotifierName = wireDefinition.getDescriptorName(JobAdditionNotifier.class)) != null) { + jobAdditionNotifierDescriptor = (ObjectDescriptor) wireDefinition.getDescriptor(jobAdditionNotifierName); + } + else { + jobAdditionNotifierDescriptor = new ObjectDescriptor(JobAdditionNotifier.class); + // inject fields + jobAdditionNotifierDescriptor.addInjection("transaction", new TransactionRefDescriptor()); + jobAdditionNotifierDescriptor.addInjection("jobExecutor", new EnvDescriptor(JobExecutor.class)); + wireDefinition.addDescriptor(jobAdditionNotifierDescriptor); + } + return jobAdditionNotifierDescriptor; + } } Modified: jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmCustomCfgTestCase.java =================================================================== --- jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmCustomCfgTestCase.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmCustomCfgTestCase.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -22,41 +22,27 @@ package org.jbpm.test; import org.jbpm.api.Configuration; -import org.jbpm.api.RepositoryService; +import org.jbpm.api.ProcessEngine; - - /** * @author Tom Baeyens */ public class JbpmCustomCfgTestCase extends JbpmTestCase { - - protected synchronized void initialize() { - String cfgResource = getClass().getPackage().getName().replace(".", "/")+"/jbpm.cfg.xml"; - + protected ProcessEngine buildProcessEngine() { + String cfgResource = getClass().getPackage().getName().replace('.', '/') + "/jbpm.cfg.xml"; + // We can't call initialize(String, String) here, since it will do a null // check on the statically cached process engine. Since the test-cfg // is meant to test different configs (and hence different process engines) // this caching is unwanted, which means we create the process engine // ourselves here. - + if (log.isDebugEnabled()) { log.debug("building ProcessEngine from resource " + cfgResource); } - Configuration configuration = new Configuration().setResource(cfgResource); - - processEngine = configuration.buildProcessEngine(); - - repositoryService = processEngine.get(RepositoryService.class); - executionService = processEngine.getExecutionService(); - historyService = processEngine.getHistoryService(); - managementService = processEngine.getManagementService(); - taskService = processEngine.getTaskService(); - identityService = processEngine.getIdentityService(); + return new Configuration().setResource(cfgResource).buildProcessEngine(); } - - } Modified: jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmTestCase.java =================================================================== --- jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmTestCase.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/test-base/src/main/java/org/jbpm/test/JbpmTestCase.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -24,9 +24,14 @@ import java.util.ArrayList; import java.util.List; import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; import javax.jms.Session; +import org.hibernate.criterion.Projections; +import org.hibernate.criterion.Restrictions; + import org.jbpm.api.Configuration; import org.jbpm.api.Execution; import org.jbpm.api.ExecutionService; @@ -38,6 +43,8 @@ import org.jbpm.api.ProcessInstance; import org.jbpm.api.RepositoryService; import org.jbpm.api.TaskService; +import org.jbpm.api.cmd.Command; +import org.jbpm.api.cmd.Environment; import org.jbpm.api.task.Task; import org.jbpm.test.assertion.CollectionAssertions; @@ -61,14 +68,14 @@ */ public abstract class JbpmTestCase extends BaseJbpmTestCase { - protected static ProcessEngine processEngine; + protected ProcessEngine processEngine; - protected static RepositoryService repositoryService; - protected static ExecutionService executionService; - protected static ManagementService managementService; - protected static TaskService taskService; - protected static HistoryService historyService; - protected static IdentityService identityService; + protected RepositoryService repositoryService; + protected ExecutionService executionService; + protected ManagementService managementService; + protected TaskService taskService; + protected HistoryService historyService; + protected IdentityService identityService; /** registered deployments. registered deployments will be deleted automatically * in the tearDown. This is a convenience function as each test is expected to clean up the DB. */ @@ -77,20 +84,18 @@ @Override protected void setUp() throws Exception { super.setUp(); - initialize(); + processEngine = buildProcessEngine(); + + repositoryService = processEngine.get(RepositoryService.class); + executionService = processEngine.getExecutionService(); + historyService = processEngine.getHistoryService(); + managementService = processEngine.getManagementService(); + taskService = processEngine.getTaskService(); + identityService = processEngine.getIdentityService(); } - protected synchronized void initialize() { - if (processEngine==null) { - processEngine = Configuration.getProcessEngine(); - - repositoryService = processEngine.get(RepositoryService.class); - executionService = processEngine.getExecutionService(); - historyService = processEngine.getHistoryService(); - managementService = processEngine.getManagementService(); - taskService = processEngine.getTaskService(); - identityService = processEngine.getIdentityService(); - } + protected ProcessEngine buildProcessEngine() { + return Configuration.getProcessEngine(); } @Override @@ -342,4 +347,48 @@ return result.toString(); } + public void waitTillNoMoreMessages() { + final long timeout = 60 * 1000; + final long checkInterval = 1000; + + // 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 + TimerTask interruptTask = new TimerTask() { + + Thread testThread = Thread.currentThread(); + + public void run() { + log.debug("test " + getName() + " took too long. going to interrupt..." + testThread); + testThread.interrupt(); + } + }; + Timer timer = new Timer(); + timer.schedule(interruptTask, timeout); + + try { + for (int jobCount; (jobCount = getJobCount()) > 0;) { + log.debug("waiting " + checkInterval + " ms for " + jobCount + " jobs to execute"); + Thread.sleep(checkInterval); + } + } catch (InterruptedException e) { + fail("test execution exceeded treshold of " + timeout + " milliseconds"); + } finally { + timer.cancel(); + } + } + + protected int getJobCount() { + return processEngine.execute(new Command<Integer>() { + private static final long serialVersionUID = 1L; + + public Integer execute(Environment environment) { + org.hibernate.Session session = environment.get(org.hibernate.Session.class); + return (Integer) session.createCriteria("org.jbpm.pvm.internal.job.JobImpl") + .add(Restrictions.gt("retries", 0)) + .setProjection(Projections.rowCount()) + .uniqueResult(); + } + }); + } } Modified: jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/idgenerator/DeploymentIdGenerationTest.java =================================================================== --- jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/idgenerator/DeploymentIdGenerationTest.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/idgenerator/DeploymentIdGenerationTest.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -31,13 +31,10 @@ import org.jbpm.api.Deployment; import org.jbpm.api.NewDeployment; import org.jbpm.api.ProcessEngine; -import org.jbpm.api.ProcessInstance; import org.jbpm.pvm.internal.id.DbidGenerator; import org.jbpm.pvm.internal.id.MemoryDbidGenerator; -import org.jbpm.pvm.internal.repository.DeploymentImpl; import org.jbpm.test.JbpmCustomCfgTestCase; - /** * @author Joram Barrez */ @@ -97,10 +94,9 @@ private void resetProcessEngineAndDbidGenerator() { // Close the process engine and recreate it ProcessEngine oldProcessEngine = processEngine; - processEngine.close(); - processEngine = null; + oldProcessEngine.close(); - initialize(); // creates a new ProcessEngine and services + processEngine = buildProcessEngine(); // creates a new ProcessEngine and services assertNotSame(oldProcessEngine, processEngine); // Reset the in memory generator and redeploy the process Added: jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java =================================================================== --- jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java (rev 0) +++ jbpm4/trunk/modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -0,0 +1,98 @@ +/* + * 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.test.jobexecutor; + +import org.jbpm.api.cmd.Command; +import org.jbpm.api.cmd.Environment; +import org.jbpm.pvm.internal.jobexecutor.JobAdditionNotifier; +import org.jbpm.pvm.internal.jobexecutor.JobExecutor; +import org.jbpm.test.JbpmCustomCfgTestCase; + +/** + * Merge job notification booleans. + * + * @author Alejandro Guizar + * @see <a href="https://jira.jboss.org/browse/JBPM-1453">JBPM-1453<... + */ +public class JobNotificationTest extends JbpmCustomCfgTestCase { + + static int notifications; + + public void testJobNotification() { + JobExecutor jobExecutor = processEngine.get(JobExecutor.class); + assert jobExecutor.getNbrOfThreads() > 1 : jobExecutor.getNbrOfThreads(); + + deployJpdlXmlString("<process name='JobNotification'>" + + " <on event='timeout'>" + + " <timer duedate='1 second'/>" + + " <script expr='do nothing' />" + + " </on>" + + " <start>" + + " <transition to='f' />" + + " </start>" + + " <fork name='f'>" + + " <transition name='a' to='a' continue='async' />" + + " <transition name='b' to='b' continue='async' />" + + " </fork>" + + " <state name='a' />" + + " <state name='b' />" + + "</process>"); + + jobExecutor.start(); + try { + String processInstanceId = executionService.startProcessInstanceByKey("JobNotification") + .getId(); + waitTillNoMoreMessages(); + assertActivitiesActive(processInstanceId, "a", "b"); + } + finally { + jobExecutor.stop(true); + } + + GetEnvironmentObject<JobAdditionNotifier> command = new GetEnvironmentObject<JobAdditionNotifier>(JobAdditionNotifier.class); + JobAdditionNotifier notifier = processEngine.execute(command); + assertNotSame(notifier, processEngine.execute(command)); + } + + public static class Executor extends JobExecutor { + private static final long serialVersionUID = 1L; + + @Override + public void jobWasAdded() { + super.jobWasAdded(); + notifications++; + } + } + + private static class GetEnvironmentObject<T> implements Command<T> { + private final Class<T> type; + private static final long serialVersionUID = 1L; + + GetEnvironmentObject(Class<T> type) { + this.type = type; + } + + public T execute(Environment environment) throws Exception { + return environment.get(type); + } + } +} Added: jbpm4/trunk/modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml =================================================================== --- jbpm4/trunk/modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml (rev 0) +++ jbpm4/trunk/modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml 2010-07-08 05:28:43 UTC (rev 6472) @@ -0,0 +1,18 @@ +<?xml version="1.0" encoding="UTF-8"?> + +<jbpm-configuration> + + <import resource="jbpm.default.cfg.xml" /> + <import resource="jbpm.tx.hibernate.cfg.xml" /> + <import resource="jbpm.jpdl.cfg.xml" /> + <import resource="jbpm.bpmn.cfg.xml" /> + <import resource="jbpm.identity.cfg.xml" /> + <import resource="jbpm.businesscalendar.cfg.xml" /> + + <process-engine-context> + <object class="org.jbpm.test.jobexecutor.JobNotificationTest$Executor"> + <field name="commandService"><ref object="txRequiredCommandService"/></field> + </object> + </process-engine-context> + +</jbpm-configuration> Modified: jbpm4/trunk/modules/test-db/src/test/java/org/jbpm/test/history/HistoryTaskDetailTest.java =================================================================== --- jbpm4/trunk/modules/test-db/src/test/java/org/jbpm/test/history/HistoryTaskDetailTest.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/test-db/src/test/java/org/jbpm/test/history/HistoryTaskDetailTest.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -23,13 +23,13 @@ import java.util.Date; import java.util.List; -import org.jbpm.api.cmd.Command; + +import org.jbpm.api.TaskService; import org.jbpm.api.cmd.Environment; -import org.jbpm.api.history.HistoryActivityInstance; +import org.jbpm.api.cmd.VoidCommand; import org.jbpm.api.task.Task; import org.jbpm.test.JbpmTestCase; - /** * @author Huisheng Xu */ @@ -85,51 +85,57 @@ assertEquals(1, historyService.createHistoryDetailQuery().list().size()); } - static class TaskReassignCmd implements Command<Void> { + private static class TaskReassignCmd extends VoidCommand { private String taskId; private String newAssignee; - public TaskReassignCmd(String taskId, String newAssignee) { + private static final long serialVersionUID = 1L; + + TaskReassignCmd(String taskId, String newAssignee) { this.taskId = taskId; this.newAssignee = newAssignee; } - public Void execute(Environment env) { + @Override + protected void executeVoid(Environment environment) throws Exception { + TaskService taskService = environment.get(TaskService.class); Task task = taskService.getTask(taskId); task.setAssignee(newAssignee); taskService.saveTask(task); - - return null; } } - static class TaskChangePriorityCmd implements Command<Void> { + private static class TaskChangePriorityCmd extends VoidCommand { private String taskId; private int newPriority; - public TaskChangePriorityCmd(String taskId, int newPriority) { + private static final long serialVersionUID = 1L; + + TaskChangePriorityCmd(String taskId, int newPriority) { this.taskId = taskId; this.newPriority = newPriority; } - public Void execute(Environment env) { + @Override + protected void executeVoid(Environment environment) throws Exception { + TaskService taskService = environment.get(TaskService.class); Task task = taskService.getTask(taskId); task.setPriority(newPriority); taskService.saveTask(task); - - return null; } } - static class TaskChangeDuedateCmd implements Command<Void> { + private static class TaskChangeDuedateCmd extends VoidCommand { private String taskId; private Date newDuedate; - public TaskChangeDuedateCmd(String taskId, Date newDuedate) { + private static final long serialVersionUID = 1L; + + TaskChangeDuedateCmd(String taskId, Date newDuedate) { this.taskId = taskId; this.newDuedate = newDuedate; } - public Void execute(Environment env) { + @Override + protected void executeVoid(Environment environment) throws Exception { + TaskService taskService = environment.get(TaskService.class); Task task = taskService.getTask(taskId); task.setDuedate(newDuedate); taskService.saveTask(task); - - return null; } } Modified: jbpm4/trunk/modules/test-load/src/test/java/org/jbpm/test/load/async/JobExecutorTestCase.java =================================================================== --- jbpm4/trunk/modules/test-load/src/test/java/org/jbpm/test/load/async/JobExecutorTestCase.java 2010-07-07 17:22:03 UTC (rev 6471) +++ jbpm4/trunk/modules/test-load/src/test/java/org/jbpm/test/load/async/JobExecutorTestCase.java 2010-07-08 05:28:43 UTC (rev 6472) @@ -21,97 +21,22 @@ */ package org.jbpm.test.load.async; -import java.util.Date; -import java.util.Timer; -import java.util.TimerTask; - -import org.hibernate.Query; -import org.hibernate.Session; -import org.jbpm.api.cmd.Command; -import org.jbpm.api.cmd.Environment; import org.jbpm.pvm.internal.cmd.CommandService; -import org.jbpm.pvm.internal.job.JobImpl; import org.jbpm.pvm.internal.jobexecutor.JobExecutor; import org.jbpm.test.JbpmTestCase; - /** * @author Tom Baeyens */ public class JobExecutorTestCase extends JbpmTestCase { - long timeoutMillis = 20 * 1000; // 20 seconds - long checkInterval = 400; - - static String jobsAvailableQueryText = - "select count(*) "+ - "from "+JobImpl.class.getName()+" as job "+ - "where ( (job.dueDate is null) or (job.dueDate <= :now) ) "+ - " and ( job.retries > 0 )"; - protected CommandService commandService; protected JobExecutor jobExecutor; protected void setUp() throws Exception { super.setUp(); - + commandService = processEngine.get(CommandService.class); jobExecutor = processEngine.get(JobExecutor.class); } - - protected void waitTillNoMoreMessages() { - - // 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 - TimerTask interruptTask = new TimerTask() { - - Thread testThread = Thread.currentThread(); - - public void run() { - log.debug("test " + getName() + " took too long. going to interrupt..." + testThread); - testThread.interrupt(); - } - }; - Timer timer = new Timer(); - timer.schedule(interruptTask, timeoutMillis); - - try { - boolean jobsAvailable = true; - while (jobsAvailable) { - log.debug("going to sleep for " + checkInterval + " millis, waiting for the job executor to process more jobs"); - Thread.sleep(checkInterval); - jobsAvailable = areJobsAvailable(); - } - - } catch (InterruptedException e) { - fail("test execution exceeded treshold of " + timeoutMillis + " milliseconds"); - } finally { - timer.cancel(); - } - } - - public boolean areJobsAvailable() { - return commandService.execute(new Command<Boolean>() { - private static final long serialVersionUID = 1L; - - public Boolean execute(Environment environment) { - Session session = environment.get(Session.class); - - Query query = session.createQuery(jobsAvailableQueryText); - query.setDate("now", new Date()); - - Long jobs = (Long) query.uniqueResult(); - - if (jobs.longValue()>0) { - log.debug("found "+jobs+" more jobs to process"); - return true; - } - log.debug("no more jobs to process"); - - return false; - } - }); - } - }