MQEmitter RabbitMQ at AMQP protocol
An Opinionated Message Queue with an emitter-style API, but with callbacks.
THIS IS NOT AN OFFICIAL MODULE FROM @mcollina.
If you need a multi process MQEmitter, check out the table below:
- [mqemitter-redis]: Redis-powered mqemitter
- [mqemitter]: Standard MQemitter
- [mqemitter-mongodb]: Mongodb based mqemitter
- [mqemitter-child-process]: Share the same mqemitter between a hierarchy of child processes
- [mqemitter-cs]: Expose a MQEmitter via a simple client/server protocol
- [mqemitter-p2p]: A P2P implementation of MQEmitter, based on HyperEmitter and a Merkle DAG
- [mqemitter-aerospike]: Aerospike mqemitter
Installation
npm install mqemitter-rabbit
Examples
import { MQEmitterAMQPLib } from '../mqemitter-rabbit';
const mqemitter = new MQEmitterAMQPLib({
separator: ':'
});
mqemitter.startConnection(
'amqps://your-url-must-come-here', ['entry:message', 'process:message', 'close:message'], 'both'
);
mqemitter.on(
'conversation:created', (
msg, done
) => {
console.log(
'Creating new conversation', msg, done
);
}
);
mqemitter.on(
'conversation:message', (
msg, done
) => {
console.log(
'Conversation received a new message', msg, done
);
}
);
setTimeout(
() => {
try {
mqemitter.emit(
{
topic: 'conversation:created'
}, 'entry:message'
);
mqemitter.emit(
{
topic: 'conversation:message'
}, 'process:message'
);
mqemitter.emit(
{
topic: 'conversation:postback'
}, 'close:messagea'
);
} catch (err) {
console.log(err);
}
}, 5000
);
new MQEmitter ([options])
- options
<object>
concurrency
<number>
maximum number of concurrent messages that can be on concurrent delivery. Default: 0
wildcardOne
<string>
a char to use for matching exactly one non-empty level word. Default: +
wildcardSome
<string>
a char to use for matching multiple level wildcards. Default: #`matchEmptyLevels
<boolean>
If true then wildcardOne
also matches an empty word. Default: true
separator
<string>
a separator character to use for separating words. Default: /
Create a new MQEmitterAMQPLib class.
MQEmitterAMQPLib is the class and function exposed by this module.
It can be created by using new MQEmitterAMQPLib({options})
.
For more information on wildcards, see this explanation or Qlobber.
emitter.emit (message, callback)
message
<object>
callback
<Function>
(error) => void
Emit the given message, which must have a topic
property, which can contain wildcards as defined on creation.
emitter.on (topic, listener, [callback])
topic
<string>
listener
<Function>
(message, done) => void
callback
<Function>
() => void
Add the given listener to the passed topic. Topic can contain wildcards, as defined on creation.
The listener
must never error and done
must not be called with an err
object.
callback
will be called when the event subscribe is done correctly.
emitter.removeListener (topic, listener, [callback])
The inverse of on
.
emitter.close (callback)
callback
<Function>
() => void
Close the given emitter. After, all writes will return an error.
Wildcards
MQEmitter supports the use of wildcards: every topic is splitted according to separator
.
The wildcard character +
matches exactly non-empty one word:
emitter.on('hello/+/world', function(message, cb) {
console.log(message)
cb()
})
emitter.on('hello/+', function(message, cb) {
console.log(message)
cb()
})
emitter.emit({ topic: 'hello/my/world', something: 'more' })
emitter.emit({ topic: 'hello//world', something: 'more' })
The wildcard character +
matches one word:
emitter.on('hello/+/world', function(message, cb) {
console.log(message)
cb()
})
emitter.on('hello/+', function(message, cb) {
console.log(message)
cb()
})
emitter.emit({ topic: 'hello/my/world', something: 'more' })
emitter.emit({ topic: 'hello//world', something: 'more' })
The wildcard character #
matches zero or more words:
emitter.on('hello/#', function(message, cb) {
console.log(message)
cb()
})
emitter.on('#', function(message, cb) {
console.log(message)
cb()
})
emitter.on('hello/my/world/#', function(message, cb) {
console.log(message)
cb()
})
emitter.emit({ topic: 'hello/my/world', something: 'more' })
Of course, you can mix #
and +
in the same subscription.
LICENSE
ISC