Stop RpcServer when its thread is interrupted

Fixes #428

(cherry picked from commit 07b86ca)

Author: acogoluegnes

src/main/java/com/rabbitmq/client/RpcServer.java

@@ -92,7 +92,8 @@ protected QueueingConsumer setupConsumer()
      * Public API - main server loop. Call this to begin processing
      * requests. Request processing will continue until the Channel
      * (or its underlying Connection) is shut down, or until
-     * terminateMainloop() is called.
+     * terminateMainloop() is called, or until the thread running the loop
+     * is interrupted.
      *
      * Note that if the mainloop is blocked waiting for a request, the
      * termination flag is not checked until a request is received, so
@@ -110,6 +111,8 @@ public ShutdownSignalException mainloop()
                 try {
                     request = _consumer.nextDelivery();
                 } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    _mainloopRunning = false;
                     continue;
                 }
                 processRequest(request);

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

@@ -34,6 +34,8 @@
 import com.rabbitmq.client.impl.recovery.RecordedQueue;
 import com.rabbitmq.client.impl.recovery.TopologyRecoveryFilter;
 import com.rabbitmq.tools.Host;
+import org.awaitility.Awaitility;
+import org.awaitility.Duration;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -47,6 +49,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.awaitility.Awaitility.waitAtMost;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -315,6 +318,28 @@ public void handleRecoveryStarted(Recoverable recoverable) {
         }
     }
 
+    @Test public void interruptingServerThreadShouldStopIt() throws Exception {
+        rpcServer = new TestRpcServer(serverChannel, queue);
+        Thread serverThread = new Thread(() -> {
+            try {
+                rpcServer.mainloop();
+            } catch (Exception e) {
+                // safe to ignore when loops ends/server is canceled
+            }
+        });
+        serverThread.start();
+        RpcClient client = new RpcClient(new RpcClientParams()
+                .channel(clientChannel).exchange("").routingKey(queue).timeout(1000));
+        RpcClient.Response response = client.doCall(null, "hello".getBytes());
+        assertEquals("*** hello ***", new String(response.getBody()));
+
+        serverThread.interrupt();
+
+        waitAtMost(Duration.ONE_SECOND).until(() -> !serverThread.isAlive()) ;
+
+        client.close();
+    }
+
     private static class TestRpcServer extends RpcServer {
 
         public TestRpcServer(Channel channel, String queueName) throws IOException {