Document RPC support

Author: acogoluegnes

src/docs/asciidoc/usage.adoc

@@ -261,7 +261,103 @@ include::{test-examples}/Api.java[tag=connection-recovery-deactivate]
 --------
 <1> Deactivate recovery
 
+=== Remote Procedure Call (RPC)
 
+Remote procedure call with RabbitMQ consists in a client sending a request message and a server replying with a response message.
+Both the RPC client and server are _client applications_ and the messages flow through the broker.
+The RPC client must send a reply-to queue address with the request.
+The RPC server uses this reply-to queue address to send the response.
+There must also be a way to correlate a request with its response, this is usually handled with a header that the RPC client and server agree on.
+
+The library provides RPC client and server support classes.
+They use sensible defaults and some of the internal mechanics are configurable.
+They should meet the requirements of most RPC use cases.
+It is still possible to implement one part or the other with regular publishers and consumers for special cases, as this is what the RPC support classes do.
+
+Here is how to create an RPC server instance:
+
+.Creating an RPC server
+[source,java,indent=0]
+--------
+include::{test-examples}/RpcApi.java[tag=rpc-server-creation]
+--------
+<1> Use builder from connection
+<2> Set the queue to consume requests from (it must exist)
+<3> Define the processing logic
+<4> Create the reply message
+
+Note the RPC server does not create the queue it waits requests on.
+It must be created beforehand.
+
+Here is how to create an RPC client:
+
+.Creating an RPC client
+[source,java,indent=0]
+--------
+include::{test-examples}/RpcApi.java[tag=rpc-client-creation]
+--------
+<1> Use builder from connection
+<2> Set the address to send request messages to
+
+The RPC client will send its request to the configured destination.
+It can be an exchange or a queue, like in the example above.
+
+Here is how to send a request:
+
+.Sending a request
+[source,java,indent=0]
+--------
+include::{test-examples}/RpcApi.java[tag=rpc-client-request]
+--------
+<1> Create the message request
+<2> Send the request
+<3> Wait for the reply (synchronously)
+
+The `RpcClient#publish(Message)` method returns a `CompletableFuture<Message>` that holds the reply message.
+It is then possible to wait for the reply asynchronously or synchronously.
+
+The RPC server uses the following defaults:
+
+* it uses the _request_ https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties[`message-id` property] for the correlation ID.
+* it assigns the correlation ID (so the _request_ `message-id` by default) to the _reply_ https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties[`correlation-id` property].
+* it assigns the _request_ https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties[`reply-to` property] to the _reply_ https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties[`to` property] if it is defined.
+This behavior is hardcoded but it is possible to cancel it thanks to a reply post-processor.
+
+The RPC client uses the following defaults:
+
+* it creates and waits for replies on an auto-delete, exclusive queue if no reply-to queue is set.
+* it uses a string-based correlation ID generator, with a fixed random UUID prefix and a strictly monotonic increasing sequence suffix (`{UUID}-{sequence}`, e.g. `6f839461-6b19-47e1-80b3-6be10d899d85-42`).
+The prefix is different for each `RpcClient` instance and the suffix is incremented by one for each request.
+* it sets the _request_ https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties[`reply-to` property] to the reply-to queue address (defined by the user or created automatically, see above).
+* it sets the _request_ https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties[`message-id` property] to the generated correlation ID.
+* it extracts the correlation ID from the _reply_ https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties[`correlation-id` property] to correlate a reply with the appropriate request.
+
+Let's see how to customize some of the RPC support mechanics.
+Imagine the request `message-id` property is a critical piece of information and we do not want to use it as the correlation ID.
+The request can use the `correlation-id` property and the RPC server just has to extract the correlation ID from this property (instead of the `message-id` property by default).
+Let's also use a random UUID for the correlation ID generation (avoids doing this in production: this is OK in terms of uniqueness but not optimal in terms of performance because randomness is not cheap).
+
+Here is how to declare the RPC client:
+
+.Customizing the RPC client
+[source,java,indent=0]
+--------
+include::{test-examples}/RpcApi.java[tag=rpc-custom-client-creation]
+--------
+<1> Declare the reply-to queue
+<2> Use a random UUID as correlation ID
+<3> Use the `correlation-id` property for the request
+<4> Set the `reply-to` property
+<5> Set the address to send request messages to
+
+We just have to tell the RPC server to get the correlation ID from the request `correlation-id` property:
+
+.Customizing the RPC server
+[source,java,indent=0]
+--------
+include::{test-examples}/RpcApi.java[tag=rpc-custom-server-creation]
+--------
+<1> Get the correlation ID from the request `correlation-id` property
 
 == Dependencies
 

