Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/job/MessageImpl.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/job/MessageImpl.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/job/MessageImpl.java	(working copy)
@@ -36,10 +36,6 @@
   public MessageImpl() {
   }
 
-  public String toString() {
-    return "message[" + dbid + "]";
-  }
-
   public MessageImpl(ExecutionImpl execution) {
     this.execution = execution;
     this.processInstance = execution.getProcessInstance();
Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/AcquireJobsCmd.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/AcquireJobsCmd.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/AcquireJobsCmd.java	(working copy)
@@ -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) {
Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAddedNotification.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAddedNotification.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAddedNotification.java	(working copy)
@@ -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() {
Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java	(revision 0)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java	(revision 0)
@@ -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;
+  }
+}

Property changes on: modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobAdditionNotifier.java
___________________________________________________________________
Added: svn:eol-style
   + native
Added: svn:keywords
   + Id Revision

Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorMessageSession.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorMessageSession.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorMessageSession.java	(working copy)
@@ -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();
     }
   }
 }
Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorTimerSession.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorTimerSession.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutorTimerSession.java	(working copy)
@@ -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();
     }
   }
 
Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/model/op/ExecuteEventListenerMessage.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/model/op/ExecuteEventListenerMessage.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/model/op/ExecuteEventListenerMessage.java	(working copy)
@@ -152,4 +152,9 @@
 
     execution.performAtomicOperationSync(AtomicOperation.EXECUTE_EVENT_LISTENER);
   }
+
+  @Override
+  public String toString() {
+    return "ExecuteEventListenerMessage[" + dbid + "]";
+  }
 }
Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/MessageSessionBinding.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/MessageSessionBinding.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/MessageSessionBinding.java	(working copy)
@@ -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;
Index: modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/TimerSessionBinding.java
===================================================================
--- modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/TimerSessionBinding.java	(revision 6438)
+++ modules/pvm/src/main/java/org/jbpm/pvm/internal/wire/binding/TimerSessionBinding.java	(working copy)
@@ -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;
+  }
 }
 
Index: modules/test-base/src/main/java/org/jbpm/test/JbpmCustomCfgTestCase.java
===================================================================
--- modules/test-base/src/main/java/org/jbpm/test/JbpmCustomCfgTestCase.java	(revision 6438)
+++ modules/test-base/src/main/java/org/jbpm/test/JbpmCustomCfgTestCase.java	(working copy)
@@ -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();
   }
-  
-
 
 }
Index: modules/test-base/src/main/java/org/jbpm/test/JbpmTestCase.java
===================================================================
--- modules/test-base/src/main/java/org/jbpm/test/JbpmTestCase.java	(revision 6464)
+++ modules/test-base/src/main/java/org/jbpm/test/JbpmTestCase.java	(working copy)
@@ -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();
 
-  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();
+  }
 
-      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();
+      }
+    });
+  }
 }
Index: modules/test-cfg/src/test/java/org/jbpm/test/idgenerator/DeploymentIdGenerationTest.java
===================================================================
--- modules/test-cfg/src/test/java/org/jbpm/test/idgenerator/DeploymentIdGenerationTest.java	(revision 6458)
+++ modules/test-cfg/src/test/java/org/jbpm/test/idgenerator/DeploymentIdGenerationTest.java	(working copy)
@@ -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
Index: modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java
===================================================================
--- modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java	(revision 0)
+++ modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java	(revision 0)
@@ -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</a>
+ */
+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);
+    }
+  }
+}

Property changes on: modules/test-cfg/src/test/java/org/jbpm/test/jobexecutor/JobNotificationTest.java
___________________________________________________________________
Added: svn:eol-style
   + native
Added: svn:keywords
   + Id Revision

Index: modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml
===================================================================
--- modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml	(revision 0)
+++ modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml	(revision 0)
@@ -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>

Property changes on: modules/test-cfg/src/test/resources/org/jbpm/test/jobexecutor/jbpm.cfg.xml
___________________________________________________________________
Added: svn:eol-style
   + native
Added: svn:keywords
   + Id Revision

Index: modules/test-db/src/test/java/org/jbpm/test/history/HistoryTaskDetailTest.java
===================================================================
--- modules/test-db/src/test/java/org/jbpm/test/history/HistoryTaskDetailTest.java	(revision 6471)
+++ modules/test-db/src/test/java/org/jbpm/test/history/HistoryTaskDetailTest.java	(working copy)
@@ -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;
     }
   }
 
Index: modules/test-load/src/test/java/org/jbpm/test/load/async/JobExecutorTestCase.java
===================================================================
--- modules/test-load/src/test/java/org/jbpm/test/load/async/JobExecutorTestCase.java	(revision 6445)
+++ modules/test-load/src/test/java/org/jbpm/test/load/async/JobExecutorTestCase.java	(working copy)
@@ -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;
-      }
-    });
-  }
-
 }
