Skip to content

Event Bus

Intent

Let a component announce that something happened, by topic, without knowing who cares, and let any number of components react without knowing who announced it. The bus in the middle owns the subscriptions and the delivery policy (inline or on a worker thread), so publisher and subscribers share nothing but a topic name and an event shape.

When to use and when not to

Use it when

  • One fact has many independent consequences that keep growing: an order placed must reserve stock, email the customer and update analytics, and next quarter feed fraud scoring, without checkout changing.
  • Publisher and subscribers live in different modules or teams; the bus points both at the event shape instead of at each other.
  • A reaction may be slow or flaky (a ledger write, a third-party call) and the publisher must neither wait for it nor fail because of it.

Leave it out when

  • The publisher needs the answer. A bus delivers and forgets; a request with a result is a function call or a Pipeline.
  • There is one subscriber and it will stay that way; a direct dependency is easier to follow in a debugger.
  • Order or atomicity matters across reactions (reserve stock, then charge, roll back on failure): that is a workflow, a Mediator or a saga.
  • Delivery must survive a crash. An in-process bus is at-most-once; durable topics with offsets are the pub/sub problem page.

Structure

Six roles: the event, the bus that routes by topic, the subscription handle, the delivery unit a worker drains, and two subscribers that never meet.

classDiagram
    class Event {
        +topic: str
        +payload: Mapping~str,Any~
        +seq: int
        +published_at: float
    }
    class EventBus {
        -_subscribers: Dict~str,List~
        -_queue: Queue~Delivery~
        -_workers: List~Thread~
        -_lock: Lock
        +unrouted: int
        +subscribe(pattern: str, handler: Handler) Subscription
        +unsubscribe(subscription: Subscription) None
        +publish(topic: str, payload: Mapping) int
        +subscriber_count(topic: str) int
        +join() None
        +close() None
    }
    class Subscription {
        +pattern: str
        +handler: Handler
        +cancel() None
    }
    class Delivery {
        +handler: Handler
        +event: Event
    }
    class InventoryService {
        -_reserved: Dict~str,int~
        +on_order_placed(event: Event) None
        +reserved() Dict~str,int~
    }
    class EmailNotifier {
        +sent: List~str~
        +on_order_event(event: Event) None
    }
    EventBus o-- "many" Subscription : by pattern
    EventBus ..> Event : creates and routes
    EventBus ..> Delivery : queues for workers
    Subscription --> EventBus : cancel
    InventoryService ..> EventBus : subscribes order.placed
    EmailNotifier ..> EventBus : subscribes order.*

EventBus is the only object that knows both sides: publishers call publish(topic, payload), subscribers hand over a callable and get a Subscription to cancel. Delivery lets a worker thread carry exactly one handler call. InventoryService and EmailNotifier import the bus and the event, nothing else.

Worker-thread delivery: publish returns once the deliveries are queued, and a handler that raises is reported without stopping the others.

sequenceDiagram
    autonumber
    participant P as Checkout
    participant B as EventBus
    participant Q as Queue
    participant W as Worker
    participant I as InventoryService
    participant A as AuditLog
    P->>B: publish(order.placed, payload)
    activate B
    Note over B: seq += 1, match order.placed and order.*
    B->>Q: put Delivery(inventory, event)
    B->>Q: put Delivery(audit, event)
    B-->>P: 2 handlers matched
    deactivate B
    W->>Q: get() blocks until a delivery arrives
    Q-->>W: Delivery(inventory, event)
    activate W
    W->>I: on_order_placed(event)
    I-->>W: stock reserved
    W->>A: audit(event)
    A-->>W: RuntimeError
    Note over W: on_error(handler, event, exc), then carry on
    W->>Q: task_done()
    deactivate W

Canonical example in Python

The event and the small types around it come first (code/patterns/event_bus.py, tested by code/patterns/tests/test_event_bus.py):

code/patterns/event_bus.py — the event, the handler types, the delivery and the subscription
@dataclass(frozen=True, slots=True)
class Event:
    """What travels: an immutable record, so every handler may keep it without copying."""

    topic: str
    payload: Mapping[str, Any]
    seq: int
    published_at: float


