[jboss-cvs] JBoss Messaging SVN: r5346 - trunk/src/main/org/jboss/messaging/core/client/impl.

[jboss-cvs-commits at lists.jboss.org]

Author: timfox
Date: 2008-11-13 03:21:44 -0500 (Thu, 13 Nov 2008)
New Revision: 5346

Modified:
   trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
Log:
Fixed race


Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java	2008-11-12 20:04:25 UTC (rev 5345)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java	2008-11-13 08:21:44 UTC (rev 5346)
@@ -69,6 +69,8 @@
 
    private volatile MessageHandler handler;
 
+   private volatile boolean closing;
+   
    private volatile boolean closed;
 
    private volatile int creditsToSend;
@@ -277,7 +279,7 @@
 
    public synchronized void handleMessage(final ClientMessage message) throws Exception
    {
-      if (closed)
+      if (closing)
       {
          // This is ok - we just ignore the message
          return;
@@ -416,7 +418,7 @@
 
    private void callOnMessage() throws Exception
    {
-      if (closed)
+      if (closing)
       {
          return;
       }
@@ -466,14 +468,19 @@
          {
             return;
          }
-
-         closed = true;
-
+             
+         //We need an extra flag closing, since we need to prevent any more messages getting queued to execute
+         //after this and we can't just set the closed flag to true here, since after/in onmessage the message
+         //might be acked and if the consumer is already closed, the ack will be ignored
+         closing = true;
+         
          // Now we wait for any current handler runners to run.
          waitForOnMessageToComplete();
-
+  
+         closed = true;
+         
          synchronized (this)
-         {
+         {                        
             if (receiverThread != null)
             {
                // Wake up any receive() thread that might be waiting