Now ignoring second runs

Author: zubairov

lib/triggers/consume.js

@@ -5,40 +5,26 @@ const amqp = require('amqplib');
 const encryptor = require('../encryptor.js');
 const debug = require('debug')('consumer');
 
+let channel;
+const queueName = `eio_consumer_${process.env.ELASTICIO_FLOW_ID}_${process.env.ELASTICIO_USER_ID}`;
+let listening;
+
 /**
  * This function will be called on component before the first message will be processed
  *
  * @param cfg
  */
 function init(cfg) {
-    console.log('Starting initialization, cfg=%j', cfg);
+    console.log('Starting initialization, cfg=%j queueName=%s', cfg, queueName);
     const amqpURI = cfg.amqpURI;
     const amqpExchange = cfg.topic;
-    const queueName = `eio_consumer_${process.env.ELASTICIO_FLOW_ID}_${process.env.ELASTICIO_USER_ID}`;
     const keys = (cfg.bindingKeys || '#').split(',').map((s) => s.trim());
-
-    /**
-     * Consumer function that will be called when new message
-     * is received
-     *
-     * @param msg
-     */
-    const consumer = (msg) => {
-        debug('Have got message fields=%j properties=%j', msg.fields, msg.properties);
-        const decrypted = encryptor.decryptMessageContent(msg.content);
-        debug('Decrypted message=%j', decrypted);
-        const newMsg = eioUtils.newMessageWithBody(decrypted.body || {});
-        newMsg.id = msg.properties.messageId;
-        newMsg.attachments = decrypted.attachments || {};
-        this.emit('data', newMsg);
-    };
-
     return co(function* initialize() {
         debug('Connecting to amqpURI=%s', amqpURI);
         const conn = yield amqp.connect(amqpURI);
 
         debug('Creating a receiver channel');
-        const channel = yield conn.createChannel();
+        channel = yield conn.createChannel();
 
         debug('Asserting topic exchange exchange=%s', amqpExchange);
         yield channel.assertExchange(amqpExchange, 'topic');
@@ -55,13 +41,6 @@ function init(cfg) {
             yield channel.bindQueue(queueName, amqpExchange, key);
         }
         console.log('Initialization completed');
-
-        console.log('Starting consuming from queue=%s', queueName);
-        yield channel.consume(queueName, consumer, {
-            noAck: true, // We can't really assert if message was consumed if we emit it yet
-            consumerTag: `consumer_${process.env.ELASTICIO_EXEC_ID}_${process.env.ELASTICIO_FLOW_ID}`
-        });
-        console.log('Consumption started');
     });
 }
 
@@ -72,8 +51,29 @@ function init(cfg) {
  * @param cfg configuration that is account information and configuration field values
  */
 function processAction(msg, cfg) {
-    console.log('Trigger started but we don not need to do anything here');
-    return Promise.resolve();
+    console.log('Trigger started, cfg=%j', cfg);
+    if (listening) {
+        console.log('Trigger was called again, we will ignore this run');
+        return Promise.resolve();
+    }
+    const consumer = (msg) => {
+        debug('Have got message fields=%j properties=%j', msg.fields, msg.properties);
+        const decrypted = encryptor.decryptMessageContent(msg.content);
+        debug('Decrypted message=%j', decrypted);
+        const newMsg = eioUtils.newMessageWithBody(decrypted.body || {});
+        newMsg.id = msg.properties.messageId;
+        newMsg.attachments = decrypted.attachments || {};
+        this.emit('data', newMsg);
+    };
+    return co(function* consume() {
+        console.log('Starting consuming from %s', queueName);
+        yield channel.consume(queueName, consumer, {
+            noAck: true, // We can't really assert if message was consumed if we emit it yet
+            consumerTag: `consumer_${process.env.ELASTICIO_EXEC_ID}_${process.env.ELASTICIO_FLOW_ID}`
+        });
+        console.log('Consumption started');
+        listening = true;
+    });
 }
 
 module.exports.process = processAction;