initial bulk publish impl for java

examples/components/pubsub/kafka.yaml

@@ -0,0 +1,12 @@
+apiVersion: dapr.io/v1alpha1
+kind: Component
+metadata:
+  name: kafka-pubsub
+spec:
+  type: pubsub.kafka
+  version: v1
+  metadata:
+    - name: brokers # Required. Kafka broker connection setting
+      value: "localhost:9092"
+    - name: "authType"
+      value: "none"

examples/src/main/java/io/dapr/examples/pubsub/grpc/BulkPublisher.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2022 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.client.DaprClientBuilder;
+import io.dapr.client.DaprPreviewClient;
+import io.dapr.client.domain.BulkPublishRequest;
+import io.dapr.client.domain.BulkPublishRequestEntry;
+import io.dapr.client.domain.BulkPublishResponse;
+import io.dapr.client.domain.BulkPublishResponseEntry;
+import io.dapr.client.domain.CloudEvent;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Message publisher.
+ * 1. Build and install jars:
+ * mvn clean install
+ * 2. cd [repo root]/examples
+ * 3. Run the program:
+ * dapr run --components-path ./components/pubsub --app-id publisher -- \
+ * java -Ddapr.grpc.port="50010" -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.BulkPublisher
+ */
+public class BulkPublisher {
+
+  private static final int NUM_MESSAGES = 10;
+
+  private static final String TOPIC_NAME = "testingtopic";
+
+  //The name of the pubsub
+  private static final String PUBSUB_NAME = "kafka-pubsub";
+
+  /**
+   * main method.
+   * @param args incoming args
+   * @throws Exception any exception
+   */
+  public static void main(String[] args) throws Exception {
+    try (DaprPreviewClient client = (new DaprClientBuilder()).withObjectSerializer(new CustomSerializer()).buildPreviewClient()) {
+      System.out.println("Using preview client...");
+      BulkPublishRequest<Object> request = new BulkPublishRequest<>(PUBSUB_NAME, TOPIC_NAME);
+      List<BulkPublishRequestEntry<Object>> entries = new ArrayList<>();
+      for (int i = 0; i < NUM_MESSAGES; i++) {
+        String message = String.format("This is message #%d", i);
+        BulkPublishRequestEntry<Object> entry = new BulkPublishRequestEntry<>("" + (i + 1),
+            message, "text/plain", new HashMap<>());
+        entries.add(entry);
+      }
+      request.setEntries(entries);
+      BulkPublishResponse res = client.publishEvents(request).block();
+      if (res != null) {
+        for (BulkPublishResponseEntry entry : res.getStatuses()) {
+          System.out.println("EntryID : " + entry.getEntryID() + " Status : " + entry.getStatus());
+        }
+      } else {
+        throw new Exception("null response from dapr");
+      }
+      System.out.println("Done");
+    }
+  }
+}
+

examples/src/main/java/io/dapr/examples/pubsub/grpc/CloudEventBulkPublisher.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2022 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.client.DaprClientBuilder;
+import io.dapr.client.DaprPreviewClient;
+import io.dapr.client.domain.BulkPublishRequest;
+import io.dapr.client.domain.BulkPublishRequestEntry;
+import io.dapr.client.domain.BulkPublishResponse;
+import io.dapr.client.domain.BulkPublishResponseEntry;
+import io.dapr.client.domain.CloudEvent;
+import io.dapr.serializer.DaprObjectSerializer;
+import io.dapr.serializer.DefaultObjectSerializer;
+import io.dapr.utils.TypeRef;
+
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Message publisher.
+ * 1. Build and install jars:
+ * mvn clean install
+ * 2. cd [repo root]/examples
+ * 3. Run the program:
+ * dapr run --components-path ./components/pubsub --app-id publisher -- \
+ * java -Ddapr.grpc.port="50010" \
+ * -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.CloudEventBulkPublisher
+ */
+public class CloudEventBulkPublisher {
+
+  private static final int NUM_MESSAGES = 10;
+
+  private static final String TOPIC_NAME = "testingtopic";
+
+  //The name of the pubsub
+  private static final String PUBSUB_NAME = "kafka-pubsub";
+
+  /**
+   * main method.
+   * @param args incoming args
+   * @throws Exception any exception
+   */
+  public static void main(String[] args) throws Exception {
+    try (DaprPreviewClient client = (new DaprClientBuilder()).withObjectSerializer(new CustomSerializer()).buildPreviewClient()) {
+      System.out.println("Using preview client...");
+      BulkPublishRequest<CloudEvent<Map<String, String>>> request = new BulkPublishRequest<>(PUBSUB_NAME, TOPIC_NAME);
+      List<BulkPublishRequestEntry<CloudEvent<Map<String, String>>>> entries = new ArrayList<>();
+      for (int i = 0; i < 1; i++) {
+        CloudEvent<Map<String, String>> cloudEvent = new CloudEvent<>();
+        cloudEvent.setId(UUID.randomUUID().toString());
+        cloudEvent.setType("example");
+        cloudEvent.setSpecversion("1");
+        cloudEvent.setDatacontenttype("application/json");
+        String val = String.format("This is message #%d", i);
+        cloudEvent.setData(new HashMap<>(){{
+          put("dataKey", val);
+        }});
+        BulkPublishRequestEntry<CloudEvent<Map<String, String>>> entry = new BulkPublishRequestEntry<>("" + (i + 1),
+
+            cloudEvent, CloudEvent.CONTENT_TYPE, new HashMap<>());
+        entries.add(entry);
+      }
+      request.setEntries(entries);
+      BulkPublishResponse res = client.publishEvents(request).block();
+      System.out.println(res);
+      if (res != null) {
+        for (BulkPublishResponseEntry entry : res.getStatuses()) {
+          System.out.println("EntryID : " + entry.getEntryID() + " Status : " + entry.getStatus());
+        }
+      } else {
+        throw new Exception("null response");
+      }
+      System.out.println("Done");
+    }
+  }
+}
+
+class CustomSerializer implements DaprObjectSerializer {
+
+  private final DefaultObjectSerializer serializer = new DefaultObjectSerializer();
+
+  @Override
+  public byte[] serialize(Object o) throws IOException {
+    return serializer.serialize(o);
+  }
+
+  @Override
+  public <T> T deserialize(byte[] data, TypeRef<T> type) throws IOException {
+    return serializer.deserialize(data, type);
+  }
+
+  @Override
+  public String getContentType() {
+    return "application/json";
+  }
+}

