Add batch size hint parameter to create batch context

And add Javadoc to BatchContext.

Author: acogoluegnes

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

@@ -135,13 +135,44 @@ interface Context {
      */
     void requeue(Map<String, Object> annotations);
 
-    BatchContext batch();
+    /**
+     * Create a batch context to accumulate message contexts and settle them at once.
+     *
+     * <p>The message context the batch context is created from is <b>not</b> added to the batch
+     * context.
+     *
+     * @return the created batch context
+     */
+    BatchContext batch(int batchSizeHint);
   }
 
+  /**
+   * Context to accumulate message contexts and settle them at once.
+   *
+   * <p>A {@link BatchContext} is also a {@link Context}: the same methods are available to settle
+   * the messages.
+   *
+   * <p>Only "simple" (not batch) message contexts can be added to a batch context. Calling {@link
+   * Context#batch()} on a batch context returns the instance itself.
+   *
+   * @see <a
+   *     href="https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-transport-v1.0-os.html#type-disposition">AMQP
+   *     1.0 Disposition performative</a>
+   */
   interface BatchContext extends Context {
 
+    /**
+     * Add a message context to the batch context.
+     *
+     * @param context the message context to add
+     */
     void add(Context context);
 
+    /**
+     * Get the current number of message contexts in the batch context.
+     *
+     * @return current number of message contexts in the batch
+     */
     int size();
   }
 }

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

@@ -447,8 +447,9 @@ public void requeue(Map<String, Object> annotations) {
     }
 
     @Override
-    public BatchContext batch() {
+    public BatchContext batch(int batchSizeHint) {
       return new BatchDeliveryContext(
+          batchSizeHint,
           protonExecutor,
           protonReceiver,
           metricsCollector,
@@ -481,7 +482,7 @@ private static final class BatchDeliveryContext implements BatchContext {
 
     private static final org.apache.qpid.protonj2.types.transport.DeliveryState REJECTED =
         new Rejected();
-    private final List<DeliveryContext> contexts = new ArrayList<>();
+    private final List<DeliveryContext> contexts;
     private final AtomicBoolean settled = new AtomicBoolean(false);
     private final Scheduler protonExecutor;
     private final ProtonReceiver protonReceiver;
@@ -491,12 +492,14 @@ private static final class BatchDeliveryContext implements BatchContext {
     private final AmqpConsumer consumer;
 
     private BatchDeliveryContext(
+        int batchSizeHint,
         Scheduler protonExecutor,
         ProtonReceiver protonReceiver,
         MetricsCollector metricsCollector,
         AtomicLong unsettledMessageCount,
         Runnable replenishCreditOperation,
         AmqpConsumer consumer) {
+      this.contexts = new ArrayList<>(batchSizeHint);
       this.protonExecutor = protonExecutor;
       this.protonReceiver = protonReceiver;
       this.metricsCollector = metricsCollector;
@@ -563,7 +566,7 @@ public void requeue(Map<String, Object> annotations) {
     }
 
     @Override
-    public BatchContext batch() {
+    public BatchContext batch(int batchSizeHint) {
       return this;
     }
 

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

@@ -27,8 +27,6 @@ final class SerialNumberUtils {
   // SERIAL_BITS = 32
   // 2 ^ SERIAL_BITS
   static final long SERIAL_SPACE = 0x100000000L;
-  // 2 ^ (SERIAL_BITS - 1) - 1
-  private static final long SERIAL_MAX_ADDEND = 0x7fffffffL;
   // 2 ^ (SERIAL_BITS - 1)
   private static final long COMPARE = 2_147_483_648L;
 

src/test/java/com/rabbitmq/client/amqp/docs/Api.java

@@ -90,17 +90,18 @@ void settlingMessagesInBatch() {
     Connection connection = null;
 
     // tag::settling-message-in-batch[]
+    int batchSize = 10;
     Consumer.MessageHandler handler = new Consumer.MessageHandler() {
       volatile Consumer.BatchContext batch = null;  // <1>
       @Override
       public void handle(Consumer.Context context, Message message) {
         if (batch == null) {
-          batch = context.batch();  // <2>
+          batch = context.batch(batchSize);  // <2>
         }
         boolean success = process(message);
         if (success) {
           batch.add(context);  // <3>
-          if (batch.size() == 10) {
+          if (batch.size() == batchSize) {
             batch.accept();  // <4>
             batch = null;  // <5>
           }

src/test/java/com/rabbitmq/client/amqp/impl/BatchDispositionTest.java

@@ -59,7 +59,7 @@ void test() {
           @Override
           public void handle(Consumer.Context context, Message message) {
             if (batch == null) {
-              batch = context.batch();
+              batch = context.batch(batchSize);
             }
 
             boolean success = processMessage(message);
@@ -154,16 +154,20 @@ private void dispose() {
     }
 
     @Override
-    public Consumer.BatchContext batch() {
-      return new TestBatchContext();
+    public Consumer.BatchContext batch(int batchSizeHint) {
+      return new TestBatchContext(batchSizeHint);
     }
   }
 
   static class TestBatchContext implements Consumer.BatchContext {
 
-    private final List<TestContext> contexts = new ArrayList<>(batchSize);
+    private final List<TestContext> contexts;
     private final AtomicBoolean disposed = new AtomicBoolean(false);
 
+    public TestBatchContext(int batchSizeHint) {
+      this.contexts = new ArrayList<>(batchSizeHint);
+    }
+
     @Override
     public void add(Consumer.Context context) {
       this.contexts.add((TestContext) context);
@@ -218,7 +222,7 @@ private void dispose() {
     }
 
     @Override
-    public Consumer.BatchContext batch() {
+    public Consumer.BatchContext batch(int batchSizeHint) {
       return this;
     }
   }

src/test/java/com/rabbitmq/client/amqp/impl/ConsumerOutcomeTest.java

@@ -128,7 +128,7 @@ void requeuedMessageWithAnnotationShouldContainAnnotationsOnRedelivery(boolean b
               messages.offer(message);
               if (deliveryCount.get() == 1) {
                 if (batch) {
-                  Consumer.BatchContext bc = ctx.batch();
+                  Consumer.BatchContext bc = ctx.batch(1);
                   bc.add(ctx);
                   ctx = bc;
                 }
@@ -195,7 +195,7 @@ void discardedMessageWithAnnotationsShouldBeDeadLeadLetteredAndContainAnnotation
         .messageHandler(
             (ctx, msg) -> {
               if (batch) {
-                Consumer.BatchContext bc = ctx.batch();
+                Consumer.BatchContext bc = ctx.batch(1);
                 bc.add(ctx);
                 ctx = bc;
               }
@@ -260,7 +260,7 @@ void batchAcceptShouldSettleMessages() {
         .messageHandler(
             (ctx, msg) -> {
               if (batchContext.get() == null) {
-                batchContext.set(ctx.batch());
+                batchContext.set(ctx.batch(batchSize));
               }
               if (random.nextInt(10) == 0) {
                 ctx.discard();

src/test/java/com/rabbitmq/client/amqp/perf/AmqpPerfTest.java

@@ -116,7 +116,7 @@ public static void main(String[] args) throws IOException {
             (context, message) -> {
               recordMessage.accept(message);
               if (batch.get() == null) {
-                batch.set(context.batch());
+                batch.set(context.batch(disposeEvery));
               }
               batch.get().add(context);
               if (batch.get().size() == disposeEvery) {