Merging bug21848 into amqp_0_9_1

Author: Simon MacMullen

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

@@ -171,7 +171,9 @@ public void releaseChannelNumber() {
 
     /**
      * Protected API - Filters the inbound command stream, processing
-     * Basic.Deliver, Basic.Return and Channel.Close specially.
+     * Basic.Deliver, Basic.Return and Channel.Close specially.  If
+     * we're in quiescing mode, all inbound commands are ignored,
+     * except for Channel.Close and Channel.CloseOk.
      */
     @Override public boolean processAsync(Command command) throws IOException
     {
@@ -180,33 +182,30 @@ public void releaseChannelNumber() {
         // If we are not, however, then we are in a quiescing, or
         // shutting-down state as the result of an application
         // decision to close this channel, and we are to discard all
-        // incoming commands except for a close-ok.
+        // incoming commands except for close and close-ok.
 
         Method method = command.getMethod();
 
-        if (method instanceof Channel.Close) {
-            // Channel should always respond to Channel.Close
-            // from the server
-            releaseChannelNumber();
-            ShutdownSignalException signal = new ShutdownSignalException(false,
-                                                                         false,
-                                                                         command,
-                                                                         this);
-            synchronized (_channelMutex) {
-                try {
-                    processShutdownSignal(signal, true, false);
-                    quiescingTransmit(new Channel.CloseOk());
-                } finally {
-                    notifyOutstandingRpc(signal);
-                }
-            }
-            notifyListeners();
-            return true;
-        }
         if (isOpen()) {
             // We're in normal running mode.
 
-            if (method instanceof Basic.Deliver) {
+            if (method instanceof Channel.Close) {
+                releaseChannelNumber();
+                ShutdownSignalException signal = new ShutdownSignalException(false,
+                                                                             false,
+                                                                             command,
+                                                                             this);
+                synchronized (_channelMutex) {
+                    try {
+                        processShutdownSignal(signal, true, false);
+                        quiescingTransmit(new Channel.CloseOk());
+                    } finally {
+                        notifyOutstandingRpc(signal);
+                    }
+                }
+                notifyListeners();
+                return true;
+            } else if (method instanceof Basic.Deliver) {
                 Basic.Deliver m = (Basic.Deliver) method;
 
                 Consumer callback = _consumers.get(m.consumerTag);
@@ -273,7 +272,13 @@ public void releaseChannelNumber() {
         } else {
             // We're in quiescing mode.
 
-            if (method instanceof Channel.CloseOk) {
+            if (method instanceof Channel.Close) {
+                // We're already shutting down, so just send back an ok.
+                synchronized (_channelMutex) {
+                    quiescingTransmit(new Channel.CloseOk());
+                }
+                return true;
+            } else if (method instanceof Channel.CloseOk) {
                 // We're quiescing, and we see a channel.close-ok:
                 // this is our signal to leave quiescing mode and
                 // finally shut down for good. Let it be handled as an