diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 416a19c6f6..60c2d2f64f 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -118,7 +118,7 @@ jobs: GOPROXY: https://proxy.golang.org JDK_VER: ${{ matrix.java }} DAPR_CLI_VER: 1.17.0-rc.2 - DAPR_RUNTIME_VER: 1.17.0-rc.6 + DAPR_RUNTIME_VER: 1.17.0-rc.8 DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.17.0-rc.2/install/install.sh DAPR_CLI_REF: DAPR_REF: diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index 26a881ae1d..555e375f81 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -38,7 +38,7 @@ jobs: GOPROXY: https://proxy.golang.org JDK_VER: ${{ matrix.java }} DAPR_CLI_VER: 1.17.0-rc.2 - DAPR_RUNTIME_VER: 1.17.0-rc.6 + DAPR_RUNTIME_VER: 1.17.0-rc.8 DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.17.0-rc.2/install/install.sh DAPR_CLI_REF: DAPR_REF: diff --git a/dapr-spring/dapr-spring-6-data/pom.xml b/dapr-spring/dapr-spring-6-data/pom.xml index 11b99c737f..5142cfdac1 100644 --- a/dapr-spring/dapr-spring-6-data/pom.xml +++ b/dapr-spring/dapr-spring-6-data/pom.xml @@ -12,6 +12,9 @@ dapr-spring-6-data + dapr-spring-data-6 + Dapr Spring Data 6 + jar 4.0.2 diff --git a/examples/src/main/java/io/dapr/examples/pubsub/BulkPublisher.java b/examples/src/main/java/io/dapr/examples/pubsub/BulkPublisher.java index 91b19f9ce9..ac3b02e9c3 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/BulkPublisher.java +++ b/examples/src/main/java/io/dapr/examples/pubsub/BulkPublisher.java @@ -15,7 +15,6 @@ import io.dapr.client.DaprClient; import io.dapr.client.DaprClientBuilder; -import io.dapr.client.DaprPreviewClient; import io.dapr.client.domain.BulkPublishResponse; import io.dapr.client.domain.BulkPublishResponseFailedEntry; import io.dapr.examples.OpenTelemetryConfig; @@ -59,13 +58,11 @@ public static void main(String[] args) throws Exception { Tracer tracer = openTelemetrySdk.getTracer(BulkPublisher.class.getCanonicalName()); Span span = tracer.spanBuilder("Bulk Publisher's Main").setSpanKind(SpanKind.CLIENT).startSpan(); - try (DaprPreviewClient client = (new DaprClientBuilder()).buildPreviewClient()) { - DaprClient c = (DaprClient) client; - - c.waitForSidecar(10000); + try (DaprClient client = (new DaprClientBuilder()).build()) { + client.waitForSidecar(10000); try (Scope scope = span.makeCurrent()) { - System.out.println("Using preview client..."); + System.out.println("Using Dapr client..."); List messages = new ArrayList<>(); diff --git a/examples/src/main/java/io/dapr/examples/pubsub/CloudEventBulkPublisher.java b/examples/src/main/java/io/dapr/examples/pubsub/CloudEventBulkPublisher.java index c30f6e684d..a6eb10a3a8 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/CloudEventBulkPublisher.java +++ b/examples/src/main/java/io/dapr/examples/pubsub/CloudEventBulkPublisher.java @@ -13,8 +13,8 @@ package io.dapr.examples.pubsub; +import io.dapr.client.DaprClient; import io.dapr.client.DaprClientBuilder; -import io.dapr.client.DaprPreviewClient; import io.dapr.client.domain.BulkPublishEntry; import io.dapr.client.domain.BulkPublishRequest; import io.dapr.client.domain.BulkPublishResponse; @@ -53,8 +53,8 @@ public class CloudEventBulkPublisher { * @throws Exception any exception */ public static void main(String[] args) throws Exception { - try (DaprPreviewClient client = (new DaprClientBuilder()).buildPreviewClient()) { - System.out.println("Using preview client..."); + try (DaprClient client = (new DaprClientBuilder()).build()) { + System.out.println("Using Dapr client..."); List>>> entries = new ArrayList<>(); for (int i = 0; i < NUM_MESSAGES; i++) { CloudEvent> cloudEvent = new CloudEvent<>(); diff --git a/examples/src/main/java/io/dapr/examples/pubsub/README.md b/examples/src/main/java/io/dapr/examples/pubsub/README.md index 6aa5d401de..5492ae0cef 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/README.md +++ b/examples/src/main/java/io/dapr/examples/pubsub/README.md @@ -275,13 +275,12 @@ Once running, the Subscriber should print the output as follows: Messages have been retrieved from the topic. -### Bulk Publish Messages -> Note : This API is currently in Alpha stage in Dapr runtime, hence the API methods in SDK are part of the DaprPreviewClient class. +### Bulk Publish Messages Another feature provided by the SDK is to allow users to publish multiple messages in a single call to the Dapr sidecar. -For this example, we have a simple Java application with a main method that uses the Dapr gPRC Preview Client to publish 10 messages to a specific topic in a single call. +For this example, we have a simple Java application with a main method that uses the Dapr Client to publish 10 messages to a specific topic in a single call. -In the `BulkPublisher.java` file, you will find the `BulkPublisher` class, containing the main method. The main method declares a Dapr Preview Client using the `DaprClientBuilder` class. Notice that this builder gets two serializer implementations in the constructor: one is for Dapr's sent and recieved objects, and the second is for objects to be persisted. +In the `BulkPublisher.java` file, you will find the `BulkPublisher` class, containing the main method. The main method declares a Dapr Client using the `DaprClientBuilder` class. Notice that this builder gets two serializer implementations in the constructor: one is for Dapr's sent and recieved objects, and the second is for objects to be persisted. The client publishes messages using `publishEvents` method. The Dapr client is also within a try-with-resource block to properly close the client at the end. See the code snippet below: Dapr's sidecar will automatically wrap the payload received into a CloudEvent object, which will later on be parsed by the subscriber. @@ -296,11 +295,10 @@ public class BulkPublisher { OpenTelemetry openTelemetry = OpenTelemetryConfig.createOpenTelemetry(); Tracer tracer = openTelemetry.getTracer(BulkPublisher.class.getCanonicalName()); Span span = tracer.spanBuilder("Bulk Publisher's Main").setSpanKind(Span.Kind.CLIENT).startSpan(); - try (DaprPreviewClient client = (new DaprClientBuilder()).buildPreviewClient()) { - DaprClient c = (DaprClient)client; - c.waitForSidecar(10000); + try (DaprClient client = (new DaprClientBuilder()).build()) { + client.waitForSidecar(10000); try (Scope scope = span.makeCurrent()) { - System.out.println("Using preview client..."); + System.out.println("Using Dapr client..."); List messages = new ArrayList<>(); System.out.println("Constructing the list of messages to publish"); for (int i = 0; i < NUM_MESSAGES; i++) { @@ -333,7 +331,7 @@ public class BulkPublisher { } } ``` -The code uses the `DaprPreviewClient` created by the `DaprClientBuilder` is used for the `publishEvents` (BulkPublish) preview API. +The code uses the `DaprClient` created by the `DaprClientBuilder` for the `publishEvents` (BulkPublish) API. In this case, when the `publishEvents` call is made, one of the arguments to the method is the content type of data, this being `text/plain` in the example. In this case, when parsing and printing the response, there is a concept of EntryID, which is automatically generated or can be set manually when using the `BulkPublishRequest` object. @@ -350,7 +348,7 @@ In this case, the application **MUST** override the content-type parameter via ` public class CloudEventBulkPublisher { ///... public static void main(String[] args) throws Exception { - try (DaprPreviewClient client = (new DaprClientBuilder()).buildPreviewClient()) { + try (DaprClient client = (new DaprClientBuilder()).build()) { // Construct request BulkPublishRequest>> request = new BulkPublishRequest<>(PUBSUB_NAME, TOPIC_NAME); List>>> entries = new ArrayList<>(); @@ -414,7 +412,7 @@ Once running, the BulkPublisher should print the output as follows: ```txt ✅ You're up and running! Both Dapr and your app logs will appear here. -== APP == Using preview client... +== APP == Using Dapr client... == APP == Constructing the list of messages to publish == APP == Going to publish message : This is message #0 == APP == Going to publish message : This is message #1 @@ -655,13 +653,15 @@ public class SubscriberGrpcService extends AppCallbackGrpc.AppCallbackImplBase { ///... } ``` -The `BulkSubscriberGrpcService.java` file is responsible for implementing the processing of bulk message subscriptions. When Dapr's sidecar successfully subscribes to bulk messages, it will call `onBulkTopicEventAlpha1` and pass them as a request parameter. You can refer to the example on how to handle bulk messages and respond correctly over gPRC. + +The `SubscriberGrpcService.java` now handles both regular and bulk message subscriptions since bulk pubsub is now stable. When Dapr's sidecar successfully subscribes to messages, it will call either `onTopicEvent` for regular messages or `onBulkTopicEvent` for bulk messages. Below is an example of the bulk handler: ```java -public class BulkSubscriberGrpcService extends AppCallbackAlphaGrpc.AppCallbackAlphaImplBase { +public class SubscriberGrpcService extends AppCallbackGrpc.AppCallbackImplBase { + // ... onTopicEvent method shown above ... @Override - public void onBulkTopicEventAlpha1(io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkRequest request, + public void onBulkTopicEvent(io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkRequest request, io.grpc.stub.StreamObserver responseObserver) { try { TopicEventBulkResponse.Builder responseBuilder = TopicEventBulkResponse.newBuilder(); @@ -669,6 +669,7 @@ public class BulkSubscriberGrpcService extends AppCallbackAlphaGrpc.AppCallbackA if (request.getEntriesCount() == 0) { responseObserver.onNext(responseBuilder.build()); responseObserver.onCompleted(); + return; } System.out.println("Bulk Subscriber received " + request.getEntriesCount() + " messages."); @@ -715,11 +716,11 @@ background: true sleep: 15 --> ```bash -// stop http subscriber if you have started one. +# stop http subscriber if you have started one. dapr stop --app-id subscriber -// start a grpc subscriber +# start a grpc subscriber dapr run --resources-path ./components/pubsub --app-id subscriber --app-port 3000 --app-protocol grpc -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.Subscriber -p 3000 ``` diff --git a/examples/src/main/java/io/dapr/examples/pubsub/grpc/BulkSubscriberGrpcService.java b/examples/src/main/java/io/dapr/examples/pubsub/grpc/BulkSubscriberGrpcService.java deleted file mode 100644 index 5422f5be75..0000000000 --- a/examples/src/main/java/io/dapr/examples/pubsub/grpc/BulkSubscriberGrpcService.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * Copyright 2023 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.pubsub.grpc; - -import io.dapr.v1.AppCallbackAlphaGrpc; -import io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkRequestEntry; -import io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkResponse; -import io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkResponseEntry; -import io.dapr.v1.DaprAppCallbackProtos.TopicEventResponse.TopicEventResponseStatus; - -/** - * Class that encapsulates all client-side logic for Grpc. - */ -public class BulkSubscriberGrpcService extends AppCallbackAlphaGrpc.AppCallbackAlphaImplBase { - - @Override - public void onBulkTopicEventAlpha1(io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkRequest request, - io.grpc.stub.StreamObserver responseObserver) { - try { - TopicEventBulkResponse.Builder responseBuilder = TopicEventBulkResponse.newBuilder(); - - if (request.getEntriesCount() == 0) { - responseObserver.onNext(responseBuilder.build()); - responseObserver.onCompleted(); - } - - System.out.println("Bulk Subscriber received " + request.getEntriesCount() + " messages."); - - for (TopicEventBulkRequestEntry entry : request.getEntriesList()) { - try { - System.out.printf("Bulk Subscriber message has entry ID: %s\n", entry.getEntryId()); - System.out.printf("Bulk Subscriber got: %s\n", entry.getCloudEvent().getData().toStringUtf8()); - TopicEventBulkResponseEntry.Builder responseEntryBuilder = TopicEventBulkResponseEntry - .newBuilder() - .setEntryId(entry.getEntryId()) - .setStatusValue(TopicEventResponseStatus.SUCCESS_VALUE); - responseBuilder.addStatuses(responseEntryBuilder); - } catch (Throwable e) { - TopicEventBulkResponseEntry.Builder responseEntryBuilder = TopicEventBulkResponseEntry - .newBuilder() - .setEntryId(entry.getEntryId()) - .setStatusValue(TopicEventResponseStatus.RETRY_VALUE); - responseBuilder.addStatuses(responseEntryBuilder); - } - } - TopicEventBulkResponse response = responseBuilder.build(); - responseObserver.onNext(response); - responseObserver.onCompleted(); - } catch (Throwable e) { - responseObserver.onError(e); - } - } - -} diff --git a/examples/src/main/java/io/dapr/examples/pubsub/grpc/Subscriber.java b/examples/src/main/java/io/dapr/examples/pubsub/grpc/Subscriber.java index 8c3c570887..052a67c8f5 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/grpc/Subscriber.java +++ b/examples/src/main/java/io/dapr/examples/pubsub/grpc/Subscriber.java @@ -51,7 +51,6 @@ public static void main(String[] args) throws Exception { Server server = ServerBuilder.forPort(port) .intercept(new SubscriberGrpcService.MetadataInterceptor()) .addService(new SubscriberGrpcService()) - .addService(new BulkSubscriberGrpcService()) .build(); server.start(); server.awaitTermination(); diff --git a/examples/src/main/java/io/dapr/examples/pubsub/grpc/SubscriberGrpcService.java b/examples/src/main/java/io/dapr/examples/pubsub/grpc/SubscriberGrpcService.java index 642d9e1b91..38d6480a09 100644 --- a/examples/src/main/java/io/dapr/examples/pubsub/grpc/SubscriberGrpcService.java +++ b/examples/src/main/java/io/dapr/examples/pubsub/grpc/SubscriberGrpcService.java @@ -16,6 +16,10 @@ import com.google.protobuf.Empty; import io.dapr.v1.AppCallbackGrpc; import io.dapr.v1.DaprAppCallbackProtos; +import io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkRequestEntry; +import io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkResponse; +import io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkResponseEntry; +import io.dapr.v1.DaprAppCallbackProtos.TopicEventResponse.TopicEventResponseStatus; import io.grpc.Context; import io.grpc.Contexts; import io.grpc.Metadata; @@ -103,6 +107,45 @@ public void onTopicEvent(DaprAppCallbackProtos.TopicEventRequest request, } } + @Override + public void onBulkTopicEvent(DaprAppCallbackProtos.TopicEventBulkRequest request, + StreamObserver responseObserver) { + try { + TopicEventBulkResponse.Builder responseBuilder = TopicEventBulkResponse.newBuilder(); + + if (request.getEntriesCount() == 0) { + responseObserver.onNext(responseBuilder.build()); + responseObserver.onCompleted(); + return; + } + + System.out.println("Bulk Subscriber received " + request.getEntriesCount() + " messages."); + + for (TopicEventBulkRequestEntry entry : request.getEntriesList()) { + try { + System.out.printf("Bulk Subscriber message has entry ID: %s\n", entry.getEntryId()); + System.out.printf("Bulk Subscriber got: %s\n", entry.getCloudEvent().getData().toStringUtf8()); + TopicEventBulkResponseEntry.Builder responseEntryBuilder = TopicEventBulkResponseEntry + .newBuilder() + .setEntryId(entry.getEntryId()) + .setStatusValue(TopicEventResponseStatus.SUCCESS_VALUE); + responseBuilder.addStatuses(responseEntryBuilder); + } catch (Throwable e) { + TopicEventBulkResponseEntry.Builder responseEntryBuilder = TopicEventBulkResponseEntry + .newBuilder() + .setEntryId(entry.getEntryId()) + .setStatusValue(TopicEventResponseStatus.RETRY_VALUE); + responseBuilder.addStatuses(responseEntryBuilder); + } + } + TopicEventBulkResponse response = responseBuilder.build(); + responseObserver.onNext(response); + responseObserver.onCompleted(); + } catch (Throwable e) { + responseObserver.onError(e); + } + } + /** * Add pubsub name and topic to topicSubscriptionList. * @@ -119,4 +162,3 @@ public void registerConsumer(String pubsubName, String topic, boolean isBulkMess .build()); } } - diff --git a/pom.xml b/pom.xml index f990109d7a..81124fb3e2 100644 --- a/pom.xml +++ b/pom.xml @@ -16,7 +16,7 @@ UTF-8 1.69.0 3.25.5 - https://raw.githubusercontent.com/dapr/dapr/v1.17.0-rc.6/dapr/proto + https://raw.githubusercontent.com/dapr/dapr/v1.17.0-rc.8/dapr/proto https://raw.githubusercontent.com/dapr/durabletask-protobuf/main/protos/orchestrator_service.proto 1.18.0-SNAPSHOT 1.7.1 diff --git a/sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java b/sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java index 57b9485e97..55352e93a4 100644 --- a/sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java @@ -17,7 +17,6 @@ import com.fasterxml.jackson.databind.ObjectMapper; import io.dapr.client.DaprClient; import io.dapr.client.DaprClientBuilder; -import io.dapr.client.DaprPreviewClient; import io.dapr.client.domain.BulkPublishEntry; import io.dapr.client.domain.BulkPublishRequest; import io.dapr.client.domain.BulkPublishResponse; @@ -120,7 +119,7 @@ public void testBulkPublishPubSubNotFound() throws Exception { this.getClass().getSimpleName(), 60000)); - try (DaprPreviewClient client = daprRun.newDaprClientBuilder().buildPreviewClient()) { + try (DaprClient client = daprRun.newDaprClientBuilder().build()) { assertThrowsDaprException( "INVALID_ARGUMENT", "INVALID_ARGUMENT: pubsub unknown pubsub is not found", @@ -152,8 +151,7 @@ public String getContentType() { return "application/json"; } }; - try (DaprClient client = daprRun.newDaprClientBuilder().withObjectSerializer(serializer).build(); - DaprPreviewClient previewClient = daprRun.newDaprClientBuilder().withObjectSerializer(serializer).buildPreviewClient()) { + try (DaprClient client = daprRun.newDaprClientBuilder().withObjectSerializer(serializer).build()) { // Only for the gRPC test // Send a multiple messages on one topic in messagebus pubsub via publishEvents API. List messages = new ArrayList<>(); @@ -161,7 +159,7 @@ public String getContentType() { messages.add(String.format("This is message #%d on topic %s", i, TOPIC_BULK)); } //Publishing 10 messages - BulkPublishResponse response = previewClient.publishEvents(PUBSUB_NAME, TOPIC_BULK, "", messages).block(); + BulkPublishResponse response = client.publishEvents(PUBSUB_NAME, TOPIC_BULK, "", messages).block(); System.out.println(String.format("Published %d messages to topic '%s' pubsub_name '%s'", NUM_MESSAGES, TOPIC_BULK, PUBSUB_NAME)); assertNotNull(response, "expected not null bulk publish response"); @@ -170,14 +168,14 @@ public String getContentType() { //Publishing an object. MyObject object = new MyObject(); object.setId("123"); - response = previewClient.publishEvents(PUBSUB_NAME, TOPIC_BULK, + response = client.publishEvents(PUBSUB_NAME, TOPIC_BULK, "application/json", Collections.singletonList(object)).block(); System.out.println("Published one object."); assertNotNull(response, "expected not null bulk publish response"); assertEquals(0, response.getFailedEntries().size(), "expected no failures in the response"); //Publishing a single byte: Example of non-string based content published - previewClient.publishEvents(PUBSUB_NAME, TOPIC_BULK, "", + client.publishEvents(PUBSUB_NAME, TOPIC_BULK, "", Collections.singletonList(new byte[]{1})).block(); System.out.println("Published one byte."); @@ -197,7 +195,7 @@ public String getContentType() { )); //Publishing a cloud event. - previewClient.publishEvents(req).block(); + client.publishEvents(req).block(); assertNotNull(response, "expected not null bulk publish response"); assertEquals(0, response.getFailedEntries().size(), "expected no failures in the response"); diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/pubsub/http/DaprPubSubIT.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/pubsub/http/DaprPubSubIT.java index 8d7111952d..eaa0f0c99f 100644 --- a/sdk-tests/src/test/java/io/dapr/it/testcontainers/pubsub/http/DaprPubSubIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/pubsub/http/DaprPubSubIT.java @@ -19,7 +19,6 @@ import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import io.dapr.client.DaprClient; -import io.dapr.client.DaprPreviewClient; import io.dapr.client.domain.BulkPublishEntry; import io.dapr.client.domain.BulkPublishRequest; import io.dapr.client.domain.BulkPublishResponse; @@ -154,7 +153,7 @@ public void shouldReceiveInvalidArgument() throws Exception { @Test @DisplayName("Should receive INVALID_ARGUMENT using bulk publish when the specified Pub/Sub name does not exist") public void shouldReceiveInvalidArgumentWithBulkPublish() throws Exception { - try (DaprPreviewClient client = DaprClientFactory.createDaprClientBuilder(DAPR_CONTAINER).buildPreviewClient()) { + try (DaprClient client = DaprClientFactory.createDaprClientBuilder(DAPR_CONTAINER).build()) { assertThrowsDaprException( "INVALID_ARGUMENT", "INVALID_ARGUMENT: pubsub unknown pubsub is not found", @@ -167,19 +166,18 @@ public void shouldReceiveInvalidArgumentWithBulkPublish() throws Exception { public void shouldPublishSomePayloadTypesWithNoError() throws Exception { try ( - DaprClient client = DaprClientFactory.createDaprClientBuilder(DAPR_CONTAINER).build(); - DaprPreviewClient previewClient = DaprClientFactory.createDaprClientBuilder(DAPR_CONTAINER).withObjectSerializer( + DaprClient client = DaprClientFactory.createDaprClientBuilder(DAPR_CONTAINER).withObjectSerializer( createJacksonObjectSerializer()) - .buildPreviewClient() + .build() ) { - publishBulkStringsAsserting(previewClient); + publishBulkStringsAsserting(client); - publishMyObjectAsserting(previewClient); + publishMyObjectAsserting(client); - publishByteAsserting(previewClient); + publishByteAsserting(client); - publishCloudEventAsserting(previewClient); + publishCloudEventAsserting(client); Thread.sleep(10000); @@ -409,10 +407,10 @@ private static void sendBulkMessagesAsText(DaprClient client, String topicName) } } - private void publishMyObjectAsserting(DaprPreviewClient previewClient) { + private void publishMyObjectAsserting(DaprClient client) { PubSubIT.MyObject object = new PubSubIT.MyObject(); object.setId("123"); - BulkPublishResponse response = previewClient.publishEvents( + BulkPublishResponse response = client.publishEvents( PUBSUB_NAME, TOPIC_BULK, "application/json", @@ -424,20 +422,20 @@ private void publishMyObjectAsserting(DaprPreviewClient previewClient) { }); } - private void publishBulkStringsAsserting(DaprPreviewClient previewClient) { + private void publishBulkStringsAsserting(DaprClient client) { List messages = new ArrayList<>(); for (int i = 0; i < NUM_MESSAGES; i++) { messages.add(String.format("This is message #%d on topic %s", i, TOPIC_BULK)); } - BulkPublishResponse response = previewClient.publishEvents(PUBSUB_NAME, TOPIC_BULK, "", messages).block(); + BulkPublishResponse response = client.publishEvents(PUBSUB_NAME, TOPIC_BULK, "", messages).block(); SoftAssertions.assertSoftly(softly -> { softly.assertThat(response).isNotNull(); softly.assertThat(response.getFailedEntries().size()).isZero(); }); } - private void publishByteAsserting(DaprPreviewClient previewClient) { - BulkPublishResponse response = previewClient.publishEvents( + private void publishByteAsserting(DaprClient client) { + BulkPublishResponse response = client.publishEvents( PUBSUB_NAME, TOPIC_BULK, "", @@ -449,7 +447,7 @@ private void publishByteAsserting(DaprPreviewClient previewClient) { }); } - private void publishCloudEventAsserting(DaprPreviewClient previewClient) { + private void publishCloudEventAsserting(DaprClient client) { CloudEvent cloudEvent = new CloudEvent<>(); cloudEvent.setId("1234"); cloudEvent.setData("message from cloudevent"); @@ -465,7 +463,7 @@ private void publishCloudEventAsserting(DaprPreviewClient previewClient) { new BulkPublishEntry<>("1", cloudEvent, "application/cloudevents+json", null) ) ); - BulkPublishResponse> response = previewClient.publishEvents(req).block(); + BulkPublishResponse> response = client.publishEvents(req).block(); SoftAssertions.assertSoftly(softly -> { softly.assertThat(response).isNotNull(); softly.assertThat(response.getFailedEntries().size()).isZero(); diff --git a/sdk/src/main/java/io/dapr/client/DaprClient.java b/sdk/src/main/java/io/dapr/client/DaprClient.java index f341344b3f..ddf89f587f 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClient.java +++ b/sdk/src/main/java/io/dapr/client/DaprClient.java @@ -13,6 +13,10 @@ package io.dapr.client; +import io.dapr.client.domain.BulkPublishEntry; +import io.dapr.client.domain.BulkPublishRequest; +import io.dapr.client.domain.BulkPublishResponse; +import io.dapr.client.domain.BulkPublishResponseFailedEntry; import io.dapr.client.domain.ConfigurationItem; import io.dapr.client.domain.DaprMetadata; import io.dapr.client.domain.DeleteJobRequest; @@ -91,6 +95,77 @@ public interface DaprClient extends AutoCloseable { */ Mono publishEvent(PublishEventRequest request); + /** + * Publish multiple events to Dapr in a single request. + * + * @param request {@link BulkPublishRequest} object. + * @return A Mono of {@link BulkPublishResponse} object. + * @param The type of events to publish in the call. + */ + Mono> publishEvents(BulkPublishRequest request); + + /** + * Publish multiple events to Dapr in a single request. + * + * @param pubsubName the pubsub name we will publish the event to. + * @param topicName the topicName where the event will be published. + * @param events the {@link List} of events to be published. + * @param contentType the content type of the event. Use Mime based types. + * @return the {@link BulkPublishResponse} containing publish status of each event. + * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be + * generated based on the order of events in the {@link List}. + * @param The type of the events to publish in the call. + */ + Mono> publishEvents(String pubsubName, String topicName, String contentType, + List events); + + /** + * Publish multiple events to Dapr in a single request. + * + * @param pubsubName the pubsub name we will publish the event to. + * @param topicName the topicName where the event will be published. + * @param events the varargs of events to be published. + * @param contentType the content type of the event. Use Mime based types. + * @return the {@link BulkPublishResponse} containing publish status of each event. + * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be + * generated based on the order of events in the {@link List}. + * @param The type of the events to publish in the call. + */ + Mono> publishEvents(String pubsubName, String topicName, String contentType, + T... events); + + /** + * Publish multiple events to Dapr in a single request. + * + * @param pubsubName the pubsub name we will publish the event to. + * @param topicName the topicName where the event will be published. + * @param events the {@link List} of events to be published. + * @param contentType the content type of the event. Use Mime based types. + * @param requestMetadata the metadata to be set at the request level for the {@link BulkPublishRequest}. + * @return the {@link BulkPublishResponse} containing publish status of each event. + * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be + * generated based on the order of events in the {@link List}. + * @param The type of the events to publish in the call. + */ + Mono> publishEvents(String pubsubName, String topicName, String contentType, + Map requestMetadata, List events); + + /** + * Publish multiple events to Dapr in a single request. + * + * @param pubsubName the pubsub name we will publish the event to. + * @param topicName the topicName where the event will be published. + * @param events the varargs of events to be published. + * @param contentType the content type of the event. Use Mime based types. + * @param requestMetadata the metadata to be set at the request level for the {@link BulkPublishRequest}. + * @return the {@link BulkPublishResponse} containing publish status of each event. + * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be + * generated based on the order of events in the {@link List}. + * @param The type of the events to publish in the call. + */ + Mono> publishEvents(String pubsubName, String topicName, String contentType, + Map requestMetadata, T... events); + /** * Invoke a service method, using serialization. * diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index cf82c5fb73..60c7568377 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -444,8 +444,25 @@ public Mono> publishEvents(BulkPublishRequest requ return Mono.deferContextual( context -> this.createMono( - it -> intercept(context, asyncStub).bulkPublishEventAlpha1(envelopeBuilder.build(), it) - ) + it -> intercept(context, asyncStub).bulkPublishEvent(envelopeBuilder.build(), it) + ).onErrorResume(throwable -> { + // Check if the error is UNIMPLEMENTED, and try the alpha API just in case + // The exception might be wrapped in DaprException, so check both + Throwable cause = throwable; + if (throwable instanceof DaprException && throwable.getCause() != null) { + cause = throwable.getCause(); + } + + if (cause instanceof io.grpc.StatusRuntimeException) { + io.grpc.StatusRuntimeException statusException = (io.grpc.StatusRuntimeException) cause; + if (statusException.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) { + return this.createMono( + it -> intercept(context, asyncStub).bulkPublishEventAlpha1(envelopeBuilder.build(), it) + ); + } + } + return Mono.error(throwable); + }) ).map( it -> { List> entries = new ArrayList<>(); diff --git a/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java b/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java index b78c1eccf8..bb5ae5e1fa 100644 --- a/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java +++ b/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java @@ -163,9 +163,11 @@ Mono> queryState(String storeName, Query query, * Publish multiple events to Dapr in a single request. * * @param request {@link BulkPublishRequest} object. - * @return A Mono of {@link BulkPublishResponse} object. * @param The type of events to publish in the call. + * @return A Mono of {@link BulkPublishResponse} object. + * @deprecated This method is now stable. Use {@link DaprClient#publishEvents(BulkPublishRequest)} instead. */ + @Deprecated Mono> publishEvents(BulkPublishRequest request); /** @@ -175,11 +177,14 @@ Mono> queryState(String storeName, Query query, * @param topicName the topicName where the event will be published. * @param events the {@link List} of events to be published. * @param contentType the content type of the event. Use Mime based types. + * @param The type of the events to publish in the call. * @return the {@link BulkPublishResponse} containing publish status of each event. * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be * generated based on the order of events in the {@link List}. - * @param The type of the events to publish in the call. + * @deprecated This method is now stable. + * Use {@link DaprClient#publishEvents(String, String, String, List)} instead. */ + @Deprecated Mono> publishEvents(String pubsubName, String topicName, String contentType, List events); @@ -190,11 +195,14 @@ Mono> publishEvents(String pubsubName, String topicNa * @param topicName the topicName where the event will be published. * @param events the varargs of events to be published. * @param contentType the content type of the event. Use Mime based types. + * @param The type of the events to publish in the call. * @return the {@link BulkPublishResponse} containing publish status of each event. * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be * generated based on the order of events in the {@link List}. - * @param The type of the events to publish in the call. + * @deprecated This method is now stable. + * Use {@link DaprClient#publishEvents(String, String, String, Object[])} instead. */ + @Deprecated Mono> publishEvents(String pubsubName, String topicName, String contentType, T... events); @@ -206,11 +214,14 @@ Mono> publishEvents(String pubsubName, String topicNa * @param events the {@link List} of events to be published. * @param contentType the content type of the event. Use Mime based types. * @param requestMetadata the metadata to be set at the request level for the {@link BulkPublishRequest}. + * @param The type of the events to publish in the call. * @return the {@link BulkPublishResponse} containing publish status of each event. * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be * generated based on the order of events in the {@link List}. - * @param The type of the events to publish in the call. + * @deprecated This method is now stable. + * Use {@link DaprClient#publishEvents(String, String, String, Map, List)} instead. */ + @Deprecated Mono> publishEvents(String pubsubName, String topicName, String contentType, Map requestMetadata, List events); @@ -222,11 +233,14 @@ Mono> publishEvents(String pubsubName, String topicNa * @param events the varargs of events to be published. * @param contentType the content type of the event. Use Mime based types. * @param requestMetadata the metadata to be set at the request level for the {@link BulkPublishRequest}. + * @param The type of the events to publish in the call. * @return the {@link BulkPublishResponse} containing publish status of each event. * The "entryID" field in {@link BulkPublishEntry} in {@link BulkPublishResponseFailedEntry} will be * generated based on the order of events in the {@link List}. - * @param The type of the events to publish in the call. + * @deprecated This method is now stable. + * Use {@link DaprClient#publishEvents(String, String, String, Map, Object[])} instead. */ + @Deprecated Mono> publishEvents(String pubsubName, String topicName, String contentType, Map requestMetadata, T... events); diff --git a/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java index 4c4955cf1f..1245c20e17 100644 --- a/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java @@ -146,7 +146,7 @@ public void tearDown() throws Exception { public void publishEventsExceptionThrownTest() { doAnswer((Answer) invocation -> { throw newStatusRuntimeException("INVALID_ARGUMENT", "bad bad argument"); - }).when(daprStub).bulkPublishEventAlpha1(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + }).when(daprStub).bulkPublishEvent(any(DaprPubsubProtos.BulkPublishRequest.class), any()); assertThrowsDaprException( StatusRuntimeException.class, @@ -163,7 +163,7 @@ public void publishEventsCallbackExceptionThrownTest() { (StreamObserver) invocation.getArguments()[1]; observer.onError(newStatusRuntimeException("INVALID_ARGUMENT", "bad bad argument")); return null; - }).when(daprStub).bulkPublishEventAlpha1(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + }).when(daprStub).bulkPublishEvent(any(DaprPubsubProtos.BulkPublishRequest.class), any()); assertThrowsDaprException( ExecutionException.class, @@ -182,7 +182,7 @@ public void publishEventsContentTypeMismatchException() throws IOException { observer.onNext(DaprPubsubProtos.BulkPublishResponse.getDefaultInstance()); observer.onCompleted(); return null; - }).when(daprStub).bulkPublishEventAlpha1(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + }).when(daprStub).bulkPublishEvent(any(DaprPubsubProtos.BulkPublishRequest.class), any()); BulkPublishEntry entry = new BulkPublishEntry<>("1", "testEntry" @@ -227,7 +227,7 @@ public void publishEventsTest() { observer.onNext(builder.build()); observer.onCompleted(); return null; - }).when(daprStub).bulkPublishEventAlpha1(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + }).when(daprStub).bulkPublishEvent(any(DaprPubsubProtos.BulkPublishRequest.class), any()); BulkPublishEntry entry = new BulkPublishEntry<>("1", "test", "text/plain", null); @@ -248,7 +248,7 @@ public void publishEventsWithoutMetaTest() { observer.onNext(builder.build()); observer.onCompleted(); return null; - }).when(daprStub).bulkPublishEventAlpha1(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + }).when(daprStub).bulkPublishEvent(any(DaprPubsubProtos.BulkPublishRequest.class), any()); Mono> result = previewClient.publishEvents(PUBSUB_NAME, TOPIC_NAME, "text/plain", Collections.singletonList("test")); @@ -266,7 +266,7 @@ public void publishEventsWithRequestMetaTest() { observer.onNext(builder.build()); observer.onCompleted(); return null; - }).when(daprStub).bulkPublishEventAlpha1(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + }).when(daprStub).bulkPublishEvent(any(DaprPubsubProtos.BulkPublishRequest.class), any()); Mono> result = previewClient.publishEvents(PUBSUB_NAME, TOPIC_NAME, "text/plain", new HashMap(){{ @@ -285,7 +285,7 @@ public void publishEventsObjectTest() { observer.onNext(DaprPubsubProtos.BulkPublishResponse.getDefaultInstance()); observer.onCompleted(); return null; - }).when(daprStub).bulkPublishEventAlpha1(ArgumentMatchers.argThat(bulkPublishRequest -> { + }).when(daprStub).bulkPublishEvent(ArgumentMatchers.argThat(bulkPublishRequest -> { DaprPubsubProtos.BulkPublishRequestEntry entry = bulkPublishRequest.getEntries(0); if (!"application/json".equals(bulkPublishRequest.getEntries(0).getContentType())) { return false; @@ -317,7 +317,7 @@ public void publishEventsContentTypeOverrideTest() { observer.onNext(DaprPubsubProtos.BulkPublishResponse.getDefaultInstance()); observer.onCompleted(); return null; - }).when(daprStub).bulkPublishEventAlpha1(ArgumentMatchers.argThat(bulkPublishRequest -> { + }).when(daprStub).bulkPublishEvent(ArgumentMatchers.argThat(bulkPublishRequest -> { DaprPubsubProtos.BulkPublishRequestEntry entry = bulkPublishRequest.getEntries(0); if (!"application/json".equals(entry.getContentType())) { return false; @@ -338,6 +338,29 @@ public void publishEventsContentTypeOverrideTest() { Assertions.assertEquals( 0, result.getFailedEntries().size(), "expected no entries to be failed"); } + @Test + public void publishEventsFallbackToAlphaOnUnimplemented() { + doAnswer((Answer) invocation -> { + throw newStatusRuntimeException("UNIMPLEMENTED", "bulkPublishEvent not supported"); + }).when(daprStub).bulkPublishEvent(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + + doAnswer((Answer) invocation -> { + StreamObserver observer = + (StreamObserver) invocation.getArguments()[1]; + observer.onNext(DaprPubsubProtos.BulkPublishResponse.getDefaultInstance()); + observer.onCompleted(); + return null; + }).when(daprStub).bulkPublishEventAlpha1(any(DaprPubsubProtos.BulkPublishRequest.class), any()); + + BulkPublishEntry entry = new BulkPublishEntry<>("1", "test", "text/plain", null); + BulkPublishRequest req = new BulkPublishRequest<>(PUBSUB_NAME, TOPIC_NAME, + Collections.singletonList(entry)); + BulkPublishResponse result = previewClient.publishEvents(req).block(); + + Assertions.assertNotNull(result); + assertEquals(0, result.getFailedEntries().size(), "expected no entries to be failed"); + } + @Test public void queryStateExceptionsTest() { assertThrows(IllegalArgumentException.class, () -> { diff --git a/spring-boot-examples/kubernetes/README.md b/spring-boot-examples/kubernetes/README.md index 3cd9440b5e..09deac6e0a 100644 --- a/spring-boot-examples/kubernetes/README.md +++ b/spring-boot-examples/kubernetes/README.md @@ -30,7 +30,7 @@ Once you have the cluster up and running you can install Dapr: helm repo add dapr https://dapr.github.io/helm-charts/ helm repo update helm upgrade --install dapr dapr/dapr \ ---version=1.17.0-rc.6 \ +--version=1.17.0-rc.8 \ --namespace dapr-system \ --create-namespace \ --wait diff --git a/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java b/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java index 4e6dc26988..c31c6d8a93 100644 --- a/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java +++ b/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java @@ -14,7 +14,7 @@ package io.dapr.testcontainers; public interface DaprContainerConstants { - String DAPR_VERSION = "1.17.0-rc.6"; + String DAPR_VERSION = "1.17.0-rc.8"; String DAPR_WORKFLOWS_DASHBOARD_VERSION = "0.0.1"; String DAPR_RUNTIME_IMAGE_TAG = "daprio/daprd:" + DAPR_VERSION; String DAPR_PLACEMENT_IMAGE_TAG = "daprio/placement:" + DAPR_VERSION;