examples/src/main/java/io/dapr/examples/pubsub/grpc/README.md

@@ -0,0 +1,10 @@
+```bash
+dapr run --components-path ./components/pubsub --app-id pubsub-test --log-level debug -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.CloudEventBulkPublisher
+```
+java -Ddapr.grpc.port="50010" -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.CloudEventBulkPublisher
+java -Ddapr.grpc.port="50010" -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.BulkPublisher
+java -Ddapr.grpc.port="50010" -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.BinaryBulkPublisher
+
+java -Ddapr.api.protocol="http" -Ddapr.http.port="3500" -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.BulkPublisher
+java -Ddapr.api.protocol="http" -Ddapr.http.port="3500" -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.CloudEventBulkPublisher
+java -Ddapr.api.protocol="http" -Ddapr.http.port="3500" -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.grpc.BinaryBulkPublisher

examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java

@@ -40,7 +40,7 @@ public class Publisher {
   private static final String TOPIC_NAME = "testingtopic";
 
   //The name of the pubsub
-  private static final String PUBSUB_NAME = "messagebus";
+  private static final String PUBSUB_NAME = "kafka-pubsub";
 
   /**
    * This is the entry point of the publisher app example.

examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java

@@ -35,7 +35,7 @@ public class SubscriberController {
    * @param cloudEvent The cloud event received.
    * @return A message containing the time.
    */
-  @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}")
+  @Topic(name = "testingtopic", pubsubName = "kafka-pubsub")
   @PostMapping(path = "/testingtopic")
   public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent<String> cloudEvent) {
     return Mono.fromRunnable(() -> {

pom.xml

@@ -16,7 +16,7 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <grpc.version>1.42.1</grpc.version>
     <protobuf.version>3.17.3</protobuf.version>
-    <dapr.proto.baseurl>https://raw.githubusercontent.com/dapr/dapr/683a436ff1367e4ff2d27da2d79069c04ec2c46d/dapr/proto</dapr.proto.baseurl>
+    <dapr.proto.baseurl>https://raw.githubusercontent.com/dapr/dapr/83a6b114667808665cab282313ee1db12eb49105/dapr/proto</dapr.proto.baseurl>
     <os-maven-plugin.version>1.6.2</os-maven-plugin.version>
     <maven-dependency-plugin.version>3.1.1</maven-dependency-plugin.version>
     <maven-antrun-plugin.version>1.8</maven-antrun-plugin.version>

sdk/src/main/java/io/dapr/client/AbstractDaprClient.java

@@ -14,6 +14,9 @@
 package io.dapr.client;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.dapr.client.domain.BulkPublishRequest;
+import io.dapr.client.domain.BulkPublishRequestEntry;
+import io.dapr.client.domain.BulkPublishResponse;
 import io.dapr.client.domain.ConfigurationItem;
 import io.dapr.client.domain.DeleteStateRequest;
 import io.dapr.client.domain.ExecuteStateTransactionRequest;
@@ -39,6 +42,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -389,6 +393,34 @@ public <T> Mono<QueryStateResponse<T>> queryState(QueryStateRequest request, Cla
     return this.queryState(request, TypeRef.get(clazz));
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public <T> Mono<BulkPublishResponse> publishEvents(String pubsubName, String topicName, List<T> events,
+                                                     String contentType) {
+    return publishEvents(pubsubName, topicName, events, contentType, null);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public <T> Mono<BulkPublishResponse> publishEvents(String pubsubName, String topicName, List<T> events,
+                                                     String contentType, Map<String, String> requestMetadata) {
+    BulkPublishRequest<T> request = new BulkPublishRequest<>(pubsubName, topicName);
+    if (events == null || events.size() == 0) {
+      throw new IllegalArgumentException("list of events cannot be null or empty");
+    }
+    List<BulkPublishRequestEntry<T>> entries = new ArrayList<>();
+    for (int i = 0; i < events.size(); i++) {
+      // entryID field is generated based on order of events in the request
+      entries.add(new BulkPublishRequestEntry<>("" + (i + 1), events.get(i), contentType, null));
+    }
+    request.setMetadata(requestMetadata);
+    return publishEvents(request.setEntries(entries));
+  }
+
   /**
    * {@inheritDoc}
    */