Publish–Subscribe

The pipeline was a fixed, linear chain: every record passed through every step, and each step’s output was the next one’s input. The library holds work that does not fit this pattern. When a shipment’s fee is finalized, a record must be opened and a route assigned in the delivery operations context, the contracted customer’s volume counter must increase, and which tariff version priced it logged to the archive. These jobs do not feed each other, do not transform the record, and have no order among them; their count also changes over time. If the module that finalizes the fee calls them itself, it must import all three.

The publish–subscribe style was built and measured in the Messaging topic of the Caching, Queues and Asynchronous Processing course; topic, subscription, producer, and consumer were defined there, and delivery semantics and retry belong there — not revisited here. The Observer pattern in the Design Patterns course builds the same idea at the object level: the object announcing the change carries the subscriber list itself. This style’s distinctive face shows up in a single file — the subscriber list lives in neither the publisher nor the consumer. A named topic and an event bus enter between them; the two sides know the bus, not each other.

The quality attribute here is maintainability, and the scenario question is: when a fourth job reacting to the finalized fee is added, how many lines change in the file that finalizes the fee. The reverse of the question is also measured: when the publisher does not know the handlers, how many files must be read to answer “who handles this event.”

Three Jobs

The three jobs are identical files in both arrangements; the only thing that changes is how they are called. Each keeps its own state and exposes a counter, so both arrangements’ effect can be counted as equal.

mkdir -p jobs direct bus

// jobs/route.mjs — delivery operations: opens a delivery record and assigns a route
const TREE = { near: ["34"], mid: ["34", "06"], far: ["34", "41", "65"] };
const opened = [];
export const assignRoute = (event) => { opened.push({ code: event.code, route: TREE[event.zone] }); };
export const openedRecords = () => opened.length;

// jobs/volume.mjs — pricing: the contracted customer's volume counter
const counter = new Map();
export const incrementVolume = (event) => { counter.set(event.customer, (counter.get(event.customer) ?? 0) + 1); };
export const volume = (customer) => counter.get(customer) ?? 0;

// jobs/archive.mjs — writes the finalized amount to the archive with its tariff version
const lines = [];
export const archive = (event) => { lines.push(`${event.code} ${event.tariffVersion} ${event.net}`); };
export const archiveLines = () => lines.length;

Direct Call

In the first arrangement, the module that finalizes the fee calls the three jobs in sequence. Three import lines and three calls are the entirety of the publisher’s knowledge obligation.

// direct/finalize.mjs — the module that finalizes the fee calls the three jobs itself
import { assignRoute } from "../jobs/route.mjs";
import { incrementVolume } from "../jobs/volume.mjs";
import { archive } from "../jobs/archive.mjs";

export function finalize(event) {
  assignRoute(event);
  incrementVolume(event);
  archive(event);
  return event.code;
}

Event Bus

In the second arrangement, a module enters between them. The bus exposes two operations: subscribe attaches a handler to a topic, publish gives an event on that topic to every attached handler. Each handler call is isolated and counted. What happens when a consumer throws, and what changes when the event crosses a system boundary, is measured in this course’s Event-Driven Architecture lesson; the bus here is single-process, and publication is synchronous.

// bus/bus.mjs — publication by topic name: the two sides know the bus, not each other
const subscriptions = new Map();
let published = 0, delivered = 0, failed = 0;

export function subscribe(topic, handler) {
  if (subscriptions.has(topic) === false) subscriptions.set(topic, []);
  subscriptions.get(topic).push(handler);
}

export function publish(topic, event) {
  published += 1;
  for (const handler of subscriptions.get(topic) ?? []) {
    try { handler(event); delivered += 1; } catch { failed += 1; }
  }
  return (subscriptions.get(topic) ?? []).length;
}

export const counts = () => ({ published, delivered, failed });

The publisher knows only one of these two operations. The other thing it knows is a string: the topic’s name.

// bus/finalize.mjs — knows only the topic name and the event shape
import { publish } from "./bus.mjs";

export function finalize(event) {
  publish("fee-finalized", event);
  return event.code;
}

