medium_zmq_pub_sub_proxy.py

# This shows how to capture data using a pub-sub proxy
import time

from random import randint
from string import ascii_uppercase as uppercase
from threading import Thread, currentThread

import zmq
from zmq.devices import monitored_queue

from multi_sub_pub.zhelpers import zpipe


# The subscriber thread requests messages starting with
# C then reads and counts incoming messages.

def subscriber_thread_2():
    ctx = zmq.Context.instance()

    # Subscribe to "C"
    subscriber = ctx.socket(zmq.SUB)
    subscriber.connect("tcp://localhost:6001")
    subscriber.setsockopt(zmq.SUBSCRIBE, b"C")

    count = 0
    while count < 5:
        try:
            msg = subscriber.recv()
        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break           # Interrupted
            else:
                raise
        count += 1

    print ("Subscriber {} received {} messages".format(currentThread().getName(), count))

# The subscriber thread requests messages starting with
# A and B, then reads and counts incoming messages.

def subscriber_thread_1():
    ctx = zmq.Context.instance()

    # Subscribe to "A" and "B"
    subscriber = ctx.socket(zmq.SUB)
    subscriber.connect("tcp://localhost:6001")
    subscriber.setsockopt(zmq.SUBSCRIBE, b"A")
    subscriber.setsockopt(zmq.SUBSCRIBE, b"B")

    count = 0
    while count < 5:
        try:
            msg = subscriber.recv()
        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break           # Interrupted
            else:
                raise
        count += 1

    print ("Subscriber {} received {} messages".format(currentThread().getName(), count))

# publisher thread
# The publisher sends random messages starting with A-J:

def publisher_thread():
    ctx = zmq.Context.instance()

    publisher = ctx.socket(zmq.PUB)
    publisher.bind("tcp://*:6000")

    while True:
        string = "%s-%05d" % (uppercase[randint(0,10)], randint(0,100000))
        try:
            publisher.send(string.encode('utf-8'))
        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break           # Interrupted
            else:
                raise

        time.sleep(0.1)         # Wait for 1/10th second

# listener thread
# The listener receives all messages flowing through the proxy, on its
# pipe. Here, the pipe is a pair of ZMQ_PAIR sockets that connects
# attached child threads via inproc. In other languages your mileage may vary:

def listener_thread (pipe):

    # Print everything that arrives on pipe
    while True:
        try:
            print (pipe.recv())
        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break           # Interrupted

# main thread
# The main task starts the subscriber and publisher, and then sets
# itself up as a listening proxy. The listener runs as a child thread:

def main ():

    # Start child threads
    ctx = zmq.Context.instance()
    p_thread = Thread(target=publisher_thread, name="publisher_thread")
    s_thread_1 = Thread(target=subscriber_thread_1, name="subscriber_thread_1")
    s_thread_2 = Thread(target=subscriber_thread_2, name="subscriber_thread_2")
    p_thread.start()
    s_thread_1.start()
    s_thread_2.start()

    pipe = zpipe(ctx)

    subscriber = ctx.socket(zmq.XSUB)
    subscriber.connect("tcp://localhost:6000")

    publisher = ctx.socket(zmq.XPUB)
    publisher.bind("tcp://*:6001")

    l_thread = Thread(target=listener_thread, args=(pipe[1],))
    l_thread.start()

    try:
        monitored_queue(subscriber, publisher, pipe[0], b'pub', b'sub')
    except KeyboardInterrupt:
        print ("Interrupted")

    del subscriber, publisher, pipe

    print("we fire this on interrupt")
    
    ctx.term()


if __name__ == '__main__':
    main()