diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index 5412dd150e..3772d719c2 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -13,7 +13,6 @@ package io.dapr.client; -import com.google.common.base.Strings; import com.google.protobuf.Any; import com.google.protobuf.ByteString; import com.google.protobuf.Empty; @@ -22,11 +21,8 @@ import io.dapr.client.domain.AppConnectionPropertiesMetadata; import io.dapr.client.domain.BatchQueryVectorsRequestAlpha1; import io.dapr.client.domain.BatchQueryVectorsResultAlpha1; -import io.dapr.client.domain.BulkPublishEntry; 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.ComponentMetadata; import io.dapr.client.domain.ConfigurationItem; import io.dapr.client.domain.ConstantFailurePolicy; @@ -86,10 +82,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; import io.dapr.utils.FailurePolicyUtils; import io.dapr.utils.TypeRef; import io.dapr.v1.CommonProtos; @@ -102,7 +95,6 @@ import io.dapr.v1.DaprLockProtos; import io.dapr.v1.DaprMetadataProtos; import io.dapr.v1.DaprProtos; -import io.dapr.v1.DaprPubsubProtos; import io.dapr.v1.DaprSecretProtos; import io.grpc.Channel; import io.grpc.Metadata; @@ -117,12 +109,9 @@ import reactor.core.Disposables; import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxSink; import reactor.core.publisher.Mono; import reactor.util.retry.Retry; -import javax.annotation.Nonnull; - import java.io.ByteArrayOutputStream; import java.io.IOException; import java.net.URI; @@ -168,11 +157,6 @@ public class DaprClientImpl extends AbstractDaprClient { */ private final GrpcChannelFacade channel; - /** - * The async gRPC stub. - */ - private final DaprGrpc.DaprStub asyncStub; - /** * The HTTP client to be used for healthz and HTTP service invocation only. * @@ -186,6 +170,8 @@ public class DaprClientImpl extends AbstractDaprClient { private final DaprConversationClient conversationClient; + private final DaprPubSubClient pubSubClient; + private final DaprStateClient stateClient; private final DaprVectorClient vectorClient; @@ -283,11 +269,11 @@ private DaprClientImpl( String daprApiToken) { super(objectSerializer, stateSerializer); this.channel = channel; - this.asyncStub = asyncStub; this.httpClient = httpClient; this.grpcInterceptors = new DaprClientGrpcInterceptors(daprApiToken, timeoutPolicy); this.grpcCalls = new DaprGrpcCalls(asyncStub, this.grpcInterceptors, retryPolicy); this.conversationClient = new DaprConversationClient(this.grpcCalls, objectSerializer); + this.pubSubClient = new DaprPubSubClient(this.grpcCalls, objectSerializer); this.stateClient = new DaprStateClient(this.grpcCalls, stateSerializer); this.vectorClient = new DaprVectorClient(this.grpcCalls); this.logger = LoggerFactory.getLogger(DaprClientImpl.class); @@ -350,48 +336,7 @@ public Mono healthCheck() { */ @Override public Mono publishEvent(PublishEventRequest request) { - try { - String pubsubName = request.getPubsubName(); - String topic = request.getTopic(); - Object data = request.getData(); - - // Content-type can be overwritten on a per-request basis. - // It allows CloudEvents to be handled differently, for example. - String contentType = request.getContentType(); - byte[] serializedData; - if (Strings.isNullOrEmpty(contentType)) { - serializedData = objectSerializer.serialize(data); - contentType = objectSerializer.getContentType(); - } else if (objectSerializer instanceof DefaultObjectSerializer - && DefaultContentTypeConverter.isStringContentType(contentType) - && (data instanceof String || data instanceof Number || data instanceof Boolean)) { - // Text content is sent as-is, otherwise the JSON serializer would quote it and the - // payload would not match the given content type. - serializedData = DefaultContentTypeConverter.convertEventToBytesForGrpc(data, contentType); - } else { - serializedData = objectSerializer.serialize(data); - } - - DaprPubsubProtos.PublishEventRequest.Builder envelopeBuilder = DaprPubsubProtos.PublishEventRequest.newBuilder() - .setTopic(topic) - .setPubsubName(pubsubName) - .setData(ByteString.copyFrom(serializedData)) - .setDataContentType(contentType); - - Map metadata = request.getMetadata(); - if (metadata != null) { - envelopeBuilder.putAllMetadata(metadata); - } - - return Mono.deferContextual( - context -> - grpcCalls.createMono( - it -> grpcCalls.intercept(context).publishEvent(envelopeBuilder.build(), it) - ) - ).then(); - } catch (Exception ex) { - return DaprException.wrapMono(ex); - } + return this.pubSubClient.publishEvent(request); } /** @@ -399,103 +344,7 @@ public Mono publishEvent(PublishEventRequest request) { */ @Override public Mono> publishEvents(BulkPublishRequest request) { - try { - String pubsubName = request.getPubsubName(); - String topic = request.getTopic(); - DaprPubsubProtos.BulkPublishRequest.Builder envelopeBuilder = DaprPubsubProtos.BulkPublishRequest.newBuilder(); - envelopeBuilder.setTopic(topic); - envelopeBuilder.setPubsubName(pubsubName); - - if (Strings.isNullOrEmpty(pubsubName) || Strings.isNullOrEmpty(topic)) { - throw new IllegalArgumentException("pubsubName and topic name cannot be null or empty"); - } - - for (BulkPublishEntry entry : request.getEntries()) { - Object event = entry.getEvent(); - byte[] data; - String contentType = entry.getContentType(); - try { - // Serialize event into bytes - if (!Strings.isNullOrEmpty(contentType) && objectSerializer instanceof DefaultObjectSerializer) { - // If content type is given by user and default object serializer is used - data = DefaultContentTypeConverter.convertEventToBytesForGrpc(event, contentType); - } else { - // perform the serialization as per user given input of serializer - // this is also the case when content type is empty - - data = objectSerializer.serialize(event); - - if (Strings.isNullOrEmpty(contentType)) { - // Only override content type if not given in input by user - contentType = objectSerializer.getContentType(); - } - } - } catch (IOException ex) { - throw DaprException.propagate(ex); - } - - DaprPubsubProtos.BulkPublishRequestEntry.Builder reqEntryBuilder = DaprPubsubProtos.BulkPublishRequestEntry - .newBuilder() - .setEntryId(entry.getEntryId()) - .setEvent(ByteString.copyFrom(data)) - .setContentType(contentType); - Map metadata = entry.getMetadata(); - if (metadata != null) { - reqEntryBuilder.putAllMetadata(metadata); - } - envelopeBuilder.addEntries(reqEntryBuilder.build()); - } - - // Set metadata if available - Map metadata = request.getMetadata(); - if (metadata != null) { - envelopeBuilder.putAllMetadata(metadata); - } - - Map> entryMap = new HashMap<>(); - for (BulkPublishEntry entry : request.getEntries()) { - entryMap.put(entry.getEntryId(), entry); - } - return Mono.deferContextual( - context -> - grpcCalls.createMono( - it -> grpcCalls.intercept(context).bulkPublishEvent(envelopeBuilder.build(), it) - ).onErrorResume(throwable -> { - // Check if the error is UNIMPLEMENTED, and try the alpha API just in case - // The exception might be wrapped in DaprException, so check both - Throwable cause = throwable; - if (throwable instanceof DaprException && throwable.getCause() != null) { - cause = throwable.getCause(); - } - - if (cause instanceof io.grpc.StatusRuntimeException) { - io.grpc.StatusRuntimeException statusException = (io.grpc.StatusRuntimeException) cause; - if (statusException.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) { - return grpcCalls.createMono( - it -> grpcCalls.intercept(context).bulkPublishEventAlpha1(envelopeBuilder.build(), it) - ); - } - } - return Mono.error(throwable); - }) - ).map( - it -> { - List> entries = new ArrayList<>(); - for (DaprPubsubProtos.BulkPublishResponseFailedEntry entry : it.getFailedEntriesList()) { - BulkPublishResponseFailedEntry domainEntry = new BulkPublishResponseFailedEntry( - entryMap.get(entry.getEntryId()), - entry.getError()); - entries.add(domainEntry); - } - if (entries.size() > 0) { - return new BulkPublishResponse<>(entries); - } - return new BulkPublishResponse<>(); - } - ); - } catch (RuntimeException ex) { - return DaprException.wrapMono(ex); - } + return this.pubSubClient.publishEvents(request); } /** @@ -504,7 +353,7 @@ public Mono> publishEvents(BulkPublishRequest requ @Override public Subscription subscribeToEvents( String pubsubName, String topic, SubscriptionListener listener, TypeRef type) { - return subscribeToEvents(pubsubName, topic, null, listener, type); + return this.pubSubClient.subscribeToEvents(pubsubName, topic, listener, type); } /** @@ -517,18 +366,7 @@ public Subscription subscribeToEvents( String deadLetterTopic, SubscriptionListener listener, TypeRef type) { - DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.Builder initialRequestBuilder = - DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.newBuilder() - .setTopic(topic) - .setPubsubName(pubsubName); - if (deadLetterTopic != null && !deadLetterTopic.isEmpty()) { - initialRequestBuilder.setDeadLetterTopic(deadLetterTopic); - } - DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request = - DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1.newBuilder() - .setInitialRequest(initialRequestBuilder.build()) - .build(); - return buildSubscription(listener, type, request); + return this.pubSubClient.subscribeToEvents(pubsubName, topic, deadLetterTopic, listener, type); } /** @@ -537,7 +375,7 @@ public Subscription subscribeToEvents( @Deprecated @Override public Flux subscribeToEvents(String pubsubName, String topic, TypeRef type) { - return subscribeToTopic(pubsubName, topic, type); + return this.pubSubClient.subscribeToEvents(pubsubName, topic, type); } /** @@ -546,7 +384,7 @@ public Flux subscribeToEvents(String pubsubName, String topic, TypeRef @Deprecated @Override public Flux subscribeToEvents(String pubsubName, String topic, TypeRef type, Map metadata) { - return subscribeToTopic(pubsubName, topic, type, metadata); + return this.pubSubClient.subscribeToEvents(pubsubName, topic, type, metadata); } /** @@ -554,7 +392,7 @@ public Flux subscribeToEvents(String pubsubName, String topic, TypeRef */ @Override public Flux subscribeToTopic(String pubsubName, String topic, TypeRef type) { - return subscribeToTopic(pubsubName, topic, null, type, null); + return this.pubSubClient.subscribeToTopic(pubsubName, topic, type); } /** @@ -562,7 +400,7 @@ public Flux subscribeToTopic(String pubsubName, String topic, TypeRef */ @Override public Flux subscribeToTopic(String pubsubName, String topic, TypeRef type, Map metadata) { - return subscribeToTopic(pubsubName, topic, null, type, metadata); + return this.pubSubClient.subscribeToTopic(pubsubName, topic, type, metadata); } /** @@ -570,7 +408,7 @@ public Flux subscribeToTopic(String pubsubName, String topic, TypeRef */ @Override public Flux subscribeToTopic(String pubsubName, String topic, String deadLetterTopic, TypeRef type) { - return subscribeToTopic(pubsubName, topic, deadLetterTopic, type, null); + return this.pubSubClient.subscribeToTopic(pubsubName, topic, deadLetterTopic, type); } /** @@ -583,81 +421,7 @@ public Flux subscribeToTopic( String deadLetterTopic, TypeRef type, Map metadata) { - DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.Builder initialRequestBuilder = - DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.newBuilder() - .setTopic(topic) - .setPubsubName(pubsubName); - - if (metadata != null && !metadata.isEmpty()) { - initialRequestBuilder.putAllMetadata(metadata); - } - - if (deadLetterTopic != null && !deadLetterTopic.isEmpty()) { - initialRequestBuilder.setDeadLetterTopic(deadLetterTopic); - } - - DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request = - DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1.newBuilder() - .setInitialRequest(initialRequestBuilder.build()) - .build(); - - return Flux.create(sink -> { - DaprGrpc.DaprStub interceptedStub = this.grpcInterceptors.intercept(this.asyncStub); - EventSubscriberStreamObserver eventSubscriber = new EventSubscriberStreamObserver<>( - interceptedStub, - sink, - type, - this.objectSerializer - ); - StreamObserver 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 Subscription buildSubscription( - SubscriptionListener listener, - TypeRef type, - DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request) { - var interceptedStub = this.grpcInterceptors.intercept(this.asyncStub); - Subscription subscription = new Subscription<>(interceptedStub, request, listener, response -> { - if (response.getEventMessage() == null) { - return null; - } - - var message = response.getEventMessage(); - if ((message.getPubsubName() == null) || message.getPubsubName().isEmpty()) { - return null; - } - - try { - CloudEvent cloudEvent = new CloudEvent<>(); - T object = null; - if (type != null) { - object = DaprClientImpl.this.objectSerializer.deserialize(message.getData().toByteArray(), type); - } - cloudEvent.setData(object); - cloudEvent.setDatacontenttype(message.getDataContentType()); - cloudEvent.setId(message.getId()); - cloudEvent.setTopic(message.getTopic()); - cloudEvent.setSpecversion(message.getSpecVersion()); - cloudEvent.setType(message.getType()); - cloudEvent.setPubsubName(message.getPubsubName()); - return cloudEvent; - } catch (IOException e) { - throw new RuntimeException(e); - } - }); - subscription.start(); - return subscription; + return this.pubSubClient.subscribeToTopic(pubsubName, topic, deadLetterTopic, type, metadata); } @Override diff --git a/sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java b/sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java index bf65057d7f..73a62c7898 100644 --- a/sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java +++ b/sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java @@ -46,6 +46,15 @@ final class DaprGrpcCalls { this.retryPolicy = retryPolicy; } + /** + * Populates GRPC client with interceptors, without a Reactor context. + * + * @return Client after adding interceptors. + */ + DaprGrpc.DaprStub intercept() { + return this.grpcInterceptors.intercept(this.asyncStub); + } + /** * Populates GRPC client with interceptors for telemetry. * diff --git a/sdk/src/main/java/io/dapr/client/DaprPubSubClient.java b/sdk/src/main/java/io/dapr/client/DaprPubSubClient.java new file mode 100644 index 0000000000..47e866b547 --- /dev/null +++ b/sdk/src/main/java/io/dapr/client/DaprPubSubClient.java @@ -0,0 +1,338 @@ +/* + * Copyright 2026 The Dapr Authors + * 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. +*/ + +package io.dapr.client; + +import com.google.common.base.Strings; +import com.google.protobuf.ByteString; +import com.google.protobuf.Empty; +import io.dapr.client.domain.BulkPublishEntry; +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.PublishEventRequest; +import io.dapr.exceptions.DaprException; +import io.dapr.internal.subscription.EventSubscriberStreamObserver; +import io.dapr.serializer.DaprObjectSerializer; +import io.dapr.serializer.DefaultObjectSerializer; +import io.dapr.utils.DefaultContentTypeConverter; +import io.dapr.utils.TypeRef; +import io.dapr.v1.DaprGrpc; +import io.dapr.v1.DaprPubsubProtos; +import io.grpc.stub.StreamObserver; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; +import reactor.core.publisher.Mono; + +import javax.annotation.Nonnull; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Runs the pub/sub calls of {@link DaprClientImpl}: publishing and streaming subscriptions. + */ +final class DaprPubSubClient { + + private final Logger logger = LoggerFactory.getLogger(DaprClientImpl.class); + + private final DaprGrpcCalls grpcCalls; + + private final DaprObjectSerializer objectSerializer; + + DaprPubSubClient(DaprGrpcCalls grpcCalls, DaprObjectSerializer objectSerializer) { + this.grpcCalls = grpcCalls; + this.objectSerializer = objectSerializer; + } + + Mono publishEvent(PublishEventRequest request) { + try { + String pubsubName = request.getPubsubName(); + String topic = request.getTopic(); + Object data = request.getData(); + + // Content-type can be overwritten on a per-request basis. + // It allows CloudEvents to be handled differently, for example. + String contentType = request.getContentType(); + byte[] serializedData; + if (Strings.isNullOrEmpty(contentType)) { + serializedData = objectSerializer.serialize(data); + contentType = objectSerializer.getContentType(); + } else if (objectSerializer instanceof DefaultObjectSerializer + && DefaultContentTypeConverter.isStringContentType(contentType) + && (data instanceof String || data instanceof Number || data instanceof Boolean)) { + // Text content is sent as-is, otherwise the JSON serializer would quote it and the + // payload would not match the given content type. + serializedData = DefaultContentTypeConverter.convertEventToBytesForGrpc(data, contentType); + } else { + serializedData = objectSerializer.serialize(data); + } + + DaprPubsubProtos.PublishEventRequest.Builder envelopeBuilder = DaprPubsubProtos.PublishEventRequest.newBuilder() + .setTopic(topic) + .setPubsubName(pubsubName) + .setData(ByteString.copyFrom(serializedData)) + .setDataContentType(contentType); + + Map metadata = request.getMetadata(); + if (metadata != null) { + envelopeBuilder.putAllMetadata(metadata); + } + + return Mono.deferContextual( + context -> + grpcCalls.createMono( + it -> grpcCalls.intercept(context).publishEvent(envelopeBuilder.build(), it) + ) + ).then(); + } catch (Exception ex) { + return DaprException.wrapMono(ex); + } + } + + Mono> publishEvents(BulkPublishRequest request) { + try { + String pubsubName = request.getPubsubName(); + String topic = request.getTopic(); + DaprPubsubProtos.BulkPublishRequest.Builder envelopeBuilder = DaprPubsubProtos.BulkPublishRequest.newBuilder(); + envelopeBuilder.setTopic(topic); + envelopeBuilder.setPubsubName(pubsubName); + + if (Strings.isNullOrEmpty(pubsubName) || Strings.isNullOrEmpty(topic)) { + throw new IllegalArgumentException("pubsubName and topic name cannot be null or empty"); + } + + for (BulkPublishEntry entry : request.getEntries()) { + Object event = entry.getEvent(); + byte[] data; + String contentType = entry.getContentType(); + try { + // Serialize event into bytes + if (!Strings.isNullOrEmpty(contentType) && objectSerializer instanceof DefaultObjectSerializer) { + // If content type is given by user and default object serializer is used + data = DefaultContentTypeConverter.convertEventToBytesForGrpc(event, contentType); + } else { + // perform the serialization as per user given input of serializer + // this is also the case when content type is empty + + data = objectSerializer.serialize(event); + + if (Strings.isNullOrEmpty(contentType)) { + // Only override content type if not given in input by user + contentType = objectSerializer.getContentType(); + } + } + } catch (IOException ex) { + throw DaprException.propagate(ex); + } + + DaprPubsubProtos.BulkPublishRequestEntry.Builder reqEntryBuilder = DaprPubsubProtos.BulkPublishRequestEntry + .newBuilder() + .setEntryId(entry.getEntryId()) + .setEvent(ByteString.copyFrom(data)) + .setContentType(contentType); + Map metadata = entry.getMetadata(); + if (metadata != null) { + reqEntryBuilder.putAllMetadata(metadata); + } + envelopeBuilder.addEntries(reqEntryBuilder.build()); + } + + // Set metadata if available + Map metadata = request.getMetadata(); + if (metadata != null) { + envelopeBuilder.putAllMetadata(metadata); + } + + Map> entryMap = new HashMap<>(); + for (BulkPublishEntry entry : request.getEntries()) { + entryMap.put(entry.getEntryId(), entry); + } + return Mono.deferContextual( + context -> + grpcCalls.createMono( + it -> grpcCalls.intercept(context).bulkPublishEvent(envelopeBuilder.build(), it) + ).onErrorResume(throwable -> { + // Check if the error is UNIMPLEMENTED, and try the alpha API just in case + // The exception might be wrapped in DaprException, so check both + Throwable cause = throwable; + if (throwable instanceof DaprException && throwable.getCause() != null) { + cause = throwable.getCause(); + } + + if (cause instanceof io.grpc.StatusRuntimeException) { + io.grpc.StatusRuntimeException statusException = (io.grpc.StatusRuntimeException) cause; + if (statusException.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) { + return grpcCalls.createMono( + it -> grpcCalls.intercept(context).bulkPublishEventAlpha1(envelopeBuilder.build(), it) + ); + } + } + return Mono.error(throwable); + }) + ).map( + it -> { + List> entries = new ArrayList<>(); + for (DaprPubsubProtos.BulkPublishResponseFailedEntry entry : it.getFailedEntriesList()) { + BulkPublishResponseFailedEntry domainEntry = new BulkPublishResponseFailedEntry( + entryMap.get(entry.getEntryId()), + entry.getError()); + entries.add(domainEntry); + } + if (entries.size() > 0) { + return new BulkPublishResponse<>(entries); + } + return new BulkPublishResponse<>(); + } + ); + } catch (RuntimeException ex) { + return DaprException.wrapMono(ex); + } + } + + Subscription subscribeToEvents( + String pubsubName, String topic, SubscriptionListener listener, TypeRef type) { + return subscribeToEvents(pubsubName, topic, null, listener, type); + } + + Subscription subscribeToEvents( + String pubsubName, + String topic, + String deadLetterTopic, + SubscriptionListener listener, + TypeRef type) { + DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.Builder initialRequestBuilder = + DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.newBuilder() + .setTopic(topic) + .setPubsubName(pubsubName); + if (deadLetterTopic != null && !deadLetterTopic.isEmpty()) { + initialRequestBuilder.setDeadLetterTopic(deadLetterTopic); + } + DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request = + DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1.newBuilder() + .setInitialRequest(initialRequestBuilder.build()) + .build(); + return buildSubscription(listener, type, request); + } + + @Deprecated + Flux subscribeToEvents(String pubsubName, String topic, TypeRef type) { + return subscribeToTopic(pubsubName, topic, type); + } + + @Deprecated + Flux subscribeToEvents(String pubsubName, String topic, TypeRef type, Map metadata) { + return subscribeToTopic(pubsubName, topic, type, metadata); + } + + Flux subscribeToTopic(String pubsubName, String topic, TypeRef type) { + return subscribeToTopic(pubsubName, topic, null, type, null); + } + + Flux subscribeToTopic(String pubsubName, String topic, TypeRef type, Map metadata) { + return subscribeToTopic(pubsubName, topic, null, type, metadata); + } + + Flux subscribeToTopic(String pubsubName, String topic, String deadLetterTopic, TypeRef type) { + return subscribeToTopic(pubsubName, topic, deadLetterTopic, type, null); + } + + Flux subscribeToTopic( + String pubsubName, + String topic, + String deadLetterTopic, + TypeRef type, + Map metadata) { + DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.Builder initialRequestBuilder = + DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.newBuilder() + .setTopic(topic) + .setPubsubName(pubsubName); + + if (metadata != null && !metadata.isEmpty()) { + initialRequestBuilder.putAllMetadata(metadata); + } + + if (deadLetterTopic != null && !deadLetterTopic.isEmpty()) { + initialRequestBuilder.setDeadLetterTopic(deadLetterTopic); + } + + DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request = + DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1.newBuilder() + .setInitialRequest(initialRequestBuilder.build()) + .build(); + + return Flux.create(sink -> { + DaprGrpc.DaprStub interceptedStub = this.grpcCalls.intercept(); + EventSubscriberStreamObserver eventSubscriber = new EventSubscriberStreamObserver<>( + interceptedStub, + sink, + type, + this.objectSerializer + ); + StreamObserver 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 Subscription buildSubscription( + SubscriptionListener listener, + TypeRef type, + DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request) { + var interceptedStub = this.grpcCalls.intercept(); + Subscription subscription = new Subscription<>(interceptedStub, request, listener, response -> { + if (response.getEventMessage() == null) { + return null; + } + + var message = response.getEventMessage(); + if ((message.getPubsubName() == null) || message.getPubsubName().isEmpty()) { + return null; + } + + try { + CloudEvent cloudEvent = new CloudEvent<>(); + T object = null; + if (type != null) { + object = DaprPubSubClient.this.objectSerializer.deserialize(message.getData().toByteArray(), type); + } + cloudEvent.setData(object); + cloudEvent.setDatacontenttype(message.getDataContentType()); + cloudEvent.setId(message.getId()); + cloudEvent.setTopic(message.getTopic()); + cloudEvent.setSpecversion(message.getSpecVersion()); + cloudEvent.setType(message.getType()); + cloudEvent.setPubsubName(message.getPubsubName()); + return cloudEvent; + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + subscription.start(); + return subscription; + } +}