Adding raw event subscription.

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

Author: artur-ciocanu

sdk-tests/src/test/java/io/dapr/it/pubsub/stream/PubSubStreamIT.java

@@ -124,4 +124,48 @@ public void onError(RuntimeException exception) {
       }
     }
   }
+
+  @Test
+  public void testPubSubRawData() throws Exception {
+    final DaprRun daprRun = closeLater(startDaprApp(
+        this.getClass().getSimpleName() + "-rawdata",
+        60000));
+
+    var runId = UUID.randomUUID().toString();
+    try (DaprClient client = daprRun.newDaprClient();
+         DaprPreviewClient previewClient = daprRun.newDaprPreviewClient()) {
+
+      // Publish messages
+      for (int i = 0; i < NUM_MESSAGES; i++) {
+        String message = String.format("Raw message #%d for run %s", i, runId);
+        client.publishEvent(PUBSUB_NAME, TOPIC_NAME, message).block();
+        System.out.println(
+            String.format("Published raw message: '%s' to topic '%s'", message, TOPIC_NAME));
+      }
+
+      System.out.println("Starting raw data subscription for " + TOPIC_NAME);
+
+      Set<String> messages = Collections.synchronizedSet(new HashSet<>());
+
+      // Use new subscribeToEventsData - receives String directly, not CloudEvent<String>
+      var disposable = previewClient.subscribeToEventsData(PUBSUB_NAME, TOPIC_NAME, TypeRef.STRING)
+          .doOnNext(rawMessage -> {
+            // rawMessage is String directly
+            if (rawMessage.contains(runId)) {
+              messages.add(rawMessage);
+              System.out.println("Received raw message: " + rawMessage);
+            }
+          })
+          .subscribe();
+
+      callWithRetry(() -> {
+        var messageCount = messages.size();
+        System.out.println(
+            String.format("Got %d raw messages out of %d for topic %s.", messageCount, NUM_MESSAGES, TOPIC_NAME));
+        assertEquals(NUM_MESSAGES, messages.size());
+      }, 60000);
+
+      disposable.dispose();
+    }
+  }
 }

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

@@ -514,6 +514,15 @@ public <T> Flux<CloudEvent<T>> subscribeToEvents(String pubsubName, String topic
     }, FluxSink.OverflowStrategy.BUFFER);
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public <T> Flux<T> subscribeToEventsData(String pubsubName, String topic, TypeRef<T> type) {
+    return subscribeToEvents(pubsubName, topic, type)
+        .map(CloudEvent::getData);
+  }
+
   @Nonnull
   private <T> Subscription<T> buildSubscription(
       SubscriptionListener<T> listener,

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

@@ -293,6 +293,18 @@ <T> Subscription subscribeToEvents(
    */
   <T> Flux<CloudEvent<T>> subscribeToEvents(String pubsubName, String topic, TypeRef<T> type);
 
+  /**
+   * Subscribe to pubsub events via streaming using Project Reactor Flux.
+   * Returns only the deserialized event data without CloudEvent metadata wrapper.
+   *
+   * @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 deserialized event payloads (no CloudEvent wrapper).
+   * @param <T> Type of the event payload.
+   */
+  <T> Flux<T> subscribeToEventsData(String pubsubName, String topic, TypeRef<T> type);
+
   /*
    * Converse with an LLM.
    *

sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java

@@ -659,6 +659,88 @@ public void onCompleted() {
     assertEquals(numEvents, eventCount.get());
   }
 
+  @Test
+  public void subscribeToEventsDataTest() throws Exception {
+    var numEvents = 100;
+    var pubsubName = "pubsubName";
+    var topicName = "topicName";
+    var data = "my message";
+    var started = new Semaphore(0);
+
+    doAnswer((Answer<StreamObserver<DaprProtos.SubscribeTopicEventsRequestAlpha1>>) invocation -> {
+      StreamObserver<DaprProtos.SubscribeTopicEventsResponseAlpha1> observer =
+              (StreamObserver<DaprProtos.SubscribeTopicEventsResponseAlpha1>) invocation.getArguments()[0];
+
+      var emitterThread = new Thread(() -> {
+        try {
+          started.acquire();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+
+        observer.onNext(DaprProtos.SubscribeTopicEventsResponseAlpha1.getDefaultInstance());
+
+        for (int i = 0; i < numEvents; i++) {
+          DaprProtos.SubscribeTopicEventsResponseAlpha1 response =
+              DaprProtos.SubscribeTopicEventsResponseAlpha1.newBuilder()
+                  .setEventMessage(DaprAppCallbackProtos.TopicEventRequest.newBuilder()
+                      .setId(Integer.toString(i))
+                      .setPubsubName(pubsubName)
+                      .setTopic(topicName)
+                      .setData(ByteString.copyFromUtf8("\"" + data + "\""))
+                      .setDataContentType("application/json")
+                      .build())
+                  .build();
+          observer.onNext(response);
+        }
+
+        observer.onCompleted();
+      });
+
+      emitterThread.start();
+
+      return new StreamObserver<>() {
+        @Override
+        public void onNext(DaprProtos.SubscribeTopicEventsRequestAlpha1 subscribeTopicEventsRequestAlpha1) {
+          started.release();
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+          // No-op
+        }
+
+        @Override
+        public void onCompleted() {
+          // No-op
+        }
+      };
+    }).when(daprStub).subscribeTopicEventsAlpha1(any(StreamObserver.class));
+
+    final AtomicInteger eventCount = new AtomicInteger(0);
+    final Semaphore gotAll = new Semaphore(0);
+
+    // Use new subscribeToEventsData - receives raw String, not CloudEvent<String>
+    var disposable = previewClient.subscribeToEventsData(pubsubName, topicName, TypeRef.STRING)
+            .doOnNext(rawData -> {
+              // rawData is String directly, not CloudEvent
+              assertEquals(data, rawData);
+              assertTrue(rawData instanceof String);
+
+              int count = eventCount.incrementAndGet();
+
+              if (count >= numEvents) {
+                gotAll.release();
+              }
+            })
+            .subscribe();
+
+    gotAll.acquire();
+    disposable.dispose();
+
+    assertEquals(numEvents, eventCount.get());
+  }
+
   @Test
   public void converseShouldThrowIllegalArgumentExceptionWhenComponentNameIsNull() throws Exception {
     List<ConversationInput> inputs = new ArrayList<>();