MIgrated to new sailor, improved publishing

Author: zubairov

lib/actions/publish.js

@@ -3,9 +3,29 @@ const co = require('co');
 const amqp = require('amqplib');
 const encryptor = require('../encryptor.js');
 
-var conn, channel;
+let channel;
 
 module.exports.process = processAction;
+module.exports.init = init;
+
+/**
+ * This methdo will be called from elastic.io platform on initialization
+ *
+ * @param cfg
+ */
+function init(cfg) {
+  console.log('Starting initialization, cfg=%j', cfg);
+  const amqpURI = cfg.amqpURI;
+  const amqpExchange = cfg.topic;
+  return co(function* gen() {
+    console.log('Connecting to amqpURI=%s', amqpURI);
+    const conn = yield amqp.connect(amqpURI);
+    console.log('Creating a confirm channel');
+    channel = yield conn.createConfirmChannel();
+    console.log('Asserting topic exchange exchange=%s', amqpExchange);
+    yield channel.assertExchange(amqpExchange, 'topic');
+  });
+}
 
 /**
  * This method will be called from elastic.io platform providing following data
@@ -14,42 +34,20 @@ module.exports.process = processAction;
  * @param cfg configuration that is account information and configuration field values
  */
 function processAction(msg, cfg) {
-  console.log('Action started');
-  const amqpURI = cfg.amqpURI;
   const amqpExchange = cfg.topic;
-  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');
-    }
+  return co(function* sendMessage() {
     console.log('Publishing message id=%s', msg.id);
     let encryptedData = encryptor.encryptMessageContent({
       body: msg.body.payload || msg.body,
       attachments: msg.attachments
     });
-    console.log(encryptedData, encryptor.decryptMessageContent(encryptedData));
-    channel.publish(amqpExchange, 'foo', encryptedData, {
+    channel.publish(amqpExchange, msg.body.routingKey || '', 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);
     this.emit('data', msg);
-    this.emit('end');
-    console.log('Closing the channel');
-    yield channel.close();
-    console.log('Closing the connnection');
-    yield conn.close();
-  }.bind(this)).catch(err => {
-    console.log('Error occurred', err.stack || err);
-    this.emit('error', err);
-    this.emit('end');
-  });
+  }.bind(this));
 }

package.json

@@ -27,7 +27,7 @@
     "co": "^4.6.0",
     "debug": "^2.4.4",
     "elasticio-node": "0.0.5",
-    "elasticio-sailor-nodejs": "1.3.0",
+    "elasticio-sailor-nodejs": "2.0.0",
     "request": "^2.75.0",
     "request-promise": "^4.1.1"
   },