[jboss-cvs] JBossAS SVN: r78001 - branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq.

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

Author: jhowell at redhat.com
Date: 2008-09-04 15:13:17 -0400 (Thu, 04 Sep 2008)
New Revision: 78001

Modified:
   branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/Connection.java
Log:
[ASPATCH-385] Migrate JBAS-4090: Race condition in Connection.close() can deadlock the JBossMQ's ping thread.

Modified: branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/Connection.java
===================================================================
--- branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/Connection.java	2008-09-04 18:53:23 UTC (rev 78000)
+++ branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/Connection.java	2008-09-04 19:13:17 UTC (rev 78001)
@@ -118,7 +118,7 @@
    /** Maps a destination to a LinkedList of Subscriptions */
    public HashMap destinationSubscriptions = new HashMap();
 
-   /* Maps a a subsction id to a Subscription */
+   /** Maps a subscription id to a Subscription */
    public HashMap subscriptions = new HashMap();
 
    /** Is the connection stopped ? */
@@ -338,7 +338,9 @@
 	 */
    public void asynchClose()
    {
-      // This obviously did something at some point?
+      // If we receive a close and we did not initiate it, then fire the exception listener
+      if (closing.get() == false)
+         asynchFailure("Asynchronous close from server.", new IOException("Close request from the server or transport layer."));
    }
 
    /**
@@ -378,23 +380,24 @@
       {
          for (int i = 0; i < requests.length; i++)
          {
+            ReceiveRequest r = requests[i];
             if (trace)
-               log.trace("Processing request=" + requests[i] + " " + this);
+               log.trace("Processing request=" + r + " " + this);
             
-            SpyConsumer consumer = (SpyConsumer) subscriptions.get(requests[i].subscriptionId);
-            requests[i].message.createAcknowledgementRequest(requests[i].subscriptionId.intValue());
+            SpyConsumer consumer = (SpyConsumer) subscriptions.get(r.subscriptionId);
+            r.message.createAcknowledgementRequest(r.subscriptionId.intValue());
 
             if (consumer == null)
             {
-               send(requests[i].message.getAcknowledgementRequest(false));
-               log.debug("WARNING: NACK issued due to non existent subscription " + requests[i].message.header.messageId);
+               send(r.message.getAcknowledgementRequest(false));
+               log.debug("WARNING: NACK issued due to non existent subscription " + r.message.header.messageId);
                continue;
             }
 
             if (trace)
-               log.trace("Delivering messageid=" + requests[i].message.header.messageId + " to consumer=" + consumer);
+               log.trace("Delivering messageid=" + r.message.header.messageId + " to consumer=" + consumer);
             
-            consumer.addMessage(requests[i].message);
+            consumer.addMessage(r.message);
          }
       }
       catch (Throwable t)
@@ -1259,9 +1262,18 @@
 		 */
       public void run()
       {
+         // Don't bother if we are closing
+         if (closing.get())
+            return;
+         
          try
          {
-            pingTaskSemaphore.acquire();
+            // If we can't aquire the semaphore then it
+            // almost certainly means the close has got it
+            // Try for 10 seconds to make sure the problem
+            // is not just a long garbage collection that has suspended threads
+            if (pingTaskSemaphore.attempt(1000 * 10) == false)
+               return;
          }
          catch (InterruptedException e)
          {