Middleware

The interface of ØMQ is pretty low-level, it only allows us to use strings and binary buffers for messages, so any encoding or custom formatting of data has to be implemented by the users of the library.

In the next example, we are going to build a middleware infrastructure to abstract the preprocessing and postprocessing of the data passing through a ØMQ socket, so that we can transparently work with JSON objects but also seamlessly compress the messages traveling over the wire.

Note

Before continuing with the example, please make sure to install the ØMQ native libraries following the instructions at this URL: http://zeromq.org/intro:get-the-software. Any version in the 4.0 branch should be enough for working on this example.

The Middleware Manager

The first step to build a middleware infrastructure around ØMQ is to create a component that is responsible for executing the middleware pipeline when a new message is received or sent. For the purpose, let's create a new module called zmqMiddlewareManager.js and let's start defining it:

function ZmqMiddlewareManager(socket) {
  this.socket = socket;
  this.inboundMiddleware = [];           //[1]
  this.outboundMiddleware = [];
  var self = this;
  socket.on('message', function(message) {       //[2]
    self.executeMiddleware(self.inboundMiddleware, {
      data: message
    });
  });
}
module.exports = ZmqMiddlewareManager;

This first code fragment defines a new constructor for our new component. It accepts a ØMQ socket as an argument and:

Creates two empty lists that will contain our middleware functions, one for the inbound messages and another one for the outbound messages.
Immediately, it starts listening for the new messages coming from the socket by attaching a new listener to the 'message' event. In the listener, we process the inbound message by executing the inboundMiddleware pipeline.

The next method of the ZmqMiddlewareManager prototype is responsible for executing the middleware when a new message is sent through the socket:

ZmqMiddlewareManager.prototype.send = function(data) {
  var self = this;
  var message = {
    data: data
  };
  
  self.executeMiddleware(self.outboundMiddleware, message,
    function() {
      self.socket.send(message.data);
    }
  );
}

This time the message is processed using the filters in the outboundMiddleware list and then passed to socket.send() for the actual network transmission.

Now, we need a small method to append new middleware functions to our pipelines; we already mentioned that such a method is conventionally called use():

ZmqMiddlewareManager.prototype.use = function(middleware) {
  if(middleware.inbound) {
    this.inboundMiddleware.push(middleware.inbound);
  }
  if(middleware.outbound) {
    this.outboundMiddleware.unshift(middleware.outbound);
  }
}

Each middleware comes in pairs; in our implementation it's an object that contains two properties, inbound and outbound, that contain the middleware functions to be added to the respective list.

It's important to observe here that the inbound middleware is pushed to the end of the inboundMiddleware list, while the outbound middleware is inserted at the beginning of the outboundMiddleware list. This is because complementary inbound/outbound middleware functions usually need to be executed in an inverted order. For example, if we want to decompress and then deserialize an inbound message using JSON, it means that for the outbound, we should instead first serialize and then compress.

Note

It's important to understand that this convention for organizing the middleware in pairs is not strictly part of the general pattern, but only an implementation detail of our specific example.

Now, it's time to define the core of our component, the function that is responsible for executing the middleware:

ZmqMiddlewareManager.prototype.executeMiddleware =
  function(middleware, arg, finish) {
    var self = this;
    (function iterator(index) {
      if(index === middleware.length) {
        return finish && finish();
      }
      middleware[index].call(self, arg, function(err) {
        if(err) {
          console.log('There was an error: ' + err.message);
        }
        iterator(++index);
      });
    })(0);
  }

The preceding code should look very familiar; in fact, it is a simple implementation of the asynchronous sequential iteration pattern that we learned in Chapter 2, Asynchronous Control Flow Patterns. Each function in the middleware array received in input is executed one after the other, and the same arg object is provided as an argument to each middleware function; this is the trick that makes it possible to propagate the data from one middleware to the next. At the end of the iteration, the finish() callback is invoked.

Note

Please note that for brevity we are not supporting an error middleware pipeline. Normally, when a middleware function propagates an error, another set of middleware specifically dedicated to handling errors is executed. This can be easily implemented using the same technique that we are demonstrating here.

A middleware to support JSON messages

