Publish and Subscribe

Suppose you have a class Publisher, whose instances can be given objects to publish, and that a number of consumers are potentially interested in consuming that "data feed." The Publisher class will need methods to allow the subscribers to subscribe when they want to start receiving the feed and unsubscribe when they no longer require it.

The consumers, in turn, have to know how the Publisher will transmit the data to them, which will normally be achieved by calling one of its methods. So consumers may need to provide an API to satisfy the requirements of the Publisher. We'll create an example.

For our purposes, we'll write a module that asks for lines of input from the user, and then distributes the lines to any subscribed consumers. The subscriber interface will have subscribe and unsubscribe methods that add and remove items from the publisher's subscriber list. Subscribers must provide a "process" method, which the publisher will call with each new input.

We will have the subscribers print the input string after processing it in basic, but distinguishable ways. In the first example, subscribers print out the uppercase version of the string they've received.

Create pubandsub.py as shown:

class Publisher:

    def __init__(self):
        self.subscribers  = []

    def subscribe(self, subscriber):
        self.subscribers.append(subscriber)

    def unsubscribe(self, subscriber):
        self.subscribers.remove(subscriber)

    def publish(self, s):
        for subscriber in self.subscribers:
            subscriber.process(s)

if __name__ == '__main__':

    class SimpleSubscriber:

        def __init__(self, publisher):
            publisher.subscribe(self)
            self.publisher = publisher

        def process(self, s):
            print(s.upper())

    publisher = Publisher()
    for i in range(3):
        newsub = SimpleSubscriber(publisher)
        line = input("Input {}: ".format(i))
        publisher.publish(line)

The program asks you for three lines of input. The first is echoed in uppercase once, the second twice, and the third three times, because each time through the loop, a new subscriber is subscribed to the publisher.

Input 0: pub
PUB
Input 1: and
AND
AND
Input 2: sub
SUB
SUB
SUB

The Publisher keeps a list of subscribers (which starts out empty). Subscribing an object appends it to the subscriber list; unsubscribing an object removes it. The SimpleSubscriber object takes a publisher as an argument to the __init__() method and immediately subscribes to the publisher.

These same principles can be applied to programs you may already use. For example, a spreadsheet program may have to process spreadsheets where there are multiple graphics based on the data, all of which must be updated as the data changes. One way to arrange that is to enlist the graphics as subscribers to an event stream publisher, which publishes an alert every time any change is made to the data. To avoid unnecessary computing, the event stream publisher might publish the event after a change only when no further changes were made to the data within a fixed (and preferably short) period of time.

We can refine this process further in various ways because it allows very loose coupling between the publisher and the subscriber: neither needs to have advance knowledge of the other, and the connections are created at run-time rather than determined in advance. We like loose coupling in systems design because it's flexible and allows dynamic relationships between objects.

Our initial implementation is defective in a couple of ways. First, there is nothing to stop a given subscriber from being subscribed multiple times. Similarly, there is nothing present to check whether a subscriber requesting unsubscription (code not yet exercised in the main program) is actually subscribed. Passing a nonexistent subscriber would cause the list's remove() method to raise an exception:

>>> [1, 2, 3].remove(4)
Traceback (most recent call last):
  File "<console>", line 1, in <module>
ValueError: list.remove(x): x not in list
>>>

In order to make the message associated with the exception easier to understand, you'll want to trap it or test beforehand for the condition that would cause the exception and then raise your own, more meaningful, exception.

Finally, the original version of our program does not identify which specific subscriber is responsible for an individual message. We want it to identify the culprit though, because that will make the operation of the program easier to understand. Let's revise it so that each subscriber instance takes an additional argument (its name), which it will then use to identify all of its output. Modify pubandsub.py to check for errors and identify subscribers:

class Publisher:

    def __init__(self):
        self.subscribers  = []

    def subscribe(self, subscriber):
        if subscriber in self.subscribers:
            raise ValueError("Multiple subscriptions are not allowed")
        self.subscribers.append(subscriber)

    def unsubscribe(self, subscriber):
        if subscriber not in self.subscribers:
            raise ValueError("Can only unsubscribe subscribers")
        self.subscribers.remove(subscriber)

    def publish(self, s):
        for subscriber in self.subscribers:
            subscriber.process(s)

if __name__ == '__main__':

    class SimpleSubscriber:

        def __init__(self, name, publisher):
            publisher.subscribe(self)
            self.name = name
            self.publisher = publisher

        def process(self, s):
            print(self.name, ":", s.upper())

    publisher = Publisher()
    for i in range(3):
        newsub = SimpleSubscriber("Sub"+str(i), publisher)
        line = input("Input {}: ".format(i))
        publisher.publish(line)

This version of the program doesn't actually trigger any of the newly-added exceptions, but the inclusion of the tests makes our code more robust. The SimpleSubscriber.process() method identifies each output line with the name of the instance that was responsible for it, which can be especially helpful in more complex situations. The code that creates the subscribers generates names such as "Sub0", "Sub1" and so on for the subscribers. You should see output that looks like this:

Input 0: sub
Sub0 : SUB
Input 1: and
Sub0 : AND
Sub1 : AND
Input 2: pub
Sub0 : PUB
Sub1 : PUB
Sub2 : PUB

If we were to write unit tests for this code, we might include an assertRaises() test to ensure that the double-subscription and attempts to remove non-subscribed objects were handled correctly. In the absence of unit tests, we should at least make sure that exceptions will be raised under expected circumstances. We can verify this at an interactive console:

