[jboss-cvs] JBoss Messaging SVN: r2724 - in trunk/src/main/org/jboss: messaging/core and 1 other directory.

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

Author: timfox
Date: 2007-05-24 11:21:04 -0400 (Thu, 24 May 2007)
New Revision: 2724

Modified:
   trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
   trunk/src/main/org/jboss/messaging/core/ChannelSupport.java
Log:
Fixed failing message consumer test


Modified: trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
===================================================================
--- trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2007-05-24 13:15:35 UTC (rev 2723)
+++ trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2007-05-24 15:21:04 UTC (rev 2724)
@@ -855,7 +855,14 @@
       promptDelivery(channels);
       
       //Close down the executor
-      executor.shutdownAfterProcessingCurrentTask();
+      
+      //Note we need to wait for ALL tasks to complete NOT just one otherwise we can end up with the following situation
+      //prompter is queued and starts to execute
+      //prompter almost finishes executing then a message is cancelled due to this session closing
+      //this causes another prompter to be queued
+      //shutdownAfterProcessingCurrentTask is then called
+      //this means the second prompter never runs and the cancelled message doesn't get redelivered
+      executor.shutdownAfterProcessingCurrentlyQueuedTasks();
 
       deliveries.clear();
       
@@ -955,7 +962,8 @@
          //cancelDeliveries because remoting one way invocations can 
          //overtake each other in flight - this problem will
          //go away when we have our own transport and our dedicated connection
-         this.executor.execute(new Runnable() { public void run() { channel.deliver(); } } );
+         this.executor.execute(new Runnable() { public void run() { channel.deliver();} } );
+         
       }
       catch (Throwable t)
       {
@@ -1532,8 +1540,6 @@
       //Now prompt delivery on the channels
       Iterator iter = channels.iterator();
       
-      if (trace) { log.trace("Prompting delivery"); }
-      
       while (iter.hasNext())
       {
          DeliveryObserver observer = (DeliveryObserver)iter.next();

Modified: trunk/src/main/org/jboss/messaging/core/ChannelSupport.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/ChannelSupport.java	2007-05-24 13:15:35 UTC (rev 2723)
+++ trunk/src/main/org/jboss/messaging/core/ChannelSupport.java	2007-05-24 15:21:04 UTC (rev 2724)
@@ -296,7 +296,7 @@
    public void deliver()
    {
       checkClosed();
-
+      
       if (router.getNumberOfReceivers() > 0)
       {                
          synchronized (refLock)
@@ -305,6 +305,7 @@
             
             deliverInternal();
          }
+         
       }
    }      
 
@@ -632,7 +633,21 @@
                   
                   // Receiver accepted the reference
                   
-                  removeReference(iter);
+                  synchronized (refLock)
+                  {
+                     if (iter == null)
+                     {
+                        if (trace) { log.trace(this + " removing first ref in memory"); } 
+                        
+                        removeFirstInMemory();
+                     }
+                     else
+                     {
+                        if (trace) { log.trace(this + " removed current message from iterator"); }                           
+                                    
+                        iter.remove();                                
+                     }
+                  }
                   
                   deliveringCount.increment();                     
                }               
@@ -887,26 +902,7 @@
    }    
    
    // Private --------------------------------------------------------------------------------------
-   
-   private void removeReference(ListIterator iter) throws Exception
-   {
-      synchronized (refLock)
-      {
-         if (iter == null)
-         {
-            if (trace) { log.trace(this + " removing first ref in memory"); } 
-            
-            removeFirstInMemory();
-         }
-         else
-         {
-            if (trace) { log.trace(this + " removed current message from iterator"); }                           
-                        
-            iter.remove();                                
-         }
-      }
-   }
-   
+      
    private MessageReference nextReference(ListIterator iter) throws Throwable
    {
       MessageReference ref;