637: Configurable correlatorId generation for RPCClient

Change-Id: I17d7a214d3336ad6e5fe890857a48b457ea599b6
(cherry picked from commit e554e4b)

Conflicts:
	src/main/java/com/rabbitmq/client/RpcClient.java
	src/main/java/com/rabbitmq/client/RpcClientParams.java

Author: janssk1

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

@@ -0,0 +1,18 @@
+package com.rabbitmq.client;
+
+import java.util.function.Supplier;
+
+public class IncrementingCorrelationIdGenerator implements Supplier<String> {
+
+    private final String _prefix;
+    private int _correlationId;
+
+    public IncrementingCorrelationIdGenerator(String _prefix) {
+        this._prefix = _prefix;
+    }
+
+    @Override
+    public String get() {
+        return _prefix + _correlationId++;
+    }
+}

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

@@ -86,7 +86,7 @@ public Response handle(Object reply) {
     /** Map from request correlation ID to continuation BlockingCell */
     private final Map<String, BlockingCell<Object>> _continuationMap = new HashMap<String, BlockingCell<Object>>();
     /** Contains the most recently-used request correlation ID */
-    private int _correlationId;
+    private final Supplier<String> _correlationIdGenerator;
 
     /** Consumer attached to our reply queue */
     private DefaultConsumer _consumer;
@@ -111,7 +111,7 @@ public RpcClient(RpcClientParams params) throws
         _timeout = params.getTimeout();
         _useMandatory = params.shouldUseMandatory();
         _replyHandler = params.getReplyHandler();
-        _correlationId = 0;
+        _correlationIdGenerator = params.getCorrelationIdGenerator();
 
         _consumer = setupConsumer();
         if (_useMandatory) {
@@ -297,8 +297,7 @@ public Response doCall(AMQP.BasicProperties props, byte[] message, int timeout)
         BlockingCell<Object> k = new BlockingCell<Object>();
         String replyId;
         synchronized (_continuationMap) {
-            _correlationId++;
-            replyId = "" + _correlationId;
+            replyId = _correlationIdGenerator.get();
             props = ((props==null) ? new AMQP.BasicProperties.Builder() : props.builder())
                 .correlationId(replyId).replyTo(_replyTo).build();
             _continuationMap.put(replyId, k);
@@ -478,14 +477,6 @@ public Map<String, BlockingCell<Object>> getContinuationMap() {
         return _continuationMap;
     }
 
-    /**
-     * Retrieve the correlation id.
-     * @return the most recently used correlation id
-     */
-    public int getCorrelationId() {
-        return _correlationId;
-    }
-
     /**
      * Retrieve the consumer.
      * @return an interface to the client's consumer object

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

@@ -52,6 +52,8 @@ public class RpcClientParams {
      */
     private RpcClient.RpcClientReplyHandler replyHandler = RpcClient.DEFAULT_REPLY_HANDLER;
 
+    private Supplier<String> correlationIdGenerator = RpcClient.DEFAULT_CORRELATION_ID_GENERATOR;
+
     /**
      * Set the channel to use for communication.
      *
@@ -172,6 +174,15 @@ public RpcClient.RpcClientReplyHandler getReplyHandler() {
         return replyHandler;
     }
 
+    public RpcClientParams correlationIdGenerator(Supplier<String> correlationIdGenerator) {
+        this.correlationIdGenerator = correlationIdGenerator;
+        return this;
+    }
+
+    public Supplier<String> getCorrelationIdGenerator() {
+        return correlationIdGenerator;
+    }
+
     /**
      * Set the behavior to use when receiving replies.
      * <p>

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

@@ -34,6 +34,7 @@
 import com.rabbitmq.client.impl.recovery.RecordedQueue;
 import com.rabbitmq.client.impl.recovery.TopologyRecoveryFilter;
 import com.rabbitmq.tools.Host;
+import org.hamcrest.CoreMatchers;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -49,9 +50,7 @@
 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;
+import static org.junit.Assert.*;
 
 public class RpcTest {
 
@@ -158,6 +157,25 @@ public void run() {
         client.close();
     }
 
+    @Test
+    public void rpcCustomCorrelatorId() throws Exception {
+        rpcServer = new TestRpcServer(serverChannel, queue);
+        new Thread(() -> {
+            try {
+                rpcServer.mainloop();
+            } catch (Exception e) {
+                // safe to ignore when loops ends/server is canceled
+            }
+        }).start();
+        RpcClient client = new RpcClient(new RpcClientParams()
+                .channel(clientChannel).exchange("").routingKey(queue).timeout(1000)
+                .correlationIdGenerator(new IncrementingCorrelationIdGenerator("myPrefix-"))
+        );
+        RpcClient.Response response = client.doCall(null, "hello".getBytes());
+        assertThat(response.getProperties().getCorrelationId(), CoreMatchers.equalTo("myPrefix-0"));
+        client.close();
+    }
+
     @Test
     public void rpcCustomReplyHandler() throws Exception {
         rpcServer = new TestRpcServer(serverChannel, queue);
@@ -182,7 +200,6 @@ public RpcClient.Response handle(Object reply) {
                     }
                 })
         );
-        assertEquals(0, replyHandlerCalls.get());
         RpcClient.Response response = client.doCall(null, "hello".getBytes());
         assertEquals(1, replyHandlerCalls.get());
         assertEquals("*** hello ***", new String(response.getBody()));