a standard interface to access different message queues. Both the publisher and the subscriber are event emitters. Currently supports redis and amqp, e.g. RabbitMQ.
var queue = require('message-queue')('redis');
var pub = queue.Publish();
var channel = pub.channel('cats');
channel.publish({meow: 'yay'}, console.log);var queue = require('message-queue')('redis');
var cats = queue.Subscribe({channel: 'cats'});
cats.on('message', function(coolCat){
console.log('message: ' + JSON.stringify(coolCat));
});Channels are streams so you can pipe to them.
In json mode message-queue expects newline delimited json. For plaintext, each line gets published.
You can use your joi schemas to validate and prevent bad messages from being sent.
var mq = require('message-queue');
var fs = require('fs');
var queue = mq('redis');
var Joi = mq.Joi;
var pub = queue.Publish();
var channel = pub.channel('cats', {
schema: {
meow : Joi.string().required()
}
});
channel.on('error', function (err) {
console.error('err: ' + err);
});
fs.createReadStream(__dirname + '/meow.json-stream.txt')
.pipe(channel);Create a connection to the server.
var mq = require('message-queue');
var queue = mq('redis');
var pub = queue.Publish(options);redis is one of the supported back end adapters. You can find the full list on mq.adapters.
pub is an EventEmitter.
The following options can be used:
host: The host for the server.port: Define the port.
Default values are specified in adapters/*/defaults.json.
Adapter specific options can be passed.
pub will emit events.
pub will emit ready when it has connected to the server, and it is not ready to be written to. You can still publish before server being ready since internally message-queue will buffer those messages and publish once a connection is established.
pub will emit error when encountering an error connecting to the server.
close is emitted when the developer calls the pub.close() function.
pub emits end when for some reason the connection was terminated.
Returns a channel named name for this publisher. See Channel for more details.
Closes the connection to the server.
var Joi = mq.Joi;
var channel = pub.channel('cats', {
schema: {
meow : Joi.string().required()
}
});The following options can be used:
schema: The joi schema that should be used to validate messages before they are published.json: Ensure only json can be published in this channel. Defaults to true.
channel is a Duplex Stream. It is created with the channel method of the Publish object.
fs.createReadStream(__dirname + '/meow.json-stream.txt')
.pipe(channel)
.pipe(process.stdout);When piping you should listen for errors:
channel.on('error', function(err) {
console.error('err: ' + err);
});channel will emit error events when validation fails, or there's a parsing problem. This only happens when you use pipe, since normal publishes use the error first in callback node.js idiom.
Publishes a message. If there is a schema, the message will be validated.
channel.publish('meow', function (err, ack) {
if (err) {
console.error('err: ' + err.message);
} else {
console.log(JSON.stringify(ack, null, 2));
}
});Errors are not emitted unless you are piping.
var cats = queue.Subscribe({channel: 'cats'});
cats.on('message', console.log);
sub is an EventEmitter.
The following options can be used:
channel: required The channel to subscribe to.json: Expect all messages to be json. Defaults to true.host: The host for the server.port: Define the port.
Default values are specified in adapters/*/defaults.json.
Adapter specific options can be passed.
sub will emit events.
sub will emit message when a new message arrives
sub will emit ready when it has connected to the server.
sub will emit error when encountering an error connecting to the server.
close is emitted when the developer calls the sub.close() function.
sub emits end when for some reason the connection was terminated.
Closes the connection to the server.
Redis does not support features such as ordering or persisting messages. You can use config files to turn these features on and off on other providers such as amqp but each adapter has its own config format.
Pull requests are welcome!
