Log warning when correlation ID isn't found in RpcClient

Add a test as well to check the outstanding request map is cleaned after
a timeout.

References #375

Author: acogoluegnes

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 */
@@ -191,10 +196,12 @@ public void handleDelivery(String consumerTag,
                 synchronized (_continuationMap) {
                     String replyId = properties.getCorrelationId();
                     BlockingCell<Object> blocker =_continuationMap.remove(replyId);
-                    if (blocker != null) {
-                        blocker.set(new Response(consumerTag, envelope, properties, body));
+                    if (blocker == null) {
+                        // Entry should have been removed if request timed out,
+                        // log a warning nevertheless.
+                        LOGGER.warn("No outstanding request for correlation ID {}", replyId);
                     } else {
-                        // Not an error. Entry will have been removed if request timed out.
+                        blocker.set(new Response(consumerTag, envelope, properties, body));
                     }
                 }
             }

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 {