[jboss-cvs] JBoss Messaging SVN: r8383 - branches/Branch_1_4/src/main/org/jboss/jms/server/endpoint.

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

Author: gaohoward
Date: 2011-07-11 22:13:49 -0400 (Mon, 11 Jul 2011)
New Revision: 8383

Modified:
   branches/Branch_1_4/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
Log:
JBMESSAGING-1885


Modified: branches/Branch_1_4/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
===================================================================
--- branches/Branch_1_4/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2011-07-07 12:44:44 UTC (rev 8382)
+++ branches/Branch_1_4/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2011-07-12 02:13:49 UTC (rev 8383)
@@ -197,6 +197,9 @@
    private boolean isCC = false;
    
    private boolean markClose = false;
+   
+   //https://issues.jboss.org/browse/JBMESSAGING-1885
+   private Object txLock = new Object();
 
    // Constructors ---------------------------------------------------------------------------------
 
@@ -512,18 +515,25 @@
          // deliveries must be cancelled in reverse order
 
          Set channels = new HashSet();
-
-         for (int i = cancels.size() - 1; i >= 0; i--)
+         
+         //https://issues.jboss.org/browse/JBMESSAGING-1885
+         synchronized (txLock)
          {
-            Cancel cancel = (Cancel)cancels.get(i);
+            for (int i = cancels.size() - 1; i >= 0; i--)
+            {
+               Cancel cancel = (Cancel)cancels.get(i);
 
-            if (trace) { log.trace(this + " cancelling delivery " + cancel.getDeliveryId()); }
+               if (trace)
+               {
+                  log.trace(this + " cancelling delivery " + cancel.getDeliveryId());
+               }
 
-            Delivery del = cancelDeliveryInternal(cancel);
+               Delivery del = cancelDeliveryInternal(cancel);
 
-            if (del != null)
-            {
-            	channels.add(del.getObserver());
+               if (del != null)
+               {
+                  channels.add(del.getObserver());
+               }
             }
          }
 
@@ -2554,11 +2564,19 @@
       {
          // Remove the deliveries from the delivery map.
          Iterator iter = delList.iterator();
+         
+         synchronized(txLock)
+         {
          while (iter.hasNext())
          {
             Long deliveryId = (Long)iter.next();
 
             DeliveryRecord del = (DeliveryRecord)deliveries.remove(deliveryId);
+            
+            if (del == null)
+            {
+               throw new TransactionException("transaction rolled back because deliveries has been cancelled by the session.");
+            }
 
             if (del != null && del.replicating)
             {
@@ -2586,6 +2604,7 @@
                }
             }
          }
+         }
       }
 
       public void afterRollback(boolean onePhase) throws TransactionException