Merge default into 19129

Author: 0x6e6562

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

@@ -236,7 +236,8 @@ public void processShutdownSignal(ShutdownSignalException signal,
                     ensureIsOpen(); // invariant: we should never be shut down more than once per instance
                 if (isOpen())
                     _shutdownCause = signal;
-                setBlockContent(true);
+                
+                notifyAll();
             }
         } finally {
             if (notifyRpc)
@@ -257,14 +258,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 +266,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

@@ -232,8 +232,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 {