why do want to use ZeroMQ

ZeroMQ is a high-performance asynchronous messaging library. It’s not a dedicated message broker but rather an embeddable concurrency framework with support for direct and fan-out endpoint connections over a variety of transports.

whenever you hear the word MQ(messaging Queue) you probably think that sits in the middle of your network and help you solve the problem and that is not zeromq, it’s not a demon, it’s not a server, its a library. Its a socket system that allows you to do concurrency, queueing and high availability.

Why ØMQ?

• To make reliable, resilient, fault tolerant distributed systems it’s for connecting programs and having them chat to each other
• it’s very low latency
• it’s not a lot of overhead
• it has a philosophy of being event-driven and non blocking
• it provides you rich patterns for interacting with programs and it offers scalability

The Network Onion

• Application, Presentation, Session
  — SSL, HTTP, SMTP, ØMQ
• Transport — — — — — TCP, UDP
• Network — — — — —-IP
• Data Link — — — — — MAC
• Physical — — — — — -wires

zeromq lives at the application presentation and session layers of the OSI model along with other things like SSL and HTTP SMTP which is a for email.

so you wouldn’t implement something like HTTP or rest on top of zeromq because it’s a peer with zeromq at that same level of things, you can use one or the other but you wouldn’t implement one on top the other

you could implement something very much like HTTP on top of a zeromq infrastructure but it wouldn’t be true HTTP because of zeromq’s interaction with the underlying sockets

Naive PUB/SUB Sockets

Beacon application: fires an event once a second.

{
  "pid": 12345,
  "timestamp": 1404168475695
}

PUB Program:

'use strict';
const
  net = require('net'),
  connections = [],
  server = net.createServer(function(connection) {
    connections.push(connection);
    connection.on('close', function() {
      let index = connections.indexOf(connection);
      connections.splice(index, 1);
    });
  });

setInterval(function() {
  let data = JSON.stringify({
    pid: process.pid,
    timestamp: Date.now()
  });
  connections.forEach(function(connection) {
    connection.write(data);
  });
}, 1000);

server.listen(5432, function() {
  console.log('Listening for subscribers...');
});

SUB Program:

'use strict';
const
  net = require('net'),
  client = net.connect({port: 5432});

client.on('data', function(data) {
  let msg = JSON.parse(data);
  console.log(msg.pid + ': ' + new Date(msg.timestamp));
});

Whats wrong with that ?

• Listener bias.
• Fault intolerant.
• Leaky buffers.
• Directionality (Publisher = Listener).

Listener bias: it is the unfortunate characteristic where if my publisher is not running my whole system doesn’t work. For example in the above code if I start up the subscriber without having the publisher running, node tells me connect ECONNREFUSED

fault intolerant: suppose if we got everything started up nice and neat and I’m receiving subscription messages and then I kill my publisher you’ll notice that the subscriber also shuts down right so this is is called fault intolerance. Even if the system start state was good as the ongoing running distributed system runs things happen sometimes the network has a hiccup or processes die in this case my whole system shuts down and my subscribers don’t start back up even if my publisher restarts itself automatically so we have to now plan for for the connection not starting up correctly and I have to plan for failures as they happen so my work is getting is getting harder all the time

leaky buffers: leaky buffers are best shown by looking at the code real quick so if I look at the publisher the publisher says connection dot writes data so I’m relying on this connection to just send out these bytes which happen to contain a JSON message and then in my subscriber code I presume that I’m getting a whole JSON message and then I’m sending it to json parse but nothing in the stack nothing in the stack below my code guarantees that I’ll get exactly one JSON message on a message boundary it could be that I get a half a message it was still a legitimate message but it comes over to events instead of one event if it got chopped up on the network on the way to my subscriber so I now as the developer have to implement buffering code I have to bring in a node module or write my own code to say I’m going to buffer up these data events then look for some sort of boundary character to know that that is a full message then try to parse it then move on to the next one, right so again now my work is getting harder as a developer trying to make a resilient system

directionality : directionality is more of an architectural concern it’s pretty clear in the publisher that this publisher is going to create a server and it’s going to manage a connection pool all of that is really baked in so when you have a publish/subscribe system it’s fair to say that most of the time typically your publisher is going to be the stable endpoint and your subscribers are going to come and go I mean most of the time that’s probably what you want but it doesn’t have to be that way I mean why couldn’t I make a system where my subscriber is a stable piece of my architecture and publishers come in and publish a message and then leave maybe I want to do that in this code I would have some serious refactoring to do to implement that change because it’s very clear that the connection handling the connection pooling is all part of the publisher code if I wanted to move that over to the subscriber and have this and have the publisher connect I’d have to do a lot of kind of refactoring.

PUB/SUB with zeromq

PUB program

var zmq = require("zeromq"),
 sock = zmq.socket("pub");
sock.bindSync("tcp://127.0.0.1:3000");
console.log("Listening for subscribers");
setInterval(function () {
 let msg = {
  pid: process.pid,
  timestamp: Date.now()
 };
 sock.send(["", JSON.stringify(msg)]);
}, 1000);

SUB Program

var zmq = require("zeromq"),
 sock = zmq.socket("sub");
sock.connect("tcp://127.0.0.1:3000");
sock.subscribe("");
sock.on("message", function (topic, message) {
 let s = message.toString();
 msg = JSON.parse(s);
 console.log(msg.pid + ': ' + new Date(msg.timestamp));
});

let’s talk about the four problems that we mentioned earlier and see if zeromq does them better so the first problem that I had mentioned earlier was listener bias. so that has to do with the start state of the system if I startup the subscriber it will wait for messages and behind the scenes zeromq is trying to connect to that endpoint it’s on its own back off schedule so I didn’t have to implement a backup algorithm or anything it’s trying to connect to the publisher and when the publisher comes up it’ll eventually work and the subscriber starts getting messages, so as a developer I don’t have to worry about the start state of the system it’ll handle that for me.

now if I kill off the publisher the subscriber keeps running and it just stops receiving messages which is what we would expect and behind the scenes it’s trying to reconnect to that same port and if I start up the publisher again I get messages.

third problem was leaky buffers so in my subscriber code it seems like I would be doing the same thing I was doing before in the TCP socket version I’m just calling json parse on the data but this is not the same kind of event that the TCP socket event was just giving me some characters of data and it was my responsibility to know whether or not it was on a boundary but in this case zeromq guarantees that when you send a message it will either deliver the whole message or nothing right so I don’t have to worry about the buffer problem because this is actually a message event that will either have the whole thing or nothing. now I could still defensively code around JSON I mean I’ve agreed with myself in my distributed system that I’m passing JSON messages, so that that solves leaky buffers problem.

The fourth problem was directionality so In the pub/sub code I’m getting a zmq socket object that socket object has methods connect and bind and I just either call connect or bind depending on what I want to do so switching this around if I really did want to change the direction of the connection would mean I would tell my subscriber instead of connect to something I would tell it to bind an endpoint and then in my publisher code I would tell it to connect instead of bind right so I could flip that around without kind of deeply re-architecting all of my code.

we have seen how zeromq solves few of the common problems in distributed messaging you can checkout the various patterns available in the zeromq docs.