feat: Add Bulk Subscribe support (#437)

* Add Bulk Subscribe changes

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Update Bulk Subscribe Support

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Add e2e test for bulkSub

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Fix build issue

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Fix time

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Use latest dapr

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Add docs

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Fix docs

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Fix pretty

Signed-off-by: Deepanshu Agarwal <[email protected]>

* incorporate review comments

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Pretty

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Modify language

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Introduce BulkSubscribeOptions

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Correct docs

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Pretty

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Fix destructure

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Use retry logic

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Fix assignment

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Test names

Signed-off-by: Deepanshu Agarwal <[email protected]>

* remove todo

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Incorporate review comments

Signed-off-by: Deepanshu Agarwal <[email protected]>

* pretty

Signed-off-by: Deepanshu Agarwal <[email protected]>

* nitties

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Comment improve

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Comment improve

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Correct proto

Signed-off-by: Deepanshu Agarwal <[email protected]>

* Comment Space

Signed-off-by: Deepanshu Agarwal <[email protected]>

---------

Signed-off-by: Deepanshu Agarwal <[email protected]>
Signed-off-by: Deepanshu Agarwal <[email protected]>
Co-authored-by: Deepanshu Agarwal <[email protected]>

Author: DeepanshuA

daprdocs/content/en/js-sdk-docs/js-server/_index.md

@@ -351,6 +351,51 @@ async function start() {
 }
 ```
 
+#### Bulk Subscribe to messages
+
+Bulk Subscription is supported and is available through following API:
+
+- Bulk subscription through the `subscribeBulk` method: `maxMessagesCount` and `maxAwaitDurationMs` are optional; and if not provided, default values for related components will be used.
+
+While listening for messages, the application receives messages from Dapr in bulk. However, like regular subscribe, the callback function receives a single message at a time, and the user can choose to return a `DaprPubSubStatusEnum` value to acknowledge successfully, retry, or drop the message. The default behavior is to return a success response.
+
+Please refer [this document](https://v1-10.docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-bulk/) for more details.
+
+```typescript
+import { DaprServer } from "@dapr/dapr";
+
+const pubSubName = "orderPubSub";
+const topic = "topicbulk";
+
+const DAPR_HOST = process.env.DAPR_HOST || "127.0.0.1";
+const DAPR_HTTP_PORT = process.env.DAPR_HTTP_PORT || "3502";
+const SERVER_HOST = process.env.SERVER_HOST || "127.0.0.1";
+const SERVER_PORT = process.env.APP_PORT || 5001;
+
+async function start() {
+  const server = new DaprServer(SERVER_HOST, SERVER_PORT, DAPR_HOST, DAPR_HTTP_PORT);
+
+  // Publish multiple messages to a topic with default config.
+  await client.pubsub.subscribeBulk(pubSubName, topic, (data) =>
+    console.log("Subscriber received: " + JSON.stringify(data)),
+  );
+
+  // Publish multiple messages to a topic with specific maxMessagesCount and maxAwaitDurationMs.
+  await client.pubsub.subscribeBulk(
+    pubSubName,
+    topic,
+    (data) => {
+      console.log("Subscriber received: " + JSON.stringify(data));
+      return DaprPubSubStatusEnum.SUCCESS; // If App doesn't return anything, the default is SUCCESS. App can also return RETRY or DROP based on the incoming message.
+    },
+    {
+      maxMessagesCount: 100,
+      maxAwaitDurationMs: 40,
+    },
+  );
+}
+```
+
 #### Dead Letter Topics
 
 Dapr supports [dead letter topic](https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-deadletter/). This means that when a message fails to be processed, it gets sent to a dead letter queue. E.g., when a message fails to be handled on `/my-queue` it will be sent to `/my-queue-failed`.

examples/pubsub/src/index.ts

@@ -29,6 +29,27 @@ async function start() {
     // The library parses JSON when possible.
     console.log(`[Dapr-JS][Example] Received on subscription: ${JSON.stringify(data)}`);
   });
+
+  // Publish multiple messages to a topic with default config.
+  await client.pubsub.subscribeBulk("my-pubsub-component", "my-topic", async (data: Record<string, any>) => {
+    // The library parses JSON when possible.
+    console.log(`[Dapr-JS][Example] Received on subscription: ${JSON.stringify(data)}`);
+  });
+
+  // Publish multiple messages to a topic with specific maxMessagesCount and maxAwaitDurationMs.
+  await client.pubsub.subscribeBulk(
+    "my-pubsub-component",
+    "my-topic",
+    async (data: Record<string, any>) => {
+      // The library parses JSON when possible.
+      console.log(`[Dapr-JS][Example] Received on subscription: ${JSON.stringify(data)}`);
+    },
+    {
+      maxMessagesCount: 100,
+      maxAwaitDurationMs: 40,
+    },
+  );
+
   await server.start();
 
   // Wait for 1 second to allow the server to start.

scripts/fetch-proto.sh

@@ -3,7 +3,7 @@ OS=$(echo `uname`|tr '[:upper:]' '[:lower:]')
 ARCH=$(uname -m)
 ORG_NAME="dapr"
 REPO_NAME="dapr"
-BRANCH_NAME="v1.10.0"
+BRANCH_NAME="v1.10.2"
 
 # Path to store output
 PATH_ROOT=$(pwd)

src/implementation/Server/GRPCServer/GRPCServer.ts

@@ -13,7 +13,7 @@ limitations under the License.
 
 import * as grpc from "@grpc/grpc-js";
 import GRPCServerImpl from "./GRPCServerImpl";
-import { AppCallbackService } from "../../../proto/dapr/proto/runtime/v1/appcallback_grpc_pb";
+import { AppCallbackService, AppCallbackAlphaService } from "../../../proto/dapr/proto/runtime/v1/appcallback_grpc_pb";
 import IServer from "../../../interfaces/Server/IServer";
 import { DaprClient } from "../../..";
 import { Logger } from "../../../logger/Logger";
@@ -52,6 +52,10 @@ export default class GRPCServer implements IServer {
     this.logger.info("Adding Service Implementation - AppCallbackService");
     // @ts-ignore
     this.server.addService(AppCallbackService, this.serverImpl);
+    // Add our implementation
+    this.logger.info("Adding Service Implementation - AppCallbackAlphaService");
+    // @ts-ignore
+    this.server.addService(AppCallbackAlphaService, this.serverImpl);
   }
 
   // See: https://cs.github.com/nestjs/nest/blob/f4e9ac6208f3e7ee7ad44c3de713c9086f657977/packages/microservices/external/grpc-options.interface.ts

src/implementation/Server/GRPCServer/GRPCServerImpl.ts

@@ -20,13 +20,18 @@ import { HTTPExtension, InvokeRequest, InvokeResponse } from "../../../proto/dap
 import {
   BindingEventRequest,
   BindingEventResponse,
+  BulkSubscribeConfig,
   ListInputBindingsResponse,
   ListTopicSubscriptionsResponse,
   TopicEventRequest,
   TopicEventResponse,
   TopicRoutes,
   TopicRule,
   TopicSubscription,
+  TopicEventBulkRequest,
+  TopicEventBulkResponse,
+  TopicEventBulkResponseEntry,
+  TopicEventCERequest,
 } from "../../../proto/dapr/proto/runtime/v1/appcallback_pb";
 import * as HttpVerbUtil from "../../../utils/HttpVerb.util";
 import { TypeDaprBindingCallback } from "../../../types/DaprBindingCallback.type";
@@ -249,6 +254,7 @@ export default class GRPCServerImpl implements IAppCallbackServer {
         metadata: metadata,
         route: this.generateDaprSubscriptionRoute(pubsubName, topic),
         deadLetterTopic: options.deadLetterTopic,
+        bulkSubscribe: options.bulkSubscribe,
       };
     } else if (typeof options.route === "string") {
       return {
@@ -257,6 +263,7 @@ export default class GRPCServerImpl implements IAppCallbackServer {
         metadata: metadata,
         route: this.generateDaprSubscriptionRoute(pubsubName, topic, options.route),
         deadLetterTopic: options.deadLetterTopic,
+        bulkSubscribe: options.bulkSubscribe,
       };
     } else {
       return {
@@ -271,6 +278,7 @@ export default class GRPCServerImpl implements IAppCallbackServer {
           })),
         },
         deadLetterTopic: options.deadLetterTopic,
+        bulkSubscribe: options.bulkSubscribe,
       };
     }
   }
@@ -454,6 +462,82 @@ export default class GRPCServerImpl implements IAppCallbackServer {
     return callback(null, res);
   }
 
+  async onBulkTopicEventAlpha1(
+    call: grpc.ServerUnaryCall<TopicEventBulkRequest, TopicEventBulkResponse>,
+    callback: grpc.sendUnaryData<TopicEventBulkResponse>,
+  ): Promise<void> {
+    const req = call.request;
+    const pubsubName = req.getPubsubName();
+    const topic = req.getTopic();
+
+    // Route is unique to pubsub and topic and has format pubsub--topic--route so we strip it since else we can't find the route
+    const route = this.generatePubSubSubscriptionTopicRouteName(req.getPath().replace(`${pubsubName}--${topic}--`, ""));
+
+    if (
+      !this.pubSubSubscriptions[pubsubName] ||
+      !this.pubSubSubscriptions[pubsubName][topic] ||
+      !this.pubSubSubscriptions[pubsubName][topic].routes[route]
+    ) {
+      this.logger.warn(
+        `The topic '${topic}' is not being subscribed to on PubSub '${pubsubName}' for route '${route}'.`,
+      );
+      return;
+    }
+
+    const resArr: TopicEventBulkResponseEntry[] = [];
+    const entries = req.getEntriesList();
+
+    for (const ind in entries) {
+      const event = entries[ind];
+      let data: any;
+      if (event.hasBytes()) {
+        data = deserializeGrpc(event.getContentType(), event.getBytes());
+      } else if (event.hasCloudEvent()) {
+        const cloudEvent = event.getCloudEvent();
+        if (cloudEvent instanceof TopicEventCERequest) {
+          data = deserializeGrpc(cloudEvent.getDataContentType(), cloudEvent.getData());
+        }
+      }
+
+      const res = new TopicEventBulkResponseEntry();
+
+      // Get the headers
+      const headers: { [key: string]: string } = {};
+
+      for (const [key, value] of Object.entries(call.metadata.toHttp2Headers())) {
+        if (value) {
+          headers[key] = value.toString();
+        }
+      }
+
+      // Process the callbacks
+      // we handle priority of status on `RETRY` > `DROP` > `SUCCESS` and default to `SUCCESS`
+      const routeObj = this.pubSubSubscriptions[pubsubName][topic].routes[route];
+      const status = await this.processPubSubCallbacks(routeObj, data, headers);
+
+      switch (status) {
+        case DaprPubSubStatusEnum.RETRY:
+          res.setStatus(TopicEventResponse.TopicEventResponseStatus.RETRY);
+          break;
+        case DaprPubSubStatusEnum.DROP:
+          res.setStatus(TopicEventResponse.TopicEventResponseStatus.DROP);
+          break;
+        case DaprPubSubStatusEnum.SUCCESS:
+        default:
+          res.setStatus(TopicEventResponse.TopicEventResponseStatus.SUCCESS);
+          break;
+      }
+
+      res.setEntryId(event.getEntryId());
+      resArr.push(res);
+    }
+
+    const totalRes = new TopicEventBulkResponse();
+    totalRes.setStatusesList(resArr);
+
+    return callback(null, totalRes);
+  }
+
   async processPubSubCallbacks(
     routeObj: PubSubSubscriptionTopicRouteType,
     data: any,
@@ -512,6 +596,21 @@ export default class GRPCServerImpl implements IAppCallbackServer {
           topicSubscription.setDeadLetterTopic(daprConfig.deadLetterTopic);
         }
 
+        if (daprConfig?.bulkSubscribe) {
+          const bulkSubscribe = new BulkSubscribeConfig();
+          bulkSubscribe.setEnabled(daprConfig.bulkSubscribe.enabled);
+
+          if (daprConfig?.bulkSubscribe?.maxMessagesCount) {
+            bulkSubscribe.setMaxMessagesCount(daprConfig.bulkSubscribe.maxMessagesCount);
+          }
+
+          if (daprConfig?.bulkSubscribe?.maxAwaitDurationMs) {
+            bulkSubscribe.setMaxAwaitDurationMs(daprConfig.bulkSubscribe.maxAwaitDurationMs);
+          }
+
+          topicSubscription.setBulkSubscribe(bulkSubscribe);
+        }
+
         if (daprConfig?.metadata) {
           for (const [mKey, mValue] of Object.entries(daprConfig.metadata)) {
             topicSubscription.getMetadataMap().set(mKey, mValue);

src/implementation/Server/GRPCServer/pubsub.ts

@@ -19,6 +19,8 @@ import { PubSubSubscriptionOptionsType } from "../../../types/pubsub/PubSubSubsc
 import { DaprPubSubRouteType } from "../../../types/pubsub/DaprPubSubRouteType.type";
 import { PubSubSubscriptionsType } from "../../../types/pubsub/PubSubSubscriptions.type";
 import { KeyValueType } from "../../../types/KeyValue.type";
+import { BulkSubscribeConfig } from "../../../types/pubsub/BulkSubscribeConfig.type";
+import { BulkSubscribeOptions } from "../../../types/pubsub/BulkSubscribeOptions.type";
 
 // https://docs.dapr.io/reference/api/pubsub_api/
 export default class DaprPubSub implements IServerPubSub {
@@ -97,6 +99,35 @@ export default class DaprPubSub implements IServerPubSub {
     this.server.getServerImpl().registerPubSubSubscriptionEventHandler(pubsubName, topic, route, cb);
   }
 
+  async subscribeBulk(
+    pubsubName: string,
+    topic: string,
+    cb: TypeDaprPubSubCallback,
+    bulkSubscribeOptions: BulkSubscribeOptions,
+  ): Promise<void> {
+    const bulkSubscribe: BulkSubscribeConfig = {
+      enabled: true,
+    };
+
+    const { route, metadata, maxMessagesCount, maxAwaitDurationMs } = bulkSubscribeOptions || {};
+
+    if (maxMessagesCount != undefined) {
+      bulkSubscribe.maxMessagesCount = maxMessagesCount;
+    }
+
+    if (maxAwaitDurationMs != undefined) {
+      bulkSubscribe.maxAwaitDurationMs = maxAwaitDurationMs;
+    }
+
+    this.server
+      .getServerImpl()
+      .registerPubsubSubscription(pubsubName, topic, { route, metadata, bulkSubscribe: bulkSubscribe });
+
+    // Add the callback to the event handlers manually
+    // @todo: we will deprecate this way of working? and require subscribeToRoute?
+    this.subscribeToRoute(pubsubName, topic, route, cb);
+  }
+
   getSubscriptions(): PubSubSubscriptionsType {
     return this.server.getServerImpl().pubSubSubscriptions;
   }