>>> from pubandsub import Publisher
>>> publisher = Publisher()
>>> publisher.unsubscribe(None)
Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "pubandsub.py", line 16, in unsubscribe
    raise ValueError("Can only unsubscribe subscribers")
ValueError: Can only unsubscribe subscribers
>>> publisher.subscribe(None)
>>> publisher.subscribe(None)
Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "pubandsub.py", line 12, in subscribe
    raise ValueError("Multiple subscriptions are not allowed")
ValueError: Multiple subscriptions are not allowed
>>>

Since exceptions appear to be raised under the correct circumstances, we could proceed without modifying the code further, but it's a good idea to copy and paste the interactive session into your source as a doctest. A simple copy-and-paste from the console panel is usually adequate.

So far our program has not tested the non-error branch of the unsubscribe code. We'll perform that test next by restricting the number of subscribers. This can be done either internally (from within the Publisher.subscribe() method, for example) or by truncating the subscription list from the main loop. We're going to do the latter. We'll add a few loops to make sure that the strategy is properly tested. After each new subscription, we'll remove the least recent if the length of the subscription list exceeds three. This will ensure that no input sees more than three responses. Modify pubandsub.py as shown below:

class Publisher:

    def __init__(self):
        self.subscribers  = []

    def subscribe(self, subscriber):
        if subscriber in self.subscribers:
            raise ValueError("Multiple subscriptions are not allowed")
        self.subscribers.append(subscriber)

    def unsubscribe(self, subscriber):
        if subscriber not in self.subscribers:
            raise ValueError("Can only unsubscribe subscribers")
        self.subscribers.remove(subscriber)

    def publish(self, s):
        for subscriber in self.subscribers:
            subscriber.process(s)

if __name__ == '__main__':

    class SimpleSubscriber:

        def __init__(self, name, publisher):
            publisher.subscribe(self)
            self.name = name
            self.publisher = publisher

        def process(self, s):
            print(self.name, ":", s.upper())

    publisher = Publisher()
    for i in range(35):
        newsub = SimpleSubscriber("Sub"+str(i), publisher)
        if len(publisher.subscribers) > 3:
            publisher.unsubscribe(publisher.subscribers[0])
        line = input("Input {}: ".format(i))
        publisher.publish(line)

This code is not much different from the last example, except that there are never more than three responses to any input which indicates that the unsubscribe function is working correctly. Each time the subscriber count exceeds three it is trimmed from the left:

Input 0: sub
Sub0 : SUB
Input 1: and
Sub0 : AND
Sub1 : AND
Input 2: pub
Sub0 : PUB
Sub1 : PUB
Sub2 : PUB
Input 3: more
Sub1 : MORE
Sub2 : MORE
Sub3 : MORE
Input 4: inputs
Sub2 : INPUTS
Sub3 : INPUTS
Sub4 : INPUTS

At present, the publisher requires subscribers to have a "process" method, which it calls to have each subscriber process the published data. This works well enough, but it does constrain the nature of the subscribers. For example, there is no way to subscribe functions, because there is no way to add a method to a function.

Let's modify the program so that it registers the callable method directly instead of registering an instance and then calling a specific method. Our program will then allow any callable to be registered. We'll verify this by defining a simple function and registering it with the publisher before the loop begins. Modify pubandsub.py to allow registration of any callable:

class Publisher:

    def __init__(self):
        self.subscribers  = []

    def subscribe(self, subscriber):
        if subscriber in self.subscribers:
            raise ValueError("Multiple subscriptions are not allowed")
        self.subscribers.append(subscriber)

    def unsubscribe(self, subscriber):
        if subscriber not in self.subscribers:
            raise ValueError("Can only unsubscribe subscribers")
        self.subscribers.remove(subscriber)

    def publish(self, s):
        for subscriber in self.subscribers:
            subscriber.process(s)

if __name__ == '__main__':

    def multiplier(s):
        print(2*s)

    class SimpleSubscriber:

        def __init__(self, name, publisher):
            publisher.subscribe(self)
            self.name = name
            self.publisher = publisher
            publisher.subscribe(self.process)

        def process(self, s):
            print(self.name, ":", s.upper())
            print(self, ":", s.upper())

        def __repr__(self):
            return self.name

    publisher = Publisher()
    for i in range(5):
    publisher.subscribe(multiplier)
    for i in range(6):
        newsub = SimpleSubscriber("Sub"+str(i), publisher)
        line = input("Input {}: ".format(i))
        publisher.publish(line)
        if len(publisher.subscribers) > 3:
            publisher.unsubscribe(publisher.subscribers[0])
        line = input("Input {}: ".format(i))
        publisher.publish(line)

The SimpleSubscriber object now registers its (bound) process method as a callable, and the Publisher.publish() method calls the subscribers directly rather than calling a method of the subscriber. This makes it possible to subscribe functions to the Publisher:

Input 0: pub
pubpub
Sub0 : PUB
Input 1: and
andand
Sub0 : AND
Sub1 : AND
Input 2: sub
subsub
Sub0 : SUB
Sub1 : SUB
Sub2 : SUB
Input 3: and
Sub0 : AND
Sub1 : AND
Sub2 : AND
Sub3 : AND
Input 4: dub
Sub1 : DUB
Sub2 : DUB
Sub3 : DUB
Sub4 : DUB
Input 5: and
Sub2 : AND
Sub3 : AND
Sub4 : AND
Sub5 : AND