First version of the triger

Author: zubairov

README.md

@@ -33,6 +33,14 @@ Now we are ready to push it:
 $ git push elasticio master
 ```
 
+## How consumer works
+
+Consumer will register a non-exclusive non-durable queue with autodelete=true without
+  any dead-letter. Name of the queue will be dynamically
+  generated based on the user ID, TASK ID prefixed with ``eio_consumer_``. 
+  This queue will be bound to the exchange with specified bound key or multiple
+   bound keys that are specified in one string separated by commas.
+
 ## Authentication
 
 This component exects user to provide a AMQP URI, username and password should be embedded

component.json

@@ -39,12 +39,12 @@
           "placeholder": "up_to_200_symbols",
           "note": "This exchange will be created on start if not exists."
         },
-        "bindingKey": {
-          "label": "Binding Key",
+        "bindingKeys": {
+          "label": "Binding Keys",
           "viewClass": "TextFieldWithNoteView",
           "required": false,
-          "placeholder": "dot.delimited.words",
-          "note": "You can use <b>#</b> or <b>*</b> to wildcard, more info <a href=\"http://www.rabbitmq.com/tutorials/tutorial-five-javascript.html\">here</a>"
+          "placeholder": "this.key,that.key",
+          "note": "Optional. You can use <b>#</b> or <b>*</b> to wildcard, more info <a href=\"http://www.rabbitmq.com/tutorials/tutorial-five-javascript.html\">here</a>"
         }
       },
       "metadata": {

lib/triggers/consume.js

@@ -4,8 +4,6 @@ const co = require('co');
 const amqp = require('amqplib');
 const encryptor = require('../encryptor.js');
 
-var conn, channel;
-
 module.exports.process = processAction;
 
 /**
@@ -18,30 +16,38 @@ function processAction(msg, cfg) {
   console.log('Trigger started, cfg=%j', cfg);
   const amqpURI = cfg.amqpURI;
   const amqpExchange = cfg.topic;
-
+  const queueName = `eio_consumer_${process.env.ELASTICIO_TASK_ID}_${process.env.ELASTICIO_USER_ID}`;
+  const keys = cfg.bindingKeys.split(',').map((s)=>s.trim());
+  const consumer = (msg) => {
+      console.log('consuming message %s in generator', JSON.stringify(msg.content.toString()));
+  };
   co(function*() {
-    if (!conn) {
-      console.log('Connecting to amqpURI=%s', amqpURI);
-      conn = yield amqp.connect(amqpURI);
-    }
-    if (!channel) {
-      console.log('Creating a confirm channel');
-      channel = yield conn.createConfirmChannel();
-      console.log('Asserting topic exchange exchange=%s', amqpExchange);
-      yield channel.assertExchange(amqpExchange, 'topic');
+    console.log('Connecting to amqpURI=%s', amqpURI);
+    const conn = yield amqp.connect(amqpURI);
+
+    console.log('Creating a receiver channel');
+    const channel = yield conn.createChannel();
+
+    console.log('Asserting topic exchange exchange=%s', amqpExchange);
+    yield channel.assertExchange(amqpExchange, 'topic');
+
+    console.log('Asserting queue');
+    yield channel.assertQueue(queueName, {
+      exclusive: false,
+      durable: false,
+      autoDelete: true
+    });
+
+    for(key of keys) {
+      console.log(`Binding queue to exchange queue=${queueName} exchange=${amqpExchange} bindingKey=${key}`);
+      yield channel.bindQueue(queueName, amqpExchange, key);
     }
-    console.log('Publishing message id=%s', msg.id);
-    let encryptedData = encryptor.encryptMessageContent({
-      id: msg.id,
-      body: msg.body,
-      attachments: msg.attachments
+
+    console.log('Start consuming');
+    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_TASK_ID}`
     });
-    channel.publish(amqpExchange, 'foo', encryptedData);
-    console.log('Message published id=%s', msg.id);
-    yield channel.waitForConfirms();
-    console.log('Message publishing confirmed id=%s', msg.id);
-    this.emit('data', msg);
-    this.emit('end');
   }.bind(this)).catch(err => {
     console.log('Error occurred', err.stack || err);
     this.emit('error', err);