[jboss-cvs] JBossAS SVN: r70205 - branches/JBoss_4_0_5_GA_CP/messaging/src/main/org/jboss/mq/server.

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

Author: bershath27
Date: 2008-02-28 03:53:59 -0500 (Thu, 28 Feb 2008)
New Revision: 70205

Modified:
   branches/JBoss_4_0_5_GA_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java
Log:
Fixed http://jira.jboss.com/jira/browse/ASPATCH-338

[back ported JBAS-5228]




Modified: branches/JBoss_4_0_5_GA_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java
===================================================================
--- branches/JBoss_4_0_5_GA_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2008-02-28 08:50:03 UTC (rev 70204)
+++ branches/JBoss_4_0_5_GA_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2008-02-28 08:53:59 UTC (rev 70205)
@@ -211,40 +211,48 @@
       if (trace)
          log.trace("addReceiver " + sub + " " + this);
       
-      MessageReference found = null;
-      synchronized (messages)
+      synchronized(receivers)
       {
-         if (messages.size() != 0)
+         MessageReference found = null;
+         synchronized (messages)
          {
-            for (Iterator it = messages.iterator(); it.hasNext();)
+            if (messages.size() != 0)
             {
-               MessageReference message = (MessageReference) it.next();
-               try
+               for (Iterator it = messages.iterator(); it.hasNext();)
                {
-                  if (message.isExpired())
+                  MessageReference message = (MessageReference) it.next();
+                  try
                   {
-                     it.remove();
-                     expireMessageAsync(message);
+                     if (message.isExpired())
+                     {
+                        it.remove();
+                        expireMessageAsync(message);
+                     }
+                     else if (sub.accepts(message.getHeaders()))
+                     {
+                        //queue message for sending to this sub
+                        it.remove();
+                        found = message;
+                        break;
+                     }
                   }
-                  else if (sub.accepts(message.getHeaders()))
+                  catch (JMSException ignore)
                   {
-                     //queue message for sending to this sub
-                     it.remove();
-                     found = message;
-                     break;
+                     log.info("Caught unusual exception in addToReceivers.", ignore);
                   }
                }
-               catch (JMSException ignore)
-               {
-                  log.info("Caught unusual exception in addToReceivers.", ignore);
-               }
             }
+            if (found != null)
+               setupMessageAcknowledgement(sub, found);
          }
+         if (found != null)
+         {
+            updateRedeliveryFlags(found);
+            queueMessageForSending(sub, found);
+         }
+         else
+            addToReceivers(sub);
       }
-      if (found != null)
-         queueMessageForSending(sub, found);
-      else
-         addToReceivers(sub);
    }
 
    /**
@@ -668,6 +676,9 @@
                   else
                      break;
                }
+               
+               if (messageRef != null)
+                  setupMessageAcknowledgement(sub, messageRef);
             }
          }
          else
@@ -691,6 +702,9 @@
                      break;
                   }
                }
+               
+               if (messageRef != null)
+                  setupMessageAcknowledgement(sub, messageRef);
             }
          }
 
@@ -699,14 +713,11 @@
             if (wait)
                addToReceivers(sub);
          }
-         else
-         {
-            setupMessageAcknowledgement(sub, messageRef);
-         }
       }
 
       if (messageRef == null)
          return null;
+      updateRedeliveryFlags(messageRef);
       return messageRef.getMessageForDelivery();
    }
 
@@ -1094,9 +1105,9 @@
                }
             }
 
-            if (found == null)
+            synchronized (messages)
             {
-               synchronized (messages)
+               if (found == null)
                {
                   messages.add(message);
 
@@ -1107,12 +1118,19 @@
                      addTimeout(message, new ExpireMessageTask(message), message.messageExpiration);
                   }
                }
+               else
+               {
+                  setupMessageAcknowledgement(found, message);
+               }
             }
          }
          
          // Queue to the receiver
          if (found != null)
+         {
+            updateRedeliveryFlags(message);
             queueMessageForSending(found, message);
+         }
       }
       catch (JMSException e)
       {
@@ -1137,7 +1155,6 @@
 
       try
       {
-         setupMessageAcknowledgement(sub, message);
          RoutedMessage r = new RoutedMessage();
          r.message = message;
          r.subscriptionId = new Integer(sub.subscriptionId);
@@ -1163,7 +1180,27 @@
       nack.destination = message.getJMSDestination();
       nack.messageID = message.getJMSMessageID();
       nack.subscriberId = sub.subscriptionId;
+      
+      UnackedMessageInfo unacked = new UnackedMessageInfo(messageRef, sub);
+      unacknowledgedMessages.put(nack, unacked);
+      unackedByMessageRef.put(messageRef, nack);
+      HashMap map = (HashMap) unackedBySubscription.get(sub);
+      if (map == null)
+      {
+         map = new HashMap();
+         unackedBySubscription.put(sub, map);
+      }
+      map.put(messageRef, nack);
+   }
 
+   /**
+    * Update redelivery flags
+    *
+    * @param messageRef the message to be acknowledged
+    * @throws JMSException for any error
+    */
+   protected void updateRedeliveryFlags(MessageReference messageRef) throws JMSException
+   {
       // Need to update the message so it recovers from crash with redelivery=true 
       if (server.isLazyRedeliveryUpdate() == false && messageRef.isPersistent())
       {
@@ -1176,20 +1213,6 @@
          // Now revert the flags back again (but only in memory)
          messageRef.revertRedelivered();
       }
-      
-      synchronized (messages)
-      {
-         UnackedMessageInfo unacked = new UnackedMessageInfo(messageRef, sub);
-         unacknowledgedMessages.put(nack, unacked);
-         unackedByMessageRef.put(messageRef, nack);
-         HashMap map = (HashMap) unackedBySubscription.get(sub);
-         if (map == null)
-         {
-            map = new HashMap();
-            unackedBySubscription.put(sub, map);
-         }
-         map.put(messageRef, nack);
-      }
    }
 
    /**