Merge in bug22083

Author: Michael Bridgen

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 Close was sent from within the MainLoop we
+                    // will 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;
     }
 }

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

@@ -0,0 +1,69 @@
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.impl.*;
+import com.rabbitmq.client.*;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.CountDownLatch;
+
+import javax.net.SocketFactory;
+
+public class CloseInMainLoop extends BrokerTestCase{
+  class SpecialConnection extends AMQConnection{
+    private AtomicBoolean validShutdown = new AtomicBoolean();
+
+    public boolean hadValidShutdown(){
+      if(isOpen()) throw new IllegalStateException("hadValidShutdown called while connection is still open");
+      return validShutdown.get();
+    }
+  
+    public SpecialConnection() throws Exception{
+      super(new ConnectionParameters(), new SocketFrameHandler(SocketFactory.getDefault().createSocket("localhost", 5672)));
+      this.start(true);
+    }
+
+    @Override
+    public boolean processControlCommand(Command c) throws IOException{
+      if(c.getMethod() instanceof AMQP.Connection.CloseOk) validShutdown.set(true);
+      return super.processControlCommand(c);
+    }
+  }
+
+
+  public void testCloseOKNormallyReceived() throws Exception{
+    SpecialConnection connection = new SpecialConnection();
+    connection.close();
+    assertTrue(connection.hadValidShutdown());
+  }
+  
+  // The thrown runtime exception should get intercepted by the
+  // consumer exception handler, and result in a clean shut down.
+  public void testCloseWithFaultyConsumer() throws Exception{
+    SpecialConnection connection = new SpecialConnection();
+    Channel channel = connection.createChannel();
+    channel.exchangeDeclare("x", "direct");
+    channel.queueDeclare("q");
+    channel.queueDelete("q");
+    channel.queueDeclare("q");
+    channel.queueBind("q", "x", "k");
+
+    final CountDownLatch latch = new CountDownLatch(1);    
+
+    channel.basicConsume("q", true, new DefaultConsumer(channel){
+      public void handleDelivery(String consumerTag,
+                                 Envelope envelope,
+                                 AMQP.BasicProperties properties,
+                                 byte[] body){
+        latch.countDown();
+        throw new RuntimeException("I am a bad consumer");
+      }
+    });
+
+    channel.basicPublish("x", "k", null, new byte[10]);
+
+    latch.await();
+    Thread.sleep(200);
+    assertTrue(connection.hadValidShutdown());
+  }
+  
+}