Pub/Sub, From First Principles: Part 3 — Message Routing with Topics

Part 3 of 4 in the Pub/Sub, From First Principles series.

A single channel doesn't scale past a toy example — every subscriber sees every message. The fix is a routing layer: publishers emit to a named topic, and subscribers express interest in specific topics (or patterns of topics) instead of receiving everything indiscriminately. Think of it as three roles: the publisher emits an event with a topic attached, a routing layer — the exchange, in message-broker terms — decides which subscriber lists match that topic, and each matching subscriber has its own filtered queue that only ever receives what it asked for.

Rather than bolting routing onto the naive synchronous loop from Part 1 — which would reintroduce every bug Part 2 just fixed — this combines topic matching with the per-subscriber queues and workers we already built. The exchange decides who a message goes to; each matched subscriber's own queue and worker still decide when it gets processed.

// topic-pubsub.ts — topic routing combined with Part 2's async workers
type Subscriber = (payload: unknown) => void;

interface TopicWorker {
  subscriber: Subscriber;
  queue: unknown[];
  processing: boolean;
}

class TopicPubSub {
  // Maps a topic pattern (e.g. "order.*") to that pattern's workers.
  // This Map is the exchange: it decides who a message reaches.
  private topics = new Map<string, TopicWorker[]>();

  subscribe(pattern: string, subscriber: Subscriber): void {
    const workers = this.topics.get(pattern) ?? [];
    workers.push({ subscriber, queue: [], processing: false });
    this.topics.set(pattern, workers);
  }

  unsubscribe(pattern: string, subscriber: Subscriber): void {
    const workers = this.topics.get(pattern);
    if (!workers) return;
    const i = workers.findIndex((w) => w.subscriber === subscriber);
    if (i !== -1) workers.splice(i, 1);
    if (workers.length === 0) this.topics.delete(pattern);
  }

  // A pattern segment of "*" matches exactly one topic segment — "order.*"
  // matches "order.created" and "order.failed", but not "order.created.retry"
  // (different segment count) and not "order" alone.
  private matches(pattern: string, topic: string): boolean {
    const patternParts = pattern.split(".");
    const topicParts = topic.split(".");
    if (patternParts.length !== topicParts.length) return false;
    return patternParts.every((p, i) => p === "*" || p === topicParts[i]);
  }

  publish(topic: string, payload: unknown): void {
    for (const [pattern, workers] of [...this.topics]) {
      if (this.matches(pattern, topic)) {
        // Snapshot: protects the loop from mid-dispatch mutations
        for (const worker of [...workers]) {
          worker.queue.push(payload);
          if (!worker.processing) this.processQueue(worker);
        }
      }
    }
  }

  private async processQueue(worker: TopicWorker): Promise<void> {
    worker.processing = true;
    while (worker.queue.length > 0) {
      const message = worker.queue.shift();
      try {
        await Promise.resolve(worker.subscriber(message));
      } catch (err) {
        console.error("Subscriber failed:", err);
      }
    }
    worker.processing = false;
  }
}

A subscriber to order.* catches both order.created and order.failed without needing to subscribe to each individually — and without receiving user.registered at all. And because delivery still goes through per-subscriber queues, everything Part 2 established carries over: a throwing subscriber is isolated to itself, a slow one only backs up its own queue, and per-subscriber ordering holds.

[Publisher] ── "order.created" ──> [Exchange]
                                       │  matches patterns
                          ┌────────────┼────────────────┐
                    "order.*"    "order.created"    "user.*"
                       ✓               ✓               ✗
                       │               │
                  [queue+worker]  [queue+worker]    (not delivered)

This publisher → exchange → filtered-queue shape is RabbitMQ's topic exchange almost literally — its routing keys use the same dot-segmented patterns, with * matching one segment. Kafka takes a coarser approach (whole topics, with consumers subscribing per topic), and SNS filters with message attributes — but the role split is the same everywhere: something in the middle decides who gets what, so publishers and subscribers never have to know about each other.

Edge case: orphaned topics. unsubscribe deletes a pattern's entry once its worker list is empty, so the routing table doesn't accumulate dead patterns forever in a long-running process.

Edge case: messages matching no subscribers. Publishing to a topic nobody's listening for is a silent no-op — it doesn't throw, but it also doesn't tell you the data went nowhere. In production systems this is worth catching deliberately: an unrouted-message log or metric (publishes per topic, publishes with zero matches), so "nothing is consuming this event" is something you find out from a dashboard rather than from a missing side effect weeks later.

Edge case: mid-dispatch mutation. publish() iterates over snapshots ([...this.topics], [...workers]) rather than the live collections. If a subscriber's callback calls unsubscribe() while a dispatch is in flight, mutating the array mid-loop would shift indices under the iterator and silently skip sibling subscribers. Snapshotting makes the in-flight dispatch immune to that: the unsubscribe still takes effect for all future publishes, it just can't corrupt the current one.

Next up: how do you make this setup production-ready — resilient to failures, and tolerant of the things that inevitably go wrong once real traffic hits it? Continue to Part 4: Production Resilience and Guarantees.