Messaging in Node.JS with RabbitMQ

$ npm install rabbit.js

This library provides a simple, socket-oriented API* for messaging in Node.JS, using RabbitMQ as a backend.

var context = require('rabbit.js').createContext();
context.on('ready', function() {
  var pub = context.socket('PUB'), sub = context.socket('SUB');
  sub.pipe(process.stdout);
  sub.connect('events', function() {
    pub.connect('events', function() {
      pub.write(JSON.stringify({welcome: 'rabbit.js'}), 'utf8');
    });
  });
});

See Github pages for documentation of the most recent release, and the branch gh-pages-next for provisional documentation of the next release (which usually corresponds to the code in master branch).

Status

Still on major version 0, though in use in a number of places, I believe.

Version 0.3.0 and on are built on amqplib. Previous versions, of which version 0.2.2 was the last, used node-amqp.

Uses

This library is suitable for co-ordinating peers (e.g., Node.JS programs), acting as a gateway to other kinds of network (e.g., relaying to browsers via SockJS), and otherwise as a really easy way to use RabbitMQ.

Fork features

Topics

SEND -- A provider for the JOB socket type that ensures that the exchange configurations can align. It is not required to use this socket type with a JOB provider when using the default exchange.

JOB

• provides access to the rabbitMQ message instead of the contents.

The JOB topic allows for asynchronous acking of messages after a task has completed. So the worker can push the completed message onto another queue and then ack the original message ensuring that rabbitMQ always has a message in a queue related to the original job until all sub-tasks are complete.

• see the test suite for more examples.

var ctx = require('rabbit.js').createContext();
ctx.on('ready', function() {
  var exchange = "testNextJobs"
  var routingKey = "testRoutingKey"

  var consumerOptions = {routing:'topic',durable:true, prefetch:3}
  var providerOptions = {routing:'topic',durable:true}

  var provider = ctx.socket('SEND');
  var nextQ = 'bar-tasks'
  provider.connect(exchange, providerOptions);

  var job = ctx.socket('JOB', {prefetch:64});
  var actionQ = 'foo-tasks';
  job.connect(actionQ, exchange, routingKey, consumerOptions);


  function recv(msg) {
    /* Do work
     ...
    */
    var nextMsg = {"data":"data"}
    job.next(msg, provider, nextMsg, routingKey)
  }

  job.on('data', recv.bind(job));
})