type Handler = Callable[[Event], None]
type ErrorHandler = Callable[[Handler, Event, Exception], None]


def log_handler_error(handler: Handler, event: Event, exc: Exception) -> None:
    """Default error policy: record and move on. Inject another to count, alert or re-raise."""
    log.warning("handler %r failed on %s #%d: %r", handler, event.topic, event.seq, exc)


@dataclass(frozen=True, slots=True)
class Delivery:
    """One unit of work for a worker thread: this handler, this event."""

    handler: Handler
    event: Event


class Subscription:
    """The handle ``subscribe`` returns. ``cancel`` is idempotent and safe from inside a handler."""

    def __init__(self, bus: EventBus, pattern: str, handler: Handler) -> None:
        self._bus = bus
        self.pattern = pattern
        self.handler = handler

    def cancel(self) -> None:
        self._bus.unsubscribe(self)

Event is frozen and carries what the bus stamps on it: a sequence number and a time from the injected Clock. A Handler is any callable, and so is the error policy.

The bus:

code/patterns/event_bus.py — routing, delivery policy, error isolation, lifecycle
class EventBus:
    """Topic-based pub/sub with synchronous or worker-thread delivery.

    ``_lock`` protects ``_subscribers``, ``_seq``, ``_unrouted`` and ``_closed``. The
    handler list is copied under the lock and handlers run outside it, so a slow
    handler never blocks ``subscribe`` and a handler may cancel its own subscription.
    With ``workers == 0``, ``publish`` runs every handler before returning. With
    ``workers >= 1`` it enqueues one ``Delivery`` per handler and returns at once;
    daemon threads drain the ``queue.Queue`` in FIFO order (one worker keeps events
    in publish order, more workers trade that order for throughput). A handler that
    raises goes to ``on_error``; the remaining handlers, and the worker, carry on.
    """

    def __init__(
        self,
        clock: Clock,
        *,
        workers: int = 0,
        on_error: ErrorHandler = log_handler_error,
    ) -> None:
        if workers < 0:
            raise ValidationError("workers cannot be negative")
        self._clock = clock
        self._on_error = on_error
        self._subscribers: defaultdict[str, list[Subscription]] = defaultdict(list)
        self._lock = threading.Lock()
        self._seq = 0
        self._unrouted = 0
        self._closed = False
        self._queue: queue.Queue[Delivery | None] = queue.Queue()
        self._workers = [
            threading.Thread(target=self._work, name=f"EventBus-worker-{n}", daemon=True)
            for n in range(workers)
        ]
        for worker in self._workers:
            worker.start()

    def subscribe(self, pattern: str, handler: Handler) -> Subscription:
        """``pattern`` is a topic (``order.placed``) or a glob (``order.*``)."""
        if not pattern:
            raise ValidationError("a subscription needs a topic or a pattern")
        subscription = Subscription(self, pattern, handler)
        with self._lock:
            self._subscribers[pattern].append(subscription)
        return subscription

    def unsubscribe(self, subscription: Subscription) -> None:
        with self._lock:
            kept = [s for s in self._subscribers.get(subscription.pattern, ()) if s is not subscription]
            if kept:
                self._subscribers[subscription.pattern] = kept
            else:
                self._subscribers.pop(subscription.pattern, None)

    def subscriber_count(self, topic: str) -> int:
        return len(self._handlers_for(topic))

    @property
    def unrouted(self) -> int:
        """Events published to a topic nobody listens to: a metric worth alarming on."""
        with self._lock:
            return self._unrouted

    def publish(self, topic: str, payload: Mapping[str, Any] | None = None) -> int:
        """Route one event to every matching handler; returns how many handlers matched."""
        with self._lock:
            if self._closed:
                raise InvalidStateError("event bus is closed")
            self._seq += 1
            event = Event(topic, dict(payload or {}), self._seq, self._clock.now())
        handlers = self._handlers_for(topic)
        if not handlers:
            with self._lock:
                self._unrouted += 1
            return 0
        for handler in handlers:
            if self._workers:
                self._queue.put(Delivery(handler, event))
            else:
                self._deliver(Delivery(handler, event))
        return len(handlers)

    def _handlers_for(self, topic: str) -> list[Handler]:
        with self._lock:
            exact = [s.handler for s in self._subscribers.get(topic, ())]
            wild = [
                s.handler
                for pattern, subs in self._subscribers.items()
                if "*" in pattern and fnmatchcase(topic, pattern)
                for s in subs
            ]
        return exact + wild

    def _deliver(self, delivery: Delivery) -> None:
        """Error isolation: one failing handler cannot stop the others or kill a worker."""
        try:
            delivery.handler(delivery.event)
        except Exception as exc:
            try:
                self._on_error(delivery.handler, delivery.event, exc)
            except Exception:
                log.exception("on_error itself failed for %s #%d", delivery.event.topic, delivery.event.seq)

    def _work(self) -> None:
        while (delivery := self._queue.get()) is not None:
            self._deliver(delivery)
            self._queue.task_done()
        self._queue.task_done()

    def join(self) -> None:
        """Block until every queued delivery has been handled (tests, graceful shutdown)."""
        self._queue.join()

    def close(self) -> None:
        """Refuse new events, let the workers drain what is queued, then stop them."""
        with self._lock:
            if self._closed:
                return
            self._closed = True
        for _ in self._workers:
            self._queue.put(None)
        for worker in self._workers:
            worker.join()

    def __enter__(self) -> EventBus:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.close()

