In-process pub/sub with topic wildcards

Q: What if there are thousands of wildcard subscriptions?

Store patterns in a trie keyed by segment, where each node has a child per literal segment plus one * child. Publishing walks the trie along the topic, branching into the * child at each level. The cost then depends on the topic's length and the matches, not on the total number of patterns. MQTT and AMQP topic exchanges use this idea, and also support a multi-level wildcard (#).

mediumScheduling and queues

Build a small in-process publish/subscribe bus, the kind a deploy tool uses so the Slack notifier, the audit logger and the metrics exporter can all react to deploy.started without the deploy code knowing about any of them.

API (Java: handlers are BiConsumer<String, String>, errors is a List<String>)

◈ DIAGRAM

PubSub()
subscribe(pattern: str, handler) -> int     # subscription id, starting at 1
unsubscribe(sub_id: int) -> bool            # False if unknown
publish(topic: str, message) -> int         # handlers that ran without raising
errors: list[str]                           # "sub <id>: <message>" per failed handler

A handler is called as handler(topic, message).

Rules

Topics are dot-separated segments, such as deploy.started. Empty segments (a..b, .a) raise ValueError.
A pattern segment * matches exactly one topic segment: deploy.* matches deploy.started but not deploy or deploy.canary.started. Publishing to a topic containing * raises ValueError.
Delivery is synchronous, and handlers run in subscription order (ascending id), whether they subscribed with an exact topic or a wildcard.
If a handler raises, record it in errors and keep delivering to the rest. The return value counts only handlers that succeeded.
A handler may subscribe or unsubscribe while a message is being delivered. The set of recipients is fixed when publish starts.
Exact-topic subscriptions must be found without scanning every subscription.

h(name) in the examples is a handler that appends "name:topic" to received.

Examples

Example 1

Input: bus.subscribe("deploy.started", h("A")) bus.subscribe("deploy.*", h("B")) bus.subscribe("alert.*", h("C")) bus.publish("deploy.started", "api v42") bus.publish("deploy.finished", "api v42") bus.publish("deploy", "x") bus.unsubscribe(2) bus.publish("deploy.started", "api v43") bus.unsubscribe(2) received

Output: 1 2 3 2 1 0 true 1 false [A:deploy.started, B:deploy.started, B:deploy.finished, A:deploy.started]

Explanation: "deploy" has one segment, so "deploy.*" does not match it. Unsubscribing the same id twice returns false the second time.

Example 2

Input: bus.subscribe("job.failed", broken) # raises RuntimeError("smtp down") bus.subscribe("*.failed", h("pager")) bus.publish("job.failed", "nightly-backup") bus.errors received

Output: 1 2 1 [sub 1: smtp down] [pager:job.failed]

Explanation: The broken email handler does not stop the pager from being notified.

Hints

Approach

Optimal

Keep three maps:

exact: topic to an insertion-ordered map of id -> handler.
wild: id -> (pattern segments, handler) for patterns containing *.
where: id -> pattern, so unsubscribe knows where to look without searching.

publish(topic):

Look up exact[topic] in O(1).
Test each wildcard pattern: same number of segments, and each segment is * or equal.
Sort the matches by id so exact and wildcard subscribers interleave in subscription order.
Call each handler in a try/except, counting successes and recording failures.

Because step 4 iterates over the list built in steps 1 to 3, handlers can change subscriptions without breaking the loop. Deleting an empty topic entry in unsubscribe keeps short-lived topics from leaking memory.

ComplexityTime O(m + W) per publish (m exact matches, W wildcard subs); O(1) subscribe/unsubscribeSpace O(subscriptions)

Python

import itertools
 
 
class PubSub:
    def __init__(self):
        self.exact = {}          # topic -> {sub_id: handler}
        self.wild = {}           # sub_id -> (segments, handler) for patterns containing "*"
        self.where = {}          # sub_id -> pattern, so unsubscribe is O(1)
        self.ids = itertools.count(1)
        self.errors = []
 
    @staticmethod
    def _segments(s):
        parts = s.split(".")
        if any(p == "" for p in parts):
            raise ValueError(f"invalid topic: {s!r}")
        return parts
 
    def subscribe(self, pattern, handler):
        segs = self._segments(pattern)
        sid = next(self.ids)
        if "*" in segs:
            self.wild[sid] = (segs, handler)
        else:
            self.exact.setdefault(pattern, {})[sid] = handler
        self.where[sid] = pattern
        return sid
 
    def unsubscribe(self, sid):
        pattern = self.where.pop(sid, None)
        if pattern is None:
            return False
        if sid in self.wild:
            del self.wild[sid]
        else:
            subs = self.exact[pattern]
            del subs[sid]
            if not subs:
                del self.exact[pattern]
        return True
 
    def publish(self, topic, message):
        segs = self._segments(topic)
        if "*" in segs:
            raise ValueError("cannot publish to a wildcard topic")
        matches = list(self.exact.get(topic, {}).items())
        for sid, (pat, handler) in self.wild.items():
            if len(pat) == len(segs) and all(p == "*" or p == s for p, s in zip(pat, segs)):
                matches.append((sid, handler))
        matches.sort(key=lambda m: m[0])          # delivery in subscription order
        delivered = 0
        for sid, handler in matches:              # a snapshot: handlers may (un)subscribe safely
            try:
                handler(topic, message)
                delivered += 1
            except Exception as exc:              # one broken subscriber must not starve the rest
                self.errors.append(f"sub {sid}: {exc}")
        return delivered

Follow-up questions

Deliver asynchronously on a worker pool. What happens to ordering per subscriber?
Add a multi-level wildcard # that matches zero or more segments.
A subscriber is too slow and messages pile up. What are your options?

Frequently asked questions

Store patterns in a trie keyed by segment, where each node has a child per literal segment plus one * child. Publishing walks the trie along the topic, branching into the * child at each level. The cost then depends on the topic's length and the matches, not on the total number of patterns. MQTT and AMQP topic exchanges use this idea, and also support a multi-level wildcard (#).

Idiomatic Go gives each subscriber a buffered channel instead of a callback: Subscribe(pattern) (<-chan Msg, func()), where the returned function unsubscribes. A slow subscriber then fills its own buffer, and you choose whether Publish blocks, drops, or disconnects it. Guard the maps with a sync.RWMutex. Across processes the same API sits on NATS, Redis pub/sub or Kafka, which adds durability and delivery guarantees an in-process bus does not have.