src/main/java/com/rabbitmq/model/Message.java

@@ -138,7 +138,9 @@ public interface Message {
   // TODO support message annotations
   // TODO support message headers
 
-  MessageAddressBuilder address();
+  MessageAddressBuilder toAddress();
+
+  MessageAddressBuilder replyToAddress();
 
   interface MessageAddressBuilder extends AddressBuilder<MessageAddressBuilder> {
 

src/main/java/com/rabbitmq/model/amqp/AmqpMessage.java

@@ -24,6 +24,7 @@
 import java.math.BigDecimal;
 import java.util.Date;
 import java.util.UUID;
+import java.util.function.BiConsumer;
 import org.apache.qpid.protonj2.client.exceptions.ClientException;
 import org.apache.qpid.protonj2.types.*;
 
@@ -359,18 +360,29 @@ public Object removeProperty(String key) {
   }
 
   @Override
-  public MessageAddressBuilder address() {
-    return new DefaultMessageAddressBuilder(this);
+  public MessageAddressBuilder toAddress() {
+    return new DefaultMessageAddressBuilder(this, DefaultMessageAddressBuilder.TO_CALLBACK);
+  }
+
+  @Override
+  public MessageAddressBuilder replyToAddress() {
+    return new DefaultMessageAddressBuilder(this, DefaultMessageAddressBuilder.REPLY_TO_CALLBACK);
   }
 
   private static class DefaultMessageAddressBuilder
       extends DefaultAddressBuilder<MessageAddressBuilder> implements MessageAddressBuilder {
 
+    private static final BiConsumer<Message, String> TO_CALLBACK = Message::to;
+    private static final BiConsumer<Message, String> REPLY_TO_CALLBACK = Message::replyTo;
+
     private final Message message;
+    private final BiConsumer<Message, String> buildCallback;
 
-    private DefaultMessageAddressBuilder(Message message) {
+    private DefaultMessageAddressBuilder(
+        Message message, BiConsumer<Message, String> buildCallback) {
       super(null);
       this.message = message;
+      this.buildCallback = buildCallback;
     }
 
     @Override
@@ -380,7 +392,7 @@ MessageAddressBuilder result() {
 
     @Override
     public Message message() {
-      this.message.to(this.address());
+      this.buildCallback.accept(this.message, this.address());
       return this.message;
     }
   }

src/main/java/com/rabbitmq/model/amqp/AmqpRpcClient.java

@@ -42,7 +42,6 @@ class AmqpRpcClient implements RpcClient {
   AmqpRpcClient(RpcSupport.AmqpRpcClientBuilder builder) {
     AmqpConnection connection = builder.connection();
 
-    // TODO request address cannot be null
     AmqpPublisherBuilder publisherBuilder = (AmqpPublisherBuilder) connection.publisherBuilder();
     ((DefaultAddressBuilder<?>) builder.requestAddress()).copyTo(publisherBuilder.addressBuilder());
     this.publisher = publisherBuilder.build();

src/main/java/com/rabbitmq/model/amqp/RpcSupport.java

@@ -78,9 +78,6 @@ Function<Message, Object> correlationIdExtractor() {
 
     @Override
     public RpcClient build() {
-      if (this.requestAddressBuilder.address() == null) {
-        throw new IllegalArgumentException("Request address cannot be null");
-      }
       return new AmqpRpcClient(this);
     }
 

src/test/java/com/rabbitmq/model/amqp/AddressFormatTest.java

@@ -179,14 +179,14 @@ void exchangeKeyInToField(TestInfo info) {
 
       CountDownLatch failedLatch = new CountDownLatch(2);
       publisher.publish(
-          publisher.message().address().exchange(e).message(),
+          publisher.message().toAddress().exchange(e).message(),
           ctx -> {
             if (ctx.status() == Publisher.Status.FAILED) {
               failedLatch.countDown();
             }
           });
       publisher.publish(
-          publisher.message().address().exchange(e).key("foo").message(),
+          publisher.message().toAddress().exchange(e).key("foo").message(),
           ctx -> {
             if (ctx.status() == Publisher.Status.FAILED) {
               failedLatch.countDown();
@@ -205,8 +205,8 @@ void exchangeKeyInToField(TestInfo info) {
               })
           .build();
 
-      publisher.publish(publisher.message().address().exchange(e).key(k).message(), ctx -> {});
-      publisher.publish(publisher.message().address().queue(q).message(), ctx -> {});
+      publisher.publish(publisher.message().toAddress().exchange(e).key(k).message(), ctx -> {});
+      publisher.publish(publisher.message().toAddress().queue(q).message(), ctx -> {});
       assertThat(consumeLatch).completes();
     } finally {
       management.queueDeletion().delete(q);

src/test/java/com/rabbitmq/model/amqp/RpcTest.java

@@ -92,10 +92,7 @@ void rpcWithDefaults() {
                       () -> {
                         String request = UUID.randomUUID().toString();
                         CompletableFuture<Message> responseFuture =
-                            rpcClient.publish(
-                                rpcClient
-                                    .message(request.getBytes(UTF_8))
-                                    .messageId(UUID.randomUUID()));
+                            rpcClient.publish(rpcClient.message(request.getBytes(UTF_8)));
                         Message response = responseFuture.get(10, TimeUnit.SECONDS);
                         assertThat(response.body()).asString(UTF_8).isEqualTo(process(request));
                         latch.countDown();
@@ -141,7 +138,7 @@ void rpcWithCustomSettings() {
               (ctx, request) -> {
                 Message reply = HANDLER.handle(ctx, request);
                 return reply
-                    .address()
+                    .toAddress()
                     .queue(request.property("reply-to-queue").toString())
                     .message();
               })
@@ -166,6 +163,54 @@ void rpcWithCustomSettings() {
     }
   }
 
+  @Test
+  void rpcUseCorrelationIdRequestProperty() {
+    try (Connection clientConnection = environment.connectionBuilder().build();
+        Connection serverConnection = environment.connectionBuilder().build()) {
+
+      String requestQueue = serverConnection.management().queue().exclusive(true).declare().name();
+
+      String replyToQueue =
+          clientConnection.management().queue().autoDelete(true).exclusive(true).declare().name();
+      RpcClient rpcClient =
+          clientConnection
+              .rpcClientBuilder()
+              .correlationIdSupplier(UUID::randomUUID)
+              .requestPostProcessor(
+                  (msg, corrId) ->
+                      msg.correlationId(corrId).replyToAddress().queue(replyToQueue).message())
+              .replyToQueue(replyToQueue)
+              .requestAddress()
+              .queue(requestQueue)
+              .rpcClient()
+              .build();
+
+      serverConnection
+          .rpcServerBuilder()
+          .correlationIdExtractor(Message::correlationId)
+          .requestQueue(requestQueue)
+          .handler(HANDLER)
+          .build();
+
+      int requestCount = 100;
+      CountDownLatch latch = new CountDownLatch(requestCount);
+      IntStream.range(0, requestCount)
+          .forEach(
+              ignored ->
+                  executorService.submit(
+                      () -> {
+                        String request = UUID.randomUUID().toString();
+                        CompletableFuture<Message> responseFuture =
+                            rpcClient.publish(rpcClient.message(request.getBytes(UTF_8)));
+                        Message response = responseFuture.get(10, TimeUnit.SECONDS);
+                        assertThat(response.body()).asString(UTF_8).isEqualTo(process(request));
+                        latch.countDown();
+                        return null;
+                      }));
+      TestUtils.assertThat(latch).completes();
+    }
+  }
+
   @Test
   void rpcShouldRecoverAfterConnectionIsClosed()
       throws ExecutionException, InterruptedException, TimeoutException {

src/test/java/com/rabbitmq/model/docs/Api.java

@@ -117,15 +117,15 @@ void targetAddressNull() {
         .build(); // <1>
 
     Message message1 = publisher.message()
-        .address().exchange("foo").key("bar") // <2>
+        .toAddress().exchange("foo").key("bar") // <2>
         .message();
 
     Message message2 = publisher.message()
-        .address().exchange("foo") // <3>
+        .toAddress().exchange("foo") // <3>
         .message();
 
     Message message3 = publisher.message()
-        .address().queue("my-queue") // <4>
+        .toAddress().queue("my-queue") // <4>
         .message();
     // end::target-address-null[]
   }

src/test/java/com/rabbitmq/model/docs/RpcApi.java

@@ -0,0 +1,71 @@
+package com.rabbitmq.model.docs;
+
+import com.rabbitmq.model.Connection;
+import com.rabbitmq.model.Message;
+import com.rabbitmq.model.RpcClient;
+import com.rabbitmq.model.RpcServer;
+
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+public class RpcApi {
+
+  void rpcWithDefaults() throws Exception {
+    Connection connection = null;
+    // tag::rpc-server-creation[]
+    RpcServer rpcServer = connection.rpcServerBuilder() // <1>
+        .requestQueue("rpc-server") // <2>
+        .handler((ctx, req) -> { // <3>
+          String in = new String(req.body(), UTF_8);
+          String out = "*** " + in + " ***";
+          return ctx.message(out.getBytes(UTF_8)); // <4>
+        }).build();
+    // end::rpc-server-creation[]
+
+    // tag::rpc-client-creation[]
+    RpcClient rpcClient = connection.rpcClientBuilder() // <1>
+        .requestAddress().queue("rpc-server") // <2>
+        .rpcClient()
+        .build();
+    // end::rpc-client-creation[]
+
+    // tag::rpc-client-request[]
+    Message request = rpcClient.message("hello".getBytes(UTF_8)); // <1>
+    CompletableFuture<Message> replyFuture = rpcClient.publish(request); // <2>
+    Message reply = replyFuture.get(10, TimeUnit.SECONDS); // <3>
+    // end::rpc-client-request[]
+  }
+
+  void rpcWithCustomSettings() throws Exception {
+    Connection connection = null;
+    // tag::rpc-custom-client-creation[]
+    String replyToQueue = connection.management().queue()
+        .autoDelete(true).exclusive(true)
+        .declare().name(); // <1>
+    RpcClient rpcClient = connection.rpcClientBuilder()
+        .correlationIdSupplier(UUID::randomUUID) // <2>
+        .requestPostProcessor((msg, corrId) ->
+            msg.correlationId(corrId) // <3>
+               .replyToAddress().queue(replyToQueue).message()) // <4>
+        .replyToQueue(replyToQueue)
+        .requestAddress().queue("rpc-server") // <5>
+        .rpcClient()
+        .build();
+    // end::rpc-custom-client-creation[]
+
+    // tag::rpc-custom-server-creation[]
+    RpcServer rpcServer = connection.rpcServerBuilder()
+        .correlationIdExtractor(Message::correlationId) // <1>
+        .requestQueue("rpc-server")
+        .handler((ctx, req) -> {
+          String in = new String(req.body(), UTF_8);
+          String out = "*** " + in + " ***";
+          return ctx.message(out.getBytes(UTF_8));
+        }).build();
+    // end::rpc-custom-server-creation[]
+  }
+}