Now that we have implemented our Middleware Manager, we can create a pair of middleware functions to demonstrate how to process inbound and outbound messages. As we said, one of the goals of our middleware infrastructure is having a filter that serializes and deserializes JSON messages, so let's create a new middleware to take care of this. In a new module called 'middleware.js' let's include the following code:

module.exports.json = function() {
  return {
    inbound: function(message, next) {
      message.data = JSON.parse(message.data.toString());
      next();
    },
    outbound:
      function(message, next) {
      message.data = new Buffer(JSON.stringify(message.data));
      next();
    }
  }
}

The json middleware that we just created is very simple:

The inbound middleware deserializes the message received as an input and assigns the result back to the data property of message, so that it can be further processed along the pipeline
The outbound middleware serializes any data found into message.data

Please note how the middleware supported by our framework is quite different from the one used in express; this is totally normal and a perfect demonstration of how we can adapt this pattern to fit our specific need.

Using the ØMQ middleware framework

We are now ready to use the middleware infrastructure that we just created. To do that, we are going to build a very simple application, with a client sending a ping to a server at regular intervals and the server echoing back the message received.

From an implementation perspective, we are going to rely on a request/reply messaging pattern using the req/rep socket pair provided by ØMQ (http://zguide.zeromq.org/page:all#Ask-and-Ye-Shall-Receive). We will then wrap the sockets with our zmqMiddlewareManager to get all the advantages from the middleware infrastructure that we built, including the middleware for serializing/deserializing JSON messages.

The server

Let's start by creating the server side (server.js). In the first part of the module we initialize our components:

var zmq = require('zmq');
var ZmqMiddlewareManager = require('./zmqMiddlewareManager');
var middleware = require('./middleware');
var reply = zmq.socket('rep');
reply.bind('tcp://127.0.0.1:5000');

In the preceding code, we loaded the required dependencies and bind a ØMQ 'rep' (reply) socket to a local port. Next, we initialize our middleware:

var zmqm = new ZmqMiddlewareManager(reply);
zmqm.use(middleware.zlib());
zmqm.use(middleware.json());

We created a new ZmqMiddlewareManager object and then added two middlewares, one for compressing/decompressing the messages and another one for parsing/serializing JSON messages.

Note

For brevity, we did not show the implementation of the zlib middleware, but you can find it in the sample code that is distributed with the book.

Now we are ready to handle a request coming from the client, we will do this by simply adding another middleware, this time using it as a request handler:

zmqm.use({
  inbound: function(message, next) {
    console.log('Received: ', message.data);
    if(message.data.action === 'ping') {
      this.send({action: 'pong', echo: message.data.echo});
    }
    next();
  }
});

Since this last middleware is defined after the zlib and json middlewares, we can transparently use the decompressed and deserialized message that is available in the message.data variable. On the other hand, any data passed to send() will be processed by the outbound middleware, which in our case will serialize then compress the data.

The client

On the client side of our little application, 'client.js', we will first have to initiate a new ØMQ 'req' (request) socket connected to the port 5000, the one used by our server:

var zmq = require('zmq');
var ZmqMiddlewareManager = require('./zmqMiddlewareManager');
var middleware = require('./middleware');

var request = zmq.socket('req');
request.connect('tcp://127.0.0.1:5000');

Then, we need to set up our middleware framework in the same way that we did for the server:

var zmqm = new ZmqMiddlewareManager(request);
zmqm.use(middleware.zlib());
zmqm.use(middleware.json());

Next, we create an inbound middleware to handle the responses coming from the server:

zmqm.use({
  inbound: function(message, next) {
    console.log('Echoed back: ', message.data);
    next();
  }
});

In the preceding code, we simply intercept any inbound response and print it to the console.

Finally, we set up a timer to send some ping requests at regular intervals, always using the zmqMiddlewareManager to get all the advantages of our middleware:

setInterval(function() {
  zmqm.send({action: 'ping', echo: Date.now()});
}, 1000);

We can now try our application by first starting the server:

node server

We can then start the client with the following command:

node client

At this point, we should see the client sending messages and the server echoing them back.

Our middleware framework did its job; it allowed us to decompress/compress and deserialize/serialize our messages transparently, leaving the handlers free to focus on their business logic!