Split close processing (channel and connection);

- start and finish parts
- defer finish until processing response from the broker for application initiated close.

Author: Steve Powell

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

@@ -41,6 +41,7 @@
 import com.rabbitmq.client.SaslConfig;
 import com.rabbitmq.client.SaslMechanism;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.AMQChannel.BlockingRpcContinuation;
 import com.rabbitmq.utility.BlockingCell;
 import com.rabbitmq.utility.Utility;
 
@@ -650,6 +651,16 @@ public ShutdownSignalException shutdown(Object reason,
                          boolean initiatedByApplication,
                          Throwable cause,
                          boolean notifyRpc)
+    {
+        ShutdownSignalException sse = startShutdown(reason, initiatedByApplication, cause, notifyRpc);
+        finishShutdown(sse);
+        return sse;
+    }
+
+    private ShutdownSignalException startShutdown(Object reason,
+                         boolean initiatedByApplication,
+                         Throwable cause,
+                         boolean notifyRpc)
     {
         ShutdownSignalException sse = new ShutdownSignalException(true,initiatedByApplication,
                                                                   reason, this);
@@ -664,10 +675,12 @@ public ShutdownSignalException shutdown(Object reason,
 
         _channel0.processShutdownSignal(sse, !initiatedByApplication, notifyRpc);
 
+        return sse;
+    }
+
+    private void finishShutdown(ShutdownSignalException sse) {
         ChannelManager cm = _channelManager;
         if (cm != null) cm.handleSignal(sse);
-
-        return sse;
     }
 
     /** Public API - {@inheritDoc} */
@@ -762,10 +775,15 @@ public void close(int closeCode,
                     .replyText(closeMessage)
                 .build();
 
-            shutdown(reason, initiatedByApplication, cause, true);
+            final ShutdownSignalException sse = startShutdown(reason, initiatedByApplication, cause, true);
             if(sync){
-              AMQChannel.SimpleBlockingRpcContinuation k =
-                  new AMQChannel.SimpleBlockingRpcContinuation();
+                BlockingRpcContinuation<AMQCommand> k = new BlockingRpcContinuation<AMQCommand>(){
+                    @Override
+                    public AMQCommand transformReply(AMQCommand command) {
+                        AMQConnection.this.finishShutdown(sse);
+                        return command;
+                    }};
+
               _channel0.quiescingRpc(reason, k);
               k.getReply(timeout);
             } else {

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

@@ -239,6 +239,7 @@ public void setDefaultConsumer(Consumer consumer) {
 
     /**
      * Sends a ShutdownSignal to all active consumers.
+     * Idempotent.
      * @param signal an exception signalling channel shutdown
      */
     private void broadcastShutdownSignal(ShutdownSignalException signal) {
@@ -250,19 +251,39 @@ private void broadcastShutdownSignal(ShutdownSignalException signal) {
     }
 
     /**
-     * Protected API - overridden to quiesce consumer work and broadcast the signal
-     * to all consumers after calling the superclass's method.
+     * Start to shutdown -- defer rest of processing until ready
      */
-    @Override public void processShutdownSignal(ShutdownSignalException signal,
+    private void startProcessShutdownSignal(ShutdownSignalException signal,
                                                 boolean ignoreClosed,
                                                 boolean notifyRpc)
+    {   super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
+    }
+
+    /**
+     * Finish shutdown processing -- idempotent
+     */
+    private void finishProcessShutdownSignal()
     {
-        super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
+        this.dispatcher.quiesce();
+        broadcastShutdownSignal(getCloseReason());
+
         synchronized (unconfirmedSet) {
             unconfirmedSet.notifyAll();
         }
     }
 
+    /**
+     * Protected API - overridden to quiesce consumer work and broadcast the signal
+     * to all consumers after calling the superclass's method.
+     */
+    @Override public void processShutdownSignal(ShutdownSignalException signal,
+                                                boolean ignoreClosed,
+                                                boolean notifyRpc)
+    {
+        startProcessShutdownSignal(signal, ignoreClosed, notifyRpc);
+        finishProcessShutdownSignal();
+    }
+
     CountDownLatch getShutdownLatch() {
         return this.finishedShutdownFlag;
     }
@@ -527,17 +548,15 @@ public void close(int closeCode,
         BlockingRpcContinuation<AMQCommand> k = new BlockingRpcContinuation<AMQCommand>(){
             @Override
             public AMQCommand transformReply(AMQCommand command) {
-                ChannelN.this.dispatcher.quiesce();
-                broadcastShutdownSignal(getCloseReason());
-
+                ChannelN.this.finishProcessShutdownSignal();
                 return command;
             }};
         boolean notify = false;
         try {
             // Synchronize the block below to avoid race conditions in case
             // connnection wants to send Connection-CloseOK
             synchronized (_channelMutex) {
-                processShutdownSignal(signal, !initiatedByApplication, true);
+                startProcessShutdownSignal(signal, !initiatedByApplication, true);
                 quiescingRpc(reason, k);
             }
 

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

@@ -144,7 +144,7 @@ public CountDownLatch handleShutdownSignal(final Map<String, Consumer> consumers
             // Execute shutdown processing even if there are no consumers.
             execute(new Runnable() {
                 public void run() {
-                    notifyConsumersOfShutdown(consumers, signal);
+                    ConsumerDispatcher.this.notifyConsumersOfShutdown(consumers, signal);
                     ConsumerDispatcher.this.shutdown(signal);
                     ConsumerDispatcher.this.workService.stopWork(ConsumerDispatcher.this.channel);
                     latch.countDown();