More Javadoc

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/Consumer.java

@@ -25,7 +25,7 @@
  * @see Connection#consumerBuilder()
  * @see ConsumerBuilder
  */
-public interface Consumer extends AutoCloseable {
+public interface Consumer extends AutoCloseable, Resource {
 
   /** Pause the consumer to stop receiving messages. */
   void pause();

src/main/java/com/rabbitmq/client/amqp/Resource.java

@@ -17,30 +17,83 @@
 // info@rabbitmq.com.
 package com.rabbitmq.client.amqp;
 
+/**
+ * Marker interface for {@link Resource}-like classes.
+ *
+ * <p>Instances of these classes have different states during their lifecycle: open, recovering,
+ * closed, etc. Application can be interested in taking some actions for a given state (e.g.
+ * stopping publishing when a {@link Publisher} is recovering after a connection problem and
+ * resuming publishing when it is open again).
+ *
+ * @see Connection
+ * @see Publisher
+ * @see Consumer
+ */
 public interface Resource {
 
+  /**
+   * Application listener for a {@link Resource}.
+   *
+   * <p>They are usually registered at creation time.
+   *
+   * @see ConnectionBuilder#listeners(StateListener...)
+   * @see PublisherBuilder#listeners(StateListener...)
+   * @see ConsumerBuilder#listeners(StateListener...)
+   */
   @FunctionalInterface
   interface StateListener {
 
+    /**
+     * Handle state change.
+     *
+     * @param context state change context
+     */
     void handle(Context context);
   }
 
+  /** Context of a resource state change. */
   interface Context {
 
+    /**
+     * The resource instance.
+     *
+     * @return resource instance
+     */
     Resource resource();
 
+    /**
+     * The failure cause, can be null.
+     *
+     * @return failure cause, null if no cause for failure
+     */
     Throwable failureCause();
 
+    /**
+     * The previous state of the resource.
+     *
+     * @return previous state
+     */
     State previousState();
 
+    /**
+     * The current (new) state of the resource.
+     *
+     * @return current state
+     */
     State currentState();
   }
 
+  /** Resource state. */
   enum State {
+    /** The resource is currently opening. */
     OPENING,
+    /** The resource is open and functional. */
     OPEN,
+    /** The resource is recovering. */
     RECOVERING,
+    /** The resource is closing. */
     CLOSING,
+    /** The resource is closed. */
     CLOSED
   }
 }

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

@@ -19,14 +19,43 @@
 
 import java.util.concurrent.CompletableFuture;
 
+/**
+ * Client support class for RPC.
+ *
+ * @see RpcClientBuilder
+ */
 public interface RpcClient extends AutoCloseable {
 
+  /**
+   * Create a message meant to be published by the underlying publisher instance.
+   *
+   * <p>Once published with the {@link #publish(Message)} the message instance should be not be
+   * modified or even reused.
+   *
+   * @return a message
+   */
   Message message();
 
+  /**
+   * Create a message meant to be published by the underlying publisher instance.
+   *
+   * <p>Once published with the {@link #publish(Message)} the message instance should be not be
+   * modified or even reused.
+   *
+   * @param body message body
+   * @return a message with the provided body
+   */
   Message message(byte[] body);
 
+  /**
+   * Publish a request message and expect a response.
+   *
+   * @param message message request
+   * @return the response as {@link CompletableFuture}
+   */
   CompletableFuture<Message> publish(Message message);
 
+  /** Close the RPC client and its resources. */
   @Override
   void close();
 }

src/main/java/com/rabbitmq/client/amqp/RpcClientBuilder.java

@@ -22,24 +22,89 @@
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+/** API to configure and create a {@link RpcClient}. */
 public interface RpcClientBuilder {
 
+  /**
+   * Builder for the request address.
+   *
+   * @return the request address builder
+   */
   RpcClientAddressBuilder requestAddress();
 
+  /**
+   * The queue the client expects responses on.
+   *
+   * <p>The queue <b>must</b> exist if it is set.
+   *
+   * <p>The RPC client will create an exclusive, auto-delete queue if it is not set.
+   *
+   * @param replyToQueue reply queue
+   * @return this builder instance
+   */
   RpcClientBuilder replyToQueue(String replyToQueue);
 
+  /**
+   * The generator for correlation ID.
+   *
+   * <p>The default generator uses a fixed random UUID prefix and a strictly monotonic increasing
+   * sequence suffix.
+   *
+   * @param correlationIdSupplier correlation ID generator
+   * @return the this builder instance
+   */
   RpcClientBuilder correlationIdSupplier(Supplier<Object> correlationIdSupplier);
 
+  /**
+   * A callback before sending a request message.
+   *
+   * <p>The callback accepts the request message and the correlation ID as parameters. It must
+   * return the message that will be sent as request, after having potentially modified it.
+   *
+   * <p>The default post-processor sets the reply-to field and assigns the correlation ID to the
+   * message ID field.
+   *
+   * @param requestPostProcessor logic to post-process request message
+   * @return this builder instance
+   */
   RpcClientBuilder requestPostProcessor(BiFunction<Message, Object, Message> requestPostProcessor);
 
+  /**
+   * Callback to extract the correlation ID from a reply message.
+   *
+   * <p>The correlation ID is then used to correlate the reply message to an outstanding request
+   * message.
+   *
+   * <p>The default implementation uses the correlation ID field.
+   *
+   * @param correlationIdExtractor correlation ID extractor
+   * @return this builder instance
+   */
   RpcClientBuilder correlationIdExtractor(Function<Message, Object> correlationIdExtractor);
 
+  /**
+   * Timeout before failing outstanding requests.
+   *
+   * @param timeout timeout
+   * @return the builder instance
+   */
   RpcClientBuilder requestTimeout(Duration timeout);
 
+  /**
+   * Build the configured instance.
+   *
+   * @return the configured instance
+   */
   RpcClient build();
 
+  /** Builder for the request address. */
   interface RpcClientAddressBuilder extends AddressBuilder<RpcClientAddressBuilder> {
 
+    /**
+     * Go back to the RPC client builder.
+     *
+     * @return the RPC client builder
+     */
     RpcClientBuilder rpcClient();
   }
 }

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

@@ -17,21 +17,53 @@
 // info@rabbitmq.com.
 package com.rabbitmq.client.amqp;
 
+/**
+ * Client server class for RPC.
+ *
+ * @see RpcServerBuilder
+ */
 public interface RpcServer extends AutoCloseable {
 
+  /** Contract to process a request message and return a reply message. */
   @FunctionalInterface
   interface Handler {
 
+    /**
+     * Process request message.
+     *
+     * @param ctx context
+     * @param request request message
+     * @return the reply message
+     */
     Message handle(Context ctx, Message request);
   }
 
+  /** Request processing context. */
   interface Context {
 
+    /**
+     * Create a message meant to be published by the underlying publisher instance.
+     *
+     * <p>Once returned in the {@link Handler#handle(Context, Message)} the message instance should
+     * be not be modified or even reused.
+     *
+     * @return a message
+     */
     Message message();
 
+    /**
+     * Create a message meant to be published by the underlying publisher instance.
+     *
+     * <p>Once returned in the {@link Handler#handle(Context, Message)} the message instance should
+     * be not be modified or even reused.
+     *
+     * @param body message body
+     * @return a message with the provided body
+     */
     Message message(byte[] body);
   }
 
+  /** Close the RPC server and its resources. */
   @Override
   void close();
 }

src/main/java/com/rabbitmq/client/amqp/RpcServerBuilder.java

@@ -20,22 +20,52 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 
+/** API to configure and create a {@link RpcServer}. */
 public interface RpcServerBuilder {
 
+  /**
+   * The queue to wait for requests on.
+   *
+   * @param requestQueue request queue
+   * @return this builder instance
+   */
   RpcServerBuilder requestQueue(String requestQueue);
 
+  /**
+   * The logic to process requests and issue replies.
+   *
+   * @param handler handler
+   * @return this builder instance
+   */
   RpcServerBuilder handler(RpcServer.Handler handler);
 
-  RpcServerAddressBuilder replyToAddress();
-
+  /**
+   * Logic to extract the correlation ID from a request message.
+   *
+   * <p>The default implementation uses the message ID.
+   *
+   * @param correlationIdExtractor logic to extract the correlation ID
+   * @return this builder instance
+   */
   RpcServerBuilder correlationIdExtractor(Function<Message, Object> correlationIdExtractor);
 
+  /**
+   * A callback called after request processing but before sending the reply message.
+   *
+   * <p>The callback accepts the request message and the correlation ID as parameters. It must
+   * return the message that will be sent as the reply, after having potentially modified it.
+   *
+   * <p>The default implementation set the correlation ID field and returns the updated message.
+   *
+   * @param replyPostProcessor logic to post-process reply message
+   * @return this builder instance
+   */
   RpcServerBuilder replyPostProcessor(BiFunction<Message, Object, Message> replyPostProcessor);
 
+  /**
+   * Create the configured instance.
+   *
+   * @return the configured instance
+   */
   RpcServer build();
-
-  interface RpcServerAddressBuilder extends AddressBuilder<RpcServerAddressBuilder> {
-
-    RpcServerBuilder rpcServer();
-  }
 }

src/main/java/com/rabbitmq/client/amqp/impl/RpcSupport.java

@@ -147,8 +147,6 @@ static class AmqpRpcServerBuilder implements RpcServerBuilder {
 
     private String requestQueue;
     private RpcServer.Handler handler;
-    private final DefaultRpcServerAddressBuilder replyToAddressBuilder =
-        new DefaultRpcServerAddressBuilder(this);
     private Function<Message, Object> correlationIdExtractor;
     private BiFunction<Message, Object, Message> replyPostProcessor;
 
@@ -168,11 +166,6 @@ public RpcServerBuilder handler(RpcServer.Handler handler) {
       return this;
     }
 
-    @Override
-    public RpcServerAddressBuilder replyToAddress() {
-      return this.replyToAddressBuilder;
-    }
-
     @Override
     public RpcServerBuilder correlationIdExtractor(
         Function<Message, Object> correlationIdExtractor) {
@@ -212,21 +205,4 @@ BiFunction<Message, Object, Message> replyPostProcessor() {
       return this.replyPostProcessor;
     }
   }
-
-  private static class DefaultRpcServerAddressBuilder
-      extends DefaultAddressBuilder<RpcServerBuilder.RpcServerAddressBuilder>
-      implements RpcServerBuilder.RpcServerAddressBuilder {
-
-    private final AmqpRpcServerBuilder builder;
-
-    private DefaultRpcServerAddressBuilder(AmqpRpcServerBuilder builder) {
-      super(null);
-      this.builder = builder;
-    }
-
-    @Override
-    public RpcServerBuilder rpcServer() {
-      return this.builder;
-    }
-  }
 }