Merge branch '5.3.x-stable' into 5.x.x-stable

acogoluegnes · acogoluegnes · commit 3f675f40fd51 · 2018-06-28T09:50:11.000+02:00
diff --git a/src/main/java/com/rabbitmq/client/RpcClient.java b/src/main/java/com/rabbitmq/client/RpcClient.java
@@ -33,6 +33,8 @@
 import com.rabbitmq.client.impl.ValueReader;
 import com.rabbitmq.client.impl.ValueWriter;
 import com.rabbitmq.utility.BlockingCell;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Convenience class which manages simple RPC-style communication.
@@ -41,6 +43,9 @@
  * and waiting for a response.
 */
 public class RpcClient {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);
+
     /** Channel we are communicating on */
     private final Channel _channel;
     /** Exchange to send requests to */
@@ -192,9 +197,12 @@ public void handleDelivery(String consumerTag,
                     String replyId = properties.getCorrelationId();
                     BlockingCell<Object> blocker =_continuationMap.remove(replyId);
                     if (blocker == null) {
-                        throw new IllegalStateException("No outstanding request for correlation ID " + replyId);
+                        // Entry should have been removed if request timed out,
+                        // log a warning nevertheless.
+                        LOGGER.warn("No outstanding request for correlation ID {}", replyId);
+                    } else {
+                        blocker.set(new Response(consumerTag, envelope, properties, body));
                     }
-                    blocker.set(new Response(consumerTag, envelope, properties, body));
                 }
             }
         };
@@ -217,15 +225,23 @@ public Response doCall(AMQP.BasicProperties props, byte[] message, int timeout)
         throws IOException, ShutdownSignalException, TimeoutException {
         checkConsumer();
         BlockingCell<Object> k = new BlockingCell<Object>();
+        String replyId;
         synchronized (_continuationMap) {
             _correlationId++;
-            String replyId = "" + _correlationId;
+            replyId = "" + _correlationId;
             props = ((props==null) ? new AMQP.BasicProperties.Builder() : props.builder())
                 .correlationId(replyId).replyTo(_replyTo).build();
             _continuationMap.put(replyId, k);
         }
         publish(props, message);
-        Object reply = k.uninterruptibleGet(timeout);
+        Object reply;
+        try {
+            reply = k.uninterruptibleGet(timeout);
+        } catch (TimeoutException ex) {
+            // Avoid potential leak.  This entry is no longer needed by caller.
+            _continuationMap.remove(replyId);
+            throw ex;
+        }
         if (reply instanceof ShutdownSignalException) {
             ShutdownSignalException sig = (ShutdownSignalException) reply;
             ShutdownSignalException wrapper =
diff --git a/src/test/java/com/rabbitmq/client/test/RpcTest.java b/src/test/java/com/rabbitmq/client/test/RpcTest.java
@@ -24,6 +24,7 @@
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeoutException;
 
 import static org.junit.Assert.assertEquals;
 
@@ -72,6 +73,17 @@ public void rpc() throws Exception {
         client.close();
     }
 
+    @Test public void rpcResponseTimeout() throws Exception {
+        RpcClient client = new RpcClient(clientChannel, "", queue);
+        try {
+            client.doCall(null, "hello".getBytes(), 200);
+        } catch (TimeoutException e) {
+            // OK
+        }
+        assertEquals(0, client.getContinuationMap().size());
+        client.close();
+    }
+
     private static class TestRpcServer extends RpcServer {
 
         public TestRpcServer(Channel channel, String queueName) throws IOException {
