diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index 8625f95e6a..cf0dcce8f0 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -122,6 +122,7 @@ jobs: command: | set -e (cd examples && mm.py ./src/main/java/io/dapr/examples/crypto/README.md) + (cd examples && mm.py ./src/main/java/io/dapr/examples/binarystore/README.md) (cd examples && mm.py ./src/main/java/io/dapr/examples/workflows/README.md) (cd spring-boot-examples/pubsub && mm.py README.md) (cd spring-boot-examples/workflows/patterns && mm.py README.md) diff --git a/examples/components/binarystore/inmemory.yaml b/examples/components/binarystore/inmemory.yaml new file mode 100644 index 0000000000..cf891a88f5 --- /dev/null +++ b/examples/components/binarystore/inmemory.yaml @@ -0,0 +1,7 @@ +apiVersion: dapr.io/v1alpha1 +kind: Component +metadata: + name: binarystore +spec: + type: binarystore.in-memory + version: v1 diff --git a/examples/src/main/java/io/dapr/examples/binarystore/BinaryStoreExample.java b/examples/src/main/java/io/dapr/examples/binarystore/BinaryStoreExample.java new file mode 100644 index 0000000000..23aa1189e8 --- /dev/null +++ b/examples/src/main/java/io/dapr/examples/binarystore/BinaryStoreExample.java @@ -0,0 +1,91 @@ +/* + * 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.examples.binarystore; + +import io.dapr.client.DaprClientBuilder; +import io.dapr.client.DaprPreviewClient; +import io.dapr.client.domain.SetBinaryFileRequestAlpha1; +import io.dapr.config.Properties; +import io.dapr.config.Property; +import reactor.core.publisher.Flux; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Map; +import java.util.Random; + +/** + * Stores, reads, replaces, and deletes files with the Dapr binary store API (alpha). + * + *

Run with the in-memory binary store component in components/binarystore. + */ +public class BinaryStoreExample { + + private static final String STORE_NAME = "binarystore"; + private static final String FILE_NAME = "hello.txt"; + private static final String LARGE_FILE_NAME = "large.bin"; + private static final int LARGE_FILE_SIZE = 1024 * 1024; + private static final int LARGE_FILE_CHUNKS = 4; + + /** + * The main method of the example. + * + * @param args Command line arguments (unused). + * @throws Exception if the client fails to close. + */ + public static void main(String[] args) throws Exception { + Map, String> overrides = Map.of( + Properties.HTTP_PORT, "3500", + Properties.GRPC_PORT, "50001" + ); + + try (DaprPreviewClient client = new DaprClientBuilder().withPropertyOverrides(overrides).buildPreviewClient()) { + System.out.println("=== Dapr Binary Store Example ==="); + + // Store a small file from a byte array. overwrite=false fails if the file exists. + byte[] content = "Hello from the Dapr binary store!".getBytes(StandardCharsets.UTF_8); + client.setBinaryFile(STORE_NAME, FILE_NAME, content, false).block(); + System.out.println("Stored file: " + FILE_NAME); + + byte[] stored = client.getBinaryFileBytes(STORE_NAME, FILE_NAME).block(); + System.out.println("Read file content: " + new String(stored, StandardCharsets.UTF_8)); + + // Replace the file. overwrite=true replaces an existing file. + byte[] updated = "Updated content".getBytes(StandardCharsets.UTF_8); + client.setBinaryFile(STORE_NAME, FILE_NAME, updated, true).block(); + byte[] replaced = client.getBinaryFileBytes(STORE_NAME, FILE_NAME).block(); + System.out.println("Replaced file content: " + new String(replaced, StandardCharsets.UTF_8)); + + // Stream a large file in chunks, then read it back as a stream of chunks. + byte[] large = new byte[LARGE_FILE_SIZE]; + new Random().nextBytes(large); + int chunkSize = LARGE_FILE_SIZE / LARGE_FILE_CHUNKS; + Flux chunks = Flux.range(0, LARGE_FILE_CHUNKS) + .map(i -> Arrays.copyOfRange(large, i * chunkSize, (i + 1) * chunkSize)); + client.setBinaryFile(new SetBinaryFileRequestAlpha1(STORE_NAME, LARGE_FILE_NAME, chunks) + .setOverwrite(true)).block(); + long size = client.getBinaryFile(STORE_NAME, LARGE_FILE_NAME) + .map(chunk -> (long) chunk.length) + .reduce(0L, Long::sum) + .block(); + System.out.println("Read large file: " + size + " bytes"); + + client.deleteBinaryFile(STORE_NAME, FILE_NAME).block(); + client.deleteBinaryFile(STORE_NAME, LARGE_FILE_NAME).block(); + System.out.println("Deleted files"); + + System.out.println("SUCCESS: The binary store example finished."); + } + } +} diff --git a/examples/src/main/java/io/dapr/examples/binarystore/README.md b/examples/src/main/java/io/dapr/examples/binarystore/README.md new file mode 100644 index 0000000000..26fba3eace --- /dev/null +++ b/examples/src/main/java/io/dapr/examples/binarystore/README.md @@ -0,0 +1,134 @@ +## Dapr Binary Store API Example (alpha) + +This example shows how to store, read, replace, and delete files with the Dapr binary store API. The binary store API is an alpha API in Dapr 1.19. Its names and behavior can change in a later release. + +### Using the Binary Store API + +The Java SDK has these methods on `DaprPreviewClient`: +* `client.setBinaryFile(...)` stores a file from a byte array or from a `Flux` stream. Set `overwrite` to `true` to replace an existing file. +* `client.getBinaryFile(...)` reads a file as a `Flux` of chunks. +* `client.getBinaryFileBytes(...)` reads a whole file into a byte array. +* `client.deleteBinaryFile(...)` deletes a file. + +## Pre-requisites + +* [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/) and Dapr runtime 1.19 or later. +* Java JDK 17 (or greater): + * [Microsoft JDK 17](https://docs.microsoft.com/en-us/java/openjdk/download#openjdk-17) + * [Oracle JDK 17](https://www.oracle.com/technetwork/java/javase/downloads/index.html#JDK17) + * [OpenJDK 17](https://jdk.java.net/17/) +* [Apache Maven](https://maven.apache.org/install.html) version 3.x. + +### Checking out the code + +Clone this repository: + +```sh +git clone https://github.com/dapr/java-sdk.git +cd java-sdk +``` + +Then build the Maven project: + +```sh +# make sure you are in the `java-sdk` directory. +mvn install +``` + +Then get into the examples directory: + +```sh +cd examples +``` + +### Initialize Dapr + +Run `dapr init` to initialize Dapr in Self-Hosted Mode if it's not already initialized. + +### The binary store component + +The example uses the in-memory binary store component in `./components/binarystore/inmemory.yaml`: + +```yaml +apiVersion: dapr.io/v1alpha1 +kind: Component +metadata: + name: binarystore +spec: + type: binarystore.in-memory + version: v1 +``` + +The in-memory store keeps files only in the sidecar's memory. Use a different binary store component, such as `binarystore.aws.s3` or `binarystore.azure.blobstorage`, to keep files. + +### Running the Example + +`BinaryStoreExample.java` stores a small file, reads it, replaces it, streams a 1 MiB file in chunks, reads it back, and deletes both files: + +```java +try (DaprPreviewClient client = new DaprClientBuilder().buildPreviewClient()) { + byte[] content = "Hello from the Dapr binary store!".getBytes(StandardCharsets.UTF_8); + client.setBinaryFile("binarystore", "hello.txt", content, false).block(); + + byte[] stored = client.getBinaryFileBytes("binarystore", "hello.txt").block(); + + client.setBinaryFile("binarystore", "hello.txt", "Updated content".getBytes(StandardCharsets.UTF_8), true) + .block(); + + Flux chunks = ...; // the file content as a stream of chunks + client.setBinaryFile(new SetBinaryFileRequestAlpha1("binarystore", "large.bin", chunks).setOverwrite(true)) + .block(); + Flux received = client.getBinaryFile("binarystore", "large.bin"); + + client.deleteBinaryFile("binarystore", "hello.txt").block(); +} +``` + +The SDK sends upload content in chunks of at most 32 KiB. + +Use the following command to run this example: + + + +```bash +dapr run --resources-path ./components/binarystore --app-id binarystore-app --dapr-http-port 3500 --dapr-grpc-port 50001 -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.binarystore.BinaryStoreExample +``` + + + +### Sample Output + +``` +=== Dapr Binary Store Example === +Stored file: hello.txt +Read file content: Hello from the Dapr binary store! +Replaced file content: Updated content +Read large file: 1048576 bytes +Deleted files +SUCCESS: The binary store example finished. +``` + +### Cleanup + +To stop the app, run (or press CTRL+C): + + + +```bash +dapr stop --app-id binarystore-app +``` + + diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/binarystore/DaprPreviewClientBinaryStoreIT.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/binarystore/DaprPreviewClientBinaryStoreIT.java new file mode 100644 index 0000000000..74d05c2d65 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/binarystore/DaprPreviewClientBinaryStoreIT.java @@ -0,0 +1,145 @@ +/* + * 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.it.testcontainers.binarystore; + +import io.dapr.client.DaprClientBuilder; +import io.dapr.client.DaprPreviewClient; +import io.dapr.client.domain.SetBinaryFileRequestAlpha1; +import io.dapr.config.Properties; +import io.dapr.exceptions.DaprException; +import io.dapr.testcontainers.Component; +import io.dapr.testcontainers.DaprContainer; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import reactor.core.publisher.Flux; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Random; +import java.util.UUID; + +import static io.dapr.it.testcontainers.ContainerConstants.DAPR_RUNTIME_IMAGE_TAG; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Integration tests for the Dapr binary store Alpha1 API. + */ +@Testcontainers +@Tag("testcontainers") +public class DaprPreviewClientBinaryStoreIT { + + private static final String STORE_NAME = "binarystore"; + + @Container + private static final DaprContainer DAPR_CONTAINER = new DaprContainer(DAPR_RUNTIME_IMAGE_TAG) + .withAppName("binarystore-test-app") + .withComponent(new Component(STORE_NAME, "binarystore.in-memory", "v1", Collections.emptyList())); + + private static DaprPreviewClient daprPreviewClient; + + @BeforeAll + static void setUp() { + daprPreviewClient = new DaprClientBuilder() + .withPropertyOverride(Properties.HTTP_ENDPOINT, DAPR_CONTAINER.getHttpEndpoint()) + .withPropertyOverride(Properties.GRPC_ENDPOINT, DAPR_CONTAINER.getGrpcEndpoint()) + .buildPreviewClient(); + } + + @AfterAll + static void tearDown() throws Exception { + if (daprPreviewClient != null) { + daprPreviewClient.close(); + } + } + + @Test + public void shouldStoreAndReadSmallFile() { + String fileName = uniqueFileName(); + byte[] content = "hello binary store".getBytes(StandardCharsets.UTF_8); + + daprPreviewClient.setBinaryFile(STORE_NAME, fileName, content, false).block(); + + assertArrayEquals(content, daprPreviewClient.getBinaryFileBytes(STORE_NAME, fileName).block()); + } + + @Test + public void shouldStoreAndReadLargeFileAsChunks() { + String fileName = uniqueFileName(); + byte[] content = new byte[1024 * 1024]; + new Random(42).nextBytes(content); + // Uneven chunks check that the SDK splits and orders them correctly. + Flux chunks = Flux.just( + Arrays.copyOfRange(content, 0, 100_000), + Arrays.copyOfRange(content, 100_000, 700_000), + Arrays.copyOfRange(content, 700_000, content.length)); + + daprPreviewClient.setBinaryFile(new SetBinaryFileRequestAlpha1(STORE_NAME, fileName, chunks)).block(); + + List received = daprPreviewClient.getBinaryFile(STORE_NAME, fileName).collectList().block(); + ByteArrayOutputStream joined = new ByteArrayOutputStream(); + received.forEach(chunk -> joined.write(chunk, 0, chunk.length)); + assertArrayEquals(content, joined.toByteArray()); + } + + @Test + public void shouldFailWhenFileExistsAndOverwriteIsFalse() { + String fileName = uniqueFileName(); + daprPreviewClient.setBinaryFile(STORE_NAME, fileName, bytes("first"), false).block(); + + DaprException error = assertThrows(DaprException.class, + () -> daprPreviewClient.setBinaryFile(STORE_NAME, fileName, bytes("second"), false).block()); + + assertEquals("ALREADY_EXISTS", error.getErrorCode()); + assertArrayEquals(bytes("first"), daprPreviewClient.getBinaryFileBytes(STORE_NAME, fileName).block()); + } + + @Test + public void shouldReplaceFileWhenOverwriteIsTrue() { + String fileName = uniqueFileName(); + daprPreviewClient.setBinaryFile(STORE_NAME, fileName, bytes("first"), false).block(); + + daprPreviewClient.setBinaryFile(STORE_NAME, fileName, bytes("second"), true).block(); + + assertArrayEquals(bytes("second"), daprPreviewClient.getBinaryFileBytes(STORE_NAME, fileName).block()); + } + + @Test + public void shouldFailToReadDeletedFile() { + String fileName = uniqueFileName(); + daprPreviewClient.setBinaryFile(STORE_NAME, fileName, bytes("to delete"), false).block(); + + daprPreviewClient.deleteBinaryFile(STORE_NAME, fileName).block(); + + DaprException error = assertThrows(DaprException.class, + () -> daprPreviewClient.getBinaryFileBytes(STORE_NAME, fileName).block()); + assertEquals("NOT_FOUND", error.getErrorCode()); + } + + private static String uniqueFileName() { + return "file-" + UUID.randomUUID() + ".bin"; + } + + private static byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } +} diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index 42a6a09977..70e5815677 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -78,6 +78,7 @@ import io.dapr.client.domain.RuleMetadata; import io.dapr.client.domain.SaveStateRequest; import io.dapr.client.domain.ScheduleJobRequest; +import io.dapr.client.domain.SetBinaryFileRequestAlpha1; import io.dapr.client.domain.State; import io.dapr.client.domain.StateOptions; import io.dapr.client.domain.SubscribeConfigurationRequest; @@ -103,6 +104,7 @@ import io.dapr.utils.TypeRef; import io.dapr.v1.CommonProtos; import io.dapr.v1.DaprAiProtos; +import io.dapr.v1.DaprBinaryStoreProtos; import io.dapr.v1.DaprBindingsProtos; import io.dapr.v1.DaprConfigurationProtos; import io.dapr.v1.DaprCryptoProtos; @@ -116,10 +118,16 @@ import io.dapr.v1.DaprStateProtos; 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.FluxSink; import reactor.core.publisher.Mono; @@ -129,6 +137,7 @@ import javax.annotation.Nonnull; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.net.URI; import java.time.Duration; @@ -144,6 +153,8 @@ 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.Consumer; import java.util.function.Function; import java.util.stream.Collectors; @@ -160,6 +171,11 @@ */ 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; /** @@ -2501,4 +2517,255 @@ public void onCompleted() { return DaprException.wrapFlux(ex); } } + + /** + * {@inheritDoc} + */ + @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 -> this.createMono( + it -> intercept(context, asyncStub).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."); + } + } + + /** + * {@inheritDoc} + */ + @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; + } + intercept(context, asyncStub).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); + } + } + + /** + * {@inheritDoc} + */ + @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); + } + + /** + * {@inheritDoc} + */ + @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)); + } + + /** + * {@inheritDoc} + */ + @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 = + intercept(context, asyncStub).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; + } } diff --git a/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java b/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java index d0d2e3ea66..9078fc80c8 100644 --- a/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java +++ b/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java @@ -26,6 +26,7 @@ import io.dapr.client.domain.LockRequest; import io.dapr.client.domain.QueryStateRequest; import io.dapr.client.domain.QueryStateResponse; +import io.dapr.client.domain.SetBinaryFileRequestAlpha1; import io.dapr.client.domain.UnlockRequest; import io.dapr.client.domain.UnlockResponseStatus; import io.dapr.client.domain.query.Query; @@ -437,4 +438,78 @@ Flux subscribeToTopic( * @throws IllegalArgumentException if required parameters are missing. */ Flux decrypt(DecryptRequestAlpha1 request); + + /** + * Deletes a file from a binary store component. + * + *

If the file does not exist, the Mono fails with the NOT_FOUND error code. + * This is an alpha API. Its name and behavior can change in a later release. + * + * @param storeName Name of the binary store component. + * @param fileName Name of the file to delete. + * @return An empty Mono that completes when the sidecar deletes the file. + * @throws IllegalArgumentException if storeName or fileName is null or blank. + */ + Mono deleteBinaryFile(String storeName, String fileName); + + /** + * Reads a file from a binary store component as a stream of chunks. + * + *

The chunks arrive in file order. Cancel the subscription to stop the download. + * If the file does not exist, the Flux fails with the NOT_FOUND error code. + * A timeout set with dapr.api.timeoutMilliseconds applies to the whole download. + * This is an alpha API. Its name and behavior can change in a later release. + * + * @param storeName Name of the binary store component. + * @param fileName Name of the file to read. + * @return A Flux of file chunks. + * @throws IllegalArgumentException if storeName or fileName is null or blank. + */ + Flux getBinaryFile(String storeName, String fileName); + + /** + * Reads a whole file from a binary store component into memory. + * + *

This method holds the whole file in memory. Use {@link #getBinaryFile(String, String)} for large files. + * This is an alpha API. Its name and behavior can change in a later release. + * + * @param storeName Name of the binary store component. + * @param fileName Name of the file to read. + * @return A Mono with the file content. An empty file gives an empty array. + * @throws IllegalArgumentException if storeName or fileName is null or blank. + */ + Mono getBinaryFileBytes(String storeName, String fileName); + + /** + * Stores a file in a binary store component. + * + *

The SDK sends the content in chunks of at most 32 KiB. If overwrite is false and the file + * exists, the call fails with the ALREADY_EXISTS error code. The SDK does not apply gRPC flow + * control: if the content Flux produces data faster than the network sends it, gRPC holds the + * extra data in memory. + * Each subscription to the returned Mono subscribes to the content Flux again. To retry, use a + * content Flux that supports more than one subscription. + * A timeout set with dapr.api.timeoutMilliseconds applies to the whole upload. + * This is an alpha API. Its name and behavior can change in a later release. + * + * @param request The request with the store name, file name, content, and overwrite flag. + * @return An empty Mono that completes when the sidecar stores the file. + * @throws IllegalArgumentException if the request, store name, file name, or content is missing. + */ + Mono setBinaryFile(SetBinaryFileRequestAlpha1 request); + + /** + * Stores a file in a binary store component from a byte array. + * + *

The SDK sends the content in chunks of at most 32 KiB. + * This is an alpha API. Its name and behavior can change in a later release. + * + * @param storeName Name of the binary store component. + * @param fileName Name of the file to store. + * @param content File content. + * @param overwrite true to replace an existing file, false to fail when the file exists. + * @return An empty Mono that completes when the sidecar stores the file. + * @throws IllegalArgumentException if storeName, fileName, or content is missing. + */ + Mono setBinaryFile(String storeName, String fileName, byte[] content, boolean overwrite); } diff --git a/sdk/src/main/java/io/dapr/client/domain/SetBinaryFileRequestAlpha1.java b/sdk/src/main/java/io/dapr/client/domain/SetBinaryFileRequestAlpha1.java new file mode 100644 index 0000000000..64b1c6cc1b --- /dev/null +++ b/sdk/src/main/java/io/dapr/client/domain/SetBinaryFileRequestAlpha1.java @@ -0,0 +1,89 @@ +/* + * 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.domain; + +import reactor.core.publisher.Flux; + +/** + * Request to store a file in a Dapr binary store component. + * + *

This is an alpha API. Its name and behavior can change in a later release. + */ +public class SetBinaryFileRequestAlpha1 { + + private final String storeName; + private final String fileName; + private final Flux content; + private boolean overwrite; + + /** + * Constructor for SetBinaryFileRequestAlpha1. + * + * @param storeName Name of the binary store component. Required. + * @param fileName Name of the file to store. Required. + * @param content File content as a stream of chunks. Required. A chunk can have any size. + */ + public SetBinaryFileRequestAlpha1(String storeName, String fileName, Flux content) { + this.storeName = storeName; + this.fileName = fileName; + this.content = content; + } + + /** + * Gets the binary store component name. + * + * @return the store name + */ + public String getStoreName() { + return storeName; + } + + /** + * Gets the name of the file to store. + * + * @return the file name + */ + public String getFileName() { + return fileName; + } + + /** + * Gets the file content. + * + * @return the content as a Flux of byte arrays + */ + public Flux getContent() { + return content; + } + + /** + * Gets whether the request replaces an existing file. + * + * @return true to replace an existing file, false to fail when the file exists + */ + public boolean isOverwrite() { + return overwrite; + } + + /** + * Sets whether the request replaces an existing file. The default is false. + * + * @param overwrite true to replace an existing file, false to fail when the file exists + * @return this request instance for method chaining + */ + public SetBinaryFileRequestAlpha1 setOverwrite(boolean overwrite) { + this.overwrite = overwrite; + return this; + } +} diff --git a/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java index b8d1f3f6b2..05b35d8d4b 100644 --- a/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java @@ -44,17 +44,20 @@ import io.dapr.client.domain.QueryStateItem; import io.dapr.client.domain.QueryStateRequest; import io.dapr.client.domain.QueryStateResponse; +import io.dapr.client.domain.SetBinaryFileRequestAlpha1; import io.dapr.client.domain.SystemMessage; import io.dapr.client.domain.ToolMessage; import io.dapr.client.domain.UnlockResponseStatus; import io.dapr.client.domain.UserMessage; import io.dapr.client.domain.query.Query; +import io.dapr.exceptions.DaprException; import io.dapr.serializer.DaprObjectSerializer; import io.dapr.serializer.DefaultObjectSerializer; import io.dapr.utils.TypeRef; import io.dapr.v1.CommonProtos; import io.dapr.v1.DaprAiProtos; import io.dapr.v1.DaprAppCallbackProtos; +import io.dapr.v1.DaprBinaryStoreProtos; import io.dapr.v1.DaprCryptoProtos; import io.dapr.v1.DaprGrpc; import io.dapr.v1.DaprLockProtos; @@ -62,6 +65,8 @@ import io.dapr.v1.DaprStateProtos; import io.grpc.Status; import io.grpc.StatusRuntimeException; +import io.grpc.stub.ClientCallStreamObserver; +import io.grpc.stub.ClientResponseObserver; import io.grpc.stub.StreamObserver; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; @@ -72,7 +77,10 @@ import org.mockito.ArgumentMatchers; import org.mockito.Mockito; import org.mockito.stubbing.Answer; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; +import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; import java.io.IOException; @@ -86,8 +94,10 @@ import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -2525,4 +2535,538 @@ public void decryptWithKeyNameVersionTest() { assertNotNull(results); assertEquals(1, results.size()); } + + // ==================== Binary Store Tests ==================== + + @Test + @DisplayName("deleteBinaryFile should send the store name and file name") + public void deleteBinaryFileTest() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(DaprBinaryStoreProtos.DeleteBinaryFileRequestAlpha1.class); + doAnswer((Answer) invocation -> { + StreamObserver observer = + (StreamObserver) invocation.getArguments()[1]; + observer.onNext(DaprBinaryStoreProtos.DeleteBinaryFileResponseAlpha1.getDefaultInstance()); + observer.onCompleted(); + return null; + }).when(daprStub).deleteBinaryFileAlpha1(captor.capture(), any()); + + previewClient.deleteBinaryFile("mystore", "file.txt").block(); + + assertEquals("mystore", captor.getValue().getComponentName()); + assertEquals("file.txt", captor.getValue().getFileName()); + } + + @Test + @DisplayName("deleteBinaryFile should reject a null or blank store name or file name") + public void deleteBinaryFileInvalidArgumentsTest() { + assertThrows(IllegalArgumentException.class, () -> previewClient.deleteBinaryFile(null, "file.txt").block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.deleteBinaryFile(" ", "file.txt").block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.deleteBinaryFile("mystore", null).block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.deleteBinaryFile("mystore", "").block()); + } + + @Test + @DisplayName("deleteBinaryFile should keep the gRPC error code") + public void deleteBinaryFileNotFoundTest() { + doAnswer((Answer) invocation -> { + StreamObserver observer = + (StreamObserver) invocation.getArguments()[1]; + observer.onError(newStatusRuntimeException("NOT_FOUND", "binary file not found")); + return null; + }).when(daprStub).deleteBinaryFileAlpha1( + any(DaprBinaryStoreProtos.DeleteBinaryFileRequestAlpha1.class), any()); + + assertThrowsDaprException( + ExecutionException.class, + "NOT_FOUND", + "NOT_FOUND: binary file not found", + () -> previewClient.deleteBinaryFile("mystore", "missing.txt").block()); + } + + @Test + @DisplayName("getBinaryFile should emit non-empty chunks in order") + public void getBinaryFileChunksTest() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class); + doAnswer((Answer) invocation -> { + StreamObserver observer = + (StreamObserver) invocation.getArguments()[1]; + observer.onNext(binaryFileResponse("first", 0)); + observer.onNext(binaryFileResponse("", 1)); + observer.onNext(binaryFileResponse("second", 2)); + observer.onCompleted(); + return null; + }).when(daprStub).getBinaryFileAlpha1(captor.capture(), any()); + + List chunks = previewClient.getBinaryFile("mystore", "file.txt").collectList().block(); + + assertNotNull(chunks); + assertEquals(2, chunks.size()); + assertArrayEquals("first".getBytes(StandardCharsets.UTF_8), chunks.get(0)); + assertArrayEquals("second".getBytes(StandardCharsets.UTF_8), chunks.get(1)); + assertEquals("mystore", captor.getValue().getComponentName()); + assertEquals("file.txt", captor.getValue().getFileName()); + } + + @Test + @DisplayName("getBinaryFileBytes should join all chunks") + public void getBinaryFileBytesTest() { + doAnswer((Answer) invocation -> { + StreamObserver observer = + (StreamObserver) invocation.getArguments()[1]; + observer.onNext(binaryFileResponse("hello ", 0)); + observer.onNext(binaryFileResponse("world", 1)); + observer.onCompleted(); + return null; + }).when(daprStub).getBinaryFileAlpha1(any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + + byte[] content = previewClient.getBinaryFileBytes("mystore", "file.txt").block(); + + assertArrayEquals("hello world".getBytes(StandardCharsets.UTF_8), content); + } + + @Test + @DisplayName("getBinaryFileBytes should return an empty array for an empty file") + public void getBinaryFileBytesEmptyFileTest() { + doAnswer((Answer) invocation -> { + StreamObserver observer = + (StreamObserver) invocation.getArguments()[1]; + observer.onCompleted(); + return null; + }).when(daprStub).getBinaryFileAlpha1(any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + + byte[] content = previewClient.getBinaryFileBytes("mystore", "empty.txt").block(); + + assertNotNull(content); + assertEquals(0, content.length); + } + + @Test + @DisplayName("getBinaryFile should keep the gRPC error code") + public void getBinaryFileNotFoundTest() { + doAnswer((Answer) invocation -> { + StreamObserver observer = + (StreamObserver) invocation.getArguments()[1]; + observer.onError(newStatusRuntimeException("NOT_FOUND", "binary file not found")); + return null; + }).when(daprStub).getBinaryFileAlpha1(any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + + assertThrowsDaprException( + StatusRuntimeException.class, + "NOT_FOUND", + "NOT_FOUND: binary file not found", + () -> previewClient.getBinaryFileBytes("mystore", "missing.txt").block()); + } + + @Test + @DisplayName("getBinaryFile should reject a null or blank store name or file name") + public void getBinaryFileInvalidArgumentsTest() { + assertThrows(IllegalArgumentException.class, () -> previewClient.getBinaryFile(null, "file.txt").blockLast()); + assertThrows(IllegalArgumentException.class, () -> previewClient.getBinaryFile("mystore", " ").blockLast()); + assertThrows(IllegalArgumentException.class, () -> previewClient.getBinaryFileBytes("", "file.txt").block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.getBinaryFileBytes("mystore", null).block()); + } + + @Test + @DisplayName("getBinaryFile should cancel the gRPC call when the subscriber cancels") + public void getBinaryFileCancelTest() { + ClientCallStreamObserver callStream = + mock(ClientCallStreamObserver.class); + doAnswer((Answer) invocation -> { + ClientResponseObserver observer = + (ClientResponseObserver) invocation.getArguments()[1]; + observer.beforeStart(callStream); + observer.onNext(binaryFileResponse("first", 0)); + observer.onNext(binaryFileResponse("second", 1)); + return null; + }).when(daprStub).getBinaryFileAlpha1(any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + + byte[] first = previewClient.getBinaryFile("mystore", "file.txt").take(1).blockLast(); + + assertArrayEquals("first".getBytes(StandardCharsets.UTF_8), first); + verify(callStream).cancel(any(), any()); + } + + @Test + @DisplayName("getBinaryFile should ask gRPC for messages only when the subscriber has demand") + public void getBinaryFileBackpressureTest() { + ClientCallStreamObserver callStream = + mock(ClientCallStreamObserver.class); + AtomicReference> observerRef = + new AtomicReference<>(); + doAnswer((Answer) invocation -> { + ClientResponseObserver observer = + (ClientResponseObserver) invocation.getArguments()[1]; + observer.beforeStart(callStream); + observerRef.set(observer); + return null; + }).when(daprStub).getBinaryFileAlpha1(any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + + List received = new ArrayList<>(); + BaseSubscriber subscriber = new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + request(2); + } + + @Override + protected void hookOnNext(byte[] value) { + received.add(value); + } + }; + previewClient.getBinaryFile("mystore", "file.txt").subscribe(subscriber); + + verify(callStream).disableAutoRequestWithInitial(0); + verify(callStream).request(2); + + // An empty chunk uses no demand, so the client asks gRPC for one more message. + observerRef.get().onNext(binaryFileResponse("", 0)); + verify(callStream).request(1); + + observerRef.get().onNext(binaryFileResponse("first", 1)); + assertEquals(1, received.size()); + assertArrayEquals("first".getBytes(StandardCharsets.UTF_8), received.get(0)); + + subscriber.dispose(); + verify(callStream).cancel(any(), any()); + } + + private static DaprBinaryStoreProtos.GetBinaryFileResponseAlpha1 binaryFileResponse(String data, long seq) { + return DaprBinaryStoreProtos.GetBinaryFileResponseAlpha1.newBuilder() + .setPayload(CommonProtos.StreamPayload.newBuilder() + .setData(ByteString.copyFromUtf8(data)) + .setSeq(seq) + .build()) + .build(); + } + + @Test + @DisplayName("setBinaryFile should send the initial request first, then payloads with seq from 0") + public void setBinaryFileMessageOrderTest() { + List sent = mockSetBinaryFileStream(); + + Flux content = Flux.just( + "ab".getBytes(StandardCharsets.UTF_8), + new byte[0], + "cd".getBytes(StandardCharsets.UTF_8)); + previewClient.setBinaryFile(new SetBinaryFileRequestAlpha1("mystore", "file.txt", content)).block(); + + assertEquals(3, sent.size()); + assertTrue(sent.get(0).hasInitialRequest()); + assertEquals("mystore", sent.get(0).getInitialRequest().getComponentName()); + assertEquals("file.txt", sent.get(0).getInitialRequest().getFileName()); + assertTrue(sent.get(0).getInitialRequest().hasOverwrite()); + assertFalse(sent.get(0).getInitialRequest().getOverwrite()); + assertEquals(0, sent.get(1).getPayload().getSeq()); + assertEquals("ab", sent.get(1).getPayload().getData().toStringUtf8()); + assertEquals(1, sent.get(2).getPayload().getSeq()); + assertEquals("cd", sent.get(2).getPayload().getData().toStringUtf8()); + } + + @Test + @DisplayName("setBinaryFile should split a large byte array into 32 KiB chunks") + public void setBinaryFileSplitsLargeContentTest() { + List sent = mockSetBinaryFileStream(); + byte[] content = new byte[100 * 1024]; + for (int i = 0; i < content.length; i++) { + content[i] = (byte) i; + } + + previewClient.setBinaryFile("mystore", "large.bin", content, true).block(); + + // 1 initial request + 4 payloads: 32 KiB, 32 KiB, 32 KiB, 4 KiB. + assertEquals(5, sent.size()); + assertTrue(sent.get(0).getInitialRequest().getOverwrite()); + ByteString joined = ByteString.EMPTY; + for (int i = 1; i < sent.size(); i++) { + CommonProtos.StreamPayload payload = sent.get(i).getPayload(); + assertEquals(i - 1, payload.getSeq()); + assertTrue(payload.getData().size() <= 32 * 1024); + joined = joined.concat(payload.getData()); + } + assertEquals(32 * 1024, sent.get(1).getPayload().getData().size()); + assertEquals(4 * 1024, sent.get(4).getPayload().getData().size()); + assertArrayEquals(content, joined.toByteArray()); + } + + @Test + @DisplayName("setBinaryFile should keep the gRPC error code") + public void setBinaryFileAlreadyExistsTest() { + doAnswer((Answer>) invocation -> { + StreamObserver responseObserver = + (StreamObserver) invocation.getArguments()[0]; + responseObserver.onError(newStatusRuntimeException("ALREADY_EXISTS", "binary file already exists")); + return mock(StreamObserver.class); + }).when(daprStub).setBinaryFileAlpha1(any()); + + assertThrowsDaprException( + StatusRuntimeException.class, + "ALREADY_EXISTS", + "ALREADY_EXISTS: binary file already exists", + () -> previewClient.setBinaryFile("mystore", "file.txt", + "data".getBytes(StandardCharsets.UTF_8), false).block()); + } + + @Test + @DisplayName("setBinaryFile should cancel the gRPC call when the content stream fails") + public void setBinaryFileContentErrorTest() { + StreamObserver requestObserver = mock(StreamObserver.class); + doAnswer((Answer>) invocation -> + requestObserver).when(daprStub).setBinaryFileAlpha1(any()); + + Flux content = Flux.concat( + Flux.just("ab".getBytes(StandardCharsets.UTF_8)), + Flux.error(new IllegalStateException("disk read failed"))); + + // DaprException.propagate wraps a non-gRPC error in a DaprException whose cause is the original error. + assertThrowsDaprException(IllegalStateException.class, () -> previewClient.setBinaryFile( + new SetBinaryFileRequestAlpha1("mystore", "file.txt", content)).block()); + verify(requestObserver).onError(any(IllegalStateException.class)); + verify(requestObserver, times(0)).onCompleted(); + } + + @Test + @DisplayName("setBinaryFile should cancel the gRPC call when the subscriber cancels") + public void setBinaryFileCancelTest() { + StreamObserver requestObserver = mock(StreamObserver.class); + doAnswer((Answer>) invocation -> + requestObserver).when(daprStub).setBinaryFileAlpha1(any()); + + reactor.core.Disposable upload = previewClient.setBinaryFile( + new SetBinaryFileRequestAlpha1("mystore", "file.txt", Flux.never())).subscribe(); + upload.dispose(); + + verify(requestObserver).onError(any(StatusRuntimeException.class)); + verify(requestObserver, times(0)).onCompleted(); + } + + @Test + @DisplayName("setBinaryFile should report only the content error when the content stream fails") + public void setBinaryFileContentErrorSignalsOnceTest() { + List dropped = new ArrayList<>(); + Hooks.onErrorDropped(dropped::add); + try { + doAnswer((Answer>) invocation -> { + StreamObserver responseObserver = + (StreamObserver) invocation.getArguments()[0]; + return new StreamObserver() { + @Override + public void onNext(DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1 value) { + } + + @Override + public void onError(Throwable t) { + // Like real gRPC: cancelling the call reports CANCELLED to the response observer. + responseObserver.onError(Status.CANCELLED.withCause(t).asRuntimeException()); + } + + @Override + public void onCompleted() { + responseObserver.onNext(DaprBinaryStoreProtos.SetBinaryFileResponseAlpha1.getDefaultInstance()); + responseObserver.onCompleted(); + } + }; + }).when(daprStub).setBinaryFileAlpha1(any()); + + Flux content = Flux.concat( + Flux.just("ab".getBytes(StandardCharsets.UTF_8)), + Flux.error(new IllegalStateException("disk read failed"))); + + assertThrowsDaprException(IllegalStateException.class, () -> previewClient.setBinaryFile( + new SetBinaryFileRequestAlpha1("mystore", "file.txt", content)).block()); + assertTrue(dropped.isEmpty()); + } finally { + Hooks.resetOnErrorDropped(); + } + } + + @Test + @DisplayName("setBinaryFile should reject invalid requests") + public void setBinaryFileInvalidArgumentsTest() { + Flux content = Flux.just("data".getBytes(StandardCharsets.UTF_8)); + assertThrows(IllegalArgumentException.class, () -> previewClient.setBinaryFile(null).block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.setBinaryFile( + new SetBinaryFileRequestAlpha1(null, "file.txt", content)).block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.setBinaryFile( + new SetBinaryFileRequestAlpha1("mystore", " ", content)).block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.setBinaryFile( + new SetBinaryFileRequestAlpha1("mystore", "file.txt", null)).block()); + assertThrows(IllegalArgumentException.class, () -> previewClient.setBinaryFile( + "mystore", "file.txt", null, false).block()); + } + + /** + * Makes {@code setBinaryFileAlpha1} return a fake request stream. The fake records every message. + * When the client completes the request stream, the fake completes the response. + */ + private List mockSetBinaryFileStream() { + List sent = new ArrayList<>(); + doAnswer((Answer>) invocation -> { + StreamObserver responseObserver = + (StreamObserver) invocation.getArguments()[0]; + return new StreamObserver() { + @Override + public void onNext(DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1 value) { + sent.add(value); + } + + @Override + public void onError(Throwable t) { + responseObserver.onError(t); + } + + @Override + public void onCompleted() { + responseObserver.onNext(DaprBinaryStoreProtos.SetBinaryFileResponseAlpha1.getDefaultInstance()); + responseObserver.onCompleted(); + } + }; + }).when(daprStub).setBinaryFileAlpha1(any()); + return sent; + } + + @Test + @DisplayName("getBinaryFile should not start the call when the subscriber cancels first") + public void getBinaryFileEarlyCancelTest() { + previewClient.getBinaryFile("mystore", "file.txt").subscribe(new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + cancel(); + } + }); + + verify(daprStub, times(0)).getBinaryFileAlpha1( + any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + } + + @Test + @DisplayName("getBinaryFile should not report the CANCELLED error that follows a subscriber cancel") + public void getBinaryFileCancelledErrorIsNotDroppedTest() { + List dropped = new ArrayList<>(); + Hooks.onErrorDropped(dropped::add); + try { + ClientCallStreamObserver callStream = + mock(ClientCallStreamObserver.class); + AtomicReference> observerRef = + new AtomicReference<>(); + doAnswer((Answer) invocation -> { + ClientResponseObserver observer = + (ClientResponseObserver) invocation.getArguments()[1]; + observer.beforeStart(callStream); + observerRef.set(observer); + return null; + }).when(daprStub).getBinaryFileAlpha1(any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + + reactor.core.Disposable download = previewClient.getBinaryFile("mystore", "file.txt").subscribe(); + download.dispose(); + // Like real gRPC: the cancelled call reports CANCELLED. + observerRef.get().onError(Status.CANCELLED.asRuntimeException()); + + verify(callStream).cancel(any(), any()); + assertTrue(dropped.isEmpty()); + } finally { + Hooks.resetOnErrorDropped(); + } + } + + @Test + @DisplayName("getBinaryFile should not ask for more messages after empty chunks when demand is unbounded") + public void getBinaryFileUnboundedDemandTest() throws Exception { + ClientCallStreamObserver callStream = + mock(ClientCallStreamObserver.class); + AtomicReference> observerRef = + new AtomicReference<>(); + doAnswer((Answer) invocation -> { + ClientResponseObserver observer = + (ClientResponseObserver) invocation.getArguments()[1]; + observer.beforeStart(callStream); + observerRef.set(observer); + return null; + }).when(daprStub).getBinaryFileAlpha1(any(DaprBinaryStoreProtos.GetBinaryFileRequestAlpha1.class), any()); + + CompletableFuture> result = + previewClient.getBinaryFile("mystore", "file.txt").collectList().toFuture(); + + verify(callStream).request(Integer.MAX_VALUE); + observerRef.get().onNext(binaryFileResponse("", 0)); + observerRef.get().onNext(binaryFileResponse("data", 1)); + observerRef.get().onCompleted(); + + verify(callStream, times(0)).request(1); + List chunks = result.get(); + assertEquals(1, chunks.size()); + assertArrayEquals("data".getBytes(StandardCharsets.UTF_8), chunks.get(0)); + } + + @Test + @DisplayName("setBinaryFile should stop reading content when the server fails") + public void setBinaryFileServerErrorDisposesContentTest() { + AtomicReference> responseRef = + new AtomicReference<>(); + StreamObserver requestObserver = mock(StreamObserver.class); + doAnswer((Answer>) invocation -> { + responseRef.set( + (StreamObserver) invocation.getArguments()[0]); + return requestObserver; + }).when(daprStub).setBinaryFileAlpha1(any()); + + AtomicBoolean contentCancelled = new AtomicBoolean(); + Flux content = Flux.never().doOnCancel(() -> contentCancelled.set(true)); + CompletableFuture result = + previewClient.setBinaryFile(new SetBinaryFileRequestAlpha1("mystore", "file.txt", content)).toFuture(); + + responseRef.get().onError(newStatusRuntimeException("ALREADY_EXISTS", "binary file already exists")); + + assertTrue(contentCancelled.get()); + ExecutionException error = assertThrows(ExecutionException.class, result::get); + assertEquals("ALREADY_EXISTS", ((DaprException) error.getCause()).getErrorCode()); + } + + @Test + @DisplayName("setBinaryFile should stop sending synchronous content when the server fails mid-upload") + public void setBinaryFileSynchronousContentStopsOnServerErrorTest() { + List sent = new ArrayList<>(); + doAnswer((Answer>) invocation -> { + StreamObserver responseObserver = + (StreamObserver) invocation.getArguments()[0]; + return new StreamObserver() { + @Override + public void onNext(DaprBinaryStoreProtos.SetBinaryFileRequestAlpha1 value) { + sent.add(value); + // The server rejects the file while the second payload is being sent. + if (value.hasPayload() && value.getPayload().getSeq() == 1) { + responseObserver.onError(newStatusRuntimeException("ALREADY_EXISTS", "binary file already exists")); + } + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + }).when(daprStub).setBinaryFileAlpha1(any()); + + // 100 KiB in one synchronous chunk: the SDK would send 4 payloads. + byte[] content = new byte[100 * 1024]; + + assertThrowsDaprException( + StatusRuntimeException.class, + "ALREADY_EXISTS", + "ALREADY_EXISTS: binary file already exists", + () -> previewClient.setBinaryFile("mystore", "large.bin", content, false).block()); + // The initial request and the first two payloads only. + assertEquals(3, sent.size()); + } }