-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathindex.js
92 lines (81 loc) · 3.1 KB
/
index.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
/* Copyright (c) 2017 Devis, MIT License */
"use strict";
const devis = require("devis");
const colors = require("colors/safe");
const amqp = require('amqplib/callback_api');
let rabbitmqOptions = [];
let channel = [];
//Producer
function listen(args, callback) {
//connect to RabbitMQ
if (args.rabbitmq.port === undefined) args.rabbitmq.port = "5672";
amqp.connect('amqp://' + args.rabbitmq.host + ":" + args.rabbitmq.port, function (err, conn) {
if (err) {
callback(err, null);
}
else {
//create a channel
conn.createChannel(function (err, ch) {
if (err) {
callback(err, null);
}
else {
ch.assertQueue(args.rabbitmq.queue + "Producer");
ch.assertQueue(args.rabbitmq.queue + "Consumer");
ch.consume(args.rabbitmq.queue + "Consumer", function (msg) {
let data = msg.content.toString();
data = JSON.parse(data);
delete data.path["queue"];
devis.call(data.path, data.args, (err, res) => {
ch.sendToQueue(args.rabbitmq.queue + "Producer", new Buffer(JSON.stringify(res)));
});
}, { noAck: true });
let transportInfors={host:args.rabbitmq.host,port:args.rabbitmq.port};
console.log(colors.green("server launched using the following informations : rabbitMQ ", JSON.stringify(transportInfors)));
callback(null, "Hello producer!");
}
});
}
});
return devis;
}
function client(args, callback) {
rabbitmqOptions[args.rabbitmq.queue] = args.rabbitmq;
if (args.rabbitmq.port === undefined) args.rabbitmq.port = "5672";
amqp.connect('amqp://' + args.rabbitmq.host + ":" + args.rabbitmq.port, function (err, conn) {
if (err) {
callback(err, null);
}
else {
//create a channel
conn.createChannel(function (err, ch) {
if (err) {
callback(err, null);
}
else {
let transportInfors={host:args.rabbitmq.host,port:args.rabbitmq.port};
console.log(colors.green("connected using the following informations : ", JSON.stringify(transportInfors)));
channel[args.rabbitmq.queue] = ch;
callback(null, channel);
}
});
}
});
return devis;
}
function call(path, args, callback) {
let dataToSend = {
path: path,
args: args
}
channel[path.queue].sendToQueue(path.queue + "Consumer", new Buffer(JSON.stringify(dataToSend)));
channel[path.queue].assertQueue(path.queue);
channel[path.queue].consume(path.queue + "Producer", function (msg) {
callback(null, msg.content.toString());
}, { noAck: true });
return devis;
}
devis.listenMQ = listen;
devis.clientMQ = client;
devis.callMQ = call;
module.exports = devis;