Subscriptions are set up in a third file. This file does the composition root’s job: it knows both the bus and the jobs, but no one knows it — the publisher does not import it.

// bus/registry.mjs — subscriptions are set up here; the publisher does not know this file
import { subscribe } from "./bus.mjs";
import { assignRoute } from "../jobs/route.mjs";
import { incrementVolume } from "../jobs/volume.mjs";
import { archive } from "../jobs/archive.mjs";

subscribe("fee-finalized", assignRoute);
subscribe("fee-finalized", incrementVolume);
subscribe("fee-finalized", archive);

Measurement

The script runs three events through both arrangements and compares the increase in the jobs’ counters. It then reads the import lines of both publisher files, extracts the known module and name counts, computes the import closure, and in its last section tries to answer “who handles this event” in both arrangements: the publisher file in the direct arrangement, every module in the source tree in the bus arrangement.

// coupling.mjs — both arrangements produce the same effect, the names the publisher knows and the files read to find handlers
import { readdirSync, readFileSync } from "node:fs";
import { dirname, join, normalize } from "node:path";
import { finalize as direct } from "./direct/finalize.mjs";
import { finalize as bus } from "./bus/finalize.mjs";
import { counts } from "./bus/bus.mjs";
import { openedRecords } from "./jobs/route.mjs";
import { volume } from "./jobs/volume.mjs";
import { archiveLines } from "./jobs/archive.mjs";
import "./bus/registry.mjs";

const EVENTS = [
  { code: "G-1", zone: "mid", customer: "MUS-1", net: 89.25, tariffVersion: "T2" },
  { code: "G-2", zone: "far", customer: "MUS-2", net: 119.7, tariffVersion: "T2" },
  { code: "G-3", zone: "near", customer: "MUS-1", net: 27, tariffVersion: "T2" },
];
const state = () => ({ route: openedRecords(), archive: archiveLines(), volume: volume("MUS-1") });
const diff = (a, b) => `route records +${b.route - a.route}, archive lines +${b.archive - a.archive}, MUS-1 volume +${b.volume - a.volume}`;

const s0 = state();
for (const e of EVENTS) direct(e);
const s1 = state();
for (const e of EVENTS) bus(e);
const s2 = state();
console.log(`direct arrangement: ${diff(s0, s1)}`);
console.log(`bus arrangement: ${diff(s1, s2)}`);
console.log(`bus counts: ${JSON.stringify(counts())}`);

const IMPORT = /import\s*\{([^}]*)\}\s*from\s*"(\.[^"]+)"/g;
const closure = (root) => {
  const seen = new Set([root]), stack = [root];
  while (stack.length > 0) {
    const path = stack.pop();
    for (const [, , rel] of readFileSync(path, "utf8").matchAll(IMPORT)) {
      const resolved = normalize(join(dirname(path), rel));
      if (seen.has(resolved) === false) { seen.add(resolved); stack.push(resolved); }
    }
  }
  return seen.size;
};

console.log("\npublisher                  known module  known name  import closure");
for (const path of ["direct/finalize.mjs", "bus/finalize.mjs"]) {
  const imports = [...readFileSync(path, "utf8").matchAll(IMPORT)];
  const names = imports.flatMap((m) => m[1].split(",").map((s) => s.trim()));
  console.log(
    `${path.padEnd(27)}${String(imports.length).padStart(13)}${String(names.length).padStart(12)}` +
      `${String(closure(path)).padStart(16)}  [${names.join(" ")}]`,
  );
}

const files = readdirSync(".", { recursive: true }).filter((y) => y.endsWith(".mjs") && y.includes("/"));
const directHandlers = (readFileSync("direct/finalize.mjs", "utf8").match(/^ {2}\w+\(event\);$/gm) ?? []).length;
const subscribedFiles = files.filter((y) => readFileSync(y, "utf8").includes('subscribe("fee-finalized"'));
const busHandlers = subscribedFiles.reduce(
  (t, y) => t + (readFileSync(y, "utf8").match(/subscribe\("fee-finalized", \w+\)/g) ?? []).length, 0);
