Node.js – How to create REP/REQ on Rabbit.js

node.jsrabbitmq

I have been working with RabbitMQ on .Net for a while already and I don't have much of a problem with it. Now I'm moving to rabbit.js with node.js and I'm not pretty much familiar with it. rabbit.js has a limited documentation. All I know is the basic PUSH/PULL or PUB/SUB. Now I wanted to do REQ/REP and I don't know how do it. Anybody can share some snippet please.

Your reply is greatly appreciated.

Best,

Best Answer

This is perhaps more then you asked for but I have a snipplet (even though it's quite long) for doing RPC using node-amqp instead of REQ/RES with rabbit.js. What I have done is similar to what you could find in the RabbitMQ tutorial about RPC

For the moment the content in the message should be an object (hash) that will get transformed by the amqp module to json.

The AmqpRpc class take an amqp connection when initialized then it should only be a matter of calling makeRequest and wait for a response in the callback. The response have the form of function(err, response) where err might be a timeout error

I'm sorry its not exactly what you asked for but it's maybe close enough. I also posted the code as a gist on github: https://gist.github.com/2720846

Edit: Samples changed to support multiple outstanding requests.

amqprpc.js

var amqp = require('amqp')
  , crypto = require('crypto')

var TIMEOUT=2000; //time to wait for response in ms
var CONTENT_TYPE='application/json';

exports = module.exports = AmqpRpc;

function AmqpRpc(connection){
  var self = this;
  this.connection = typeof(connection) != 'undefined' ? connection : amqp.createConnection();
  this.requests = {}; //hash to store request in wait for response
  this.response_queue = false; //plaseholder for the future queue
}

AmqpRpc.prototype.makeRequest = function(queue_name, content, callback){
  var self = this;
  //generate a unique correlation id for this call
  var correlationId = crypto.randomBytes(16).toString('hex');

  //create a timeout for what should happen if we don't get a response
  var tId = setTimeout(function(corr_id){
    //if this ever gets called we didn't get a response in a 
    //timely fashion
    callback(new Error("timeout " + corr_id));
    //delete the entry from hash
    delete self.requests[corr_id];
  }, TIMEOUT, correlationId);

  //create a request entry to store in a hash
  var entry = {
    callback:callback,
    timeout: tId //the id for the timeout so we can clear it
  };

  //put the entry in the hash so we can match the response later
  self.requests[correlationId]=entry;

  //make sure we have a response queue
  self.setupResponseQueue(function(){
    //put the request on a queue
    self.connection.publish(queue_name, content, {
      correlationId:correlationId,
      contentType:CONTENT_TYPE,
      replyTo:self.response_queue});
  });
}


AmqpRpc.prototype.setupResponseQueue = function(next){
  //don't mess around if we have a queue
  if(this.response_queue) return next();

  var self = this;
  //create the queue
  self.connection.queue('', {exclusive:true}, function(q){  
    //store the name
    self.response_queue = q.name;
    //subscribe to messages
    q.subscribe(function(message, headers, deliveryInfo, m){
      //get the correlationId
      var correlationId = m.correlationId;
      //is it a response to a pending request
      if(correlationId in self.requests){
        //retreive the request entry
        var entry = self.requests[correlationId];
        //make sure we don't timeout by clearing it
        clearTimeout(entry.timeout);
        //delete the entry from hash
        delete self.requests[correlationId];
        //callback, no err
        entry.callback(null, message);
      }
    });
    return next();    
  });
}

A small example on how to use it can be found below. Save both code parts and just run with...

node client.js

If you don't have a server to provide the reply the request will time out.

client.js

//exmaple on how to use amqprpc
var amqp = require('amqp');
var connection = amqp.createConnection({host:'127.0.0.1'});

var rpc = new (require('./amqprpc'))(connection);

connection.on("ready", function(){
  console.log("ready");
  var outstanding=0; //counter of outstanding requests

  //do a number of requests
  for(var i=1; i<=10 ;i+=1){
    //we are about to make a request, increase counter
    outstanding += 1;
    rpc.makeRequest('msg_queue', {foo:'bar', index:outstanding}, function response(err, response){
      if(err)
        console.error(err);
      else
        console.log("response", response);
      //reduce for each timeout or response
      outstanding-=1;
      isAllDone();
    });
  }

  function isAllDone() {
    //if no more outstanding then close connection
    if(outstanding === 0){
      connection.end();
    }
  }

});

I'll even throw in a sample server for good measure

server.js 

//super simple rpc server example
var amqp = require('amqp')
  , util = require('util');

var cnn = amqp.createConnection({host:'127.0.0.1'});

cnn.on('ready', function(){
  console.log("listening on msg_queue");
  cnn.queue('msg_queue', function(q){
      q.subscribe(function(message, headers, deliveryInfo, m){
        util.log(util.format( deliveryInfo.routingKey, message));
        //return index sent
        cnn.publish(m.replyTo, {response:"OK", index:message.index}, {
            contentType:'application/json',
            contentEncoding:'utf-8',
            correlationId:m.correlationId
          });
      });
  });
});
Related Topic