From 1c019c91dba11e532edb4b8b9e368f061dc0a049 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Fri, 13 Nov 2026 10:53:18 +0100 Subject: [PATCH] refactor: move binary files out of DaprClientImpl Move deleteBinaryFile, getBinaryFile, getBinaryFileBytes and both setBinaryFile overloads, with their validation, chunking helper and chunk size constant, into a package-private DaprBinaryFileClient. The public methods on DaprClientImpl now delegate to it. Part of #1875 Signed-off-by: Javier Aliaga --- .../io/dapr/client/DaprBinaryFileClient.java | 283 ++++++++++++++++++ .../java/io/dapr/client/DaprClientImpl.java | 239 +-------------- 2 files changed, 291 insertions(+), 231 deletions(-) create mode 100644 sdk/src/main/java/io/dapr/client/DaprBinaryFileClient.java diff --git a/sdk/src/main/java/io/dapr/client/DaprBinaryFileClient.java b/sdk/src/main/java/io/dapr/client/DaprBinaryFileClient.java new file mode 100644 index 0000000000..3b72677cce --- /dev/null +++ b/sdk/src/main/java/io/dapr/client/DaprBinaryFileClient.java @@ -0,0 +1,283 @@ +/* + * 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.protobuf.ByteString; +import io.dapr.client.domain.SetBinaryFileRequestAlpha1; +import io.dapr.exceptions.DaprException; +import io.dapr.v1.CommonProtos; +import io.dapr.v1.DaprBinaryStoreProtos; +import io.grpc.Status; +import io.grpc.stub.ClientCallStreamObserver; +import io.grpc.stub.ClientResponseObserver; +import io.grpc.stub.StreamObserver; +import reactor.core.Disposable; +import reactor.core.Disposables; +import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.io.ByteArrayOutputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Runs the binary file calls (alpha) of {@link DaprClientImpl}: get, set and delete. + */ +final class DaprBinaryFileClient { + + /** + * Maximum size of one upload chunk sent to the binary store API. + */ + private static final int BINARY_FILE_CHUNK_SIZE = 32 * 1024; + + private final DaprGrpcCalls grpcCalls; + + DaprBinaryFileClient(DaprGrpcCalls grpcCalls) { + this.grpcCalls = grpcCalls; + } + + Mono deleteBinaryFile(String storeName, String fileName) { + try { + validateBinaryFileArgs(storeName, fileName); + + DaprBinaryStoreProtos.DeleteBinaryFileRequestAlpha1 request = + DaprBinaryStoreProtos.DeleteBinaryFileRequestAlpha1.newBuilder() + .setComponentName(storeName) + .setFileName(fileName) + .build(); + + return Mono.deferContextual( + context -> grpcCalls.createMono( + it -> grpcCalls.intercept(context).deleteBinaryFileAlpha1(request, it) + ) + ).then(); + } catch (Exception ex) { + return DaprException.wrapMono(ex); + } + } + + private static void validateBinaryFileArgs(String storeName, String fileName) { + if (storeName == null || storeName.trim().isEmpty()) { + throw new IllegalArgumentException("Store name cannot be null or empty."); + } + if (fileName == null || fileName.trim().isEmpty()) { + throw new IllegalArgumentException("File name cannot be null or empty."); + } + } + + Flux getBinaryFile(String storeName, String fileName) { + try { + validateBinaryFileArgs(storeName, fileName); + + DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1 request = + DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.newBuilder() + .setComponentName(storeName) + .setFileName(fileName) + .build(); + + return Flux.deferContextual(context -> Flux.create(sink -> { + final AtomicReference> callStream = + new AtomicReference<>(); + final AtomicBoolean unboundedDemand = new AtomicBoolean(); + + final ClientResponseObserver responseObserver = + new ClientResponseObserver() { + @Override + public void beforeStart( + ClientCallStreamObserver requestStream) { + // Receive messages only when the subscriber asks for them, so a slow subscriber + // does not make the client buffer the whole file. + requestStream.disableAutoRequestWithInitial(0); + callStream.set(requestStream); + } + + @Override + public void onNext(DaprBinaryStoreProtos.GetBinaryFileResponseAlpha1 response) { + byte[] data = response.getPayload().getData().toByteArray(); + if (data.length > 0) { + sink.next(data); + return; + } + // An empty chunk uses no subscriber demand, so ask for one more message. + // Unbounded demand has already asked for all messages. + ClientCallStreamObserver stream = callStream.get(); + if (stream != null && !unboundedDemand.get()) { + stream.request(1); + } + } + + @Override + public void onError(Throwable t) { + // After a cancel, gRPC reports CANCELLED. The subscriber is gone, so drop it. + if (!sink.isCancelled()) { + sink.error(DaprException.propagate(t)); + } + } + + @Override + public void onCompleted() { + sink.complete(); + } + }; + + // The subscriber can cancel before the call starts. Do not start the call then. + if (sink.isCancelled()) { + return; + } + grpcCalls.intercept(context).getBinaryFileAlpha1(request, responseObserver); + + // The call has started, so gRPC now accepts cancel and request calls. + ClientCallStreamObserver stream = callStream.get(); + if (stream != null) { + sink.onCancel(() -> stream.cancel("Subscriber cancelled the download.", null)); + sink.onRequest(n -> { + if (n == Long.MAX_VALUE) { + unboundedDemand.set(true); + } + stream.request((int) Math.min(n, Integer.MAX_VALUE)); + }); + } + })); + } catch (Exception ex) { + return DaprException.wrapFlux(ex); + } + } + + Mono getBinaryFileBytes(String storeName, String fileName) { + return getBinaryFile(storeName, fileName) + .collect(ByteArrayOutputStream::new, (out, chunk) -> out.write(chunk, 0, chunk.length)) + .map(ByteArrayOutputStream::toByteArray); + } + + Mono setBinaryFile(String storeName, String fileName, byte[] content, boolean overwrite) { + if (content == null) { + return DaprException.wrapMono(new IllegalArgumentException("Content cannot be null.")); + } + return setBinaryFile( + new SetBinaryFileRequestAlpha1(storeName, fileName, Flux.just(content)).setOverwrite(overwrite)); + } + + Mono setBinaryFile(SetBinaryFileRequestAlpha1 request) { + try { + if (request == null) { + throw new IllegalArgumentException("SetBinaryFileRequestAlpha1 cannot be null."); + } + validateBinaryFileArgs(request.getStoreName(), request.getFileName()); + if (request.getContent() == null) { + throw new IllegalArgumentException("Content cannot be null."); + } + + final DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1 initialMessage = + DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1.newBuilder() + .setInitialRequest(DaprBinaryStoreProtos.SetBinaryFileRequestInitialAlpha1.newBuilder() + .setComponentName(request.getStoreName()) + .setFileName(request.getFileName()) + .setOverwrite(request.isOverwrite()) + .build()) + .build(); + + return Mono.deferContextual(context -> Mono.create(sink -> { + final Disposable.Swap contentSubscription = Disposables.swap(); + // Set when the upload ends on the client side (subscriber cancel or content error). + // gRPC then reports CANCELLED, which must not replace the real outcome. + final AtomicBoolean terminated = new AtomicBoolean(); + + final StreamObserver responseObserver = + new StreamObserver() { + @Override + public void onNext(DaprBinaryStoreProtos.SetBinaryFileResponseAlpha1 response) { + // The response has no fields. + } + + @Override + public void onError(Throwable t) { + contentSubscription.dispose(); + if (!terminated.get()) { + sink.error(DaprException.propagate(t)); + } + } + + @Override + public void onCompleted() { + sink.success(); + } + }; + + final StreamObserver requestObserver = + grpcCalls.intercept(context).setBinaryFileAlpha1(responseObserver); + + // The sidecar must receive the initial request within 5 seconds, so send it before any content. + requestObserver.onNext(initialMessage); + + sink.onCancel(() -> { + terminated.set(true); + contentSubscription.dispose(); + requestObserver.onError( + Status.CANCELLED.withDescription("Subscriber cancelled the upload.").asRuntimeException()); + }); + + final long[] sequenceNumber = {0}; + final BaseSubscriber contentSubscriber = new BaseSubscriber() { + @Override + protected void hookOnNext(ByteString chunk) { + requestObserver.onNext(DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1.newBuilder() + .setPayload(CommonProtos.StreamPayload.newBuilder() + .setData(chunk) + .setSeq(sequenceNumber[0]++) + .build()) + .build()); + } + + @Override + protected void hookOnError(Throwable error) { + terminated.set(true); + requestObserver.onError(error); + sink.error(DaprException.propagate(error)); + } + + @Override + protected void hookOnComplete() { + requestObserver.onCompleted(); + } + }; + // Register the subscriber before it receives content, so a server error can stop even a + // synchronous content Flux. + contentSubscription.update(contentSubscriber); + request.getContent() + .concatMapIterable(DaprBinaryFileClient::splitBinaryFileChunk) + .subscribe(contentSubscriber); + })); + } catch (Exception ex) { + return DaprException.wrapMono(ex); + } + } + + /** + * Splits one content chunk into parts of at most {@link #BINARY_FILE_CHUNK_SIZE} bytes. + * An empty chunk gives no parts. + */ + private static List splitBinaryFileChunk(byte[] chunk) { + List parts = new ArrayList<>(); + for (int offset = 0; offset < chunk.length; offset += BINARY_FILE_CHUNK_SIZE) { + int length = Math.min(BINARY_FILE_CHUNK_SIZE, chunk.length - offset); + parts.add(ByteString.copyFrom(chunk, offset, length)); + } + return parts; + } +} diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index 3772d719c2..da4409b1a7 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -86,7 +86,6 @@ import io.dapr.utils.FailurePolicyUtils; import io.dapr.utils.TypeRef; import io.dapr.v1.CommonProtos; -import io.dapr.v1.DaprBinaryStoreProtos; import io.dapr.v1.DaprBindingsProtos; import io.dapr.v1.DaprConfigurationProtos; import io.dapr.v1.DaprCryptoProtos; @@ -98,21 +97,14 @@ import io.dapr.v1.DaprSecretProtos; import io.grpc.Channel; import io.grpc.Metadata; -import io.grpc.Status; import io.grpc.stub.AbstractStub; -import io.grpc.stub.ClientCallStreamObserver; -import io.grpc.stub.ClientResponseObserver; import io.grpc.stub.StreamObserver; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import reactor.core.Disposable; -import reactor.core.Disposables; -import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.util.retry.Retry; -import java.io.ByteArrayOutputStream; import java.io.IOException; import java.net.URI; import java.time.Duration; @@ -128,8 +120,6 @@ import java.util.List; import java.util.Map; import java.util.TreeMap; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.stream.Collectors; @@ -145,11 +135,6 @@ */ public class DaprClientImpl extends AbstractDaprClient { - /** - * Maximum size of one upload chunk sent to the binary store API. - */ - private static final int BINARY_FILE_CHUNK_SIZE = 32 * 1024; - private final Logger logger; /** @@ -170,6 +155,8 @@ public class DaprClientImpl extends AbstractDaprClient { private final DaprConversationClient conversationClient; + private final DaprBinaryFileClient binaryFileClient; + private final DaprPubSubClient pubSubClient; private final DaprStateClient stateClient; @@ -273,6 +260,7 @@ private DaprClientImpl( this.grpcInterceptors = new DaprClientGrpcInterceptors(daprApiToken, timeoutPolicy); this.grpcCalls = new DaprGrpcCalls(asyncStub, this.grpcInterceptors, retryPolicy); this.conversationClient = new DaprConversationClient(this.grpcCalls, objectSerializer); + this.binaryFileClient = new DaprBinaryFileClient(this.grpcCalls); this.pubSubClient = new DaprPubSubClient(this.grpcCalls, objectSerializer); this.stateClient = new DaprStateClient(this.grpcCalls, stateSerializer); this.vectorClient = new DaprVectorClient(this.grpcCalls); @@ -1484,32 +1472,7 @@ public void onCompleted() { */ @Override public Mono deleteBinaryFile(String storeName, String fileName) { - try { - validateBinaryFileArgs(storeName, fileName); - - DaprBinaryStoreProtos.DeleteBinaryFileRequestAlpha1 request = - DaprBinaryStoreProtos.DeleteBinaryFileRequestAlpha1.newBuilder() - .setComponentName(storeName) - .setFileName(fileName) - .build(); - - return Mono.deferContextual( - context -> grpcCalls.createMono( - it -> grpcCalls.intercept(context).deleteBinaryFileAlpha1(request, it) - ) - ).then(); - } catch (Exception ex) { - return DaprException.wrapMono(ex); - } - } - - private static void validateBinaryFileArgs(String storeName, String fileName) { - if (storeName == null || storeName.trim().isEmpty()) { - throw new IllegalArgumentException("Store name cannot be null or empty."); - } - if (fileName == null || fileName.trim().isEmpty()) { - throw new IllegalArgumentException("File name cannot be null or empty."); - } + return this.binaryFileClient.deleteBinaryFile(storeName, fileName); } /** @@ -1517,83 +1480,7 @@ private static void validateBinaryFileArgs(String storeName, String fileName) { */ @Override public Flux getBinaryFile(String storeName, String fileName) { - try { - validateBinaryFileArgs(storeName, fileName); - - DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1 request = - DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.newBuilder() - .setComponentName(storeName) - .setFileName(fileName) - .build(); - - return Flux.deferContextual(context -> Flux.create(sink -> { - final AtomicReference> callStream = - new AtomicReference<>(); - final AtomicBoolean unboundedDemand = new AtomicBoolean(); - - final ClientResponseObserver responseObserver = - new ClientResponseObserver() { - @Override - public void beforeStart( - ClientCallStreamObserver requestStream) { - // Receive messages only when the subscriber asks for them, so a slow subscriber - // does not make the client buffer the whole file. - requestStream.disableAutoRequestWithInitial(0); - callStream.set(requestStream); - } - - @Override - public void onNext(DaprBinaryStoreProtos.GetBinaryFileResponseAlpha1 response) { - byte[] data = response.getPayload().getData().toByteArray(); - if (data.length > 0) { - sink.next(data); - return; - } - // An empty chunk uses no subscriber demand, so ask for one more message. - // Unbounded demand has already asked for all messages. - ClientCallStreamObserver stream = callStream.get(); - if (stream != null && !unboundedDemand.get()) { - stream.request(1); - } - } - - @Override - public void onError(Throwable t) { - // After a cancel, gRPC reports CANCELLED. The subscriber is gone, so drop it. - if (!sink.isCancelled()) { - sink.error(DaprException.propagate(t)); - } - } - - @Override - public void onCompleted() { - sink.complete(); - } - }; - - // The subscriber can cancel before the call starts. Do not start the call then. - if (sink.isCancelled()) { - return; - } - grpcCalls.intercept(context).getBinaryFileAlpha1(request, responseObserver); - - // The call has started, so gRPC now accepts cancel and request calls. - ClientCallStreamObserver stream = callStream.get(); - if (stream != null) { - sink.onCancel(() -> stream.cancel("Subscriber cancelled the download.", null)); - sink.onRequest(n -> { - if (n == Long.MAX_VALUE) { - unboundedDemand.set(true); - } - stream.request((int) Math.min(n, Integer.MAX_VALUE)); - }); - } - })); - } catch (Exception ex) { - return DaprException.wrapFlux(ex); - } + return this.binaryFileClient.getBinaryFile(storeName, fileName); } /** @@ -1601,9 +1488,7 @@ public void onCompleted() { */ @Override public Mono getBinaryFileBytes(String storeName, String fileName) { - return getBinaryFile(storeName, fileName) - .collect(ByteArrayOutputStream::new, (out, chunk) -> out.write(chunk, 0, chunk.length)) - .map(ByteArrayOutputStream::toByteArray); + return this.binaryFileClient.getBinaryFileBytes(storeName, fileName); } /** @@ -1611,11 +1496,7 @@ public Mono getBinaryFileBytes(String storeName, String fileName) { */ @Override public Mono setBinaryFile(String storeName, String fileName, byte[] content, boolean overwrite) { - if (content == null) { - return DaprException.wrapMono(new IllegalArgumentException("Content cannot be null.")); - } - return setBinaryFile( - new SetBinaryFileRequestAlpha1(storeName, fileName, Flux.just(content)).setOverwrite(overwrite)); + return this.binaryFileClient.setBinaryFile(storeName, fileName, content, overwrite); } /** @@ -1623,111 +1504,7 @@ public Mono setBinaryFile(String storeName, String fileName, byte[] conten */ @Override public Mono setBinaryFile(SetBinaryFileRequestAlpha1 request) { - try { - if (request == null) { - throw new IllegalArgumentException("SetBinaryFileRequestAlpha1 cannot be null."); - } - validateBinaryFileArgs(request.getStoreName(), request.getFileName()); - if (request.getContent() == null) { - throw new IllegalArgumentException("Content cannot be null."); - } - - final DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1 initialMessage = - DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1.newBuilder() - .setInitialRequest(DaprBinaryStoreProtos.SetBinaryFileRequestInitialAlpha1.newBuilder() - .setComponentName(request.getStoreName()) - .setFileName(request.getFileName()) - .setOverwrite(request.isOverwrite()) - .build()) - .build(); - - return Mono.deferContextual(context -> Mono.create(sink -> { - final Disposable.Swap contentSubscription = Disposables.swap(); - // Set when the upload ends on the client side (subscriber cancel or content error). - // gRPC then reports CANCELLED, which must not replace the real outcome. - final AtomicBoolean terminated = new AtomicBoolean(); - - final StreamObserver responseObserver = - new StreamObserver() { - @Override - public void onNext(DaprBinaryStoreProtos.SetBinaryFileResponseAlpha1 response) { - // The response has no fields. - } - - @Override - public void onError(Throwable t) { - contentSubscription.dispose(); - if (!terminated.get()) { - sink.error(DaprException.propagate(t)); - } - } - - @Override - public void onCompleted() { - sink.success(); - } - }; - - final StreamObserver requestObserver = - grpcCalls.intercept(context).setBinaryFileAlpha1(responseObserver); - - // The sidecar must receive the initial request within 5 seconds, so send it before any content. - requestObserver.onNext(initialMessage); - - sink.onCancel(() -> { - terminated.set(true); - contentSubscription.dispose(); - requestObserver.onError( - Status.CANCELLED.withDescription("Subscriber cancelled the upload.").asRuntimeException()); - }); - - final long[] sequenceNumber = {0}; - final BaseSubscriber contentSubscriber = new BaseSubscriber() { - @Override - protected void hookOnNext(ByteString chunk) { - requestObserver.onNext(DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1.newBuilder() - .setPayload(CommonProtos.StreamPayload.newBuilder() - .setData(chunk) - .setSeq(sequenceNumber[0]++) - .build()) - .build()); - } - - @Override - protected void hookOnError(Throwable error) { - terminated.set(true); - requestObserver.onError(error); - sink.error(DaprException.propagate(error)); - } - - @Override - protected void hookOnComplete() { - requestObserver.onCompleted(); - } - }; - // Register the subscriber before it receives content, so a server error can stop even a - // synchronous content Flux. - contentSubscription.update(contentSubscriber); - request.getContent() - .concatMapIterable(DaprClientImpl::splitBinaryFileChunk) - .subscribe(contentSubscriber); - })); - } catch (Exception ex) { - return DaprException.wrapMono(ex); - } - } - - /** - * Splits one content chunk into parts of at most {@link #BINARY_FILE_CHUNK_SIZE} bytes. - * An empty chunk gives no parts. - */ - private static List splitBinaryFileChunk(byte[] chunk) { - List parts = new ArrayList<>(); - for (int offset = 0; offset < chunk.length; offset += BINARY_FILE_CHUNK_SIZE) { - int length = Math.min(BINARY_FILE_CHUNK_SIZE, chunk.length - offset); - parts.add(ByteString.copyFrom(chunk, offset, length)); - } - return parts; + return this.binaryFileClient.setBinaryFile(request); } /**