message-queue

a standard interface to access message queues. Both the publisher and the subscriber are event emitters.

Table of Contents

• Usage
  • Publish
  • Subscribe
  • Channels & Validation
• API
  • Publish
    • Events
    • channel()
    • close()
  • Channel
    • Events
    • publish()
  • Subscribe
    • Events
    • close()
• License
• Contributing
• Examples

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.