Add consumer flow strategy Javadoc

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/ConsumerBuilder.java

@@ -166,6 +166,8 @@ public interface ConsumerBuilder {
    *
    * @return the flow configuration
    * @since 0.11.0
+   * @see ConsumerFlowStrategy#creditOnChunkArrival()
+   * @see MessageHandler.Context#processed()
    */
   FlowConfiguration flow();
 
@@ -231,7 +233,11 @@ interface AutoTrackingStrategy {
   /**
    * Message flow configuration.
    *
+   * <p>The default configuration uses {@link ConsumerFlowStrategy#creditOnChunkArrival()}.
+   *
    * @since 0.11.0
+   * @see ConsumerFlowStrategy#creditOnChunkArrival()
+   * @see MessageHandler.Context#processed()
    */
   interface FlowConfiguration {
 
@@ -240,11 +246,27 @@ interface FlowConfiguration {
      *
      * <p>Default is 1.
      *
+     * <p>This calls uses {@link ConsumerFlowStrategy#creditOnChunkArrival(int)}.
+     *
      * @param initialCredits the number of initial credits
      * @return this configuration instance
+     * @see ConsumerFlowStrategy#creditOnChunkArrival(int)
      */
     FlowConfiguration initialCredits(int initialCredits);
 
+    /**
+     * Flow strategy to use
+     *
+     * @param strategy the strategy to use
+     * @return this configuration instance
+     * @since 0.12.0
+     * @see ConsumerFlowStrategy
+     * @see ConsumerFlowStrategy#creditOnChunkArrival()
+     * @see ConsumerFlowStrategy#creditOnChunkArrival(int)
+     * @see ConsumerFlowStrategy#creditWhenHalfMessagesProcessed()
+     * @see ConsumerFlowStrategy#creditWhenHalfMessagesProcessed(int)
+     * @see ConsumerFlowStrategy#creditOnProcessedMessageCount(int, double)
+     */
     FlowConfiguration strategy(ConsumerFlowStrategy strategy);
 
     /**

src/main/java/com/rabbitmq/stream/ConsumerFlowStrategy.java

@@ -15,45 +15,159 @@
 
 import java.util.concurrent.atomic.AtomicLong;
 
+/**
+ * Contract to determine when a subscription provides credits to get more messages.
+ *
+ * <p>The broker delivers "chunks" of messages to consumers. A chunk can contain from 1 to several
+ * thousands of messages. The broker send chunks as long as the subscription has <em>credits</em>. A
+ * client connection can provide credits for a given subscription and the broker will send the
+ * corresponding number of chunks (1 credit = 1 chunk).
+ *
+ * <p>This credit mechanism avoids overwhelming a consumer with messages. A consumer does not want
+ * to provide a credit only when it is done with messages of a chunk, because it will be idle
+ * between its credit request and the arrival of the next chunk. The idea is to keep consumers busy
+ * as much as possible, without accumulating an in-memory backlog on the client side. There is no
+ * ideal solution, it depends on the use cases and several parameters (processing time, network,
+ * etc).
+ *
+ * @since 0.12.0
+ * @see MessageHandler.Context#processed()
+ * @see ConsumerBuilder#flow()
+ */
 public interface ConsumerFlowStrategy {
 
+  /**
+   * The initial number of credits for a subscription.
+   *
+   * <p>It must be greater than 0. Values are usually between 1 and 10.
+   *
+   * @return initial number of credits
+   */
   int initialCredits();
 
+  /**
+   * Return the behavior for {@link MessageHandler.Context#processed()} calls.
+   *
+   * <p>This method is called for each chunk of messages. Implementations return a callback that
+   * will be called when applications consider a message dealt with and call {@link
+   * MessageHandler.Context#processed()}. The callback can count messages and provide credits
+   * accordingly.
+   *
+   * @param context chunk context
+   * @return the message processed callback
+   */
   MessageProcessedCallback start(Context context);
 
+  /** Chunk context. */
   interface Context {
 
+    /**
+     * Provide credits for the subscription.
+     *
+     * <p>{@link ConsumerFlowStrategy} implementation should always provide 1 credit a given chunk.
+     *
+     * @param credits the number of credits provided, usually 1
+     */
     void credits(int credits);
 
+    /**
+     * The number of messages in the chunk.
+     *
+     * @return number of messages in the chunk
+     */
     long messageCount();
   }
 
+  /** Behavior for {@link MessageHandler.Context#processed()} calls. */
   @FunctionalInterface
   interface MessageProcessedCallback {
 
+    /**
+     * Method called when {@link MessageHandler.Context#processed()} is called.
+     *
+     * <p>There is one instance of this class for a given chunk and it is called for the <code>
+     * processed()</code> calls of the message of this chunk.
+     *
+     * <p>Implementations can count messages and call {@link Context#credits(int)} when appropriate.
+     *
+     * <p>Note calls to {@link MessageHandler.Context#processed()} are not idempotent: an
+     * application can call the method several times for the same message and implementations must
+     * deal with these multiple calls if they impact their logic.
+     *
+     * @param messageContext context of the message
+     */
     void processed(MessageHandler.Context messageContext);
   }
 
+  /**
+   * Strategy that provides 1 initial credit and a credit on each new chunk.
+   *
+   * <p>Calls to {@link MessageHandler.Context#processed()} are ignored.
+   *
+   * @return flow strategy
+   */
   static ConsumerFlowStrategy creditOnChunkArrival() {
     return creditOnChunkArrival(1);
   }
 
+  /**
+   * Strategy that provides the specified number of initial credits and a credit on each new chunk.
+   *
+   * <p>Calls to {@link MessageHandler.Context#processed()} are ignored.
+   *
+   * @param initialCredits number of initial credits
+   * @return flow strategy
+   */
   static ConsumerFlowStrategy creditOnChunkArrival(int initialCredits) {
     return new CreditOnChunkArrivalConsumerFlowStrategy(initialCredits);
   }
 
+  /**
+   * Strategy that provides 1 initial credit and a credit when half of the chunk messages are
+   * processed.
+   *
+   * <p>Make sure to call {@link MessageHandler.Context#processed()} on every message when using
+   * this strategy, otherwise the broker may stop sending messages to the consumer.
+   *
+   * @return flow strategy
+   */
   static ConsumerFlowStrategy creditWhenHalfMessagesProcessed() {
     return creditOnProcessedMessageCount(1, 0.5);
   }
 
+  /**
+   * Strategy that provides the specified number of initial credits and a credit when half of the
+   * chunk messages are processed.
+   *
+   * <p>Make sure to call {@link MessageHandler.Context#processed()} on every message when using
+   * this strategy, otherwise the broker may stop sending messages to the consumer.
+   *
+   * @param initialCredits number of initial credits
+   * @return flow strategy
+   */
   static ConsumerFlowStrategy creditWhenHalfMessagesProcessed(int initialCredits) {
     return creditOnProcessedMessageCount(initialCredits, 0.5);
   }
 
+  /**
+   * Strategy that provides the specified number of initial credits and a credit when the specified
+   * ratio of the chunk messages are processed.
+   *
+   * <p>Make sure to call {@link MessageHandler.Context#processed()} on every message when using
+   * this strategy, otherwise the broker may stop sending messages to the consumer.
+   *
+   * @param initialCredits number of initial credits
+   * @return flow strategy
+   */
   static ConsumerFlowStrategy creditOnProcessedMessageCount(int initialCredits, double ratio) {
     return new MessageCountConsumerFlowStrategy(initialCredits, ratio);
   }
 
+  /**
+   * Strategy that provides the specified number of initial credits and a credit on each new chunk.
+   *
+   * <p>Calls to {@link MessageHandler.Context#processed()} are ignored.
+   */
   class CreditOnChunkArrivalConsumerFlowStrategy implements ConsumerFlowStrategy {
 
     private final int initialCredits;
@@ -74,6 +188,13 @@ public MessageProcessedCallback start(Context context) {
     }
   }
 
+  /**
+   * Strategy that provides the specified number of initial credits and a credit when the specified
+   * ratio of the chunk messages are processed.
+   *
+   * <p>Make sure to call {@link MessageHandler.Context#processed()} on every message when using
+   * this strategy, otherwise the broker may stop sending messages to the consumer.
+   */
   class MessageCountConsumerFlowStrategy implements ConsumerFlowStrategy {
 
     private final int initialCredits;

src/main/java/com/rabbitmq/stream/MessageHandler.java

@@ -85,6 +85,22 @@ interface Context {
      */
     Consumer consumer();
 
+    /**
+     * Mark the message as processed, potentially asking for more messages from the broker.
+     *
+     * <p>The exact behavior depends on the {@link ConsumerFlowStrategy} chosen when creating the
+     * consumer with {@link ConsumerBuilder#flow()}.
+     *
+     * <p>The call is a no-op for strategies like {@link
+     * ConsumerFlowStrategy#creditOnChunkArrival()} and {@link
+     * ConsumerFlowStrategy#creditOnChunkArrival(int)}.
+     *
+     * <p>Calling this method for each message is mandatory for strategies like {@link
+     * ConsumerFlowStrategy#creditWhenHalfMessagesProcessed()}, {@link
+     * ConsumerFlowStrategy#creditWhenHalfMessagesProcessed(int)}, and {@link
+     * ConsumerFlowStrategy#creditOnProcessedMessageCount(int, double)}, otherwise the broker may
+     * stop sending messages to the consumer.
+     */
     void processed();
   }
 }

src/main/java/com/rabbitmq/stream/impl/StreamConsumerBuilder.java

@@ -425,7 +425,7 @@ private DefaultFlowConfiguration(ConsumerBuilder consumerBuilder) {
       this.consumerBuilder = consumerBuilder;
     }
 
-    private ConsumerFlowStrategy strategy = ConsumerFlowStrategy.creditOnChunkArrival(1);
+    private ConsumerFlowStrategy strategy = ConsumerFlowStrategy.creditOnChunkArrival();
 
     @Override
     public FlowConfiguration initialCredits(int initialCredits) {