Retrieving subscriber count using zeromq PUB/SUB sockets

Question

Is it possible to get the total count of subscribers from a PUB socket in zeromq?

Thanks!

Answer 1

Yes, but unfortunately not via any simple property or method.

You need to use the zmq_socket_monitor() function to connect an inproc service socket to the main socket you want to observe. From there you can listen to events regarding connect/disconnect and keep your own count of subscribers. It may not be a trivial task though, since it seems (to me at least) a bit hard to know when to consider a subscriber (or any remote connection) to be up/down (closed/disconnected/retry etc.). You will have to play around a bit.

The link includes samples and event descriptions.

Answer 2

This is implementation on NodeJS for rep, I think for pub it's the same.

Like Jakob Möllås said, need to use monitor.

const zmq = require('zmq')
        , rep = zmq.socket('rep');

let counter = 0;

rep.bind('tcp://*:5560', function (err) {
    if (err) {
        console.log(err);
    } else {
        console.log("Listening on 5560…");
        rep.monitor(500, 0);
    }
});

// Register to monitoring events
rep.on('connect', function (fd, ep) {
    console.log('connect, endpoint:', ep);
});
rep.on('connect_delay', function (fd, ep) {
    console.log('connect_delay, endpoint:', ep);
});
rep.on('connect_retry', function (fd, ep) {
    console.log('connect_retry, endpoint:', ep);
});
rep.on('listen', function (fd, ep) {
    console.log('listen, endpoint:', ep);
});
rep.on('bind_error', function (fd, ep) {
    console.log('bind_error, endpoint:', ep);
});
rep.on('accept', function (fd, ep) {
    console.log('accept, endpoint:', ep);
    counter++;
});
rep.on('accept_error', function (fd, ep) {
    console.log('accept_error, endpoint:', ep);
});
rep.on('close', function (fd, ep) {
    console.log('close, endpoint:', ep);
});
rep.on('close_error', function (fd, ep) {
    console.log('close_error, endpoint:', ep);
});
rep.on('disconnect', function (fd, ep) {
    console.log('disconnect, endpoint:', ep);
    counter--;
});

// Handle monitor error
rep.on('monitor_error', function(err) {
    console.log('Error in monitoring: %s, will restart monitoring in 5 seconds', err);
    setTimeout(function() { rep.monitor(500, 0); }, 5000);
});

rep.on('message', function (msg) {
    console.log(`recieve: `, JSON.parse(msg));
    rep.send(JSON.stringify({ "status": FAIL, "code": 3666 }));
});

---

console

recieve:  { method: 'login', login: 'a', password: 'b1' }
accept, endpoint: tcp://0.0.0.0:5560
accept, endpoint: tcp://0.0.0.0:5560
login: a, password: b1
recieve:  { method: 'login', login: 'a', password: 'b1' }
disconnect, endpoint: tcp://0.0.0.0:5560
login: a, password: b1
disconnect, endpoint: tcp://0.0.0.0:5560

Answer 3

There doesn't seem to be any direct way. Below is Python code to monitor socket events which can be used to maintain count:

import zmq
from zmq.eventloop import ioloop, zmqstream
import zmq.utils.monitor

class Publication:
    def start(self, port, host):
        context = zmq.Context()
        self._socket = context.socket(zmq.PUB)
        self._socket.bind("tcp://%s:%d" % (host, port))
        self._mon_socket = self._socket.get_monitor_socket(zmq.EVENT_CONNECTED | zmq.EVENT_DISCONNECTED)
        self._mon_stream = zmqstream.ZMQStream(self._mon_socket)
        self._mon_stream.on_recv(self._on_mon)

    def _on_mon(self, msg):
        ev = zmq.utils.monitor.parse_monitor_message(msg)
        event = ev['event']
        endpoint = ev['endpoint']
        if event == zmq.EVENT_CONNECTED:
            pass
            # print(endpoint)
        elif event == zmq.EVENT_DISCONNECTED:
            pass
            #print(endpoint)

One issue is that for some reason CONNECTED event is not firing. Another issue is that even when event fires, you only get endpoint ID which is like tcp://ip:port string. So for multiple clients on same node you get same endpoint ID.