Add SubscriptionListener

References #38

Author: acogoluegnes

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

@@ -64,6 +64,8 @@ public interface ConsumerBuilder {
    */
   ConsumerBuilder name(String name);
 
+  ConsumerBuilder subscriptionListener(SubscriptionListener subscriptionListener);
+
   /**
    * Enable {@link ManualTrackingStrategy}.
    *

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

@@ -0,0 +1,26 @@
+// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.stream;
+
+public interface SubscriptionListener {
+
+  void preSubscribe(SubscriptionContext subscriptionContext);
+
+  interface SubscriptionContext {
+
+    OffsetSpecification offsetSpecification();
+
+    void offsetSpecification(OffsetSpecification offsetSpecification);
+  }
+}

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

@@ -23,6 +23,8 @@
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.StreamDoesNotExistException;
 import com.rabbitmq.stream.StreamException;
+import com.rabbitmq.stream.SubscriptionListener;
+import com.rabbitmq.stream.SubscriptionListener.SubscriptionContext;
 import com.rabbitmq.stream.impl.Client.ChunkListener;
 import com.rabbitmq.stream.impl.Client.CreditNotification;
 import com.rabbitmq.stream.impl.Client.MessageListener;
@@ -83,6 +85,7 @@ Runnable subscribe(
       String stream,
       OffsetSpecification offsetSpecification,
       String trackingReference,
+      SubscriptionListener subscriptionListener,
       MessageHandler messageHandler) {
     // FIXME fail immediately if there's no locator (can provide a supplier that does not retry)
     List<Client.Broker> candidates = findBrokersForStream(stream);
@@ -95,7 +98,12 @@ Runnable subscribe(
     // we keep this instance when we move the subscription from a client to another one
     SubscriptionTracker subscriptionTracker =
         new SubscriptionTracker(
-            consumer, stream, offsetSpecification, trackingReference, messageHandler);
+            consumer,
+            stream,
+            offsetSpecification,
+            trackingReference,
+            subscriptionListener,
+            messageHandler);
 
     String key = keyForClientSubscription(newNode);
 
@@ -212,6 +220,7 @@ private static class SubscriptionTracker {
     private final String offsetTrackingReference;
     private final MessageHandler messageHandler;
     private final StreamConsumer consumer;
+    private final SubscriptionListener subscriptionListener;
     private volatile long offset;
     private volatile boolean hasReceivedSomething = false;
     private volatile byte subscriptionIdInClient;
@@ -223,11 +232,13 @@ private SubscriptionTracker(
         String stream,
         OffsetSpecification initialOffsetSpecification,
         String offsetTrackingReference,
+        SubscriptionListener subscriptionListener,
         MessageHandler messageHandler) {
       this.consumer = consumer;
       this.stream = stream;
       this.initialOffsetSpecification = initialOffsetSpecification;
       this.offsetTrackingReference = offsetTrackingReference;
+      this.subscriptionListener = subscriptionListener;
       this.messageHandler = messageHandler;
     }
 
@@ -635,7 +646,7 @@ synchronized void add(
             update(previousSubscriptions, subscriptionId, subscriptionTracker);
 
         String offsetTrackingReference = subscriptionTracker.offsetTrackingReference;
-        if (subscriptionTracker.offsetTrackingReference != null) {
+        if (offsetTrackingReference != null) {
           long trackedOffset =
               client.queryOffset(offsetTrackingReference, subscriptionTracker.stream);
           if (trackedOffset != 0) {
@@ -666,12 +677,20 @@ synchronized void add(
           subscriptionProperties.put("name", subscriptionTracker.offsetTrackingReference);
         }
 
+        SubscriptionContext subscriptionContext =
+            new DefaultSubscriptionContext(offsetSpecification);
+        subscriptionTracker.subscriptionListener.preSubscribe(subscriptionContext);
+        LOGGER.info(
+            "Computed offset specification {}, offset specification used after subscription listener {}",
+            offsetSpecification,
+            subscriptionContext.offsetSpecification());
+
         // FIXME consider using fewer initial credits
         Client.Response subscribeResponse =
             client.subscribe(
                 subscriptionId,
                 subscriptionTracker.stream,
-                offsetSpecification,
+                subscriptionContext.offsetSpecification(),
                 10,
                 subscriptionProperties);
         if (!subscribeResponse.isOk()) {
@@ -767,4 +786,28 @@ synchronized void close() {
       }
     }
   }
+
+  private static final class DefaultSubscriptionContext implements SubscriptionContext {
+
+    private volatile OffsetSpecification offsetSpecification;
+
+    private DefaultSubscriptionContext(OffsetSpecification computedOffsetSpecification) {
+      this.offsetSpecification = computedOffsetSpecification;
+    }
+
+    @Override
+    public OffsetSpecification offsetSpecification() {
+      return this.offsetSpecification;
+    }
+
+    @Override
+    public void offsetSpecification(OffsetSpecification offsetSpecification) {
+      this.offsetSpecification = offsetSpecification;
+    }
+
+    @Override
+    public String toString() {
+      return "SubscriptionContext{" + "offsetSpecification=" + offsetSpecification + '}';
+    }
+  }
 }

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

@@ -17,6 +17,7 @@
 import com.rabbitmq.stream.MessageHandler;
 import com.rabbitmq.stream.MessageHandler.Context;
 import com.rabbitmq.stream.OffsetSpecification;
+import com.rabbitmq.stream.SubscriptionListener;
 import com.rabbitmq.stream.impl.StreamConsumerBuilder.TrackingConfiguration;
 import com.rabbitmq.stream.impl.StreamEnvironment.TrackingConsumerRegistration;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -57,7 +58,8 @@ class StreamConsumer implements Consumer {
       String name,
       StreamEnvironment environment,
       TrackingConfiguration trackingConfiguration,
-      boolean lazyInit) {
+      boolean lazyInit,
+      SubscriptionListener subscriptionListener) {
 
     try {
       this.name = name;
@@ -100,6 +102,7 @@ class StreamConsumer implements Consumer {
                     stream,
                     offsetSpecification,
                     this.name,
+                    subscriptionListener,
                     messageHandlerWithOrWithoutTracking);
 
             this.status = Status.RUNNING;

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

@@ -18,6 +18,7 @@
 import com.rabbitmq.stream.MessageHandler;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.StreamException;
+import com.rabbitmq.stream.SubscriptionListener;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 import java.time.Duration;
@@ -34,6 +35,7 @@ class StreamConsumerBuilder implements ConsumerBuilder {
   private DefaultAutoTrackingStrategy autoTrackingStrategy;
   private DefaultManualTrackingStrategy manualTrackingStrategy;
   private boolean lazyInit = false;
+  private SubscriptionListener subscriptionListener = subscriptionContext -> {};
 
   public StreamConsumerBuilder(StreamEnvironment environment) {
     this.environment = environment;
@@ -77,6 +79,15 @@ public ConsumerBuilder name(String name) {
     return this;
   }
 
+  @Override
+  public ConsumerBuilder subscriptionListener(SubscriptionListener subscriptionListener) {
+    if (subscriptionListener == null) {
+      throw new IllegalArgumentException("The subscription listener cannot be null");
+    }
+    this.subscriptionListener = subscriptionListener;
+    return this;
+  }
+
   @Override
   public ManualTrackingStrategy manualTrackingStrategy() {
     this.manualTrackingStrategy = new DefaultManualTrackingStrategy(this);
@@ -142,7 +153,8 @@ public Consumer build() {
               this.name,
               this.environment,
               trackingConfiguration,
-              this.lazyInit);
+              this.lazyInit,
+              this.subscriptionListener);
       environment.addConsumer((StreamConsumer) consumer);
     } else {
       consumer =

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

@@ -28,6 +28,7 @@
 import com.rabbitmq.stream.ProducerBuilder;
 import com.rabbitmq.stream.StreamCreator;
 import com.rabbitmq.stream.StreamException;
+import com.rabbitmq.stream.SubscriptionListener;
 import com.rabbitmq.stream.compression.CompressionCodecFactory;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import com.rabbitmq.stream.impl.OffsetTrackingCoordinator.Registration;
@@ -467,10 +468,16 @@ Runnable registerConsumer(
       String stream,
       OffsetSpecification offsetSpecification,
       String trackingReference,
+      SubscriptionListener subscriptionListener,
       MessageHandler messageHandler) {
     Runnable closingCallback =
         this.consumersCoordinator.subscribe(
-            consumer, stream, offsetSpecification, trackingReference, messageHandler);
+            consumer,
+            stream,
+            offsetSpecification,
+            trackingReference,
+            subscriptionListener,
+            messageHandler);
     return closingCallback;
   }