How can zmq (zeromq) be used to send multiple requests and route async responses between two NodeJS servers?

Question

I have a NodeJS API web server (let's call it WS1) that receives RESTful HTTP requests from clients, and to respond needs to first query another local server (let's call it WS2).

The flow is pretty much like this:

1. WS1 receives an HTTP request from a client and parses it.
2. WS1 sends a request to WS2 for some information.
3. When WS1 receives the response from WS2, it finishes processing the original request and sends a response back to the client.

Until now all communication between WS1 and WS2 has been done through HTTP requests, since the two machines are on the same local network.

To speed things up though I'm considering to start using zmq instead. I've looked at the patterns they show on the docs, but still haven't figured out a concurrency problem.

WS1 can send many requests per second to WS2, and there's no guarantee that WS2 replies in the same order as it receives the requests, since some async operations can internally take longer than others.

So, using zmq with NodeJS, how do I make sure that when WS1 receives a message from WS2 it knows to what original client request it belongs to? Is there a built-in mechanism to take care of it?

Thanks!

Answer 1

I ended up implementing some sort of "middleware" to support this functionality with zmq.

In the example below for simplicity I've used Express with Node >= v4.0.0 (supporting native JS promises), but you can obviously substitute it with any HTTP server you like (these days I prefer Koa) and promises library you prefer. This is the code for the two servers.

WS1 (requester)

var zmq = require('zmq');
var mem = {};
var requester = zmq.socket('req');
requester.on("message", function(reply) {
  reply = reply.toString().split('*');
  mem[reply.pop()](reply);
});
requester.connect("tcp://localhost:5555");


var app = require('express')();

app.get('/', function (req, res) {
  var id = Date.now() + Math.random();
  new Promise(function (resolve, reject) {
    mem[id] = function (reply) {
      reply[0] === 'success' ? resolve(reply[1]) : reject(reply[1]); 
    }
  })
  .then(function (data) {
    res.send(data);
  })
  .catch(function (err) {
    console.log(err);
    res.send(500);
  })
  requester.send(id + '*' + message);
});

var server = app.listen(3000);

WS2 (responder)

var zmq = require('zmq');
var responder = zmq.socket('rep');
responder.on('message', function(message) {
  message = message.split('*');
  var reqId = message[0];
  // Do whatever async stuff you need with message[1]
  // Then at the end of your callbacks you'll have something like this
  if (err) {
    responder.send('err' + '*' + JSON.stringify(err) + '*' + reqId);
  } else {
    responder.send('success' + '*' + JSON.stringify(yourData) + '*' + reqId);
  }
});
responder.bind('tcp://*:5555');

Answer 2

0MQ is an interesting tool set that helps abstract socket communication. There are mechanism (should you choose the correct socket types) that allow the server to respond to the right client, and it is handled within the confines of 0mq.

The basic API types are:

1. PUSH-PULL
2. PUB-SUB
3. REQUEST-REPLY

IF you want to be able to have one machine respond to the originator, then I believe you want REQ-REP api type.

then you need to consider the multi-plexing on each side to get the connectors correct. But keep it one to one for simplicity sake at first:

Sample Client (from http://zguide.zeromq.org/js:rrclient

// Hello World client in Node.js
// Connects REQ socket to tcp://localhost:5559
// Sends "Hello" to server, expects "World" back

var zmq       = require('zmq')
  , requester = zmq.socket('req');

requester.connect('tcp://localhost:5559');
var replyNbr = 0;
requester.on('message', function(msg) {
  console.log('got reply', replyNbr, msg.toString());
  replyNbr += 1;
});

for (var i = 0; i < 10; ++i) {
  requester.send("Hello");
}

sample server (from http://zguide.zeromq.org/js:rrserver)

// Hello World server in Node.js
// Connects REP socket to tcp://*:5560
// Expects "Hello" from client, replies with "World"

var zmq = require('zmq')
  , responder = zmq.socket('rep');

responder.connect('tcp://localhost:5560');
responder.on('message', function(msg) {

  console.log('received request:', msg.toString());
  setTimeout(function() {
    responder.send("World");
  }, 1000);
});

The routing of the reply back to the client is handled automatically by 0MQ. it is part of the message (although I don't remember if you see the address buffer in these examples - it maybe abstracted away). Here is what the request envelope looks like:

it is the first frame, which allows 0MQ to be able to reply to the correct client.

Once that is running you can then consider 1..* *..1 and ... All it really does is require you to change the socket types to DEALER and ROUTER where appropriate.