how to recover dropped messages from publisher when a subscriber is disconnected?

Question

There are:

1. a publisher that binds on localhost:5556 and sends message (topic1, topic2) every 2 seconds.
2. two subscribers that connect to localhost:5556 and respectively receives "topic1" and "topic2"

All works fine when the PUB sends message and subscribers receive them.

My problem is when one of two subscribers stops working. I would like that all the messages dropped by the PUB would be enqueued and send to the 'dead' subscriber when it comes alive. This doesn't happen. When the subscriber reconnect to the publisher, all the messages sent in that period of time (from dead to alive) are dropped away.

I'm trying to coding in python using pyzmq module.

Morover, I noticed that if instead to close the python subscriber process, try to add a time.sleep(10), the PUB enqueues the messages and when the subscriber wakes up, all the messages are sent. This doesn't happen if the process is closed (CTRL+C) and restarted.

But if I try to invert the bind() with the connect() between publisher and subscriber, the application works as I want. But in this case there is a big problem. I couldn't have more subscribers associated to one publisher, because the subscribers bind on different port and the publisher can connect only on one port.

How can I solve it?

Answer 1

"My problem is ..." solvable

ZeroMQ is a smart and high performance Signalling / Messaging Framework built for providing a sevice-layer, composed from Scalable Formal Communication Archetypes. This was for years optimised for low latency, high performance, _almost linear scaling and ease of use of distributed-behaviours of agents, that resemble human behaviours ( like one REQ-uests, the other will REP-ly ).

All this was, by desing, BROKER-LESS - i.e., there is no such thing like a "man-in-the-middle" storage, for resending messages, that were dispatched across the Framework infrastructure during such times an agent was out-of-service.

That does no mean, one cannot make such feature on the application layer, if one indeed needs such set of properties. It would be the designer's role to define what all features to add ( timestamping, message-store, index-counters-flags of NACK/POSACK-ed message deliveries, private bidirectional protocols for requesting missed messages, etc, etc ).

So this part is solvable, yet it does not consume neither resources nor increases latency for the basic framework, which was designed with highest achievable performance and lowest possible latency, using a Zen-of-Zero ( Zero-copy, where possible, Zero-warranty - either deliver the exact copy of the original message or nothing, if corrupted parts were present, etc ) philosophy.

There is even a directly opposing feature in newer ZeroMQ versions, a .setsockopt( ZMQ_CONFLATE, 1 ) method, which permits the Context-instance data-manager to discard all but the most "fresh" message from the individual counterparty's message-queue, thus delivering, upon the next request, but the most recent message and no other. That is very handy for many application scenarios, where "old" messages simply lost their value if not delivered "now" and CONFLATE-mode allows not to move them all through a slow or unstable connection, thus ( indirectly ) prioritising a distribution only of the "newest" state of the world message ( and dropping obsoleted messages also reduces queue-management resources and workloads, doesn't it? ).

In case ZeroMQ concepts are new to your design efforts, be sure to read the Pieter HINTJENS' book, the bible not only of ZeroMQ framework itself, or may read at least about the ZeroMQ main conceptual differences to start with in [ ZeroMQ hierarchy in less than a five seconds ] Section.

Morover, I noticed ...

Yes, this will work fine, because the both distributed Context-instances use separate threads, that remain in contact, even though the Python Interpreter was instructed to .sleep() ( which it does, but not the connection-maintaining signalling efforts of both the sides of the protocol-specific data-pumps )

if I try to invert the .bind() with the .connect()

Oh sure it must behave this way. Just imagine a case with the BBC Radio Station - all people round the world do know the Radio Station to listen to, whereas neither BBC Station emplyee could ever know all the Radio Station listeners, so could never "build" a connection from BBC to all of 'em, round the world ( right due to the reason each of 'em has a unique address, that is a priori unknown to the Broadcaster, who thus cannot setup the same infrastructure from its side ). ZeroMQ PUB-lisher is the same sort of the story - you advertise the "central" address to connect to and everyone, who want and can, will. Not vice versa.

Anyway, enjoy the ZeroMQ worlds in your future design works. It is worth mastering it.

Answer 2

ZeroMQ will not support this without a lot of additional work/code.

If you really need a pub/sub message system where subscribers can come and go with no loss it would be a lot easier to use a system that was built to provide that functionality.

One such system is kafka: https://kafka.apache.org/intro

If you really want to use zeromq there is a projects along those lines recently started. https://github.com/zeromq/dafka