zeromq persistence patterns

Question

Who has to manages the persistent in the ZeroMQ?

When we use the ZeroMQ clients in Python language, what are the plug-ins/modules available to manage the persistent?

I would like to know the patterns to use the ZeroMQ.

Answer 1

As far as I know, Zeromq does not have any persistence. It is out of scope for it and needs to be handled by the end user. Just like serializing the message. In C#, I have used db4o to add persistence. Typically I persist the object in its raw state, then serialize it and send it to ZMQ socket. Btw, this was for PUB/SUB pair.

Answer 2

On the application ends you can persist accordingly, for example I've built a persistance layer in node.js which communicated to back-end php calls and via websockets.

The persistance aspect held messages for a certain period of time (http://en.wikipedia.org/wiki/Time_to_live) this was to give clients a chance to connect. I used in-memory data structures but I toyed with the idea of using redis to gain on-disk persistance.

Answer 3

We needed to persist the received messages from a subscriber before processing them. The messages are received in a separate thread and stored on disk, while the persisted message queue is manipulated in the main thread.

The module is available at: https://pypi.org/project/persizmq. From the documentation:

import pathlib

import zmq

import persizmq

context = zmq.Context()
subscriber = context.socket(zmq.SUB)
subscriber.setsockopt_string(zmq.SUBSCRIBE, "")
subscriber.connect("ipc:///some-queue.zeromq")

persistent_dir = pathlib.Path("/some/dir")
storage = persizmq.PersistentStorage(persistent_dir=persistent_dir)

def on_exception(exception: Exception)->None:
    print("an exception in the listening thread: {}".format(exception))

with persizmq.ThreadedSubscriber(
    callback=storage.add_message, subscriber=subscriber, 
    on_exception=on_exception):

    msg = storage.front()  # non-blocking
    if msg is not None:
        print("Received a persistent message: {}".format(msg))
        storage.pop_front()