Publish / subscribe — heyrian.dev

Definition

A publish-subscribe channel delivers a copy of each message published to it to every receiver that has subscribed, so the sender announces once and never needs to know who is listening.

After Hohpe and Woolf, Enterprise Integration Patterns (2003), Publish-Subscribe Channel

01Save without knowing every listener

A direct call is a good first choice.

A writing app’s editor saves drafts. When a save succeeds, the activity panel lists it and the search index adds the draft, so a writer can find it. Here is the first version, a fragment: Save holds both listeners and calls them itself (the activity panel and the search index are the app’s own types).

editor.ts

// Save calls every part of the app that cares about a saved draft.
class Editor {
  activity: ActivityPanel;
  search: SearchIndex;
  constructor(activity: ActivityPanel, search: SearchIndex) {
    this.activity = activity;
    this.search = search;
  }
  save(draft: string) {
    this.activity.add(draft);
    this.search.index(draft);
  }
}

A direct call such as activity.add(draft) is a good first choice: Save can see whether it worked and when it finished. The design gets harder when consumers are optional and change over time. The activity panel exists only while it is open. The search index runs in the background and would rather catch up in batches than on every keystroke’s save. Next month a backup service wants saves too. Save then owns a list of listeners as well as the draft, waits for each of them, and fails when one does.

Publish / subscribe puts a broker between the code that announces and the code that reacts. With it, Save publishes draft.saved after storage succeeds. A topic names the kind of event; a broker matches that name to subscriptions. Save knows the event contract, but not the activity panel or the search index.

The broker routes copies. Each subscriber owns its subscription and response. This separates the publisher from its listeners, while leaving the topic name and payload schema as shared contracts.

02Route a copy to each inbox

Matching subscribers receive independent copies.

The broker keeps one subscription per subscriber: the topics it wants and its own inbox. Publishing leaves a copy in every inbox whose topics match and returns; nobody has read anything yet. Draining hands one subscriber its unread copies, oldest first, and empties its inbox. The editor holds only the broker. This is the whole program the lab runs: the broker, the editor, and two subscribers. The activity panel wants saves and reads at once; the search index wants saves and deletes, in one subscription, and catches up after the writer has deleted the pricing page and restored it from the trash.

broker.ts

// A message on a topic: what happened, and to which draft.
type Message = { topic: string; draft: string };

// A subscription: the topics it wants, and its own inbox of unread copies.
type Subscription = { id: number; topics: string[]; inbox: Message[] };

// The broker: publishers and subscribers know it, never each other.
class Broker {
  nextId = 1;
  subscriptions: Subscription[] = [];

  subscribe(topics: string[]): number {
    const id = this.nextId++;
    this.subscriptions.push({ id, topics, inbox: [] });
    return id;
  }

  // Leaves a copy in every subscription that wants this topic.
  publish(topic: string, draft: string): number {
    let copies = 0;
    for (const subscription of this.subscriptions) {
      if (subscription.topics.includes(topic)) {
        subscription.inbox.push({ topic, draft });
        copies++;
      }
    }
    return copies;
  }

  // Hands one subscriber its unread copies, oldest first, and empties its inbox.
  drain(id: number): Message[] {
    const subscription = this.subscriptions.find((s) => s.id === id);
    if (!subscription) return [];
    const messages = subscription.inbox;
    subscription.inbox = [];
    return messages;
  }
}

// The publisher: it knows the broker and the topic names, nothing else.
class Editor {
  broker: Broker;
  constructor(broker: Broker) {
    this.broker = broker;
  }
  save(draft: string) {
    this.broker.publish('draft.saved', draft);
  }
  remove(draft: string) {
    this.broker.publish('draft.deleted', draft);
  }
}

const broker = new Broker();
const editor = new Editor(broker);

// Two subscribers. Each knows the broker and its topics, not the editor.
const activity = broker.subscribe(['draft.saved']);
const search = broker.subscribe(['draft.saved', 'draft.deleted']);

// The writer saves the pricing page; the activity panel reads at once.
editor.save('pricing');
const feed = broker.drain(activity);

// Then deletes the page, and restores it from the trash.
editor.remove('pricing');
editor.save('pricing');

// The search index catches up later, in one batch.
const indexed = new Set<string>();
for (const message of broker.drain(search)) {
  if (message.topic === 'draft.saved') indexed.add(message.draft);
  else indexed.delete(message.draft);
}
const found = indexed.has('pricing');
const later = broker.drain(activity);

console.log(`activity: read ${feed.length}, then ${later.length}`);
console.log(`search index has pricing: ${found}`);

Who owns what
Role	Responsibility	What it handles
Publisher	Announce a saved draft.	The editor decides when the fact is true. It holds the broker and a topic name, and no subscriber.
Broker	Match topic subscriptions.	It routes each publication to current subscribers, and holds each copy until its subscriber reads it.
Subscriber	Read its own copy.	One panel does not consume another panel’s message, and each reads when it is ready.

