Move channel shutdown processing to after close-ok reply received.

Move WorkPoolTests to com.rabbitmq.client.impl package space.

Author: Steve Powell

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

@@ -241,12 +241,12 @@ public void setDefaultConsumer(Consumer consumer) {
      * Sends a ShutdownSignal to all active consumers.
      * @param signal an exception signalling channel shutdown
      */
-    private CountDownLatch broadcastShutdownSignal(ShutdownSignalException signal) {
+    private void broadcastShutdownSignal(ShutdownSignalException signal) {
         Map<String, Consumer> snapshotConsumers;
         synchronized (_consumers) {
             snapshotConsumers = new HashMap<String, Consumer>(_consumers);
         }
-        return this.dispatcher.handleShutdownSignal(snapshotConsumers, signal);
+        this.finishedShutdownFlag = this.dispatcher.handleShutdownSignal(snapshotConsumers, signal);
     }
 
     /**
@@ -257,9 +257,7 @@ private CountDownLatch broadcastShutdownSignal(ShutdownSignalException signal) {
                                                 boolean ignoreClosed,
                                                 boolean notifyRpc)
     {
-        this.dispatcher.quiesce();
         super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
-        this.finishedShutdownFlag = broadcastShutdownSignal(signal);
         synchronized (unconfirmedSet) {
             unconfirmedSet.notifyAll();
         }
@@ -526,7 +524,14 @@ public void close(int closeCode,
             signal.initCause(cause);
         }
 
-        BlockingRpcContinuation<AMQCommand> k = new SimpleBlockingRpcContinuation();
+        BlockingRpcContinuation<AMQCommand> k = new BlockingRpcContinuation<AMQCommand>(){
+            @Override
+            public AMQCommand transformReply(AMQCommand command) {
+                ChannelN.this.dispatcher.quiesce();
+                broadcastShutdownSignal(getCloseReason());
+
+                return command;
+            }};
         boolean notify = false;
         try {
             // Synchronize the block below to avoid race conditions in case
@@ -536,10 +541,10 @@ public void close(int closeCode,
                 quiescingRpc(reason, k);
             }
 
-            // Now that we're in quiescing state, channel.close was sent and
-            // we wait for the reply. We ignore the result.
-            // (It's NOT always close-ok.)
-            notify = true;
+                // Now that we're in quiescing state, channel.close was sent and
+                // we wait for the reply. We ignore the result.
+                // (It's NOT always close-ok.)
+                notify = true;
             k.getReply(-1);
         } catch (TimeoutException ise) {
             // Will never happen since we wait infinitely

test/src/com/rabbitmq/client/test/impl/WorkPoolTests.java renamed to test/src/com/rabbitmq/client/impl/WorkPoolTests.java

@@ -1,6 +1,5 @@
-package com.rabbitmq.client.test.impl;
+package com.rabbitmq.client.impl;
 
-import com.rabbitmq.client.impl.WorkPool;
 import junit.framework.TestCase;
 
 import java.util.ArrayList;

test/src/com/rabbitmq/client/test/functional/FunctionalTests.java

@@ -17,8 +17,8 @@
 
 package com.rabbitmq.client.test.functional;
 
+import com.rabbitmq.client.impl.WorkPoolTests;
 import com.rabbitmq.client.test.Bug20004Test;
-import com.rabbitmq.client.test.impl.WorkPoolTests;
 
 import junit.framework.TestCase;
 import junit.framework.TestSuite;