Tests with real MQ

Author: goldbergyoni

example-application/data-access/order-repository.js

@@ -48,7 +48,6 @@ module.exports = class OrderRepository {
   }
 
   async deleteOrder(orderToDelete) {
-    console.log('About to delete', orderToDelete);
     await orderModel.destroy({ where: { id: orderToDelete } });
     return;
   }

example-application/entry-points/api.js

@@ -96,7 +96,7 @@ const defineRoutes = (expressApp) => {
       }
 
       // We should notify others that a new order was added - Let's put a message in a queue
-      new MessageQueueClient().sendMessage('new-order', req.body);
+      await new MessageQueueClient().publish('order.events','order.events.new', req.body);
 
       res.json(DBResponse);
     } catch (error) {

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

@@ -14,17 +14,17 @@ class MessageQueueStarter {
   }
 
   async consumeUserDeletionQueue() {
-    console.log("consume register", this.queueName)
     // Let's now register to new delete messages from the queue
     return await this.messageQueueClient.consume(
       this.queueName,
       async (message) => {
+       
         // Validate to ensure it is not a poisoned message (invalid) that will loop into the queue
         const newMessageAsObject = JSON.parse(message);
 
         // ️️️✅ Best Practice: Validate incoming MQ messages using your validator framework (simplistic implementation below)
         if (!newMessageAsObject.id) {
-          throw new AppError('invalid-message', true);
+          throw new AppError('invalid-message', 'Unknown message schema');
         }
 
         const orderRepository = new OrderRepository();

example-application/libraries/fake-message-queue-provider.js

@@ -20,24 +20,24 @@ class FakeMessageQueueProvider extends EventEmitter {
     this.emit('message-sent', message);
   }
 
+  async publish(exchangeName, routingKey, message) {
+    this.emit('message-published', message);
+    this.pushMessageToQueue('unknown', message);
+  }
+
   async assertQueue() {}
 
   async consume(queueName, messageHandler) {
     // We just save the callback (handler) locally, whenever a message will put into this queue
     // we will fire this handler
-    console.info(`Received request to listen to the queue ${queueName}`);
-    console.log("6");
     this.messageHandler = messageHandler;
   }
 
   // This is the only method that does not exist in the MQ client library
   // It allows us to fake like there is a new message in the queue and start a flow
   async pushMessageToQueue(queue, newMessage) {
     if (this.messageHandler) {
-      const wrappedMessage = {
-        content: Buffer.from(JSON.stringify(newMessage)),
-      };
-      this.messageHandler(wrappedMessage);
+      this.messageHandler({content: newMessage})
     } else {
       // Just warning and no exception because the test might want to simulate that
       console.error(

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

@@ -1,5 +1,6 @@
 const amqplib = require('amqplib');
 const { EventEmitter } = require('events');
+const { throwIfEmpty } = require('rxjs/operators');
 const { AppError, errorHandler } = require('../error-handling');
 const { FakeMessageQueueProvider } = require('./fake-message-queue-provider');
 
@@ -9,12 +10,17 @@ class MessageQueueClient extends EventEmitter {
   constructor(customMessageQueueProvider) {
     super();
     this.isReady = false;
+    this.requeue = true; // Tells whether to return failed messages to the queue
 
     // To facilitate testing, the client allows working with a fake MQ provider
     // It can get one in the constructor here or even change by environment variables
     if (customMessageQueueProvider) {
       this.messageQueueProvider = customMessageQueueProvider;
-    } else {
+    } 
+    else if(process.env.USE_FAKE_MQ === 'true'){
+      this.messageQueueProvider = new FakeMessageQueueProvider();
+    }
+    else {
       this.messageQueueProvider = amqplib;
     }
 
@@ -63,14 +69,14 @@ class MessageQueueClient extends EventEmitter {
     if (!this.channel) {
       await this.connect();
     }
-    console.log('publish', exchangeName, routingKey);
 
     const sendResponse = await this.channel.publish(
       exchangeName,
       routingKey,
       Buffer.from(JSON.stringify(message)),
       { messageId }
     );
+    this.emit('publish', { exchangeName, routingKey, message });
 
     return sendResponse;
   }
@@ -80,7 +86,6 @@ class MessageQueueClient extends EventEmitter {
       await this.connect();
     }
     const queueDeletionResult = await this.channel.deleteQueue(queueName);
-    console.log(queueDeletionResult);
 
     return;
   }
@@ -118,20 +123,17 @@ class MessageQueueClient extends EventEmitter {
       await this.connect();
     }
     this.channel.assertQueue(queueName);
-    console.log('consume start', queueName);
 
     await this.channel.consume(queueName, async (theNewMessage) => {
       //Not awaiting because some MQ client implementation get back to fetch messages again only after handling a message
       onMessageCallback(theNewMessage.content.toString())
         .then(() => {
-          console.log('ack');
           this.emit('ack', theNewMessage);
           this.channel.ack(theNewMessage);
         })
         .catch((error) => {
-          this.channel.nack(theNewMessage, false, true);
+          this.channel.nack(theNewMessage, false, this.requeue);
           this.emit('nack', theNewMessage);
-          console.log('nack', error.message);
           error.isTrusted = true; //Since it's related to a single message, there is no reason to let the process crash
           //errorHandler.handleError(error);
         });
@@ -140,38 +142,51 @@ class MessageQueueClient extends EventEmitter {
     return;
   }
 
+  setRequeue(newValue) {
+    this.requeue = newValue;
+  }
+
+  // This function stores all the MQ events in a local data structure so later
+  // one query this
   countEvents() {
-    const eventsToListen = ['nack', 'ack'];
-    if (this.eventsCounter === undefined) {
-      this.eventsCounter = {};
-      eventsToListen.forEach((eventToListenTo) => {
-        this.eventsCounter[eventToListenTo] = 0;
-        this.on(eventToListenTo, (eventData) => {
-          this.eventsCounter[eventToListenTo]++;
-          console.log('events counting', this.eventsCounter);
-          this.emit('event-counted', {
-            name: eventToListenTo,
-            lastEventData: eventData,
-            count: this.eventsCounter[eventToListenTo],
-          });
+    const eventsToListen = ['nack', 'ack', 'publish'];
+    if (this.eventsRecorder !== undefined) {
+      return; // Already initialized and set up
+    }
+    this.eventsRecorder = {};
+    eventsToListen.forEach((eventToListenTo) => {
+      this.eventsRecorder[eventToListenTo] = {
+        count: 0,
+        lastEventData: null,
+        name: eventToListenTo,
+      };
+      this.on(eventToListenTo, (eventData) => {
+        this.eventsRecorder[eventToListenTo].count++;
+        this.eventsRecorder[eventToListenTo].lastEventData = eventData;
+        this.emit('message-queue-event', {
+          name: eventToListenTo,
+          eventsRecorder: this.eventsRecorder,
         });
       });
-    }
+    });
   }
 
-  // Helper methods for testing
+  resolveIfEventExceededThreshold(eventName, threshold, resolve) {
+    if (this.eventsRecorder[eventName].count >= threshold) {
+      resolve({
+        name: eventName,
+        lastEventData: this.eventsRecorder[eventName].lastEventData,
+        count: this.eventsRecorder[eventName].count,
+      });
+    }
+  }
+  // Helper methods for testing - Resolves/fires when some event happens
   async waitFor(eventName, howMuch) {
     return new Promise((resolve, reject) => {
-      this.on('event-counted', (eventInfo) => {
-        if (eventInfo.name !== eventName) {
-          return;
-        }
-        if (eventInfo.count >= howMuch) {
-          resolve({
-            lastEventData: eventInfo.lastEventData,
-            count: eventInfo.count,
-          });
-        }
+      // The first resolve is for cases where the caller has approached AFTER the event has already happen
+      this.resolveIfEventExceededThreshold(eventName, howMuch, resolve);
+      this.on('message-queue-event', (eventInfo) => {
+        this.resolveIfEventExceededThreshold(eventName, howMuch, resolve);
       });
     });
   }

recipes/message-queue/benchmark-queue-creation-time.js

@@ -17,7 +17,6 @@ const {
   const numOfIterations = 200;
 
   for (let index = 0; index < numOfIterations; index++) {
-    console.time('measure');
     const startTime = new Date();
     const userDeletedQueueName = `user-deleted-${getShortUnique()}`;
     await mqClient.assertQueue(userDeletedQueueName);
@@ -31,9 +30,6 @@ const {
     const endTime = new Date();
     const singleTime = endTime.getTime() - startTime.getTime();
     overallTime += singleTime;
-    console.log('Single time', singleTime);
-    console.timeEnd('measure');
   }
 
-  console.log('Overall time', overallTime / numOfIterations);
 })();

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

@@ -1,19 +1,17 @@
 const axios = require('axios');
 const sinon = require('sinon');
 const nock = require('nock');
-const { once } = require('events');
+const testHelpers = require('./test-helpers');
+
 const {
   getNextMQConfirmation,
   startFakeMessageQueue,
 } = require('./test-helpers');
-const {
-  FakeMessageQueueProvider,
-} = require('../../example-application/libraries/fake-message-queue-provider');
-
 const {
   initializeWebServer,
   stopWebServer,
 } = require('../../example-application/entry-points/api');
+const MessageQueueClient = require('../../example-application/libraries/message-queue-client');
 
 let axiosAPIClient;
 
@@ -30,6 +28,8 @@ beforeAll(async (done) => {
   };
   axiosAPIClient = axios.create(axiosConfig);
 
+  process.env.USE_FAKE_MQ = 'true';
+
   done();
 });
 
@@ -51,46 +51,55 @@ afterAll(async (done) => {
   await stopWebServer();
   //await messageQueueClient.close();
   nock.enableNetConnect();
+  process.env.USE_FAKE_MQ = undefined;
   done();
 });
 
 // ️️️✅ Best Practice: Test a flow that starts via a queue message and ends with removing/confirming the message
 test('Whenever a user deletion message arrive, then his orders are deleted', async () => {
   // Arrange
-    const orderToAdd = {
-      userId: 1,
-      productId: 2,
-      mode: 'approved',
-    };
-    const addedOrderId = (await axiosAPIClient.post('/order', orderToAdd)).data
-      .id;
-  const fakeMessageQueue = await startFakeMessageQueue();
-  const getNextMQEvent = getNextMQConfirmation(fakeMessageQueue); //Store the MQ actions in a promise
+  const orderToAdd = {
+    userId: 1,
+    productId: 2,
+    mode: 'approved',
+  };
+  const addedOrderId = (await axiosAPIClient.post('/order', orderToAdd)).data
+    .id;
+  const messageQueueClient = await testHelpers.startMQSubscriber(
+    'fake',
+    'user.deleted'
+  );
 
   // Act
-  fakeMessageQueue.pushMessageToQueue('deleted-user', { id: addedOrderId });
+  await messageQueueClient.publish('user.events', 'user.deleted', {
+    id: addedOrderId,
+  });
 
   // Assert
-  const eventFromMessageQueue = await getNextMQEvent;
-  expect(eventFromMessageQueue).toEqual([{ event: 'message-acknowledged' }]);
+  await messageQueueClient.waitFor('ack', 1);
   const aQueryForDeletedOrder = await axiosAPIClient.get(
     `/order/${addedOrderId}`
   );
   expect(aQueryForDeletedOrder.status).toBe(404);
 });
 
-test.only('When a poisoned message arrives, then it is being rejected back', async () => {
+test('When a poisoned message arrives, then it is being rejected back', async () => {
   // Arrange
   const messageWithInvalidSchema = { nonExistingProperty: 'invalid' };
-  const fakeMessageQueue = await startFakeMessageQueue();
-  const getNextMQEvent = getNextMQConfirmation(fakeMessageQueue);
+  const messageQueueClient = await testHelpers.startMQSubscriber(
+    'fake',
+    'user.deleted'
+  );
 
   // Act
-  fakeMessageQueue.pushMessageToQueue('deleted-user', messageWithInvalidSchema);
+  await messageQueueClient.publish(
+    'user.events',
+    'user.deleted',
+    messageWithInvalidSchema
+  );
 
   // Assert
-  const eventFromMessageQueue = await getNextMQEvent;
-  expect(eventFromMessageQueue).toEqual([{ event: 'message-rejected' }]);
+  await messageQueueClient.waitFor('nack', 1);
 });
 
 // ️️️✅ Best Practice: Verify that messages are put in queue whenever the requirements state so
@@ -101,17 +110,14 @@ test('When a valid order is added, then a message is emitted to the new-order qu
     productId: 2,
     mode: 'approved',
   };
-  const spyOnSendMessage = sinon.stub(
-    FakeMessageQueueProvider.prototype,
-    'sendToQueue'
-  );
+  const spyOnSendMessage = sinon.spy(MessageQueueClient.prototype, 'publish');
 
   //Act
   await axiosAPIClient.post('/order', orderToAdd);
 
   // Assert
-  expect(spyOnSendMessage.called).toBe(true);
-  expect(spyOnSendMessage.lastCall.args).toMatchObject({}); //TODO - Be more explicit here
+  expect(spyOnSendMessage.lastCall.args[0]).toBe('order.events');
+  expect(spyOnSendMessage.lastCall.args[1]).toBe('order.events.new');
 });
 
 test.todo('When an error occurs, then the message is not acknowledged');
@@ -129,4 +135,4 @@ test.todo(
 );
 test.todo(
   'When multiple user deletion message arrives and one fails, then only the failed message is not acknowledged'
-);
+);