console.log(`\nthe "who handles the fee-finalized event" question`);
console.log(`  direct: 1 file read, ${directHandlers} handlers visible`);
console.log(`  bus: 0 visible in the publisher, ${files.length} files scanned, ${busHandlers} handlers found in ${subscribedFiles.length} file(s)`);

node coupling.mjs

direct arrangement: route records +3, archive lines +3, MUS-1 volume +2
bus arrangement: route records +3, archive lines +3, MUS-1 volume +2
bus counts: {"published":3,"delivered":9,"failed":0}

publisher                  known module  known name  import closure
direct/finalize.mjs                    3           3               4  [assignRoute incrementVolume archive]
bus/finalize.mjs                       1           1               2  [publish]

the "who handles the fee-finalized event" question
  direct: 1 file read, 3 handlers visible
  bus: 0 visible in the publisher, 7 files scanned, 3 handlers found in 1 file(s)

Reading the Numbers

The first two lines are the measurement’s validity condition: three events produced three route records, three archive lines, and two volume increases for MUS-1 in both arrangements. Both arrangements do the same work.

The third line gives publication’s payoff. Three publish calls turned into nine handler calls: the publisher made 3 calls, but 9 jobs were actually done. The multiplier is a number the publisher does not know, and since failed is zero, all nine completed.

The table counts the knowledge obligation. In the direct arrangement the publisher knows three modules and three names, and the closure is 4 files: loading the publisher loads all three jobs. In the bus arrangement one module and one name are known, and the closure is 2 files. The three jobs stepped outside the closure; the only new thing remaining inside it is the bus itself. The topic name "fee-finalized" is a string, not an import, so it is not checked at compile time — the first half of the cost.

The last three lines are the second half. In the direct arrangement, one file answers the question: three calls sit there, read in sequence. In the bus arrangement, the publisher file shows no handler; finding the answer required scanning the tree’s seven modules and searching for subscription lines. It was found — three handlers in one file — but the search cost grows with the file count. This is the measure of loose coupling: as the names the publisher knows drop from 3 to 1, the files read to find handlers rise from 1 to 7.

Fourth Consumer

The scenario question is now tested with a fourth job: the finalized fee will be notified to the carrier.

// jobs/carrier.mjs — fourth job: notifies the carrier of the finalized fee
const notifications = [];
export const notifyCarrier = (event) => { notifications.push(`${event.code} ${event.net}`); };
export const notificationCount = () => notifications.length;

The script copies both arrangements, attaches the fourth job to both, counts how many lines changed in which file, and shows lines changed in the publisher file in a separate column. It then runs both arrangements in the new tree to verify the fourth job actually runs.

// extend.mjs — changed lines, changed lines in the publisher and delivery count when a fourth consumer is added
import { cpSync, readFileSync, writeFileSync } from "node:fs";

const edit = (path, transform) => writeFileSync(path, transform(readFileSync(path, "utf8")));
const changedLines = (previous, updated) => {
  const tally = (m) => m.split("\n").reduce((h, s) => h.set(s, (h.get(s) ?? 0) + 1), new Map());
  const a = tally(previous), b = tally(updated);
  const missing = (x, y) => [...x].reduce((t, [s, n]) => t + Math.max(0, n - (y.get(s) ?? 0)), 0);
  return missing(b, a) + missing(a, b);
};
const ARCHIVE = 'import { archive } from "../jobs/archive.mjs";';
const CARRIER = 'import { notifyCarrier } from "../jobs/carrier.mjs";';

for (const dir of ["jobs", "direct", "bus"]) cpSync(dir, `new/${dir}`, { recursive: true });
edit("new/direct/finalize.mjs", (m) =>
  m.replace(ARCHIVE, `${ARCHIVE}\n${CARRIER}`).replace("  archive(event);", "  archive(event);\n  notifyCarrier(event);"));
edit("new/bus/registry.mjs", (m) =>
  m.replace(ARCHIVE, `${ARCHIVE}\n${CARRIER}`)
    .replace('subscribe("fee-finalized", archive);', 'subscribe("fee-finalized", archive);\nsubscribe("fee-finalized", notifyCarrier);'));

