Updated consumer

Author: zubairov

lib/actions/publish.js

@@ -34,7 +34,10 @@ function processAction(msg, cfg) {
       body: msg.body,
       attachments: msg.attachments
     });
-    channel.publish(amqpExchange, 'foo', encryptedData);
+    channel.publish(amqpExchange, 'foo', encryptedData, {
+      contentType: "application/octet-stream",
+      messageId: msg.id
+    });
     console.log('Message published id=%s', msg.id);
     yield channel.waitForConfirms();
     console.log('Message publishing confirmed id=%s', msg.id);

lib/cipher.js

@@ -31,11 +31,9 @@ function encryptIV(rawData) {
     return Buffer.concat([cipher.update(new Buffer.from(rawData)),cipher.final()]);
 }
 
-function decryptIV(encData, options) {
+function decryptIV(encData) {
     debug('About to decrypt:', encData);
 
-    options = options || {};
-
     if (!_.isString(encData)) {
         throw new Error('RabbitMQ message cipher.decryptIV() accepts only string as parameter.');
     }

lib/encryptor.js

@@ -12,7 +12,7 @@ function decryptMessageContent(messagePayload, messageHeaders) {
         return null;
     }
     try {
-        return JSON.parse(cipher.decrypt(messagePayload.toString(), messageHeaders));
+        return JSON.parse(cipher.decrypt(messagePayload.toString()));
     } catch (err) {
         console.error(err.stack);
         throw Error('Failed to decrypt message: ' + err.message);

lib/triggers/consume.js

@@ -3,6 +3,7 @@ const eioUtils = require('elasticio-node').messages;
 const co = require('co');
 const amqp = require('amqplib');
 const encryptor = require('../encryptor.js');
+const debug = require('debug')('consumer');
 
 module.exports.process = processAction;
 
@@ -19,28 +20,33 @@ function processAction(msg, cfg) {
   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('Have got message %s', JSON.stringify(msg.content.toString()));
-    this.emit('data', eioUtils.newMessageWithBody({}));
+    console.log('Have got message id=%s fields=%j', msg.id, msg.fields);
+    const decrypted = encryptor.decryptMessageContent(msg.content);
+    debug('Decrypted message=%j', decrypted);
+    const newMsg = eioUtils.newMessageWithBody(decrypted.body || {});
+    newMsg.id = msg.id;
+    newMsg.attachments = decrypted.attachments || {};
+    this.emit('data', newMsg);
   };
   co(function*() {
-    console.log('Connecting to amqpURI=%s', amqpURI);
+    debug('Connecting to amqpURI=%s', amqpURI);
     const conn = yield amqp.connect(amqpURI);
 
-    console.log('Creating a receiver channel');
+    debug('Creating a receiver channel');
     const channel = yield conn.createChannel();
 
-    console.log('Asserting topic exchange exchange=%s', amqpExchange);
+    debug('Asserting topic exchange exchange=%s', amqpExchange);
     yield channel.assertExchange(amqpExchange, 'topic');
 
-    console.log('Asserting queue');
+    debug('Asserting queue');
     yield channel.assertQueue(queueName, {
       exclusive: false,
       durable: false,
       autoDelete: true
     });
 
     for (let key of keys) {
-      console.log(`Binding queue to exchange queue=${queueName} exchange=${amqpExchange} bindingKey=${key}`);
+      debug(`Binding queue to exchange queue=${queueName} exchange=${amqpExchange} bindingKey=${key}`);
       yield channel.bindQueue(queueName, amqpExchange, key);
     }
 
@@ -49,6 +55,7 @@ function processAction(msg, cfg) {
       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}`
     });
+    console.log('After consume!');
   }.bind(this)).catch(err => {
     console.log('Error occurred', err.stack || err);
     this.emit('error', err);