feat: Implement PubSub Deadletter & Single/Multi Routing (#342)

* Initial implementation for Pub/Sub with Single and Multi routing + DeadLetter support - WIP

Signed-off-by: Xavier Geerinck <[email protected]>

* Implement event handlers and routes

Signed-off-by: Xavier Geerinck <[email protected]>

* Re-enable all server test

Signed-off-by: Xavier Geerinck <[email protected]>

* Add gRPC tests (disabled for now)

Signed-off-by: Xavier Geerinck <[email protected]>

* Recode for clarity

Signed-off-by: Xavier Geerinck <[email protected]>

* Fix tests

Signed-off-by: Xavier Geerinck <[email protected]>

* Add gRPC

Signed-off-by: Xavier Geerinck <[email protected]>

* Add deadletter support

Signed-off-by: Xavier Geerinck <[email protected]>

* Stabilize pubsub more, add more examples and tests and docs

Signed-off-by: Xavier Geerinck <[email protected]>

* Fix linting and fix build

Signed-off-by: Xavier Geerinck <[email protected]>

* Resolve comments

Signed-off-by: Xavier Geerinck <[email protected]>

* Minor nit corrections

Signed-off-by: Xavier Geerinck <[email protected]>

Signed-off-by: Xavier Geerinck <[email protected]>

Author: XavierGeerinck

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

@@ -277,7 +277,17 @@ start().catch((e) => {
 });
 ```
 
-##### Subscribe to messages
+#### Subscribe to messages
+
+Subscribing to messages can be done in several ways to offer flexibility of receiving messages on your topics:
+
+* Direct subscription through the `subscribe` method
+* Direct susbcription with options through the `subscribeWithOptions` method
+* Subscription afterwards through the `susbcribeOnEvent` method
+
+> Dapr requires subscriptions to be set up on startup, but in the JS SDK we allow event handlers to be added afterwards as well, providing you the flexibility of programming.
+
+An example is provided below
 
 ```javascript
 import { DaprServer } from "@dapr/dapr";
@@ -294,14 +304,117 @@ async function start() {
   const topic = "topic-a";
 
   // Configure Subscriber for a Topic
-  await server.pubsub.subscribe(pubSubName, topic, async (data: any) => console.log(`Got Data: ${JSON.stringify(data)}`));
+  // Method 1: Direct subscription through the `subscribe` method
+  await server.pubsub.subscribe(pubSubName, topic, async (data: any) => console.log(`Received Data: ${JSON.stringify(data)}`));
 
+  // Method 2: Direct susbcription with options through the `subscribeWithOptions` method
+  await server.pubsub.subscribeWithOptions(pubSubName, topic, { 
+    callback: async (data: any) => console.log(`Received Data: ${JSON.stringify(data)}`) 
+  });
+
+  // Method 3: Subscription afterwards through the `susbcribeOnEvent` method
+  // Note: we use default, since if no route was passed (empty options) we utilize "default" as the route name
+  await server.pubsub.subscribeWithOptions('pubsub-redis', 'topic-options-1', {});
+  server.pubsub.subscribeToRoute("pubsub-redis", "topic-options-1", "default", async (data) => { console.log(`Received Data: ${JSON.stringify(data)}`) });
+
+  // Start the server
   await server.start();
 }
 ```
 
 > For a full list of state operations visit [How-To: Publish & subscribe]({{< ref howto-publish-subscribe.md >}}).
 
+#### Subscribe to messages rule based
+
+Dapr [supports routing messages](https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages/
+) to different handlers (routes) based on rules. 
+
+> E.g., you are writing an application that needs to handle messages depending on their "type" with Dapr, you can send them to different routes `handlerType1` and `handlerType2` with the default route being `handlerDefault`
+
+```javascript
+import { DaprServer } from "@dapr/dapr";
+
+const daprHost = "127.0.0.1"; // Dapr Sidecar Host
+const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
+const serverHost = "127.0.0.1"; // App Host of this Example Server
+const serverPort = "50051"; // App Port of this Example Server "
+
+async function start() {
+  const server = new DaprServer(serverHost, serverPort, daprHost, daprPort);
+
+  const pubSubName = "my-pubsub-name";
+  const topic = "topic-a";
+
+  // Configure Subscriber for a Topic with rule set
+  // Note: the default route and match patterns are optional
+  await server.pubsub.subscribe('pubsub-redis', 'topic-1', {
+    default: "/default",
+    rules: [
+      {
+        match: `event.type == "my-type-1"`,
+        path: "/type-1"
+      },
+      {
+        match: `event.type == "my-type-2"`,
+        path: "/type-2"
+      }
+    ]
+  });
+
+  // Add handlers for each route
+  server.pubsub.subscribeToRoute("pubsub-redis", "topic-1", "default", async (data) => { console.log(`Handling Default`) });
+  server.pubsub.subscribeToRoute("pubsub-redis", "topic-1", "type-1", async (data) => { console.log(`Handling Type 1`) });
+  server.pubsub.subscribeToRoute("pubsub-redis", "topic-1", "type-2", async (data) => { console.log(`Handling Type 2`) });
+
+  // Start the server
+  await server.start();
+}
+```
+
+#### 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`.
+E.g., when a message fails to be handled on `/my-queue` it will be sent to `/my-queue-failed`.
+
+You can use the following options with `subscribeWithOptions` method:
+* `deadletterTopic`: Specify a deadletter topic name (note: if none is provided we create one named `deadletter`)
+* `deadletterCallback`: The method to trigger as handler for our deadletter
+
+Implementing Deadletter support in the JS SDK can be done by either
+* Passing the `deadletterCallback` as an option 
+* By subscribing to route manually with `subscribeToRoute`
+
+An example is provided below
+
+```javascript
+import { DaprServer } from "@dapr/dapr";
+
+const daprHost = "127.0.0.1"; // Dapr Sidecar Host
+const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
+const serverHost = "127.0.0.1"; // App Host of this Example Server
+const serverPort = "50051"; // App Port of this Example Server "
+
+async function start() {
+  const server = new DaprServer(serverHost, serverPort, daprHost, daprPort);
+
+  const pubSubName = "my-pubsub-name";
+
+  // Method 1 (direct subscribing through subscribeWithOptions)
+  await server.pubsub.subscribeWithOptions('pubsub-redis', 'topic-options-5', { 
+    callback: async (data: any) => { throw new Error("Triggering Deadletter") }, 
+    deadLetterCallback: async (data: any) => { console.log("Handling Deadletter message") } 
+  });
+
+  // Method 2 (subscribe afterwards)
+  await server.pubsub.subscribeWithOptions('pubsub-redis', 'topic-options-1', { deadletterTopic: "my-deadletter-topic" });
+  server.pubsub.subscribeToRoute("pubsub-redis", "topic-options-1", "default", async () => { throw new Error("Triggering Deadletter") });
+  server.pubsub.subscribeToRoute("pubsub-redis", "topic-options-1", "my-deadletter-topic", async () => { console.log("Handling Deadletter message") });
+
+  // Start server
+  await server.start();
+}
+```
+
 ### Bindings API
 
 #### Invoke Output Binding

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

@@ -141,8 +141,8 @@ async function start() {
   const pubSubName = "my-pubsub-name";
   const topic = "topic-a";
 
-  // Configure Subscriber for a Topic
-  await server.pubsub.subscribe(pubSubName, topic, async (data: any) => console.log(`Got Data: ${JSON.stringify(data)}`));
+  // Method 1: Direct subscription through the `subscribe` method
+  await server.pubsub.subscribe(pubSubName, topic, async (data: any) => console.log(`Received Data: ${JSON.stringify(data)}`));
 
   await server.start();
 }