Instead of nacking messages on service call failures, queued them in separate exchange with different routing queue. So that they don't overflow the server resources by continuously trying with failed attempts. It will further allow manually fixing the failed lead creation by looking at the error queue.

Author: Vikas Agarwal

consumer/config/constants.js

@@ -4,5 +4,6 @@ export const EVENT = {
     PROJECT_DRAFT_CREATED: 'project.draft-created',
     PROJECT_UPDATED: 'project.updated',
     PROJECT_DELETED: 'project.deleted',
+    CONNECT_TO_SF_FAILED: 'connect2sf.failed'
   },
 };

consumer/src/services/ConfigurationService.js

@@ -24,7 +24,6 @@ class ConfigurationService {
         },
       },
     }).promise();
-    console.log('node env: ' + process.env.NODE_ENV);
     if (!result.Items.length) {
       throw new Error('Configuration for AppXpressConfig not found');
     }

consumer/src/worker.js

@@ -35,7 +35,7 @@ export function initHandlers(handlers) {
  * @param {String} exchangeName the exchange name
  * @param {String} queue the queue name
  */
-export async function consume(channel, exchangeName, queue) {
+export async function consume(channel, exchangeName, queue, publishChannel) {
   channel.assertExchange(exchangeName, 'topic', { durable: true });
   channel.assertQueue(queue, { durable: true });
   const bindings = _.keys(EVENT_HANDLERS);
@@ -73,8 +73,16 @@ export async function consume(channel, exchangeName, queue) {
         if (e.shouldAck) {
           channel.ack(msg);
         } else {
-          // acking for debugging issue on production. this would prevent log pile up
+          // ack the message but copy it to other queue where no consumer is listening
+          // we can listen to that queue on adhoc basis when we see error case like lead not created in SF
+          // we can use cloudamqp console to check the messages and may be manually create SF lead
+          // nacking here was causing flood of messages to the worker and it keep on consuming high resources
           channel.ack(msg);
+          publishChannel.publish(
+            exchangeName,
+            EVENT.ROUTING_KEY.CONNECT_TO_SF_FAILED,
+            new Buffer(msg.content.toString())
+          );
         }
       }
     });
@@ -91,7 +99,13 @@ async function start() {
     debug('created connection successfully with URL: ' + config.rabbitmqURL);
     const channel = await connection.createConfirmChannel();
     debug('Channel confirmed...');
-    consume(channel, config.rabbitmq.projectsExchange, config.rabbitmq.queues.project);
+    const publishChannel = await connection.createConfirmChannel();
+    consume(
+      channel,
+      config.rabbitmq.projectsExchange,
+      config.rabbitmq.queues.project,
+      publishChannel
+    );
   } catch (e) {
     debug('Unable to connect to RabbitMQ');
   }

consumer/test/worker.spec.js

@@ -9,7 +9,7 @@ import './setup';
 describe('worker', () => {
   describe('consume', () => {
     const queueName = 'sample-queue';
-    const exchangeName = EVENT.ROUTING_KEY.PROJECT_DRAFT_CREATED//'sample-exchange';
+    const exchangeName = 'sample-exchange';
     const validMessage = {
       content: JSON.stringify({ sampleData: 'foo' }),
       properties: { correlationId : 'unit-tests'},
@@ -24,6 +24,7 @@ describe('worker', () => {
     let rabbitConsume;
     let exchangeHandlerSpy = sinon.spy();
     let fakeExchangeHandlerSpy = sinon.spy();
+    let channelPublishSpy = sinon.spy();
 
     beforeEach(() => {
       handler = sinon.spy();
@@ -58,7 +59,10 @@ describe('worker', () => {
             done(e);
           }
         },
-      }, exchangeName, queueName);
+      }, exchangeName, queueName,
+      {
+        publish: channelPublishSpy
+      });
     }
 
     it('should consume and ack a message successfully', (done) => {
@@ -91,15 +95,16 @@ describe('worker', () => {
       invokeConsume(done);
     });
 
-    xit('should nack if error is thrown', (done) => {
+    it('should ack, with message being copied to temp queue, if error is thrown', (done) => {
       initHandlers({
         [exchangeName] : () => {
           throw new Error('foo');
         }
       })
       rabbitConsume = async (queue, fn) => {
         await fn(validMessage);
-        nack.should.have.been.calledWith(validMessage);
+        ack.should.have.been.calledWith(validMessage);
+        channelPublishSpy.should.have.been.calledWith(exchangeName, EVENT.ROUTING_KEY.CONNECT_TO_SF_FAILED, new Buffer(validMessage.content));
       };
       invokeConsume(done);
     });