Notify the channels that await Channel.Flow

about connection/channel shutdown.

Author: Hubert Plociniczak

src/com/rabbitmq/client/impl/AMQChannel.java

@@ -55,8 +55,7 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
     /** The current outstanding RPC request, if any. (Could become a queue in future.) */
     public RpcContinuation _activeRpc = null;
 
-    /** Whether transmission of content-bearing methods should be blocked
-     */
+    /** Whether transmission of content-bearing methods should be blocked */
     public boolean _blockContent = false;
 
     /**
@@ -237,6 +236,7 @@ public void processShutdownSignal(ShutdownSignalException signal,
                     ensureIsOpen(); // invariant: we should never be shut down more than once per instance
                 if (isOpen())
                     _shutdownCause = signal;
+                setBlockContent(true);
             }
         } finally {
             if (notifyRpc)
@@ -252,12 +252,19 @@ public void notifyOutstandingRpc(ShutdownSignalException signal) {
     }
 
     public synchronized void transmit(Method m) throws IOException {
-        ensureIsOpen();
-        quiescingTransmit(m);
+        transmit(new AMQCommand(m));
     }
 
     public synchronized void transmit(AMQCommand c) throws IOException {
         ensureIsOpen();
+        if (c.getMethod().hasContent()) {
+            while (_blockContent) {
+                try {
+                    wait();
+                } catch (InterruptedException e) {}
+                ensureIsOpen();
+            }
+        }
         quiescingTransmit(c);
     }
 
@@ -266,15 +273,13 @@ public synchronized void quiescingTransmit(Method m) throws IOException {
     }
 
     public synchronized void quiescingTransmit(AMQCommand c) throws IOException {
-        if (c.getMethod().hasContent()) {
-            while (_blockContent) {
-                try {
-                    wait();
-                } catch (InterruptedException e) {}
-            }
-        }
         c.transmit(this);
     }
+    
+    public synchronized void setBlockContent(boolean active) {
+        _blockContent = !active;
+        notifyAll();
+    }
 
     public AMQConnection getAMQConnection() {
         return _connection;

src/com/rabbitmq/client/impl/ChannelN.java

@@ -222,9 +222,8 @@ public void releaseChannelNumber() {
             } else if (method instanceof Channel.Flow) {
                 Channel.Flow channelFlow = (Channel.Flow) method;
                 synchronized(this) {
-                    _blockContent = !channelFlow.active;
+                    setBlockContent(channelFlow.active);
                     transmit(new Channel.FlowOk(channelFlow.active));
-                    notifyAll();
                 }
                 return true;
             } else if (method instanceof Channel.Close) {