Five decisions to say out loud:

  • Topics are strings, matched exactly or by glob. order.* catches order.refunded before anyone has written it; an exact topic stays a dict lookup. Observer has no topic: you subscribe to an object.
  • Copy under the lock, call outside it. A slow handler never blocks subscribe, and a handler may cancel itself mid-dispatch.
  • The delivery policy belongs to the bus. With workers=0 handlers run inline, in subscription order, before publish returns. With workers=1 each handler call becomes a Delivery on a queue.Queue, publish returns at once and one daemon thread drains in FIFO order, so events stay in publish order; more workers trade order for throughput, as partitions do.
  • Error isolation is the contract. _deliver catches everything, hands it to on_error and moves on: a failing audit handler cannot stop inventory from reserving or kill the worker.
  • join and close. queue.join blocks until every delivery is handled; close refuses new events, then queues one sentinel per worker behind the pending deliveries, so FIFO order drains them before any worker stops.

The subscribers import Event and nothing else, and each guards its own state because a worker bus calls them off-thread:

code/patterns/event_bus.py — two subscribers that never meet
class InventoryService:
    """Reserves stock when an order is placed; knows nothing about checkout or email.

    ``_lock`` protects ``_reserved``: with a worker-thread bus the handler runs off-thread.
    """

    def __init__(self) -> None:
        self._reserved: dict[str, int] = {}
        self._lock = threading.Lock()

    def on_order_placed(self, event: Event) -> None:
        lines: Mapping[str, int] = event.payload["lines"]
        with self._lock:
            for sku, quantity in lines.items():
                self._reserved[sku] = self._reserved.get(sku, 0) + quantity

    def reserved(self) -> dict[str, int]:
        with self._lock:
            return dict(self._reserved)


class EmailNotifier:
    """Subscribed to ``order.*``: one handler for every order event, present and future."""

    def __init__(self) -> None:
        self._sent: list[str] = []
        self._lock = threading.Lock()

    def on_order_event(self, event: Event) -> None:
        with self._lock:
            self._sent.append(f"{event.topic} -> {event.payload['order_id']}")

    @property
    def sent(self) -> list[str]:
        with self._lock:
            return list(self._sent)

Running python -m patterns.event_bus prints:

--- synchronous bus: handlers run on the publisher's thread, failures are isolated ---
  order.placed   -> 3 handler(s)
  order.paid     -> 1 handler(s)
  payment.failed -> 0 handler(s)
  inventory reserved: {'sku-1': 2, 'sku-2': 1}
  emails: ['order.placed -> o-100', 'order.paid -> o-100']
  isolated failures: ["order.placed #1: RuntimeError('audit store unavailable')"]
  unrouted events: 1
  after emails.cancel(): order.shipped -> 0 handler(s)
--- worker-thread bus: publish returns at once, delivery happens on the worker ---
  publisher on MainThread continued; deliveries so far: 0
  after join(): 2 deliveries on EventBus-worker-0; worker survived 2 failures
  after close(): InvalidStateError: event bus is closed
