Event Bus With Topic Wildcards
Problem
Design an in-process event bus. subscribe(pattern, handler) registers a handler and returns a token, and unsubscribe(token) removes it again. Topics are dot-separated, such as order.paid, and in a pattern a * matches exactly one segment, so order.* matches order.paid but neither order nor order.paid.late. publish(topic, payload) calls every matching handler in the order they subscribed and returns how many were called. A handler that raises must not stop the handlers after it; the bus counts the failure in failures and carries on.
Examples
Input: subscribe("order.*", audit), subscribe("order.paid", broken), subscribe("order.paid", ship)
publish("order.paid", "o7")
Output: 3, log = ['audit o7', 'ship o7'], failures = 1
Why: broken raised, but ship still ran after itInput: then unsubscribe(the audit token), publish("order.created", "o8")
Output: 0
Why: the only handler for order.* is gone, and order.paid does not matchInput: then publish("order.paid.late", "o9")
Output: 0
Why: edge case, a * never stretches over two segmentsHints
0 / 3
The bus is the subject and the handlers are observers. Keep them in a structure that remembers insertion order and can drop one entry by its token.
Split both the pattern and the topic on dots once. Two lists of segments match when they have the same length and every pattern segment is either * or equal to the topic segment.
Store token -> (pattern segments, handler) in a dict, which keeps insertion order. publish walks a copy of the values, so a handler may unsubscribe safely, calls each match inside try/except, and counts calls and failures.
Solution
A dict from token to (pattern, handler) gives both requirements at once: Python dicts keep insertion order, so handlers run in subscription order, and a token removes one subscription in O(1) without disturbing the others. Patterns are split into segments when they are registered, so matching a topic is a same-length check plus one comparison per segment, and a * can only ever stand for one segment. Each handler runs inside its own try/except, which isolates a failing observer from the rest, the main promise of an event bus. publish iterates over a copy of the subscriptions so a handler that unsubscribes during delivery cannot break the loop. A publish costs O(s × d) for s subscriptions and d segments.
class EventBus:
def __init__(self):
self.subs, self.next_id, self.failures = {}, 0, 0
def subscribe(self, pattern, handler):
self.next_id += 1
self.subs[self.next_id] = (pattern.split("."), handler)
return self.next_id # the token to unsubscribe with
def unsubscribe(self, token):
self.subs.pop(token, None)
def publish(self, topic, payload):
called, parts = 0, topic.split(".")
for pattern, handler in list(self.subs.values()): # subscription order
if len(pattern) == len(parts) and all(p in ("*", t) for p, t in zip(pattern, parts)):
called += 1
try:
handler(payload)
except Exception:
self.failures += 1 # one bad handler never silences the rest
return called
log = []
bus = EventBus()
audit = bus.subscribe("order.*", lambda p: log.append("audit " + p))
bus.subscribe("order.paid", lambda p: 1 / 0)
bus.subscribe("order.paid", lambda p: log.append("ship " + p))
print(bus.publish("order.paid", "o7"), log, bus.failures) # -> 3 ['audit o7', 'ship o7'] 1
bus.unsubscribe(audit)
print(bus.publish("order.created", "o8")) # -> 0
print(bus.publish("order.paid.late", "o9")) # -> 0Stuck on the idea rather than the code? Observer Pattern covers it.