All content-bearing methods should be blocked

the client got Channel.Flow{active=false} message,
even in closing state.

Author: Hubert Plociniczak

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

@@ -236,7 +236,11 @@ public void processShutdownSignal(ShutdownSignalException signal,
                     ensureIsOpen(); // invariant: we should never be shut down more than once per instance
                 if (isOpen())
                     _shutdownCause = signal;
-                setBlockContent(true);
+
+                synchronized(this) {
+                    _blockContent = false;
+                    notifyAll();
+                }
             }
         } finally {
             if (notifyRpc)
@@ -257,14 +261,6 @@ public synchronized void transmit(Method m) throws IOException {
 
     public synchronized void transmit(AMQCommand c) throws IOException {
         ensureIsOpen();
-        if (c.getMethod().hasContent()) {
-            while (_blockContent) {
-                try {
-                    wait();
-                } catch (InterruptedException e) {}
-                ensureIsOpen();
-            }
-        }
         quiescingTransmit(c);
     }
 
@@ -273,13 +269,20 @@ 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) {}
+                
+                // This is to catch a situation when the thread wakes up during
+                // shutdown. Currently, no command that has content is allowed
+                // to send anything in a closing state.
+                ensureIsOpen();
+            }
+        }
         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,8 +222,9 @@ public void releaseChannelNumber() {
             } else if (method instanceof Channel.Flow) {
                 Channel.Flow channelFlow = (Channel.Flow) method;
                 synchronized(this) {
-                    setBlockContent(channelFlow.active);
+                    _blockContent = !channelFlow.active;
                     transmit(new Channel.FlowOk(channelFlow.active));
+                    notifyAll();
                 }
                 return true;
             } else if (method instanceof Channel.Close) {