rebranch off default

Author: David R. MacIver

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

@@ -523,9 +523,13 @@ public boolean processControlCommand(Command c)
                 return false;
             } else {
                 // Quiescing.
-                if (method instanceof AMQP.Connection.CloseOk) {
-                    // It's our final "RPC".
-                    return false;
+                if (method instanceof AMQP.Connection.CloseOk) {    
+                    // It's our final "RPC". Time to shut down.
+                    _running = false;
+                    // If this was called from within the MainLoop we 
+                    // may not have a continuation to return to, so we 
+                    // treat this as processed in that case.
+                    return _channel0._activeRpc == null;
                 } else {
                     // Ignore all others.
                     return true;
@@ -680,14 +684,21 @@ public void close(int closeCode,
                       boolean abort)
         throws IOException
     {
+        final boolean sync = !(Thread.currentThread() instanceof MainLoop);
+
         try {
             AMQImpl.Connection.Close reason =
                 new AMQImpl.Connection.Close(closeCode, closeMessage, 0, 0);
+
             shutdown(reason, initiatedByApplication, cause, true);
-            AMQChannel.SimpleBlockingRpcContinuation k =
-                new AMQChannel.SimpleBlockingRpcContinuation();
-            _channel0.quiescingRpc(reason, k);
-            k.getReply(timeout);
+            if(sync){
+              AMQChannel.SimpleBlockingRpcContinuation k =
+                  new AMQChannel.SimpleBlockingRpcContinuation();
+              _channel0.quiescingRpc(reason, k);
+              k.getReply(timeout);
+            } else {
+              _channel0.quiescingTransmit(reason);
+            }
         } catch (TimeoutException tte) {
             if (!abort)
                 throw new ShutdownSignalException(true, true, tte, this);
@@ -698,7 +709,7 @@ public void close(int closeCode,
             if (!abort)
                 throw ioe;
         } finally {
-            _frameHandler.close();
+            if(sync) _frameHandler.close();
         }
     }
 

test/src/com/rabbitmq/client/test/ClientTests.java

@@ -45,6 +45,7 @@ public static TestSuite suite() {
         suite.addTest(BrokenFramesTest.suite());
         suite.addTest(ClonePropertiesTest.suite());
         suite.addTestSuite(Bug20004Test.class);
+        suite.addTestSuite(CloseInMainLoop.class);
         return suite;
     }
 }