Skip to content

Commit cec1047

Browse files
kirill-levitskiykirill-levitskiy
authored
update sailor to 2.6.7 (#2)
* update sailor to 2.6.7
1 parent b808995 commit cec1047

File tree

13 files changed

+3429
-261
lines changed

13 files changed

+3429
-261
lines changed

‎.eslintrc.js‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
module.exports = {
2+
'extends': 'airbnb-base',
3+
'env': {
4+
'mocha': true,
5+
'node': true,
6+
}
7+
};

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
node_modules
22
coverage
33
.idea
4+
.env

‎.travis.yml‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
language: node_js
22
node_js:
3-
- v5
4-
- v6
5-
- v7
6-
script: npm run integration-test
3+
- v14
4+
- v13
5+
- v12
6+
script: npm run integration-test

‎CHANGELOG.md‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
## 1.3.1 (May 22, 2020)
2+
3+
* Update sailor version to 2.6.7
4+
5+
## 1.3.0 (April 21, 2017)
6+
7+
* Update sailor version to 2.1.0
8+
9+
## 1.2.0 (February 21, 2017)
10+
11+
* Update sailor version to 2.0.0
12+
13+
## 1.1.0 (December 16, 2016)
14+
15+
* Add Consume Trigger
16+
17+
## 1.0.0 (December 15, 2016)
18+
19+
* Initial release
20+
* Add Publish Action

‎component.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
{
22
"title": "AMQP",
3+
"buildType": "docker",
34
"description": "Pub/Sub Component for async communication with queues and topics",
45
"credentials": {
56
"fields": {

‎lib/actions/publish.js‎

Lines changed: 33 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1,31 +1,27 @@
1-
'use strict';
21
const co = require('co');
32
const amqp = require('amqplib');
3+
const logger = require('@elastic.io/component-logger')();
44
const encryptor = require('../encryptor.js');
5-
const debug = require('debug')('publish');
65

76
let channel;
87

9-
module.exports.process = processAction;
10-
module.exports.init = init;
11-
128
/**
13-
* This methdo will be called from elastic.io platform on initialization
9+
* This method will be called from elastic.io platform on initialization
1410
*
1511
* @param cfg
1612
*/
1713
function init(cfg) {
18-
console.log('Starting initialization');
19-
const amqpURI = cfg.amqpURI;
20-
const amqpExchange = cfg.topic;
21-
return co(function* gen() {
22-
debug('Connecting to amqpURI=%s', amqpURI);
23-
const conn = yield amqp.connect(amqpURI);
24-
debug('Creating a confirm channel');
25-
channel = yield conn.createConfirmChannel();
26-
debug('Asserting topic exchange exchange=%s', amqpExchange);
27-
yield channel.assertExchange(amqpExchange, 'topic');
28-
});
14+
logger.info('Starting initialization');
15+
const { amqpURI } = cfg;
16+
const amqpExchange = cfg.topic;
17+
return co(function* gen() {
18+
logger.debug('Connecting to amqpURI=%s', amqpURI);
19+
const conn = yield amqp.connect(amqpURI);
20+
logger.debug('Creating a confirm channel');
21+
channel = yield conn.createConfirmChannel();
22+
logger.debug('Asserting topic exchange exchange=%s', amqpExchange);
23+
yield channel.assertExchange(amqpExchange, 'topic');
24+
});
2925
}
3026

3127
/**
@@ -35,20 +31,24 @@ function init(cfg) {
3531
* @param cfg configuration that is account information and configuration field values
3632
*/
3733
function processAction(msg, cfg) {
38-
const amqpExchange = cfg.topic;
39-
return co(function* sendMessage() {
40-
console.log('Publishing message id=%s', msg.id);
41-
let encryptedData = encryptor.encryptMessageContent({
42-
body: msg.body.payload || msg.body,
43-
attachments: msg.attachments
44-
});
45-
channel.publish(amqpExchange, msg.body.routingKey || '', encryptedData, {
46-
contentType: "application/octet-stream",
47-
messageId: msg.id
48-
});
49-
console.log('Message published id=%s', msg.id);
50-
yield channel.waitForConfirms();
51-
console.log('Message publishing confirmed id=%s', msg.id);
52-
return msg;
53-
}.bind(this));
34+
const self = this;
35+
const amqpExchange = cfg.topic;
36+
return co(function* sendMessage() {
37+
self.logger.info('Publishing message id=%s', msg.id);
38+
const encryptedData = encryptor.encryptMessageContent(self, {
39+
body: msg.body.payload || msg.body,
40+
attachments: msg.attachments,
41+
});
42+
channel.publish(amqpExchange, msg.body.routingKey || '', encryptedData, {
43+
contentType: 'application/octet-stream',
44+
messageId: msg.id,
45+
});
46+
self.logger.info('Message published id=%s', msg.id);
47+
yield channel.waitForConfirms();
48+
self.logger.info('Message publishing confirmed id=%s', msg.id);
49+
return msg;
50+
});
5451
}
52+
53+
module.exports.process = processAction;
54+
module.exports.init = init;

‎lib/cipher.js‎

Lines changed: 38 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,52 +1,53 @@
1-
var _ = require('lodash');
2-
var crypto = require('crypto');
3-
var debug = require('debug')('sailor:cipher');
1+
/* eslint-disable new-cap */
42

5-
var ALGORYTHM = 'aes-256-cbc';
6-
var PASSWORD = process.env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD;
7-
var VECTOR = process.env.ELASTICIO_MESSAGE_CRYPTO_IV;
3+
const _ = require('lodash');
4+
const crypto = require('crypto');
85

9-
exports.id = 1;
10-
exports.encrypt = encryptIV;
11-
exports.decrypt = decryptIV;
6+
const ALGORYTHM = 'aes-256-cbc';
7+
const PASSWORD = process.env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD;
8+
const VECTOR = process.env.ELASTICIO_MESSAGE_CRYPTO_IV;
129

13-
function encryptIV(rawData) {
14-
debug('About to encrypt:', rawData);
10+
function encryptIV(self, rawData) {
11+
self.logger.debug('About to encrypt:', rawData);
1512

16-
if (!_.isString(rawData)) {
17-
throw new Error('RabbitMQ message cipher.encryptIV() accepts only string as parameter.');
18-
}
13+
if (!_.isString(rawData)) {
14+
throw new Error('RabbitMQ message cipher.encryptIV() accepts only string as parameter.');
15+
}
1916

20-
if (!PASSWORD) {
21-
console.log('Encryption will be skipped as ELASTICIO_MESSAGE_CRYPTO_PASSWORD env is empty');
22-
return new Buffer.from(rawData);
23-
}
17+
if (!PASSWORD) {
18+
self.logger.info('Encryption will be skipped as ELASTICIO_MESSAGE_CRYPTO_PASSWORD env is empty');
19+
return new Buffer.from(rawData);
20+
}
2421

25-
if (!VECTOR) {
26-
throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
27-
}
22+
if (!VECTOR) {
23+
throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
24+
}
2825

29-
var encodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
30-
var cipher = crypto.createCipheriv(ALGORYTHM, encodeKey, VECTOR);
31-
return Buffer.concat([cipher.update(new Buffer.from(rawData)),cipher.final()]);
26+
const encodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
27+
const cipher = crypto.createCipheriv(ALGORYTHM, encodeKey, VECTOR);
28+
return Buffer.concat([cipher.update(new Buffer.from(rawData)), cipher.final()]);
3229
}
3330

34-
function decryptIV(encData) {
35-
debug('About to decrypt:', encData);
31+
function decryptIV(self, encData) {
32+
self.logger.debug('About to decrypt:', encData);
3633

37-
if (!PASSWORD) {
38-
console.log('Decryption will be skipped as ELASTICIO_MESSAGE_CRYPTO_PASSWORD env is empty');
39-
return encData;
40-
}
34+
if (!PASSWORD) {
35+
self.logger.info('Decryption will be skipped as ELASTICIO_MESSAGE_CRYPTO_PASSWORD env is empty');
36+
return encData;
37+
}
4138

42-
if (!VECTOR) {
43-
throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
44-
}
39+
if (!VECTOR) {
40+
throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
41+
}
4542

46-
var decodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
47-
var cipher = crypto.createDecipheriv(ALGORYTHM, decodeKey, VECTOR);
43+
const decodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
44+
const cipher = crypto.createDecipheriv(ALGORYTHM, decodeKey, VECTOR);
4845

49-
var result = cipher.update(encData, 'base64', 'utf-8') + cipher.final('utf-8');
46+
const result = cipher.update(encData, 'base64', 'utf-8') + cipher.final('utf-8');
5047

51-
return result;
48+
return result;
5249
}
50+
51+
exports.id = 1;
52+
exports.encrypt = encryptIV;
53+
exports.decrypt = decryptIV;

‎lib/encryptor.js‎

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,20 @@
1-
var cipher = require('./cipher.js');
1+
const cipher = require('./cipher.js');
22

3-
exports.encryptMessageContent = encryptMessageContent;
4-
exports.decryptMessageContent = decryptMessageContent;
5-
6-
function encryptMessageContent(messagePayload) {
7-
return cipher.encrypt(JSON.stringify(messagePayload));
3+
function encryptMessageContent(self, messagePayload) {
4+
return cipher.encrypt(self, JSON.stringify(messagePayload));
85
}
96

10-
function decryptMessageContent(messagePayload) {
11-
if (!messagePayload || ! Buffer.isBuffer(messagePayload)) {
12-
throw new Error("Message payload supplied for decryption is either empty or not a Buffer");
13-
}
14-
try {
15-
return JSON.parse(cipher.decrypt(messagePayload));
16-
} catch (err) {
17-
console.error(err.stack);
18-
throw Error('Failed to decrypt message: ' + err.message);
19-
}
7+
function decryptMessageContent(self, messagePayload) {
8+
if (!messagePayload || !Buffer.isBuffer(messagePayload)) {
9+
throw new Error('Message payload supplied for decryption is either empty or not a Buffer');
10+
}
11+
try {
12+
return JSON.parse(cipher.decrypt(self, messagePayload));
13+
} catch (err) {
14+
self.logger.error(err.stack);
15+
throw Error(`Failed to decrypt message: ${err.message}`);
16+
}
2017
}
18+
19+
exports.encryptMessageContent = encryptMessageContent;
20+
exports.decryptMessageContent = decryptMessageContent;

‎lib/triggers/consume.js‎

Lines changed: 52 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,8 @@
1-
'use strict';
2-
const eioUtils = require('elasticio-node').messages;
1+
const { messages } = require('elasticio-node');
32
const co = require('co');
43
const amqp = require('amqplib');
4+
const logger = require('@elastic.io/component-logger')();
55
const encryptor = require('../encryptor.js');
6-
const debug = require('debug')('consumer');
76

87
let channel;
98
const queueName = `eio_consumer_${process.env.ELASTICIO_FLOW_ID}_${process.env.ELASTICIO_USER_ID}`;
@@ -15,33 +14,34 @@ let listening;
1514
* @param cfg
1615
*/
1716
function init(cfg) {
18-
console.log('Starting initialization, queueName=%s', queueName);
19-
const amqpURI = cfg.amqpURI;
20-
const amqpExchange = cfg.topic;
21-
const keys = (cfg.bindingKeys || '#').split(',').map((s) => s.trim());
22-
return co(function* initialize() {
23-
debug('Connecting to amqpURI=%s', amqpURI);
24-
const conn = yield amqp.connect(amqpURI);
17+
logger.info('Starting initialization, queueName=%s', queueName);
18+
const { amqpURI } = cfg;
19+
const amqpExchange = cfg.topic;
20+
const keys = (cfg.bindingKeys || '#').split(',').map((s) => s.trim());
21+
return co(function* initialize() {
22+
logger.debug('Connecting to amqpURI=%s', amqpURI);
23+
const conn = yield amqp.connect(amqpURI);
2524

26-
debug('Creating a receiver channel');
27-
channel = yield conn.createChannel();
25+
logger.debug('Creating a receiver channel');
26+
channel = yield conn.createChannel();
2827

29-
debug('Asserting topic exchange exchange=%s', amqpExchange);
30-
yield channel.assertExchange(amqpExchange, 'topic');
28+
logger.debug('Asserting topic exchange exchange=%s', amqpExchange);
29+
yield channel.assertExchange(amqpExchange, 'topic');
3130

32-
debug('Asserting queue');
33-
yield channel.assertQueue(queueName, {
34-
exclusive: false,
35-
durable: false,
36-
autoDelete: true
37-
});
38-
39-
for (let key of keys) {
40-
debug(`Binding queue to exchange queue=${queueName} exchange=${amqpExchange} bindingKey=${key}`);
41-
yield channel.bindQueue(queueName, amqpExchange, key);
42-
}
43-
console.log('Initialization completed');
31+
logger.debug('Asserting queue');
32+
yield channel.assertQueue(queueName, {
33+
exclusive: false,
34+
durable: false,
35+
autoDelete: true,
4436
});
37+
38+
// eslint-disable-next-line no-restricted-syntax
39+
for (const key of keys) {
40+
logger.debug(`Binding queue to exchange queue=${queueName} exchange=${amqpExchange} bindingKey=${key}`);
41+
yield channel.bindQueue(queueName, amqpExchange, key);
42+
}
43+
logger.info('Initialization completed');
44+
});
4545
}
4646

4747
/**
@@ -50,31 +50,34 @@ function init(cfg) {
5050
* @param msg incoming message object that contains ``body`` with payload
5151
* @param cfg configuration that is account information and configuration field values
5252
*/
53+
// eslint-disable-next-line no-unused-vars
5354
function processAction(msg, cfg) {
54-
console.log('Trigger started');
55-
if (listening) {
56-
console.log('Trigger was called again, we will ignore this run');
57-
return Promise.resolve();
58-
}
59-
const consumer = (msg) => {
60-
debug('Have got message fields=%j properties=%j', msg.fields, msg.properties);
61-
const decrypted = encryptor.decryptMessageContent(msg.content);
62-
debug('Decrypted message=%j', decrypted);
63-
const newMsg = eioUtils.newMessageWithBody(decrypted.body || {});
64-
newMsg.id = msg.properties.messageId;
65-
newMsg.attachments = decrypted.attachments || {};
66-
this.emit('data', newMsg);
67-
};
68-
return co(function* consume() {
69-
console.log('Starting consuming from %s', queueName);
70-
yield channel.consume(queueName, consumer, {
71-
noAck: true, // We can't really assert if message was consumed if we emit it yet
72-
consumerTag: `consumer_${process.env.ELASTICIO_EXEC_ID}_${process.env.ELASTICIO_FLOW_ID}`
73-
});
74-
console.log('Consumption started');
75-
listening = true;
55+
const self = this;
56+
self.logger.info('Trigger started');
57+
if (listening) {
58+
self.logger.info('Trigger was called again, we will ignore this run');
59+
return Promise.resolve();
60+
}
61+
// eslint-disable-next-line no-shadow
62+
const consumer = (msg) => {
63+
self.logger.debug('Have got message fields=%j properties=%j', msg.fields, msg.properties);
64+
const decrypted = encryptor.decryptMessageContent(self, msg.content);
65+
self.logger.debug('Decrypted message=%j', decrypted);
66+
const newMsg = messages.newMessageWithBody(decrypted.body || {});
67+
newMsg.id = msg.properties.messageId;
68+
newMsg.attachments = decrypted.attachments || {};
69+
self.emit('data', newMsg);
70+
};
71+
return co(function* consume() {
72+
self.logger.info('Starting consuming from %s', queueName);
73+
yield channel.consume(queueName, consumer, {
74+
noAck: true, // We can't really assert if message was consumed if we emit it yet
75+
consumerTag: `consumer_${process.env.ELASTICIO_EXEC_ID}_${process.env.ELASTICIO_FLOW_ID}`,
7676
});
77+
self.logger.info('Consumption started');
78+
listening = true;
79+
});
7780
}
7881

7982
module.exports.process = processAction;
80-
module.exports.init = init;
83+
module.exports.init = init;

0 commit comments

Comments
 (0)