Call it consumer, not subscriber

Author: Yoni Goldberg

example-application/business-logic/order-service.js

@@ -66,7 +66,6 @@ async function getUserFromUsersService(userId) {
         },
       }
     );
-    console.log(getUserResponse.data);
     return getUserResponse.data;
   } catch (error) {
     if (error?.code === 'ECONNABORTED') {

example-application/entry-points/message-queue-consumer.js

@@ -0,0 +1,36 @@
+const MessageQueueClient = require('../libraries/message-queue-client');
+const { errorHandler, AppError } = require('../error-handling');
+const orderService = require('../business-logic/order-service');
+
+// This is message queue entry point. Like API routes but for message queues.
+class QueueConsumer {
+  constructor(messageQueueClient) {
+    this.messageQueueClient = messageQueueClient;
+  }
+
+  async start() {
+    await this.consumeUserDeletionQueue();
+  }
+
+  async consumeUserDeletionQueue() {
+    await this.messageQueueClient.consume('user.deleted', async (message) => {
+      // ️️️Validate message
+      const newMessageAsObject = JSON.parse(message);
+      if (!newMessageAsObject.id) {
+        throw new AppError('invalid-message', 'Unknown message schema');
+      }
+
+      await orderService.deleteOrder(newMessageAsObject.id);
+    });
+  }
+}
+
+process.on('uncaughtException', (error) => {
+  errorHandler.handleError(error);
+});
+
+process.on('unhandledRejection', (reason) => {
+  errorHandler.handleError(reason);
+});
+
+module.exports = { QueueConsumer };

example-application/entry-points/message-queue-starter.js

@@ -1,7 +1,7 @@
 const { EventEmitter } = require('events');
 const amqplib = require('amqplib');
-
 const { FakeMessageQueueProvider } = require('./fake-message-queue-provider');
+const { errorHandler } = require('../error-handling');
 
 // This is a simplistic client for a popular message queue product - RabbitMQ
 // It's generic in order to be used by any service in the organization
@@ -24,6 +24,7 @@ class MessageQueueClient extends EventEmitter {
       this.messageQueueProvider = new FakeMessageQueueProvider();
     }
 
+    this.setMaxListeners(50);
     this.countEvents();
   }
 
@@ -113,7 +114,7 @@ class MessageQueueClient extends EventEmitter {
           this.channel.nack(theNewMessage, false, this.requeue);
           this.emit('nack', { queueName, message: theNewMessage });
           error.isTrusted = true; //Since it's related to a single message, there is no reason to let the process crash
-          //errorHandler.handleError(error);
+          errorHandler.handleError(error);
         });
     });
   }

example-application/libraries/message-queue-client.js

@@ -1,9 +1,9 @@
 const { initializeWebServer } = require('./entry-points/api');
-const { MessageQueueStarter } = require('./entry-points/message-queue-starter');
+const { QueueConsumer } = require('./entry-points/message-queue-consumer');
 
 async function start() {
   await initializeWebServer();
-  await new MessageQueueStarter().start();
+  await new QueueConsumer().start();
 }
 
 start()

example-application/start.js

@@ -12,8 +12,8 @@ const {
   FakeMessageQueueProvider,
 } = require('../../example-application/libraries/fake-message-queue-provider');
 const {
-  QueueSubscriber,
-} = require('../../example-application/entry-points/message-queue-starter');
+  QueueConsumer,
+} = require('../../example-application/entry-points/message-queue-consumer');
 
 let axiosAPIClient;
 
@@ -64,10 +64,7 @@ test('Whenever a user deletion message arrive, then his orders are deleted', asy
   };
   const addedOrderId = (await axiosAPIClient.post('/order', orderToAdd)).data
     .id;
-  const messageQueueClient = await testHelpers.startMQSubscriber(
-    'fake',
-    'user.deleted'
-  );
+  const messageQueueClient = await testHelpers.startMQConsumer('fake');
 
   // Act
   await messageQueueClient.publish('user.events', 'user.deleted', {
@@ -85,10 +82,7 @@ test('Whenever a user deletion message arrive, then his orders are deleted', asy
 test('When a poisoned message arrives, then it is being rejected back', async () => {
   // Arrange
   const messageWithInvalidSchema = { nonExistingProperty: 'invalid❌' };
-  const messageQueueClient = await testHelpers.startMQSubscriber(
-    'fake',
-    'user.deleted'
-  );
+  const messageQueueClient = await testHelpers.startMQConsumer('fake');
 
   // Act
   await messageQueueClient.publish(
@@ -109,7 +103,7 @@ test('When user deleted message arrives, then all corresponding orders are delet
   const messageQueueClient = new MessageQueueClient(
     new FakeMessageQueueProvider()
   );
-  await new QueueSubscriber(messageQueueClient, 'user.deleted').start();
+  await new QueueConsumer(messageQueueClient, 'user.deleted').start();
 
   // Act
   await messageQueueClient.publish('user.events', 'user.deleted', {
@@ -158,4 +152,3 @@ test.todo(
 test.todo(
   'When multiple user deletion message arrives and one fails, then only the failed message is not acknowledged'
 );
-  

recipes/message-queue/fake-message-queue.test.js

@@ -2,15 +2,13 @@ const {
   FakeMessageQueueProvider,
 } = require('../../example-application/libraries/fake-message-queue-provider');
 const {
-  QueueSubscriber: MessageQueueStarter,
-} = require('../../example-application/entry-points/message-queue-starter');
+  QueueConsumer,
+} = require('../../example-application/entry-points/message-queue-consumer');
 const amqplib = require('amqplib');
 const MessageQueueClient = require('../../example-application/libraries/message-queue-client');
 
-module.exports.startMQSubscriber = async (
+module.exports.startMQConsumer = async (
   fakeOrReal,
-  queueName,
-  deadLetterQueueName = undefined,
   messageQueueClient = undefined
 ) => {
   if (!messageQueueClient) {
@@ -19,11 +17,7 @@ module.exports.startMQSubscriber = async (
     messageQueueClient = new MessageQueueClient(messageQueueProvider);
   }
 
-  await new MessageQueueStarter(
-    messageQueueClient,
-    queueName,
-    deadLetterQueueName
-  ).start();
+  await new QueueConsumer(messageQueueClient).start();
 
   return messageQueueClient;
 };