Support stream offset specification in consumer builder

Author: acogoluegnes

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

@@ -17,6 +17,8 @@
 // [email protected].
 package com.rabbitmq.client.amqp;
 
+import java.time.Instant;
+
 public interface ConsumerBuilder {
 
   ConsumerBuilder queue(String queue);
@@ -27,5 +29,30 @@ public interface ConsumerBuilder {
 
   ConsumerBuilder listeners(Resource.StateListener... listeners);
 
+  StreamOptions stream();
+
   Consumer build();
+
+  interface StreamOptions {
+
+    StreamOptions offset(long offset);
+
+    StreamOptions offset(Instant timestamp);
+
+    StreamOptions offset(StreamOffsetSpecification specification);
+
+    StreamOptions offset(String interval);
+
+    StreamOptions filterValues(String... values);
+
+    StreamOptions filterMatchUnfiltered(boolean matchUnfiltered);
+
+    ConsumerBuilder builder();
+  }
+
+  enum StreamOffsetSpecification {
+    FIRST,
+    LAST,
+    NEXT
+  }
 }

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

@@ -22,6 +22,8 @@
 import com.rabbitmq.client.amqp.AmqpException;
 import com.rabbitmq.client.amqp.Consumer;
 import com.rabbitmq.client.amqp.metrics.MetricsCollector;
+import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -51,6 +53,7 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
   private final MessageHandler messageHandler;
   private final Long id;
   private final String address;
+  private final Map<String, Object> filters;
   private final AmqpConnection connection;
   private final AtomicReference<PauseStatus> pauseStatus =
       new AtomicReference<>(PauseStatus.UNPAUSED);
@@ -76,9 +79,11 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
             .observationCollector()
             .subscribe(builder.queue(), builder.messageHandler());
     this.address = "/queue/" + builder.queue();
+    this.filters = Collections.unmodifiableMap(builder.filters());
     this.connection = builder.connection();
     this.sessionHandler = this.connection.createSessionHandler();
-    this.nativeReceiver = this.createNativeReceiver(this.sessionHandler.session(), this.address);
+    this.nativeReceiver =
+        this.createNativeReceiver(this.sessionHandler.session(), this.address, this.filters);
     this.initStateFromNativeReceiver(this.nativeReceiver);
     this.metricsCollector = this.connection.metricsCollector();
     this.startReceivingLoop();
@@ -134,19 +139,20 @@ public void close() {
 
   // internal API
 
-  private ClientReceiver createNativeReceiver(Session nativeSession, String address) {
+  private ClientReceiver createNativeReceiver(
+      Session nativeSession, String address, Map<String, Object> filters) {
     try {
+      ReceiverOptions receiverOptions =
+          new ReceiverOptions()
+              .deliveryMode(DeliveryMode.AT_LEAST_ONCE)
+              .autoAccept(false)
+              .autoSettle(false)
+              .creditWindow(0);
+      if (!filters.isEmpty()) {
+        receiverOptions.sourceOptions().filters(filters);
+      }
       return (ClientReceiver)
-          ExceptionUtils.wrapGet(
-              nativeSession
-                  .openReceiver(
-                      address,
-                      new ReceiverOptions()
-                          .deliveryMode(DeliveryMode.AT_LEAST_ONCE)
-                          .autoAccept(false)
-                          .autoSettle(false)
-                          .creditWindow(0))
-                  .openFuture());
+          ExceptionUtils.wrapGet(nativeSession.openReceiver(address, receiverOptions).openFuture());
     } catch (ClientException e) {
       throw ExceptionUtils.convert(e, "Error while creating receiver from '%s'", address);
     }
@@ -194,7 +200,8 @@ private void startReceivingLoop() {
   }
 
   void recoverAfterConnectionFailure() {
-    this.nativeReceiver = createNativeReceiver(this.sessionHandler.sessionNoCheck(), this.address);
+    this.nativeReceiver =
+        createNativeReceiver(this.sessionHandler.sessionNoCheck(), this.address, this.filters);
     this.initStateFromNativeReceiver(this.nativeReceiver);
     this.pauseStatus.set(PauseStatus.UNPAUSED);
     this.unsettledCount.set(0);

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

@@ -17,11 +17,13 @@
 // [email protected].
 package com.rabbitmq.client.amqp.impl;
 
+import static com.rabbitmq.client.amqp.impl.Assert.notNull;
+
 import com.rabbitmq.client.amqp.Consumer;
 import com.rabbitmq.client.amqp.ConsumerBuilder;
 import com.rabbitmq.client.amqp.Resource;
-import java.util.ArrayList;
-import java.util.List;
+import java.time.Instant;
+import java.util.*;
 
 class AmqpConsumerBuilder implements ConsumerBuilder {
 
@@ -30,6 +32,8 @@ class AmqpConsumerBuilder implements ConsumerBuilder {
   private Consumer.MessageHandler messageHandler;
   private int initialCredits = 100;
   private final List<Resource.StateListener> listeners = new ArrayList<>();
+  private final Map<String, Object> filters = new HashMap<>();
+  private final StreamOptions streamOptions = new DefaultStreamOptions(this, this.filters);
 
   AmqpConsumerBuilder(AmqpConnection connection) {
     this.connection = connection;
@@ -63,6 +67,11 @@ public ConsumerBuilder listeners(Resource.StateListener... listeners) {
     return this;
   }
 
+  @Override
+  public StreamOptions stream() {
+    return this.streamOptions;
+  }
+
   AmqpConnection connection() {
     return connection;
   }
@@ -83,6 +92,10 @@ List<Resource.StateListener> listeners() {
     return listeners;
   }
 
+  Map<String, Object> filters() {
+    return this.filters;
+  }
+
   @Override
   public Consumer build() {
     if (this.queue == null || this.queue.isBlank()) {
@@ -91,6 +104,68 @@ public Consumer build() {
     if (this.messageHandler == null) {
       throw new IllegalArgumentException("Message handler cannot be null");
     }
+
+    // TODO validate stream (filtering) configuration
+
     return this.connection.createConsumer(this);
   }
+
+  private static class DefaultStreamOptions implements StreamOptions {
+
+    private final Map<String, Object> filters;
+    private final ConsumerBuilder builder;
+
+    private DefaultStreamOptions(ConsumerBuilder builder, Map<String, Object> filters) {
+      this.builder = builder;
+      this.filters = filters;
+    }
+
+    @Override
+    public StreamOptions offset(long offset) {
+      this.filters.put("rabbitmq:stream-offset-spec", offset);
+      return this;
+    }
+
+    @Override
+    public StreamOptions offset(Instant timestamp) {
+      notNull(timestamp, "Timestamp offset cannot be null");
+      this.offsetSpecification(Date.from(timestamp));
+      return this;
+    }
+
+    @Override
+    public StreamOptions offset(StreamOffsetSpecification specification) {
+      notNull(specification, "Offset specification cannot be null");
+      this.offsetSpecification(specification.name().toLowerCase(Locale.ENGLISH));
+      return this;
+    }
+
+    @Override
+    public StreamOptions offset(String interval) {
+      notNull(interval, "Interval offset cannot be null");
+      this.offsetSpecification(interval);
+      return this;
+    }
+
+    @Override
+    public StreamOptions filterValues(String... values) {
+      this.filters.put("rabbitmq:stream-filter", values);
+      return this;
+    }
+
+    @Override
+    public StreamOptions filterMatchUnfiltered(boolean matchUnfiltered) {
+      this.filters.put("rabbitmq:stream-match-unfiltered", matchUnfiltered);
+      return this;
+    }
+
+    @Override
+    public ConsumerBuilder builder() {
+      return this.builder;
+    }
+
+    private void offsetSpecification(Object value) {
+      this.filters.put("rabbitmq:stream-offset-spec", value);
+    }
+  }
 }

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

@@ -0,0 +1,29 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.amqp.impl;
+
+abstract class Assert {
+
+  private Assert() {}
+
+  static void notNull(Object object, String message) {
+    if (object == null) {
+      throw new IllegalArgumentException(message);
+    }
+  }
+}