--- pythonic: a defaultdict of callables is the whole bus ---
  lambda saw order.placed #1
  routed to 1 handler(s)

Pythonic variant

Strip the threads, wildcards, cancellation and counters and what remains is a dict of lists and a loop:

code/patterns/event_bus.py — the ten-line bus
type Subscribe = Callable[[str, Handler], None]
type Publish = Callable[[Event], int]


def simple_bus() -> tuple[Subscribe, Publish]:
    """The whole pattern in ten lines: a dict of lists and a loop.

    No threads, wildcards, cancellation, isolation or metrics; enough inside one
    module, in a test, or until the second team subscribes.
    """
    subscribers: defaultdict[str, list[Handler]] = defaultdict(list)

    def subscribe(topic: str, handler: Handler) -> None:
        subscribers[topic].append(handler)

    def publish(event: Event) -> int:
        handlers = list(subscribers[event.topic])
        for handler in handlers:
            handler(event)
        return len(handlers)

    return subscribe, publish
  • defaultdict(list) of callables is the whole data structure; list(...) before the loop lets a handler subscribe another without corrupting the iteration.
  • What you give up: a raising handler stops the rest, nothing runs off-thread, and an event nobody listens to vanishes uncounted.
  • The async form is the same bus over asyncio.Queue with a task per worker; blinker.Signal is the packaged version of the synchronous one.
Reach for When
simple_bus() or a dict of lists One module, a test, a CLI
A Signal per event type (Observer page) A few known events, no topics, subscribers are objects
EventBus(workers=0) Wildcards, counters and isolation, and the publisher can afford to wait
EventBus(workers=n) Slow or flaky subscribers behind a publisher on the request path
A broker (Kafka, Redis Streams, SQS) Across processes, with durability, replay and consumer groups

Draw the class diagram, then say "in Python I would start from a dict of callables and add the bus class the day I need isolation, a worker or a second team subscribing".

Real-world usage

  • Django signals. post_save.connect(handler, sender=Order) is a registry per signal with sender as the topic, delivered synchronously on the saving thread; send_robust is the error-isolated form.
  • blinker (Flask's signals): Namespace().signal("order-placed"), connect and send, weak references by default so a forgotten receiver does not leak.
  • logging is a bus keyed by logger name: a record for app.orders reaches the handlers of app.orders, app and the root, a prefix-wildcard topic hierarchy with propagate as the cancel.
  • Elsewhere: Guava EventBus and AsyncEventBus(executor) (the workers switch), Spring ApplicationEventPublisher, DOM CustomEvent, Qt signals.
Looks like Event Bus How to tell them apart
Observer You subscribe to an object (feed.subscribe(watchlist)), which holds its observers and defines the notification. With a bus you subscribe to a topic string on a third party; publisher and subscriber never reference each other, and one bus carries many unrelated event types: M publishers to N subscribers, not one subject to N observers.
Mediator A mediator coordinates: it knows its colleagues, encodes the workflow (stop the elevator, open the doors, reset the button), calls them in order and waits for answers. A bus understands nothing about its events; it routes and forgets. Handlers that depend on each other's order are a mediator without the name.
Pipeline and Middleware Serial and ordered, returns a result, any stage may stop the rest. A bus is fan-out: every subscriber gets the event, none can stop the others, nothing comes back.
Message queue, pub/sub system The same idea across processes, with durability, offsets, consumer groups and at-least-once delivery.
Command A request to do one thing, addressed to one receiver, often undoable. An event states that something happened and is addressed to nobody.

Where it appears in LLD problems

Interview tips

Interview tip

Draw the bus as a third box and name the two things it owns: the subscription table keyed by topic and the delivery policy. Then the three sentences that separate it from its neighbours: subscribers know a topic, not an object (Observer); the bus routes without understanding the events (Mediator); it is in-process and at-most-once (a message queue is neither).

Common mistake

Treating the bus as a function call with extra steps: publishing, then assuming the subscriber has already run (with a worker it has not), or expecting something back. A publisher that depends on a subscriber has hidden coupling nobody can trace. Runner-up: a dispatch loop without error isolation, where one raising handler silences the rest or kills the worker.