Pub/sub with PyZMQ: Part 2

Hello everyone! I hope everyone is keeping healthy and staying safe. I hope that you have had a good meal before this, as I will be enticing you with a couple of breakfast analogies for this PyZMQ tutorial :P

This post is a continuation in the "Pub/sub with PyZMQ" series. Part 1 can be found here. This part will build upon what we have discussed in the previous one. Do feel free to revisit Part 1 as a recap or if you are new to this series.

The overview for this post will be as follows:

• Multiple Subscribers to One Publisher
• Single Subscriber to Multiple Publishers
• Using Poller to Prevent Code Blocking

Multiple Subscribers to One Publisher

This pattern is extremely useful for when the same data is required by multiple subscribers from a common publisher. An example use case would be a home automation system that consists of appliances connected to smart plugs. Let's say you wake up in the morning and would like your breakfast to be prepared automatically while you have your shower. You would like for your toaster and coffee maker to be started as soon as you turn on the lights in your bathroom.

Instead of using the server/client pattern in which both the toaster and coffee maker constantly requests for the status of the light switch in your bathroom, the light switch instead tells both the toaster and coffee maker when it is time to turn on.

Using the Thread class from the threading module in Python, we can simulate this scenario. First, we create the two subscribers like so:

We create two threads which simulate the coffee maker and toaster listeners. The listeners are subscribed to the 'light' topic and await for any messages coming from the 'light' publisher.

We can then fire this script up and continue to the next step - creating the single 'light' switch publisher. As you will notice again, the threads will be blocked due to the recv_string() function in both the coffee make and toaster threads.

Once the single_pub.py script has also been run, it will have a sleep for 1 second (reason for sleep here) before the publisher sends the message to the coffee maker and toaster threads. The results can then be seen:

>>> python multi_sub.py
COFFEE MAKER received 'light is ON' from light.
TOASTER received 'light is ON' from light.

NOTE: It is better to create a new socket instance in each thread as demonstrated. This is because the socket objects are not thread-safe. The context however, can be shared. More info here.

Single Subscriber to Multiple Publishers

We can bring in a new element into our coffee maker and toaster analogy. This time, let's include a smart mirror in your bathroom. After you are done showering, you get ready for your day ahead and prepare yourself using the smart mirror. The mirror tells you if your coffee and toast are ready or not.

For this set of code, we shall utilise the multipart messaging of PyZMQ. This way, we can separate the topic and the actual message that we want. Recap on the multipart messaging can be found here.

Let's start off with the single subscriber code:

The main difference between single_sub.py and single_pub.py is the multipart message implementation and that single_sub.py is receiving messages instead of sending.

The multi_pub.py code now utilises the Thread class and instantiates two threads to send messages concurrently.

We now have the topic and status of each appliance sent in separate parts of the messages. Again, the flag zmq.SNDMORE is the one that establishes that there are more parts to the message, and the receiver needs to prepare to receive the remaining messages. The topic is always the first part of the message, which makes it easier for us to distinguish topic from message. This results in the following:

>>>python single_sub.py 
Topic: COFFEE MAKER => {'status': True}
Topic: TOASTER => {'status': True}

Using Poller to Prevent Code Blocking

As you might have noticed in previous examples, all variants of the recv() function blocks the code execution. As such, it can be annoying to have to force close your terminal each time you run your code and the recv() statement is within an infinite loop.

We can overcome this by using the zmq.Poller() object. It mirrors the built-in Python poll interface. Essentially, it checks whether the file descriptor created during the socket creation in the context.socket() function call, has any pending I/O events.

We can instantiate a Poller object like so:

poller = zmq.Poller()

We then have to register a socket to the poller object so that it knows which file descriptors to check for I/O events.

poller.register(socket, zmq.POLLIN)

zmq.POLLIN indicates that the check whether any data is ready to be read or not at the socket's file descriptor.

We can then use our previous single_sub.py and build upon it. Let's call the new script simple_poller.py. We can introduce a check for events with a timeout in the while loop.

Each iteration of the loop, once the timeout limit has been reached for the poll event, the if code block is executed, before continuing into the next iteration of the loop. This way, break/interrupt events can cause the script to exit within this small time frame.

Future Guides

I hope that this second segment of the series has served as a good build on Part 1. As a recap, this post has covered:

• Multiple subscriber to single publisher
• Single subscriber to multiple publisher
• Using poller to prevent code blocking

With these small parts, I hope that you will be able to apply some of these concepts into your own personal projects. For the next post, I do hope to cover the following:

• N-to-N publisher/subscriber messaging
• Implementing a Web Socket forwarder

Here are the links for the codes:

• Multi Sub to Single Pub
• Single Sub to Multi Pub (includes poller)

Thanks for reading! I do hope to learn from your feedback and comments. Also, do share if you have implemented any of these patterns to your projects! I would love to see them in action :D