Merged 19559 into 19129

Author: 0x6e6562

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

@@ -55,6 +55,9 @@ 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 */
+    public boolean _blockContent = false;
+
     /**
      * Construct a channel on the given connection, with the given channel number.
      * @param connection the underlying connection for this channel
@@ -233,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)
@@ -248,22 +252,34 @@ 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);
     }
 
     public synchronized void quiescingTransmit(Method m) throws IOException {
-        new AMQCommand(m).transmit(this);
+        quiescingTransmit(new AMQCommand(m));
     }
 
     public synchronized void quiescingTransmit(AMQCommand c) throws IOException {
         c.transmit(this);
     }
+    
+    public synchronized void setBlockContent(boolean active) {
+        _blockContent = !active;
+        notifyAll();
+    }
 
     public AMQConnection getAMQConnection() {
         return _connection;

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

@@ -229,6 +229,13 @@ public void releaseChannelNumber() {
                     }
                 }
                 return true;
+            } else if (method instanceof Channel.Flow) {
+                Channel.Flow channelFlow = (Channel.Flow) method;
+                synchronized(this) {
+                    setBlockContent(channelFlow.active);
+                    transmit(new Channel.FlowOk(channelFlow.active));
+                }
+                return true;
             } else {
                 return false;
             }