Merging bug22895 into default

Author: Simon MacMullen

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

@@ -544,9 +544,18 @@ public boolean processControlCommand(Command c)
         // See the detailed comments in ChannelN.processAsync.
 
         Method method = c.getMethod();
-        
+
         if (method instanceof AMQP.Connection.Close) {
-            handleConnectionClose(c);
+            if (isOpen()) {
+                handleConnectionClose(c);
+            } else {
+                // Already shutting down, so just send back a CloseOk.
+                try {
+                    _channel0.quiescingTransmit(new AMQImpl.Connection.CloseOk());
+                } catch (IOException ioe) {
+                    Utility.emptyStatement();
+                }
+            }
             return true;
         } else {
             if (isOpen()) {

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

@@ -189,8 +189,9 @@ public void releaseChannelNumber() {
 
     /**
      * Protected API - Filters the inbound command stream, processing
-     * Basic.Deliver, Basic.Return, Channel.Flow 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
     {
@@ -199,33 +200,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 a 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);
@@ -300,7 +298,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