Now do everything in init

Author: zubairov

lib/triggers/consume.js

@@ -5,9 +5,6 @@ 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}`;
-
 /**
  * This function will be called on component before the first message will be processed
  *
@@ -17,13 +14,31 @@ function init(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');
-        channel = yield conn.createChannel();
+        const channel = yield conn.createChannel();
 
         debug('Asserting topic exchange exchange=%s', amqpExchange);
         yield channel.assertExchange(amqpExchange, 'topic');
@@ -40,6 +55,13 @@ 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');
     });
 }
 
@@ -50,24 +72,8 @@ function init(cfg) {
  * @param cfg configuration that is account information and configuration field values
  */
 function processAction(msg, cfg) {
-    console.log('Trigger started, cfg=%j', cfg);
-    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');
-    });
+    console.log('Trigger started but we don not need to do anything here');
+    return Promise.resolve();
 }
 
 module.exports.process = processAction;