Adding the backup service is a subscribe call where the service starts; the editor does not change. Neither does the activity panel when the search index moves to a schedule of its own.

What a production broker addsBounded inboxes, a report, validation and cleanup

The same broker, with a delivery contract. Each subscription’s inbox holds at most a fixed number of messages; a full inbox drops the new copy for that subscriber and keeps its older messages, and another matching inbox may still accept its copy. Publishing returns a report of which subscriptions accepted and which dropped: accepted means enqueued, not processed. Topics must be non-empty and match exactly, with no wildcards. A new subscriber starts empty, unsubscribing discards that subscription’s unread messages, and retrying a publication can add a duplicate where there is room: this broker has no deduplication or replay. Its fields are #private, so a subscriber holding the broker can only call its methods, and a snapshot is a copy.

broker.ts

export type Message = Readonly<{ topic: string; draft: string }>;

export type Subscription = { id: number; topics: string[]; inbox: Message[] };

export type PublishReport = {
  accepted: number[]; // Enqueued, not yet read by anyone.
  dropped: number[]; // This subscription's inbox was already full.
};

export class Broker {
  #capacity: number;
  #nextId = 1;
  #subscriptions: Subscription[] = []; // In registration order.

  // Each inbox holds at most `capacity` unread messages.
  constructor(capacity: number) {
    if (!Number.isSafeInteger(capacity) || capacity < 1) {
      throw new Error('capacity must be a positive integer');
    }
    this.#capacity = capacity;
  }

  subscribe(topics: readonly string[]): number {
    if (topics.length === 0) {
      throw new Error('a subscription needs at least one topic');
    }
    if (topics.includes('')) throw new Error('topic must not be empty');
    const id = this.#nextId++;
    this.#subscriptions.push({ id, topics: [...topics], inbox: [] });
    return id;
  }

  publish(topic: string, draft: string): PublishReport {
    if (topic === '') throw new Error('topic must not be empty');
    const report: PublishReport = { accepted: [], dropped: [] };
    for (const subscription of this.#subscriptions) {
      if (!subscription.topics.includes(topic)) continue;
      if (subscription.inbox.length === this.#capacity) {
        // Full: drop the new copy, keep the older unread ones.
        report.dropped.push(subscription.id);
      } else {
        subscription.inbox.push({ topic, draft });
        report.accepted.push(subscription.id);
      }
    }
    return report;
  }

  drain(id: number): Message[] {
    const subscription = this.#subscriptions.find((s) => s.id === id);
    if (!subscription) throw new Error('unknown subscription');
    const messages = subscription.inbox;
    subscription.inbox = [];
    return messages;
  }

  // Ends a subscription, and discards its unread messages.
  unsubscribe(id: number): boolean {
    const index = this.#subscriptions.findIndex((s) => s.id === id);
    if (index === -1) return false;
    this.#subscriptions.splice(index, 1);
    return true;
  }

  // A copy for inspection; it reads nothing.
  snapshot(): Subscription[] {
    return this.#subscriptions.map((s) => ({
      id: s.id,
      topics: [...s.topics],
      inbox: [...s.inbox]
    }));
  }
}

This in-memory broker has no persistence, acknowledgment or retry. The section on choosing a delivery contract, below, says what a real one has to decide.

03Predict what the index ends with

The search index catches up later. What does it end with?

Experiment

One editor, two subscribers

broker.ts

// The search index subscribes to each topic on its own:
const saved = broker.subscribe(['draft.saved']);
const deleted = broker.subscribe(['draft.deleted']);

// …and catches up on one inbox, then the other:
for (const message of broker.drain(saved)) indexed.add(message.draft);
for (const message of broker.drain(deleted)) indexed.delete(message.draft);

Step through it in the lab

Reading the TypeScriptHand-offs, includes, and a Set

drain returns the inbox array and gives the subscription a new empty one, so the caller owns what it was handed: a later publish pushes into the new array, never into messages already read. topics.includes(topic) compares whole strings, so draft.saved does not match draft or Draft.saved.

The search index is a Set: adding a draft twice keeps one entry, and deleting one it does not hold does nothing. The fields are public so the lab’s memory view can show every inbox; the production broker makes them #private.

04Meet it in your language

You already publish to names you never call.

Node’s BroadcastChannel (from node:worker_threads; Node 18 and later also make it a global) is publish / subscribe across worker threads: every channel object opened with the same name hears what another posts. The name is the topic and Node is the broker. A common use is cache invalidation: each worker caches prices, and a worker that changes one posts the SKU so the others forget their copy. The bug readers meet: a channel object never receives its own messages, so updatePriceBefore leaves the old price in the cache of the one worker that just changed it. The fix: the publisher updates its own state directly, then tells everyone else.

prices.ts

import { BroadcastChannel } from 'node:worker_threads';

type PriceStore = { save(sku: string, price: number): Promise<void> };

// Each worker thread keeps its own cache, and opens the same named channel.
const cache = new Map<string, number>();
const prices = new BroadcastChannel('prices');
prices.unref(); // an open channel would otherwise keep the thread alive

