amqprpc

RPC library built on top of amqplib

Features

• Built on solid foundations
• Client timeout support
• Events for analytics or error handling

Usage

$ npm install amqprpc

RPC Server

Example

var amqp = require('amqplib');

amqp.connect('amqp://localhost/test').then(function(conn) {
  conn.createChannel().then(function(channel) {

  // magic happens here
  rpc.Server({ channel: channel, api: api });
}, console.error);

// any object with methods that return a promise
var api = {
  add: function(one, two) {
    this.resolve(one + two);
  },
  json: function(j) {
    this.resolve(j);
  },
  args: function(a, b, c) {
    this.resolve([a, b, c]);
  },
  reject: function(err) {
    this.reject(err || 'some error');
  },
  timeout: function() {
    setTimeout(this.resolve, 1000);
  }
};

Options

 * {String}   [queue.name=rpc]                   RPC queue name
 * {Object}   [queue.options={ durable:true }]   RPC queue options used to create the queue on the server

Events

connected

Emitted when the server successfully binds to the amqp queue

request

Emitted when a new request is received from a client

response

Emitted after the request has been processed and before the response is sent to the client

error

Errors thrown by the API functions are caught and passed to the client. These errors are thrown on the client side.

However, if any other error occurs, it is emitted as the error event. You can catch this error if needed, otherwise it will bubble up as an uncaughtException.

server.on('error', function(err) {
  log.critical(err);
  process.exit(1);
});

RPC Client

Example

var amqp = require('amqplib');

amqp.connect('amqp://localhost/test').then(function(conn) {
  conn.createChannel().then(function(channel) {

  var client = new rpc.Client({ channel: channel, timeout: 100 });

  // call api method
  client.call('add', [1, 2])
    .then(console.info)
    .catch(console.error);

  // optional
  rpc.Client.addMethods(client, ['add', 'json', 'args', 'reject', 'timeout']);
  client.add(1, 2)
    .then(console.info)
    .catch(console.error);
}, console.error);

var api = {
  add: function(one, two) {
    this.resolve(one + two);
  },
  json: function(j) {
    this.resolve(j);
  },
  args: function(a, b, c) {
    this.resolve(a, b, c);
  },
  reject: function(err) {
    this.reject(err || 'some error');
  },
  timeout: function() {
    setTimeout(this.resolve, 1000);
  }
};

Options

 * {Number}        [timeout=10000]                    Timeout in milliseconds, defaults to 10 seconds
 * {String}        [queue.name=rpc]                   RPC queue name
 * {Object}        [responseQueue.options={exclusive:true}] Options used for creating the response queue
 * {Object}        [responseQueue.consumeOptions={noAck:true}] Options used for consuming the response queue

Events

connected

Emitted when the client successfully binds to the amqp queue

request

Emitted when a new request is received

response

Emitted once a response is recived from the server

invalid id

Emitted if the request id does not match a pending promise

error

Errors thrown by the API functions are caught and passed to the application code.

However, if any other error occurs such as failing to bind to the amqp queue, it is emitted as the error event. You can catch this error if needed, otherwise it will bubble up as an uncaughtException.

client.on('error', function(err) {
  log.critical(err);
  process.exit(1);
});

Changelog

v0.0.2 (17 Feb 2015)

• Minor docco updates

v0.0.1 (12 Feb 2015)

• Initial commit