original version of bulk subscriber over gPRC

Signed-off-by: MregXN <mregxn@gmail.com>

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

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2021 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<io.dapr.v1.DaprAppCallbackProtos.TopicEventBulkResponse> 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);
+    }
+  }
+
+}

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

@@ -50,6 +50,7 @@ public static void main(String[] args) throws Exception {
     //start a grpc server
     Server server = ServerBuilder.forPort(port)  
         .addService(new SubscriberGrpcService())  
+        .addService(new BulkSubscriberGrpcService())  
         .build();  
     server.start();  
     server.awaitTermination();  

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

@@ -30,7 +30,9 @@ public class SubscriberGrpcService extends AppCallbackGrpc.AppCallbackImplBase {
   @Override
   public void listTopicSubscriptions(Empty request,
       StreamObserver<DaprAppCallbackProtos.ListTopicSubscriptionsResponse> responseObserver) {
-    registerConsumer("messagebus","testingtopic");
+    registerConsumer("messagebus", "testingtopic", false);
+    registerConsumer("messagebus", "bulkpublishtesting", false);
+    registerConsumer("messagebus", "testingtopicbulk", true);
     try {
       DaprAppCallbackProtos.ListTopicSubscriptionsResponse.Builder builder = DaprAppCallbackProtos
           .ListTopicSubscriptionsResponse.newBuilder();
@@ -65,12 +67,14 @@ public void onTopicEvent(DaprAppCallbackProtos.TopicEventRequest request,
    * 
    * @param topic the topic
    * @param pubsubName the pubsub name
+   * @param isBulkMessage flag to enable/disable bulk subscribe
    */
-  public void registerConsumer(String pubsubName, String topic) {
+  public void registerConsumer(String pubsubName, String topic, boolean isBulkMessage) {
     topicSubscriptionList.add(DaprAppCallbackProtos.TopicSubscription
         .newBuilder()
         .setPubsubName(pubsubName)
         .setTopic(topic)
+        .setBulkSubscribe(DaprAppCallbackProtos.BulkSubscribeConfig.newBuilder().setEnabled(isBulkMessage))
         .build());
   }
 }