svn commit: r1236664 - /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java

[gtully]

Author: gtully
Date: Fri Jan 27 12:53:09 2012
New Revision: 1236664

URL: http://svn.apache.org/viewvc?rev=1236664&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-3684 - Potential deadlock in vm 
transport setListener when sender is blocked pending space
Deal with the case of blocked producer

Modified:
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java?rev=1236664&r1=1236663&r2=1236664&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
 Fri Jan 27 12:53:09 2012
@@ -226,12 +226,20 @@ public class VMTransport implements Tran
 
     public void setTransportListener(TransportListener commandListener) {
         try {
-            try {
-                enqueueValve.turnOff();
+            // enqueue can block on blocking queue, preventing turnOff
+            // so avoid in that case: 
https://issues.apache.org/jira/browse/AMQ-3684
+            if (async && getMessageQueue().remainingCapacity() == 0) {
+                // enqueue blocked or will be
                 this.transportListener = commandListener;
                 wakeup();
-            } finally {
-                enqueueValve.turnOn();
+            } else {
+                try {
+                    enqueueValve.turnOff();
+                    this.transportListener = commandListener;
+                    wakeup();
+                } finally {
+                    enqueueValve.turnOn();
+                }
             }
         } catch (InterruptedException e) {
             throw new RuntimeException(e);