Merge pull request #269 from XavierGeerinck/issue_267

Add req.body to the subscribe response as well

Author: XavierGeerinck

src/implementation/Server/HTTPServer/pubsub.ts

@@ -32,9 +32,19 @@ export default class HTTPServerPubSub implements IServerPubSub {
     await this.server.getServerImpl().registerPubSubSubscriptionRoute(pubsubName, topic, route);
 
     this.server.getServer().post(`/${route}`, async (req, res) => {
-      // Process our callback
       // @ts-ignore
-      await cb(req?.body?.data);
+      // Parse the data of the body, we prioritize fetching the data key in body if possible
+      // i.e. Redis returns { data: {} } and other services return {}
+      // @todo: This will be deprecated in an upcoming major version and only req.body will be returned
+      const data = req?.body?.data || req?.body;
+
+      // Process our callback
+      try {
+        await cb(data);
+      } catch (e) {
+        console.error(e);
+        return res.send({ success: false });
+      }
 
       // Let Dapr know that the message was processed correctly
       // console.log(`[Dapr API][PubSub][route-${topic}] Ack'ing the message`);

test/e2e/main.grpc.test.ts

@@ -26,6 +26,7 @@ describe('grpc/main', () => {
   let client: DaprClient;
   const mockBindingReceive = jest.fn(async (_data: object) => console.log('mockBindingReceive'));
   const mockPubSubSubscribe = jest.fn(async (_data: object) => console.log('mockPubSubSubscribe'));
+  const mockPubSubSubscribeError = jest.fn(async (_data: object) => { throw new Error("mockPubSubSubscribeError") });
 
   // We need to start listening on some endpoints already
   // this because Dapr is not dynamic and registers endpoints on boot
@@ -38,6 +39,7 @@ describe('grpc/main', () => {
     // Test with:
     // dapr publish --publish-app-id test-suite --pubsub pubsub-redis --topic test-topic --data '{ "hello": "world" }'
     await server.pubsub.subscribe('pubsub-redis', 'test-topic', mockPubSubSubscribe);
+    await server.pubsub.subscribe('pubsub-redis', 'test-topic-error', mockPubSubSubscribeError);
 
     // Start server
     await server.start();
@@ -109,6 +111,19 @@ describe('grpc/main', () => {
       const res = await client.pubsub.publish('pubsub-redis', 'test-topic', { hello: 'world' });
       expect(res).toEqual(true);
     });
+
+    it('should not crash if the callback throws an error', async () => {
+      await client.pubsub.publish('pubsub-redis', 'test-topic-error', { hello: 'world' });
+
+      // Delay a bit for event to arrive
+      await new Promise((resolve, _reject) => setTimeout(resolve, 250));
+
+      expect(mockPubSubSubscribeError.mock.calls.length).toBe(1);
+
+      // Also test for receiving data
+      // @ts-ignore
+      expect(mockPubSubSubscribeError.mock.calls[0][0]['hello']).toEqual('world');
+    });
   });
 
   describe('invoker', () => {

test/e2e/main.http.test.ts

@@ -24,6 +24,7 @@ describe('http/main', () => {
   let client: DaprClient;
   const mockBindingReceive = jest.fn(async (_data: object) => console.log('mockBindingReceive'));
   const mockPubSubSubscribe = jest.fn(async (_data: object) => console.log('mockPubSubSubscribe'));
+  const mockPubSubSubscribeError = jest.fn(async (_data: object) => { throw new Error("mockPubSubSubscribeError") });
 
   // We need to start listening on some endpoints already
   // this because Dapr is not dynamic and registers endpoints on boot
@@ -40,6 +41,7 @@ describe('http/main', () => {
     // Test with:
     // dapr publish --publish-app-id test-suite --pubsub pubsub-redis --topic test-topic --data '{ "hello": "world" }'
     await server.pubsub.subscribe('pubsub-redis', 'test-topic', mockPubSubSubscribe);
+    await server.pubsub.subscribe('pubsub-redis', 'test-topic-error', mockPubSubSubscribeError);
 
     // Start server
     await server.start();
@@ -108,6 +110,19 @@ describe('http/main', () => {
       const res = await client.pubsub.publish('pubsub-redis', 'test-topic', { hello: 'world' });
       expect(res).toEqual(true);
     });
+
+    it('should not crash if the callback throws an error', async () => {
+      await client.pubsub.publish('pubsub-redis', 'test-topic-error', { hello: 'world' });
+
+      // Delay a bit for event to arrive
+      await new Promise((resolve, _reject) => setTimeout(resolve, 250));
+
+      expect(mockPubSubSubscribeError.mock.calls.length).toBe(1);
+
+      // Also test for receiving data
+      // @ts-ignore
+      expect(mockPubSubSubscribeError.mock.calls[0][0]['hello']).toEqual('world');
+    });
   });
 
   describe('invoker', () => {