message-bus

This package implements message bus using redis-based queue (resque). It provides a convenient way of building asynchronous services in node.js based architecture.

Install

npm install message-bus

Use

var MessgeBus = require('message-bus');

APIs

• new MessageBus
• messageBus.fire
• messageBus.addListener
• messageBus.stop
• messageBus.garbageCollect
• messageBus.retry

new MessageBus(cfg)

This creates a message bus instance. The configuration object is of the following format:

{
    "type": "Object",
    "fields": {
        "mysql_config": {
            "type": "Alias",
            "alias": "MYSQL_CONFIG"
        },
        "resque_configs": {
            "type": "Array",
            "element": {
                "type": "Alias",
                "alias": "COFFEE_RESQUE_CONFIG"
            }
        },
        "retry_limit": {
            "type": "Number",
            "nullable": true,
            "default": 5
        }
    }
}

Please refer to the following packages:

• mysql
• coffee-resque

for the format of the mysql configuration and coffee-resque configuration included in messge-bus configuration. The configuration field "retry_limit" will be used in the retry API.

Example

var mb = new MessageBus({
    mysql_config: {
        host     : 'localhost',
        user     : 'root',
        password : '',
        database: "message_bus"
    },
    resque_configs: [
        {
            "host": "localhost",
            "port": 6379,
            "timeout": 3000
        }
    ],
    retry_limit: 10
});

A message bus relies on two types of storage: mysql and resque, where mysql is used to store the message body and resque is used to trigger the listeners registered on the message. For the mysql storage, we enforce the following schema:

create table if not exists tasks (
  id bigint(20) not null primary key auto_increment,
  version bigint(20) not null default 0,
  date_created datetime not null,
  last_updated datetime not null,
  name varchar(255) not null,
  args text not null,
  status varchar(255) not null,
  retry_times int(11) not null default 0
);

to exist in the database specified in the configuration object. For resque, we allow muliple queues to be included in the configuration, which will be used in a round-robin fashion to increased the availability and stability of the message bus. Specifically, if one resque fails, the message load will be falling back to the rest resque(s).

mb.fire('test:foobar100', {foo: 'bar', text: 'This is the payload.'}, cb);

mb.addListener('test:foobar100', function(args, cb) {
    console.log('start to process args: ', args);
    setTimeout(function(){
        console.log('finish to process args: ', args);
    }, 1000);
}, 5);

mb.stop(cb);

mb.garbageCollect(cb);

messageBus.retry(callback)

This API retries the messages that had failed with recoverable errors. It is usually used in a cron job to increase the reliability of the message bus. The number of retries before declaring fatal can be configured in the constructor with the field "retry_limit", the default value of which is 5.

Example

mb.retry(cb);