Add test case for broken RpcClient on broker restart

The RpcClient registers a reply consumer on the direct-reply
pseudo-queue and doesn't cancel it  when the connection fails. The RpcClient
is then unusable by design (it nullifies the consumer on connection
failure). Nevertheless, the consumer must be cancelled after recovery to be
able to create a new RpcClient on the same channel.

Disabling topology recovery makes this work: the consumer isn't
re-registered during topology recovery and a new RpcClient can be
created on the same (recovered) channel.

References #382

Author: acogoluegnes

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

@@ -17,16 +17,21 @@
 package com.rabbitmq.client.test;
 
 import com.rabbitmq.client.*;
+import com.rabbitmq.client.impl.NetworkConnection;
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
+import com.rabbitmq.tools.Host;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
 import java.io.IOException;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 public class RpcTest {
 
@@ -35,7 +40,6 @@ public class RpcTest {
     String queue = "rpc.queue";
     RpcServer rpcServer;
 
-
     @Before public void init() throws Exception {
         clientConnection = TestUtils.connectionFactory().newConnection();
         clientChannel = clientConnection.createChannel();
@@ -77,6 +81,54 @@ public void run() {
         client.close();
     }
 
+    @Test public void brokenAfterBrokerRestart() throws Exception {
+        // see https://github.com/rabbitmq/rabbitmq-java-client/issues/382
+        rpcServer = new TestRpcServer(serverChannel, queue);
+        new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    rpcServer.mainloop();
+                } catch (Exception e) {
+                    // safe to ignore when loops ends/server is canceled
+                }
+            }
+        }).start();
+
+        ConnectionFactory cf = TestUtils.connectionFactory();
+        cf.setTopologyRecoveryEnabled(false);
+        cf.setNetworkRecoveryInterval(2000);
+        Connection connection = null;
+        try {
+            connection = cf.newConnection();
+            Channel channel = connection.createChannel();
+            RpcClient client = new RpcClient(channel, "", queue, 1000);
+            RpcClient.Response response = client.doCall(null, "hello".getBytes());
+            assertEquals("*** hello ***", new String(response.getBody()));
+            final CountDownLatch recoveryLatch = new CountDownLatch(1);
+            ((AutorecoveringConnection) connection).addRecoveryListener(new RecoveryListener() {
+                @Override
+                public void handleRecovery(Recoverable recoverable) {
+                    recoveryLatch.countDown();
+                }
+                @Override
+                public void handleRecoveryStarted(Recoverable recoverable) {
+
+                }
+            });
+            Host.closeConnection((NetworkConnection) connection);
+            assertTrue("Connection should have recovered by now", recoveryLatch.await(10, TimeUnit.SECONDS));
+            client = new RpcClient(channel, "", queue, 1000);
+            response = client.doCall(null, "hello".getBytes());
+            assertEquals("*** hello ***", new String(response.getBody()));
+        } finally {
+            if (connection != null) {
+                connection.close();
+            }
+        }
+
+    }
+
     private static class TestRpcServer extends RpcServer {
 
         public TestRpcServer(Channel channel, String queueName) throws IOException {