README
message-queue
a standard interface to access message queues. Both the publisher and the subscriber are event emitters.
Table of Contents
Usage
Publish
var queue = require('message-queue')('redis');
var pub = queue.Publish();
var channel = pub.channel('cats');
channel.publish({meow: 'yay'}, console.log);
Subscribe
var queue = require('message-queue')('redis');
var cats = queue.Subscribe({channel: 'cats'});
cats.on('message', function(coolCat){
console.log('message: ' + JSON.stringify(coolCat));
});
Channels & Validation
Channels are streams so you can pipe to them.
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);
API
Publish API
Create a connection to the server.
var mq = require('message-queue');
var queue = mq('redis');
var pub = queue.Publish(options);
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.
Publish Events
pub will emit events.
Ready
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.
Error
pub will emit error when encountering an error connecting to the server.
Close
close is emitted when the developer calls the pub.close() function.
End
pub emits end when for some reason the connection was terminated.
pub.channel(name, options)
Returns a channel named name for this publisher. See Channel for more details.
pub.close(cb)
Closes the connection to the server.
Channel API
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 Events
Error
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.
channel.publish(message, cb)
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.
Subscribe API
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.
Subscribe Events
sub will emit events.
Message
sub will emit message when a new message arrives
Ready
sub will emit ready when it has connected to the server.
Error
sub will emit error when encountering an error connecting to the server.
Close
close is emitted when the developer calls the sub.close() function.
End
sub emits end when for some reason the connection was terminated.
sub.close(cb)
Closes the connection to the server.