svn commit: r723614 - /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java

[rajdavies]

Author: rajdavies
Date: Thu Dec  4 22:44:41 2008
New Revision: 723614

URL: http://svn.apache.org/viewvc?rev=723614&view=rev
Log:
Fix for https://issues.apache.org/activemq/browse/AMQ-2025

Modified:
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java?rev=723614&r1=723613&r2=723614&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java
 Thu Dec  4 22:44:41 2008
@@ -19,6 +19,7 @@
 
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.jms.JMSException;
 
@@ -45,6 +46,7 @@
     private boolean dispatchedBySessionPool;
     private TaskRunner taskRunner;
     private boolean startedOrWarnedThatNotStarted;
+    private AtomicBoolean taskRunnerCreated = new AtomicBoolean();
 
     ActiveMQSessionExecutor(ActiveMQSession session) {
         this.session = session;
@@ -88,15 +90,18 @@
         if (!dispatchedBySessionPool) {
             if (session.isSessionAsyncDispatch()) {
                 try {
-                    if (taskRunner == null) {
-                        taskRunner = 
session.connection.getSessionTaskRunner().createTaskRunner(this, "ActiveMQ 
Session: " + session.getSessionId());
+                    if (taskRunnerCreated.compareAndSet(false, true)) {
+                        if (taskRunner == null) {
+                            taskRunner = 
session.connection.getSessionTaskRunner().createTaskRunner(this,
+                                    "ActiveMQ Session: " + 
session.getSessionId());
+                        }
                     }
                     taskRunner.wakeup();
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                 }
             } else {
-                while (iterate()) {                    
+                while (iterate()) {
                 }
             }
         }
@@ -141,6 +146,7 @@
                 if (taskRunner != null) {
                     taskRunner.shutdown();
                     taskRunner = null;
+                    taskRunnerCreated.set(false);
                 }
             }
         } catch (InterruptedException e) {