PubSub pattern for twisted

twisted_pubsub.py

from twisted.internet.defer import DeferredQueue


class PubSub(object):

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

    def subscribe(self, queue=None):
        if queue is None:
            queue = DeferredQueue()
        self.queues.append(queue)
        return queue

    def unsubscribe(self, queue):
        self.queues.remove(queue)

    def publish(self, data):
        for queue in self.queues:
            reactor.callLater(0, queue.put, data)