Skip to content

refactor: move pub/sub out of DaprClientImpl #1905

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dapr-bot merged 1 commit into from
Nov 12, 2026
Merged

refactor: move pub/sub out of DaprClientImpl #1905

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
262 changes: 13 additions & 249 deletions sdk/src/main/java/io/dapr/client/DaprClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -168,11 +157,6 @@
*/
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.
*
Expand All @@ -186,6 +170,8 @@

private final DaprConversationClient conversationClient;

private final DaprPubSubClient pubSubClient;

private final DaprStateClient stateClient;

private final DaprVectorClient vectorClient;
Expand Down Expand Up @@ -283,11 +269,11 @@
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);
Expand Down Expand Up @@ -350,152 +336,15 @@
*/
@Override
public Mono<Void> 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<String, String> metadata = request.getMetadata();
if (metadata != null) {
envelopeBuilder.putAllMetadata(metadata);
}

return Mono.deferContextual(
context ->
grpcCalls.<Empty>createMono(
it -> grpcCalls.intercept(context).publishEvent(envelopeBuilder.build(), it)
)
).then();
} catch (Exception ex) {
return DaprException.wrapMono(ex);
}
return this.pubSubClient.publishEvent(request);
}

/**
* {@inheritDoc}
*/
@Override
public <T> Mono<BulkPublishResponse<T>> publishEvents(BulkPublishRequest<T> 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<String, String> metadata = entry.getMetadata();
if (metadata != null) {
reqEntryBuilder.putAllMetadata(metadata);
}
envelopeBuilder.addEntries(reqEntryBuilder.build());
}

// Set metadata if available
Map<String, String> metadata = request.getMetadata();
if (metadata != null) {
envelopeBuilder.putAllMetadata(metadata);
}

Map<String, BulkPublishEntry<T>> entryMap = new HashMap<>();
for (BulkPublishEntry<T> entry : request.getEntries()) {
entryMap.put(entry.getEntryId(), entry);
}
return Mono.deferContextual(
context ->
grpcCalls.<DaprPubsubProtos.BulkPublishResponse>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.<DaprPubsubProtos.BulkPublishResponse>createMono(
it -> grpcCalls.intercept(context).bulkPublishEventAlpha1(envelopeBuilder.build(), it)
);
}
}
return Mono.error(throwable);
})
).map(
it -> {
List<BulkPublishResponseFailedEntry<T>> entries = new ArrayList<>();
for (DaprPubsubProtos.BulkPublishResponseFailedEntry entry : it.getFailedEntriesList()) {
BulkPublishResponseFailedEntry<T> domainEntry = new BulkPublishResponseFailedEntry<T>(
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);
}

/**
Expand All @@ -504,7 +353,7 @@
@Override
public <T> Subscription subscribeToEvents(
String pubsubName, String topic, SubscriptionListener<T> listener, TypeRef<T> type) {
return subscribeToEvents(pubsubName, topic, null, listener, type);
return this.pubSubClient.subscribeToEvents(pubsubName, topic, listener, type);
}

/**
Expand All @@ -517,18 +366,7 @@
String deadLetterTopic,
SubscriptionListener<T> listener,
TypeRef<T> 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);
}

/**
Expand All @@ -537,7 +375,7 @@
@Deprecated
@Override
public <T> Flux<T> subscribeToEvents(String pubsubName, String topic, TypeRef<T> type) {
return subscribeToTopic(pubsubName, topic, type);
return this.pubSubClient.subscribeToEvents(pubsubName, topic, type);
}

/**
Expand All @@ -546,31 +384,31 @@
@Deprecated
@Override
public <T> Flux<T> subscribeToEvents(String pubsubName, String topic, TypeRef<T> type, Map<String, String> metadata) {
return subscribeToTopic(pubsubName, topic, type, metadata);
return this.pubSubClient.subscribeToEvents(pubsubName, topic, type, metadata);
}

/**
* {@inheritDoc}
*/
@Override
public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, TypeRef<T> type) {
return subscribeToTopic(pubsubName, topic, null, type, null);
return this.pubSubClient.subscribeToTopic(pubsubName, topic, type);
}

/**
* {@inheritDoc}
*/
@Override
public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, TypeRef<T> type, Map<String, String> metadata) {
return subscribeToTopic(pubsubName, topic, null, type, metadata);
return this.pubSubClient.subscribeToTopic(pubsubName, topic, type, metadata);
}

/**
* {@inheritDoc}
*/
@Override
public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, String deadLetterTopic, TypeRef<T> type) {
return subscribeToTopic(pubsubName, topic, deadLetterTopic, type, null);
return this.pubSubClient.subscribeToTopic(pubsubName, topic, deadLetterTopic, type);
}

/**
Expand All @@ -583,81 +421,7 @@
String deadLetterTopic,
TypeRef<T> type,
Map<String, String> 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<T> eventSubscriber = new EventSubscriberStreamObserver<>(
interceptedStub,
sink,
type,
this.objectSerializer
);
StreamObserver<DaprPubsubProtos.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,
TypeRef<T> type,
DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request) {
var interceptedStub = this.grpcInterceptors.intercept(this.asyncStub);
Subscription<T> 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<T> 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
Expand Down Expand Up @@ -1417,7 +1181,7 @@
@Deprecated(forRemoval = true)
@Override
public Mono<ConversationResponse> converse(ConversationRequest conversationRequest) {
return this.conversationClient.converse(conversationRequest);

Check warning on line 1184 in sdk/src/main/java/io/dapr/client/DaprClientImpl.java

View workflow job for this annotation

GitHub Actions / Validate Javadocs generation 

converse(io.dapr.client.domain.ConversationRequest) in io.dapr.client.DaprConversationClient has been deprecated and marked for removal
}

/**
Expand Down
Loading
Loading