-
Notifications
You must be signed in to change notification settings - Fork 5
/
index.js
106 lines (89 loc) · 2.77 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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
"use strict";
var Bluebird = require('bluebird');
var exchanges = {};
var queues = {};
var channel = {
assertQueue: function (queue, qOptions) {
return new Bluebird(function (resolve) {
setIfUndef(queues, queue, { messages: [], subscribers: [], options: qOptions });
return resolve();
});
},
assertExchange: function (exchange, type, exchOptions) {
return new Bluebird(function (resolve) {
exchOptions = exchOptions || {};
setIfUndef(exchanges, exchange, { bindings: [], options: exchOptions, type: type });
return resolve();
});
},
bindQueue: function (queue, exchange, key, args) {
return new Bluebird(function (resolve, reject) {
if (!exchanges[exchange])
return reject("Bind to non-existing exchange " + exchange);
var re = "^" + key.replace(".", "\\.").replace("#", "(\\w|\\.)+").replace("*", "\\w+") + "$";
exchanges[exchange].bindings.push({ regex: new RegExp(re), queueName: queue });
return resolve();
});
},
publish: function (exchange, routingKey, content, props) {
return new Bluebird(function (resolve, reject) {
if (!exchanges[exchange])
return reject("Publish to non-existing exchange " + exchange);
var bindings = exchanges[exchange].bindings;
var matchingBindings = bindings.filter(function (b) { return b.regex.test(routingKey); });
matchingBindings.forEach(function (binding) {
var subscribers = queues[binding.queueName] ? queues[binding.queueName].subscribers : [];
subscribers.forEach(function (sub) {
var message = { fields: { routingKey: routingKey }, properties: props, content: content };
sub(message);
});
});
return resolve();
})
},
consume: function (queue, handler) {
queues[queue].subscribers.push(handler);
},
deleteQueue: function (queue) {
setImmediate(function () {
delete queues[queue];
});
},
ack: function () { },
nack: function () { },
prefetch: function () { },
on: function () { },
once: function () { },
close: function () {
return Bluebird.resolve();
}
};
function createChannel() {
return new Bluebird(function (resolve) {
return resolve(channel);
});
};
function connect(url, options) {
return new Bluebird(function (resolve) {
var connection = {
createChannel: createChannel,
createConfirmChannel: createChannel,
on: function () { },
once: function () { },
close: function () {
return Bluebird.resolve();
}
};
return resolve(connection);
});
}
function resetMock() {
queues = {};
exchanges = {};
}
module.exports = { connect: connect, resetMock: resetMock };
function setIfUndef(object, prop, value) {
if (!object[prop]) {
object[prop] = value;
}
}