persizmq 1.0.2

Threaded Subscriber

The threaded subscriber is implemented as persizmq.ThreadedSubscriber. You need to specify a callback which is called upon each received message.

You also need to specify on-exception callback in order to handle exceptions raised in the listening thread.

Example:

import time

import zmq

import persizmq

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

def callback(msg: bytes)->None:
    print("received a message: {}".format(msg))

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

with persizmq.ThreadedSubscriber(callback=callback, subscriber=subscriber, on_exception=on_exception):
    # do something while we are listening on messages...
    time.sleep(10)

Storage

We provide two storage modes for the received messages:

1. persizmq.PersistentStorage: stores messages in a FIFO queue on disk.

2. persizmq.PersistentLatestStorage: solely stores the newest message on disk.

The storage component is passed directly to the threaded subscriber as a callback.

Example:

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 was raised 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()

Filtering

We also provide filtering components which can be chained on the threaded subscriber. The filtering chains are particularly handy if you want to persist only a small amount of messages and ignore the rest.

The filters are implemented in persizmq.filter module.

Example:

import pathlib

import zmq

import persizmq
import persizmq.filter

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 was raised in the listening thread: {}".format(exception))

with persizmq.ThreadedSubscriber(
    lambda msg: storage.add_message(persizmq.filter.MaxSize(max_size=1000)(msg)),
    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()