Let’s start with an event publisher. The NewUserCreated object is an event publisher that hides the details of how we send event details to the hub.

# publish.py
from . import hub # implemention comes later

class NewUserCreated: # event class
    topics = {'new_user_created'} # event can have multiple topics

    @classmethod
    def publish(cls, user_id): 
        hub(cls.topics, user_id)

# how events are triggered in another module
from .publish import NewUserCreated

def signup_handler(username, email, password):
    user = make_new_user(username, email, password)
    NewUserCreated.publish(user.id)
    do_some_other_things()

When a new user signs up, we send the new_user_created topic and event details(user_id) to the hub.
The hub is an instance of the PubSubHub class:

# hub.py
from collections import defaultdict

class PubSubHub:

    def __init__(self):
        self.topic_subscriber_mapper = defaultdict(set) #1

    def subscribe_to(self, *topics): # a decorator that takes topics as parameters
        def real_subscribe(subscriber_function): #2
            for topic in topics:
                self.topic_subscriber_mapper[topic].add(subscriber_function) #3
            return subscriber_function

        return real_subscribe


    def __call__(self, topics, **kwargs): #4
        for topic in topics: #5
            for subscriber in self.topic_subscriber_mapper[topic]:
                subscriber(**kwargs) #6

Explanation

#1 The hub has a mapping of topics to subscribers. A subscriber can register interest in one or more topics using the subscribe_to method of the PubSubHubclass.
#2 The subscribe_to method is a decorator that takes one or more topics as its argument. In Python, a decorator takes a function, does something to it, and returns it. In our case, the subscribe_to method takes one or more topics and the function(subscriber) that is interested in those topics and populates the topic to subscriber maps in the hub. You will see how this decorator is used later on.
#3 This line creates a mapping from a topic to an interested subscriber.
#4 This special “dunder” method makes an instance of the PubSubHub class callabe. That’s why we can send events to the hub using hub(cls.topics, auction_id)defined as part of the NewUserCreated class implemented above.
#5 We get the topics of an event from the event publisher. An event can have more than one topic. This loop ensures that we call subscribers interested in each topic.
#6 We call each subscriber with parameters we received from the event publisher.
Now, let’s look at a subscriber:

# subscribe.py
import emailer

from . import hub

@router.subscribe_to('new_user_created') #1
def email_new_user(user_id):
    user = get_user(user_id)
    context = {
        'subject': 'Welcome!',
        'body': 'Hello{}! Thanks for signing up.'.format(user.first_name),
    }
    emailer.send(user.email, context)

#1 We’re using the subscribe_to decorator on top of our subscriber function. subscribe_to creates a mapping from the new_user_created topic to the email_new_user function.
Note: The code above doesn’t run as is because we haven’t created a router object yet. Also, we need to trigger the decorators somehow. To see a full impolementation, go to [this repo].

Analysis

This implementation is simple and many frameworks are implemented this way. However, our requirements mention that some tasks need to ben run asyncronously and that scalability should not require a rewrite of the whole application.

Async

There are two reasons why we want asynchrony:

1

Some tasks like sending an email take a long time to finish and users don’t like waiting. We could use an async web framework to enable async in our application, but that would only work if all of our long-running tasks are IO-bound. Plus, the most popular Python web frameworks like Django and Flask don’t have async built-in. In short, we don’t want to learn and use a totally new framework which doesn’t even guarantee to address all the async issues.

2

Even if we were okay with waiting for long-running tasks to finish, there’s still a big reason why we want to have asynchrony.
Did you notice that if anything in the above implementation breaks every component of the PubSub architecture is affected? Let’s say we have 3 subscribers for the new_user_created topic. The hub tries to notify the first subscriber, but an exception is thrown. This means that none of the 3 subscribers get notified.
One way of fixing this would be to catch the exceptions. Since we have 3 major components and a wide range of subscribers, the number of exceptions we’d have to hanlde will be big. This can affect performance and reduce code quality.

Scalability

To scale this implementation, we’d have to scale the whole application(add more servers running the same application). As a result, we’d get more publishers, hubs, and subscribers in addition to the main web application. This is a waste of resources because it’s rarely the case that we need to scale those 4 components at once. And, sometimes scaling everything can be dangerous. For example, if our subscribers have too many events to process, then adding more publishers would make matters worse.

How to Fix These Issues

We can fix the weaknesses of the first implementation by running each PubSub component in a separate application or process. However, this would require the introduction of task queues and buffers to our architecture. In the next part of this series, we’ll use tools like Redis, RQ, rqworker, and rqscheduler to do just that.

Article by Mikaeil Orfanian