// Another thread changed a price: forget ours.
prices.onmessage = (message) => {
  const { sku } = (message as { data: { sku: string } }).data;
  cache.delete(sku);
};

// Before: relies on the message to clear this thread's cache too.
export async function updatePriceBefore(db: PriceStore, sku: string, price: number) {
  await db.save(sku, price);
  prices.postMessage({ sku });
}

// After: the channel never delivers to the object that posted, so the
// publisher clears its own copy, then tells everyone else.
export async function updatePrice(db: PriceStore, sku: string, price: number) {
  await db.save(sku, price);
  cache.delete(sku);
  prices.postMessage({ sku });
}

export function cachedPrice(sku: string) {
  return cache.get(sku);
}

Messages arrive as later tasks, not during postMessage, and a worker that opens the channel afterwards hears nothing posted before: on start, read the current state rather than wait for an announcement. An open channel also keeps its thread running; unref() or close() lets it exit.

05Choose a delivery contract

Ask who needs the result, and what a miss means.

Publish / subscribe fits when independent consumers may join or leave without changing the publisher. It makes reactions less visible at the call site, so name topics clearly and decide whether dropped or missed messages can be recovered.

Start from the caller’s requirement
Requirement	Starting point	Reason
Several optional views refresh independently	Publish / subscribe	Each current subscriber gets its own delivery opportunity.
One required step must report success to Save	Direct call or facade	The caller can handle the operation’s result and partial failure.
One worker should take each job	Work queue or consumer group	Group members share jobs instead of each receiving a copy.
An offline consumer must recover missed events	Durable queue or event log	Retention, acknowledgment, retry, and ordering become explicit.

The pattern leaves decisions open, and each broker answers them differently. What does a full inbox do: drop the new copy (the production broker), drop the oldest, or make the publisher wait? Whose order is kept: one subscription’s, as the experiment showed, or nothing at all across subscribers? Does a late subscriber get what was published before it arrived? And is a copy delivered at most once, or at least once, so that a subscriber must expect duplicates? Write the answers down where the topic is defined.

What goes where: the broker routes copies and holds them until they are read; it never decides what a subscriber does with one. A rule about how several participants respond together belongs to a Mediator, which coordinates them under a policy. What one subscriber does with its copy stays in that subscriber.

When delivery needs stronger guaranteesRecovery and commit boundaries

Bound subscriber count and payload size as well as message count; choose what a slow consumer should do.
Draining removes messages before application work runs, so consumer failure needs its own retry rule.
Late consumers need a state read or a retained event log if they must catch up.
A database save and a remote publish can fail between steps. A transactional outbox records the event with the save; retries still need duplicate handling. See AWS’s outbox guidance.

Recognize the same shape in messaging APIsTheir delivery promises differ

Redis Pub/Sub documents at-most-once delivery; disconnected subscribers cannot replay missed messages through plain Pub/Sub. In Core NATS, ordinary subscriptions receive copies, while a queue group sends a message to one group member. Neither guarantee should be inferred from this local lab. See the Redis documentation, NATS Pub/Sub, and NATS queue groups.

Its neighbours: Observer notifies a subject’s own listeners, synchronously, while the change happens; publish / subscribe routes by topic into independent inboxes through a broker that neither side owns. A work queue hands each message to one consumer instead of a copy to each. Command objects are often what a topic carries, when a message asks for work rather than reporting a fact.

If one stable caller has one required reaction, keep the direct call. A topic would hide that dependency without removing it.

06Make the call

Keep required work in the result path.

The pattern fits optional reactions. If Save must know that a step succeeded, its result belongs in a caller-visible operation.

Before you open the answer below, explain the boundary in your own words: what can Save stop knowing about each listener, what does the broker keep for them, and what must Save still see for itself?

One way to say itWho announces, who holds, and who decides

Save announces a fact, a draft was saved, to a topic, and stops knowing who listens, how many there are, and when they read. The broker gives each subscription its own copy and holds it until that subscriber reads, so a slow, closed or busy listener costs Save nothing, and a new one is a subscription rather than an edit to Save. Each inbox keeps its own order; messages whose order matters to one subscriber belong in one subscription. What Save cannot hand to the broker is work it must see succeed: a subscriber can miss, drop or fail a copy and Save would never know, so required steps stay in Save’s own result path, and the topic carries only the reactions that may miss it.

Practice in the lab

A parcel depot’s scanners report every scan: sorted, out for delivery, delivered, or failed. A customer text notifier reacts to them, an audit service logs them all, and later a proof-of-delivery archive joins. Like the lesson’s editor, the depot should announce each scan once and let each consumer read on its own schedule. Each level starts from the previous level’s finished code, so you can begin at any of them.

PUBSUB-101 (easy) The SMS gateway went down, and so did every delivery scan
PUBSUB-102 (medium) A long gateway outage, and texts about an hour ago Pro
PUBSUB-103 (hard) Review: proofs of delivery that go missing, and texts that arrive twice Pro