-
-
Notifications
You must be signed in to change notification settings - Fork 17
/
mqemitter.js
120 lines (90 loc) · 2.36 KB
/
mqemitter.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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
'use strict'
const { Qlobber } = require('qlobber')
const assert = require('assert')
const fastparallel = require('fastparallel')
function MQEmitter (opts) {
if (!(this instanceof MQEmitter)) {
return new MQEmitter(opts)
}
const that = this
opts = opts || {}
opts.matchEmptyLevels = opts.matchEmptyLevels === undefined ? true : !!opts.matchEmptyLevels
opts.separator = opts.separator || '/'
opts.wildcardOne = opts.wildcardOne || '+'
opts.wildcardSome = opts.wildcardSome || '#'
this._messageQueue = []
this._messageCallbacks = []
this._parallel = fastparallel({
results: false,
released
})
this.concurrency = opts.concurrency || 0
this.current = 0
this._matcher = new Qlobber({
match_empty_levels: opts.matchEmptyLevels,
separator: opts.separator,
wildcard_one: opts.wildcardOne,
wildcard_some: opts.wildcardSome
})
this.closed = false
this._released = released
function released () {
that.current--
const message = that._messageQueue.shift()
const callback = that._messageCallbacks.shift()
if (message) {
that._do(message, callback)
}
}
}
Object.defineProperty(MQEmitter.prototype, 'length', {
get: function () {
return this._messageQueue.length
},
enumerable: true
})
MQEmitter.prototype.on = function on (topic, notify, done) {
assert(topic)
assert(notify)
this._matcher.add(topic, notify)
if (done) {
setImmediate(done)
}
return this
}
MQEmitter.prototype.removeListener = function removeListener (topic, notify, done) {
assert(topic)
assert(notify)
this._matcher.remove(topic, notify)
if (done) {
setImmediate(done)
}
return this
}
MQEmitter.prototype.emit = function emit (message, cb) {
assert(message)
cb = cb || noop
if (this.closed) {
return cb(new Error('mqemitter is closed'))
}
if (this.concurrency > 0 && this.current >= this.concurrency) {
this._messageQueue.push(message)
this._messageCallbacks.push(cb)
} else {
this._do(message, cb)
}
return this
}
MQEmitter.prototype.close = function close (cb) {
this.closed = true
setImmediate(cb)
return this
}
MQEmitter.prototype._do = function (message, callback) {
const matches = this._matcher.match(message.topic)
this.current++
this._parallel(this, matches, message, callback)
return this
}
function noop () {}
module.exports = MQEmitter