const PUBLISHER = { direct: "direct/finalize.mjs", bus: "bus/finalize.mjs" };
const FILES = {
  direct: ["direct/finalize.mjs"],
  bus: ["bus/finalize.mjs", "bus/bus.mjs", "bus/registry.mjs"],
};
console.log("arrangement  edited file               changed lines  changed in publisher");
for (const [arrangement, paths] of Object.entries(FILES)) {
  const lineDiff = (y) => changedLines(readFileSync(y, "utf8"), readFileSync(`new/${y}`, "utf8"));
  const edited = paths.filter((y) => lineDiff(y) > 0);
  const total = edited.reduce((t, y) => t + lineDiff(y), 0);
  const inPublisher = edited.includes(PUBLISHER[arrangement]) ? lineDiff(PUBLISHER[arrangement]) : 0;
  console.log(`${arrangement.padEnd(10)}${edited.join(" ").padEnd(26)}${String(total).padStart(13)}${String(inPublisher).padStart(25)}`);
}

const EVENTS = [
  { code: "G-4", zone: "mid", customer: "MUS-1", net: 89.25, tariffVersion: "T2" },
  { code: "G-5", zone: "far", customer: "MUS-2", net: 119.7, tariffVersion: "T2" },
];
const { notificationCount } = await import("./new/jobs/carrier.mjs");
const { finalize: direct } = await import("./new/direct/finalize.mjs");
const { finalize: bus } = await import("./new/bus/finalize.mjs");
const { counts } = await import("./new/bus/bus.mjs");
await import("./new/bus/registry.mjs");

for (const e of EVENTS) direct(e);
const directNotifications = notificationCount();
for (const e of EVENTS) bus(e);
console.log(`\nfourth job ran: direct ${directNotifications}/${EVENTS.length}, bus ${notificationCount() - directNotifications}/${EVENTS.length}`);
console.log(`bus counts: ${JSON.stringify(counts())}, delivery per event = ${counts().delivered / counts().published}`);

node extend.mjs

arrangement  edited file               changed lines  changed in publisher
direct    direct/finalize.mjs                   2                        2
bus       bus/registry.mjs                      2                        0

fourth job ran: direct 2/2, bus 2/2
bus counts: {"published":2,"delivered":8,"failed":0}, delivery per event = 4

The number of changed lines is 2 in both arrangements: one import line and one binding line. The difference is in the last column: in the direct arrangement these two lines sit inside the file that finalizes the fee; in the bus arrangement, that file was not touched. Delivery per event rose from 3 to 4 without the publisher file changing.

This shows where the gain lands: the amount of change stayed the same, but its location changed. The publisher sits inside the pricing context; bus/registry.mjs is a binding file carrying no field rule at all. With ten consumers, every new job is still two lines, but the file where the fee calculation sits is never opened.

Summary

• In the publish–subscribe style, the subscriber list lives in neither the publisher nor the consumer; a topic name and an event bus enter between them, and the two sides know the bus, not each other.
• Three events produced the same effect in both arrangements; three publish calls turned into nine handler calls, a multiplier the publisher does not know.
• The modules the publisher knows dropped from 3 to 1, names from 3 to 1, and the import closure from 4 files to 2; the topic name is a string, not an import, so it is not checked at compile time.
• The cost is traceability: “who handles this event” was answered by reading 1 file directly, and by scanning 7 modules through the bus.
• The fourth consumer changed 2 lines in both arrangements, but those lines fell in the publisher file directly and in the binding file through the bus; delivery per event rose from 3 to 4 without the publisher file changing.

Next Step

This bus has an unmeasured property: the publish call does not return until every attached handler finishes. Four handlers run in sequence on the same call stack, so the module that finalizes the fee takes as long as the slowest handler. If archive writing slows down, fee finalization slows down too, even though the archive’s write time does not matter for pricing. The handler can be taken out of waiting by putting a store in between — dropping the event somewhere and returning. The next lesson builds this store and measures it: how many consumer steps have run by the time the publisher drops the event, how much work piles up waiting when the consumer is slower than the producer, whether order is preserved with two consumers on the same store, and how many times a delivery record opens when the same message is processed twice.