From 9825a3ac62e4c7113241fa73051224424e0ed4c0 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Thu, 29 Oct 2026 10:54:12 +0100 Subject: [PATCH 1/2] refactor: move gRPC call plumbing out of DaprClientImpl Move intercept, createMono, createFlux and the stream observers into a package-private DaprGrpcCalls class. DaprClientImpl calls it instead. No public API change. Part of #1875 Signed-off-by: Javier Aliaga --- .../java/io/dapr/client/DaprClientImpl.java | 198 ++++++------------ .../java/io/dapr/client/DaprGrpcCalls.java | 126 +++++++++++ 2 files changed, 185 insertions(+), 139 deletions(-) create mode 100644 sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index 70e5815677..1f4be9acef 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -131,8 +131,6 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; import reactor.core.publisher.Mono; -import reactor.core.publisher.MonoSink; -import reactor.util.context.ContextView; import reactor.util.retry.Retry; import javax.annotation.Nonnull; @@ -202,6 +200,8 @@ public class DaprClientImpl extends AbstractDaprClient { private final DaprClientGrpcInterceptors grpcInterceptors; + private final DaprGrpcCalls grpcCalls; + /** * Default access level constructor, in order to create an instance of this class use io.dapr.client.DaprClientBuilder * @@ -299,6 +299,7 @@ private DaprClientImpl( this.httpClient = httpClient; this.retryPolicy = retryPolicy; this.grpcInterceptors = new DaprClientGrpcInterceptors(daprApiToken, timeoutPolicy); + this.grpcCalls = new DaprGrpcCalls(asyncStub, this.grpcInterceptors, retryPolicy); this.logger = LoggerFactory.getLogger(DaprClientImpl.class); } @@ -416,8 +417,8 @@ public Mono publishEvent(PublishEventRequest request) { return Mono.deferContextual( context -> - this.createMono( - it -> intercept(context, asyncStub).publishEvent(envelopeBuilder.build(), it) + grpcCalls.createMono( + it -> grpcCalls.intercept(context).publishEvent(envelopeBuilder.build(), it) ) ).then(); } catch (Exception ex) { @@ -489,8 +490,8 @@ public Mono> publishEvents(BulkPublishRequest requ } return Mono.deferContextual( context -> - this.createMono( - it -> intercept(context, asyncStub).bulkPublishEvent(envelopeBuilder.build(), it) + 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 @@ -502,8 +503,8 @@ public Mono> publishEvents(BulkPublishRequest requ if (cause instanceof io.grpc.StatusRuntimeException) { io.grpc.StatusRuntimeException statusException = (io.grpc.StatusRuntimeException) cause; if (statusException.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) { - return this.createMono( - it -> intercept(context, asyncStub).bulkPublishEventAlpha1(envelopeBuilder.build(), it) + return grpcCalls.createMono( + it -> grpcCalls.intercept(context).bulkPublishEventAlpha1(envelopeBuilder.build(), it) ); } } @@ -803,9 +804,9 @@ public Mono invokeBinding(InvokeBindingRequest request, TypeRef type) Metadata responseMetadata = new Metadata(); return Mono.deferContextual( - context -> this.createMono( + context -> grpcCalls.createMono( responseMetadata, - it -> intercept(context, asyncStub, m -> responseMetadata.merge(m)).invokeBinding(envelope, it) + it -> grpcCalls.intercept(context, m -> responseMetadata.merge(m)).invokeBinding(envelope, it) ) ).flatMap( it -> { @@ -863,8 +864,8 @@ public Mono> getState(GetStateRequest request, TypeRef type) { return Mono.deferContextual( context -> - this.createMono( - it -> intercept(context, asyncStub).getState(envelope, it) + grpcCalls.createMono( + it -> grpcCalls.intercept(context).getState(envelope, it) ) ).map( it -> { @@ -911,7 +912,7 @@ public Mono>> getBulkState(GetBulkStateRequest request, TypeRe DaprStateProtos.GetBulkStateRequest envelope = builder.build(); return Mono.deferContextual( - context -> this.createMono(it -> intercept(context, asyncStub) + context -> grpcCalls.createMono(it -> grpcCalls.intercept(context) .getBulkState(envelope, it) ) ).map( @@ -1005,7 +1006,7 @@ public Mono executeStateTransaction(ExecuteStateTransactionRequest request DaprStateProtos.ExecuteStateTransactionRequest req = builder.build(); return Mono.deferContextual( - context -> this.createMono(it -> intercept(context, asyncStub) + context -> grpcCalls.createMono(it -> grpcCalls.intercept(context) .executeStateTransaction(req, it)) ).then(); } catch (Exception e) { @@ -1032,7 +1033,7 @@ public Mono saveBulkState(SaveStateRequest request) { DaprStateProtos.SaveStateRequest req = builder.build(); return Mono.deferContextual( - context -> this.createMono(it -> intercept(context, asyncStub).saveState(req, it)) + context -> grpcCalls.createMono(it -> grpcCalls.intercept(context).saveState(req, it)) ).then(); } catch (Exception ex) { return DaprException.wrapMono(ex); @@ -1116,7 +1117,7 @@ public Mono deleteState(DeleteStateRequest request) { DaprStateProtos.DeleteStateRequest req = builder.build(); return Mono.deferContextual( - context -> this.createMono(it -> intercept(context, asyncStub).deleteState(req, it)) + context -> grpcCalls.createMono(it -> grpcCalls.intercept(context).deleteState(req, it)) ).then(); } catch (Exception ex) { return DaprException.wrapMono(ex); @@ -1152,8 +1153,8 @@ public Mono> getSecret(GetSecretRequest request) { DaprSecretProtos.GetSecretRequest req = requestBuilder.build(); return Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).getSecret(req, it)) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).getSecret(req, it)) ).map(DaprSecretProtos.GetSecretResponse::getDataMap); } @@ -1179,8 +1180,8 @@ public Mono>> getBulkSecret(GetBulkSecretRequest return Mono.deferContextual( context -> - this.createMono( - it -> intercept(context, asyncStub).getBulkSecret(envelope, it) + grpcCalls.createMono( + it -> grpcCalls.intercept(context).getBulkSecret(envelope, it) ) ).map(it -> { Map secretsMap = it.getDataMap(); @@ -1230,8 +1231,8 @@ public Mono tryLock(LockRequest request) { DaprLockProtos.TryLockRequest tryLockRequest = builder.build(); return Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).tryLockAlpha1(tryLockRequest, it) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).tryLockAlpha1(tryLockRequest, it) ) ).flatMap(response -> { try { @@ -1273,8 +1274,8 @@ public Mono unlock(UnlockRequest request) { DaprLockProtos.UnlockRequest unlockRequest = builder.build(); return Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).unlockAlpha1(unlockRequest, it) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).unlockAlpha1(unlockRequest, it) ) ).flatMap(response -> { try { @@ -1322,8 +1323,8 @@ public Mono> queryState(QueryStateRequest request, Typ DaprStateProtos.QueryStateRequest envelope = builder.build(); return Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).queryStateAlpha1(envelope, it) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).queryStateAlpha1(envelope, it) ) ).map( it -> { @@ -1394,8 +1395,8 @@ public void close() throws Exception { public Mono shutdown() { DaprProtos.ShutdownRequest shutdownRequest = DaprProtos.ShutdownRequest.newBuilder().build(); return Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).shutdown(shutdownRequest, it)) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).shutdown(shutdownRequest, it)) ).then(); } @@ -1431,8 +1432,8 @@ private Mono> getConfiguration(DaprConfigurationP .GetConfigurationRequest envelope) { return Mono.deferContextual( context -> - this.createMono( - it -> intercept(context, asyncStub).getConfiguration(envelope, it) + grpcCalls.createMono( + it -> grpcCalls.intercept(context).getConfiguration(envelope, it) ) ).map( it -> { @@ -1473,8 +1474,8 @@ public Flux subscribeConfiguration(SubscribeConf } DaprConfigurationProtos.SubscribeConfigurationRequest envelope = builder.build(); - return this.createFlux( - it -> intercept(null, asyncStub).subscribeConfiguration(envelope, it) + return grpcCalls.createFlux( + it -> grpcCalls.intercept(null).subscribeConfiguration(envelope, it) ).map( it -> { Map configMap = new HashMap<>(); @@ -1513,8 +1514,8 @@ public Mono unsubscribeConfiguration(Unsubscri DaprConfigurationProtos.UnsubscribeConfigurationRequest envelope = builder.build(); - return this.createMono( - it -> intercept(null, asyncStub).unsubscribeConfiguration(envelope, it) + return grpcCalls.createMono( + it -> grpcCalls.intercept(null).unsubscribeConfiguration(envelope, it) ).map( it -> new UnsubscribeConfigurationResponse(it.getOk(), it.getMessage()) ); @@ -1568,13 +1569,13 @@ public Mono scheduleJob(ScheduleJobRequest scheduleJobRequest) { Mono scheduleJobResponseMono = Mono.deferContextual(context -> - this.createMono( - it -> intercept(context, asyncStub).scheduleJob(request, it) + grpcCalls.createMono( + it -> grpcCalls.intercept(context).scheduleJob(request, it) ).onErrorResume(throwable -> { // The stable ScheduleJob RPC is unavailable on pre-1.18 sidecars, fall back to Alpha1. if (isUnimplemented(throwable)) { - return this.createMono( - it -> intercept(context, asyncStub).scheduleJobAlpha1(request, it)); + return grpcCalls.createMono( + it -> grpcCalls.intercept(context).scheduleJobAlpha1(request, it)); } return Mono.error(throwable); }) @@ -1599,13 +1600,13 @@ public Mono getJob(GetJobRequest getJobRequest) { Mono getJobResponseMono = Mono.deferContextual(context -> - this.createMono( - it -> intercept(context, asyncStub).getJob(request, it) + grpcCalls.createMono( + it -> grpcCalls.intercept(context).getJob(request, it) ).onErrorResume(throwable -> { // The stable GetJob RPC is unavailable on pre-1.18 sidecars, fall back to Alpha1. if (isUnimplemented(throwable)) { - return this.createMono( - it -> intercept(context, asyncStub).getJobAlpha1(request, it)); + return grpcCalls.createMono( + it -> grpcCalls.intercept(context).getJobAlpha1(request, it)); } return Mono.error(throwable); }) @@ -1672,13 +1673,13 @@ public Mono deleteJob(DeleteJobRequest deleteJobRequest) { Mono deleteJobResponseMono = Mono.deferContextual(context -> - this.createMono( - it -> intercept(context, asyncStub).deleteJob(request, it) + grpcCalls.createMono( + it -> grpcCalls.intercept(context).deleteJob(request, it) ).onErrorResume(throwable -> { // The stable DeleteJob RPC is unavailable on pre-1.18 sidecars, fall back to Alpha1. if (isUnimplemented(throwable)) { - return this.createMono( - it -> intercept(context, asyncStub).deleteJobAlpha1(request, it)); + return grpcCalls.createMono( + it -> grpcCalls.intercept(context).deleteJobAlpha1(request, it)); } return Mono.error(throwable); }) @@ -1749,93 +1750,12 @@ private ConfigurationItem buildConfigurationItem( configurationItem.getMetadataMap()); } - /** - * Populates GRPC client with interceptors for telemetry. - * - * @param context Reactor's context. - * @param client GRPC client for Dapr. - * @return Client after adding interceptors. - */ - private DaprGrpc.DaprStub intercept(ContextView context, DaprGrpc.DaprStub client) { - return this.grpcInterceptors.intercept(client, context); - } - - /** - * Populates GRPC client with interceptors for telemetry. - * - * @param context Reactor's context. - * @param client GRPC client for Dapr. - * @param metadataConsumer Consumer of gRPC metadata. - * @return Client after adding interceptors. - */ - private DaprGrpc.DaprStub intercept( - ContextView context, DaprGrpc.DaprStub client, Consumer metadataConsumer) { - return this.grpcInterceptors.intercept(client, context, metadataConsumer); - } - - private Mono createMono(Consumer> consumer) { - return this.createMono(null, consumer); - } - - private Mono createMono(Metadata metadata, Consumer> consumer) { - return retryPolicy.apply( - Mono.create(sink -> DaprException.wrap(() -> consumer.accept( - createStreamObserver(sink, metadata))).run())); - } - - private Flux createFlux(Consumer> consumer) { - return this.createFlux(null, consumer); - } - - private Flux createFlux(Metadata metadata, Consumer> consumer) { - return retryPolicy.apply( - Flux.create(sink -> DaprException.wrap(() -> consumer.accept(createStreamObserver(sink, metadata))).run())); - } - - private StreamObserver createStreamObserver(MonoSink sink, Metadata grpcMetadata) { - return new StreamObserver() { - @Override - public void onNext(T value) { - sink.success(value); - } - - @Override - public void onError(Throwable t) { - sink.error(DaprException.propagate(DaprHttpException.fromGrpcExecutionException(grpcMetadata, t))); - } - - @Override - public void onCompleted() { - sink.success(); - } - }; - } - - private StreamObserver createStreamObserver(FluxSink sink, final Metadata grpcMetadata) { - return new StreamObserver() { - @Override - public void onNext(T value) { - sink.next(value); - } - - @Override - public void onError(Throwable t) { - sink.error(DaprException.propagate(DaprHttpException.fromGrpcExecutionException(grpcMetadata, t))); - } - - @Override - public void onCompleted() { - sink.complete(); - } - }; - } - @Override public Mono getMetadata() { DaprMetadataProtos.GetMetadataRequest metadataRequest = DaprMetadataProtos.GetMetadataRequest.newBuilder().build(); return Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).getMetadata(metadataRequest, it))) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).getMetadata(metadataRequest, it))) .map( it -> { try { @@ -1882,8 +1802,8 @@ public Mono converse(ConversationRequest conversationReque } Mono conversationResponseMono = Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context) .converseAlpha1(protosConversationRequestBuilder.build(), it) ) ); @@ -1937,8 +1857,8 @@ public Mono converseAlpha2(ConversationRequestAlpha2 DaprAiProtos.ConversationRequestAlpha2 protoRequest = buildConversationRequestProto(conversationRequestAlpha2); Mono conversationResponseMono = Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).converseAlpha2(protoRequest, it) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).converseAlpha2(protoRequest, it) ) ); @@ -2393,7 +2313,7 @@ public void onCompleted() { // Get the request stream observer from gRPC final StreamObserver requestObserver = - intercept(null, asyncStub).encryptAlpha1(responseObserver); + grpcCalls.intercept(null).encryptAlpha1(responseObserver); // Subscribe to the plaintext stream and send chunks request.getPlainTextStream() @@ -2484,7 +2404,7 @@ public void onCompleted() { // Get the request stream observer from gRPC final StreamObserver requestObserver = - intercept(null, asyncStub).decryptAlpha1(responseObserver); + grpcCalls.intercept(null).decryptAlpha1(responseObserver); // Subscribe to the ciphertext stream and send chunks request.getCipherTextStream() @@ -2533,8 +2453,8 @@ public Mono deleteBinaryFile(String storeName, String fileName) { .build(); return Mono.deferContextual( - context -> this.createMono( - it -> intercept(context, asyncStub).deleteBinaryFileAlpha1(request, it) + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).deleteBinaryFileAlpha1(request, it) ) ).then(); } catch (Exception ex) { @@ -2616,7 +2536,7 @@ public void onCompleted() { if (sink.isCancelled()) { return; } - intercept(context, asyncStub).getBinaryFileAlpha1(request, responseObserver); + grpcCalls.intercept(context).getBinaryFileAlpha1(request, responseObserver); // The call has started, so gRPC now accepts cancel and request calls. ClientCallStreamObserver stream = callStream.get(); @@ -2708,7 +2628,7 @@ public void onCompleted() { }; final StreamObserver requestObserver = - intercept(context, asyncStub).setBinaryFileAlpha1(responseObserver); + grpcCalls.intercept(context).setBinaryFileAlpha1(responseObserver); // The sidecar must receive the initial request within 5 seconds, so send it before any content. requestObserver.onNext(initialMessage); diff --git a/sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java b/sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java new file mode 100644 index 0000000000..bf65057d7f --- /dev/null +++ b/sdk/src/main/java/io/dapr/client/DaprGrpcCalls.java @@ -0,0 +1,126 @@ +/* + * 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 io.dapr.exceptions.DaprException; +import io.dapr.internal.exceptions.DaprHttpException; +import io.dapr.internal.grpc.DaprClientGrpcInterceptors; +import io.dapr.internal.resiliency.RetryPolicy; +import io.dapr.v1.DaprGrpc; +import io.grpc.Metadata; +import io.grpc.stub.StreamObserver; +import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; +import reactor.core.publisher.Mono; +import reactor.core.publisher.MonoSink; +import reactor.util.context.ContextView; + +import java.util.function.Consumer; + +/** + * Turns async gRPC calls to the Dapr sidecar into Reactor types, with the client's interceptors and retry policy. + */ +final class DaprGrpcCalls { + + private final DaprGrpc.DaprStub asyncStub; + + private final DaprClientGrpcInterceptors grpcInterceptors; + + private final RetryPolicy retryPolicy; + + DaprGrpcCalls( + DaprGrpc.DaprStub asyncStub, DaprClientGrpcInterceptors grpcInterceptors, RetryPolicy retryPolicy) { + this.asyncStub = asyncStub; + this.grpcInterceptors = grpcInterceptors; + this.retryPolicy = retryPolicy; + } + + /** + * Populates GRPC client with interceptors for telemetry. + * + * @param context Reactor's context. + * @return Client after adding interceptors. + */ + DaprGrpc.DaprStub intercept(ContextView context) { + return this.grpcInterceptors.intercept(this.asyncStub, context); + } + + /** + * Populates GRPC client with interceptors for telemetry. + * + * @param context Reactor's context. + * @param metadataConsumer Consumer of gRPC metadata. + * @return Client after adding interceptors. + */ + DaprGrpc.DaprStub intercept(ContextView context, Consumer metadataConsumer) { + return this.grpcInterceptors.intercept(this.asyncStub, context, metadataConsumer); + } + + Mono createMono(Consumer> consumer) { + return this.createMono(null, consumer); + } + + Mono createMono(Metadata metadata, Consumer> consumer) { + return retryPolicy.apply( + Mono.create(sink -> DaprException.wrap(() -> consumer.accept( + createStreamObserver(sink, metadata))).run())); + } + + Flux createFlux(Consumer> consumer) { + return this.createFlux(null, consumer); + } + + Flux createFlux(Metadata metadata, Consumer> consumer) { + return retryPolicy.apply( + Flux.create(sink -> DaprException.wrap(() -> consumer.accept(createStreamObserver(sink, metadata))).run())); + } + + private StreamObserver createStreamObserver(MonoSink sink, Metadata grpcMetadata) { + return new StreamObserver() { + @Override + public void onNext(T value) { + sink.success(value); + } + + @Override + public void onError(Throwable t) { + sink.error(DaprException.propagate(DaprHttpException.fromGrpcExecutionException(grpcMetadata, t))); + } + + @Override + public void onCompleted() { + sink.success(); + } + }; + } + + private StreamObserver createStreamObserver(FluxSink sink, final Metadata grpcMetadata) { + return new StreamObserver() { + @Override + public void onNext(T value) { + sink.next(value); + } + + @Override + public void onError(Throwable t) { + sink.error(DaprException.propagate(DaprHttpException.fromGrpcExecutionException(grpcMetadata, t))); + } + + @Override + public void onCompleted() { + sink.complete(); + } + }; + } +} From c90bd29c0c7d058cd47f2cadfa86f3ba688483d2 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Thu, 29 Oct 2026 11:10:15 +0100 Subject: [PATCH 2/2] refactor: drop dead retryPolicy field and Consumer import Both became unused when the gRPC helpers moved to DaprGrpcCalls. Part of #1875 Signed-off-by: Javier Aliaga --- sdk/src/main/java/io/dapr/client/DaprClientImpl.java | 7 ------- 1 file changed, 7 deletions(-) diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index 1f4be9acef..7beaaf3fdc 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -153,7 +153,6 @@ import java.util.TreeMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; @@ -181,11 +180,6 @@ public class DaprClientImpl extends AbstractDaprClient { */ private final GrpcChannelFacade channel; - /** - * The retry policy. - */ - private final RetryPolicy retryPolicy; - /** * The async gRPC stub. */ @@ -297,7 +291,6 @@ private DaprClientImpl( this.channel = channel; this.asyncStub = asyncStub; this.httpClient = httpClient; - this.retryPolicy = retryPolicy; this.grpcInterceptors = new DaprClientGrpcInterceptors(daprApiToken, timeoutPolicy); this.grpcCalls = new DaprGrpcCalls(asyncStub, this.grpcInterceptors, retryPolicy); this.logger = LoggerFactory.getLogger(DaprClientImpl.class);