Adding a Flux based subscribeToEvents method (#1598)

* Adding a Flux based subscribeToEvents method

Signed-off-by: Artur Ciocanu <[email protected]>

* Simplify GRPC stream handling

Signed-off-by: Artur Ciocanu <[email protected]>

* Simplify Javadoc

Signed-off-by: Artur Ciocanu <[email protected]>

* Fix unit tests and simplify implementation

Signed-off-by: Artur Ciocanu <[email protected]>

* Adding event subscriber stream observer to simplify subscription logic

Signed-off-by: Artur Ciocanu <[email protected]>

* Use start() method to start stream subscription

Signed-off-by: Artur Ciocanu <[email protected]>

* Add unit test for event suscriber observer

Signed-off-by: Artur Ciocanu <[email protected]>

* Improve the tests a little bit

Signed-off-by: Artur Ciocanu <[email protected]>

* Remove the unnecessary method

Signed-off-by: Artur Ciocanu <[email protected]>

* Improve error handling and use CloudEvent wrapper

Signed-off-by: Artur Ciocanu <[email protected]>

* Fix unit tests asserts

Signed-off-by: Artur Ciocanu <[email protected]>

* Adjust Java examples for Subscriber

Signed-off-by: Artur Ciocanu <[email protected]>

---------

Signed-off-by: Artur Ciocanu <[email protected]>

Author: artur-ciocanu

examples/src/main/java/io/dapr/examples/pubsub/stream/README.md

@@ -49,7 +49,7 @@ The subscriber uses the `DaprPreviewClient` interface to use a new feature where
 
 The publisher is a simple Java application with a main method that uses the Dapr gRPC Client to publish 10 messages to a specific topic.
 
-In the `Subscriber.java` file, you will find the `Subscriber` class, containing the main method. The main method declares a `DaprPreviewClient` using the `DaprClientBuilder` class. When invoking `subscribeToEvents`, the subscriber provides an implementation of the `SubscriptionListener` interface, receiving a `Subscription` object. The `Subscription` object implements the `Closeable` interface and the `close()` method must be used to stop the subscription.
+In the `Subscriber.java` file, you will find the `Subscriber` class, containing the main method. The main method declares a `DaprPreviewClient` using the `DaprClientBuilder` class. When invoking `subscribeToEvents`, the method returns a `Flux<CloudEvent<T>>` that can be processed using reactive operators like `doOnNext()` for event handling and `doOnError()` for error handling. The example uses `blockLast()` to keep the subscriber running indefinitely. For production use cases requiring explicit subscription lifecycle control, you can use `.subscribe()` which returns a `Disposable` that can be disposed via `disposable.dispose()`.
 
 ```java
 public class Subscriber {
@@ -59,25 +59,19 @@ public class Subscriber {
   public static void main(String[] args) throws Exception {
     String topicName = getTopicName(args);
     try (var client = new DaprClientBuilder().buildPreviewClient()) {
-      var subscription = client.subscribeToEvents(
+      // Subscribe to events using the Flux-based reactive API
+      // The stream will emit CloudEvent<String> objects as they arrive
+      client.subscribeToEvents(
           PUBSUB_NAME,
           topicName,
-          new SubscriptionListener<>() {
-
-            @Override
-            public Mono<Status> onEvent(CloudEvent<String> event) {
-              System.out.println("Subscriber got: " + event.getData());
-              return Mono.just(Status.SUCCESS);
-            }
-
-            @Override
-            public void onError(RuntimeException exception) {
-              System.out.println("Subscriber got exception: " + exception.getMessage());
-            }
-          },
-          TypeRef.STRING);
-
-      subscription.awaitTermination();
+          TypeRef.STRING)
+          .doOnNext(event -> {
+            System.out.println("Subscriber got: " + event.getData());
+          })
+          .doOnError(throwable -> {
+            System.out.println("Subscriber got exception: " + throwable.getMessage());
+          })
+          .blockLast();  // Blocks indefinitely until the stream completes (keeps the subscriber running)
     }
   }
 

examples/src/main/java/io/dapr/examples/pubsub/stream/Subscriber.java

@@ -14,10 +14,7 @@
 package io.dapr.examples.pubsub.stream;
 
 import io.dapr.client.DaprClientBuilder;
-import io.dapr.client.SubscriptionListener;
-import io.dapr.client.domain.CloudEvent;
 import io.dapr.utils.TypeRef;
-import reactor.core.publisher.Mono;
 
 /**
  * Subscriber using bi-directional gRPC streaming, which does not require an app port.
@@ -44,25 +41,19 @@ public class Subscriber {
   public static void main(String[] args) throws Exception {
     String topicName = getTopicName(args);
     try (var client = new DaprClientBuilder().buildPreviewClient()) {
-      var subscription = client.subscribeToEvents(
+      // Subscribe to events using the Flux-based reactive API
+      // The stream will emit CloudEvent<String> objects as they arrive
+      client.subscribeToEvents(
           PUBSUB_NAME,
           topicName,
-          new SubscriptionListener<>() {
-
-            @Override
-            public Mono<Status> onEvent(CloudEvent<String> event) {
-              System.out.println("Subscriber got: " + event.getData());
-              return Mono.just(Status.SUCCESS);
-            }
-
-            @Override
-            public void onError(RuntimeException exception) {
-              System.out.println("Subscriber got exception: " + exception.getMessage());
-            }
-          },
-          TypeRef.STRING);
-
-      subscription.awaitTermination();
+          TypeRef.STRING)
+          .doOnNext(event -> {
+            System.out.println("Subscriber got: " + event.getData());
+          })
+          .doOnError(throwable -> {
+            System.out.println("Subscriber got exception: " + throwable.getMessage());
+          })
+          .blockLast();  // Blocks indefinitely until the stream completes (keeps the subscriber running)
     }
   }
 

sdk/src/main/java/io/dapr/client/DaprClientImpl.java

@@ -91,6 +91,7 @@
 import io.dapr.internal.grpc.DaprClientGrpcInterceptors;
 import io.dapr.internal.resiliency.RetryPolicy;
 import io.dapr.internal.resiliency.TimeoutPolicy;
+import io.dapr.internal.subscription.EventSubscriberStreamObserver;
 import io.dapr.serializer.DaprObjectSerializer;
 import io.dapr.serializer.DefaultObjectSerializer;
 import io.dapr.utils.DefaultContentTypeConverter;
@@ -475,6 +476,42 @@ public <T> Subscription subscribeToEvents(
     return buildSubscription(listener, type, request);
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public <T> Flux<CloudEvent<T>> subscribeToEvents(String pubsubName, String topic, TypeRef<T> type) {
+    DaprProtos.SubscribeTopicEventsRequestInitialAlpha1 initialRequest =
+        DaprProtos.SubscribeTopicEventsRequestInitialAlpha1.newBuilder()
+            .setTopic(topic)
+            .setPubsubName(pubsubName)
+            .build();
+    DaprProtos.SubscribeTopicEventsRequestAlpha1 request =
+        DaprProtos.SubscribeTopicEventsRequestAlpha1.newBuilder()
+            .setInitialRequest(initialRequest)
+            .build();
+
+    return Flux.create(sink -> {
+      DaprGrpc.DaprStub interceptedStub = this.grpcInterceptors.intercept(this.asyncStub);
+      EventSubscriberStreamObserver<T> eventSubscriber = new EventSubscriberStreamObserver<>(
+          interceptedStub,
+          sink,
+          type,
+          this.objectSerializer
+      );
+      StreamObserver<DaprProtos.SubscribeTopicEventsRequestAlpha1> requestStream = eventSubscriber.start(request);
+
+      // Cleanup when Flux is cancelled or completed
+      sink.onDispose(() -> {
+        try {
+          requestStream.onCompleted();
+        } catch (Exception e) {
+          logger.debug("Completing the subscription stream resulted in an error: {}", e.getMessage());
+        }
+      });
+    }, FluxSink.OverflowStrategy.BUFFER);
+  }
+
   @Nonnull
   private <T> Subscription<T> buildSubscription(
       SubscriptionListener<T> listener,

sdk/src/main/java/io/dapr/client/DaprPreviewClient.java

@@ -17,6 +17,7 @@
 import io.dapr.client.domain.BulkPublishRequest;
 import io.dapr.client.domain.BulkPublishResponse;
 import io.dapr.client.domain.BulkPublishResponseFailedEntry;
+import io.dapr.client.domain.CloudEvent;
 import io.dapr.client.domain.ConversationRequest;
 import io.dapr.client.domain.ConversationRequestAlpha2;
 import io.dapr.client.domain.ConversationResponse;
@@ -32,6 +33,7 @@
 import io.dapr.client.domain.UnlockResponseStatus;
 import io.dapr.client.domain.query.Query;
 import io.dapr.utils.TypeRef;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.List;
@@ -271,12 +273,24 @@ <T> Mono<BulkPublishResponse<T>> publishEvents(String pubsubName, String topicNa
    * @param topic Name of the topic to subscribe to.
    * @param listener Callback methods to process events.
    * @param type Type for object deserialization.
-   * @return An active subscription.
    * @param <T> Type of object deserialization.
+   * @return An active subscription.
+   * @deprecated Use {@link #subscribeToEvents(String, String, TypeRef)} instead for a more reactive approach.
    */
+  @Deprecated
   <T> Subscription subscribeToEvents(
       String pubsubName, String topic, SubscriptionListener<T> listener, TypeRef<T> type);
 
+  /**
+   * Subscribe to pubsub events via streaming using Project Reactor Flux.
+   * @param pubsubName Name of the pubsub component.
+   * @param topic Name of the topic to subscribe to.
+   * @param type Type for object deserialization.
+   * @return A Flux of CloudEvents containing deserialized event payloads and metadata.
+   * @param <T> Type of the event payload.
+   */
+  <T> Flux<CloudEvent<T>> subscribeToEvents(String pubsubName, String topic, TypeRef<T> type);
+
   /**
    * Schedules a job using the provided job request details.
    *

sdk/src/main/java/io/dapr/client/Subscription.java

@@ -35,6 +35,7 @@
  * Streaming subscription of events for Dapr's pubsub.
  * @param <T> Application's object type.
  */
+@Deprecated
 public class Subscription<T> implements Closeable {
 
   private final BlockingQueue<DaprProtos.SubscribeTopicEventsRequestAlpha1> ackQueue = new LinkedBlockingQueue<>(50);

sdk/src/main/java/io/dapr/client/SubscriptionListener.java

@@ -20,6 +20,7 @@
  * Callback interface to receive events from a streaming subscription of events.
  * @param <T> Object type for deserialization.
  */
+@Deprecated
 public interface SubscriptionListener<T> {
 
   /**