activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject svn commit: r924350 - /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
Date Wed, 17 Mar 2010 16:18:15 GMT
Author: rajdavies
Date: Wed Mar 17 16:18:14 2010
New Revision: 924350

URL: http://svn.apache.org/viewvc?rev=924350&view=rev
Log:
Fix for https://issues.apache.org/activemq/browse/AMQ-2645

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java?rev=924350&r1=924349&r2=924350&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
Wed Mar 17 16:18:14 2010
@@ -186,7 +186,7 @@ public class ActiveMQConnection implemen
     private DestinationSource destinationSource;
     private final Object ensureConnectionInfoSentMutex = new Object();
     private boolean useDedicatedTaskRunner;
-    protected CountDownLatch transportInterruptionProcessingComplete;
+    protected volatile CountDownLatch transportInterruptionProcessingComplete;
     private long consumerFailoverRedeliveryWaitPeriod;
 
     /**
@@ -1840,7 +1840,7 @@ public class ActiveMQConnection implemen
 	}
 
     public void transportInterupted() {
-        transportInterruptionProcessingComplete = new CountDownLatch(dispatchers.size() -
(advisoryConsumer != null ? 1:0));
+        this.transportInterruptionProcessingComplete = new CountDownLatch(dispatchers.size()
- (advisoryConsumer != null ? 1:0));
         if (LOG.isDebugEnabled()) {
             LOG.debug("transport interrupted, dispatchers: " + transportInterruptionProcessingComplete.getCount());
         }
@@ -2245,17 +2245,20 @@ public class ActiveMQConnection implemen
 	}
 	
 	protected void waitForTransportInterruptionProcessingToComplete() throws InterruptedException
{
-        if (transportInterruptionProcessingComplete != null) {
-            while (!closed.get() && !transportFailed.get() && !transportInterruptionProcessingComplete.await(10,
TimeUnit.SECONDS)) {
-                LOG.warn("dispatch paused, waiting for outstanding dispatch interruption
processing (" + transportInterruptionProcessingComplete.getCount() + ") to complete..");
+	    CountDownLatch cdl = this.transportInterruptionProcessingComplete;
+	    if (cdl != null) {
+            if (!closed.get() && !transportFailed.get() && cdl.getCount()>0)
{
+                LOG.warn("dispatch paused, waiting for outstanding dispatch interruption
processing (" + cdl.getCount() + ") to complete..");
+                cdl.await(10, TimeUnit.SECONDS);
             }
             signalInterruptionProcessingComplete();
         }
     }
 	
-	protected synchronized void transportInterruptionProcessingComplete() {
-	    if (transportInterruptionProcessingComplete != null) {
-	        transportInterruptionProcessingComplete.countDown();
+	protected void transportInterruptionProcessingComplete() {
+	    CountDownLatch cdl = this.transportInterruptionProcessingComplete;
+	    if (cdl != null) {
+	        cdl.countDown();
 	        try {
 	            signalInterruptionProcessingComplete();
 	        } catch (InterruptedException ignored) {}
@@ -2263,20 +2266,22 @@ public class ActiveMQConnection implemen
 	}
 
     private void signalInterruptionProcessingComplete() throws InterruptedException {
-        if (transportInterruptionProcessingComplete.await(0, TimeUnit.SECONDS)) {
+        CountDownLatch cdl = this.transportInterruptionProcessingComplete;
+        if (cdl.getCount()==0) {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("transportInterruptionProcessingComplete for: " + this.getConnectionInfo().getConnectionId());
             }
-            synchronized (this) {
-                transportInterruptionProcessingComplete = null;
-                FailoverTransport failoverTransport = transport.narrow(FailoverTransport.class);
-                if (failoverTransport != null) {
-                    failoverTransport.connectionInterruptProcessingComplete(this.getConnectionInfo().getConnectionId());
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("notified failover transport (" + failoverTransport +")
of interruption completion for: " + this.getConnectionInfo().getConnectionId());
-                    }
-                } 
+            this.transportInterruptionProcessingComplete = null;
+
+            FailoverTransport failoverTransport = transport.narrow(FailoverTransport.class);
+            if (failoverTransport != null) {
+                failoverTransport.connectionInterruptProcessingComplete(this.getConnectionInfo().getConnectionId());
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("notified failover transport (" + failoverTransport
+                            + ") of interruption completion for: " + this.getConnectionInfo().getConnectionId());
+                }
             }
+
         }
     }
 



Mime
View raw message