Build Your Own Pub/Sub System
Build a pub/sub system from 60 lines of JavaScript: subscribe, publish, unsubscribe, once(), namespace wildcards, error isolation, and async publish.
Build Your Own Pub/Sub System
A user signs up. You need to send a welcome email, create a default workspace, start a trial, and fire an analytics event. The naïve version puts all of that in one signUp function. Three months later that function is 200 lines long, has eight imports, and every test touches all of them.
The pattern that cuts this knot is Publish/Subscribe. The signup code fires one event — user:signup — and has no idea who's listening. The email service, workspace service, billing service, and analytics service each subscribe independently. Add a new listener without touching the publisher. Remove one without breaking the others.
It's not magic. The entire mechanism is a Map of channel names to handler functions, plus a bit of bookkeeping. Here's how it works.
What We're Building
subscribe(channel, handler)— register a listener, get a token backpublish(channel, data)— call every listener on that channelunsubscribe(token)— remove a specific listener by its tokenonce(channel, handler)— subscribe but auto-remove after the first fire- Namespace wildcards:
user:*catchesuser:signup,user:login,user:deleted - Error isolation: one bad handler doesn't crash the others
publishAsync(channel, data)— await all handlers settling before continuing
The publisher doesn't import the subscribers. The subscribers don't import each other. The PubSub engine is the only shared dependency, and it's a tiny thing.
Step 1: The Skeleton
Start with the constructor and subscribe. The key insight is the token: instead of asking callers to hold a reference to their handler function to unsubscribe later, we return an opaque string. This means you can subscribe an inline arrow function and still cancel it.
Constructor and subscribe
// pubsub.js
class PubSub {
constructor() {
// { [channel]: { [token]: { handler, once } } }
this._channels = Object.create(null);
// { [token]: channel } — lets unsubscribe run in O(1)
this._tokenMap = Object.create(null);
this._counter = 0;
}
_nextToken() {
return `sub_${++this._counter}`;
}
subscribe(channel, handler) {
if (typeof handler !== 'function') {
throw new TypeError(`handler must be a function, got ${typeof handler}`);
}
if (!this._channels[channel]) {
this._channels[channel] = Object.create(null);
}
const token = this._nextToken();
this._channels[channel][token] = { handler, once: false };
this._tokenMap[token] = channel;
return token;
}
}
module.exports = { PubSub };Object.create(null) gives us a dictionary with no prototype, so there's no accidental collision with keys like "constructor" or "__proto__". A channel named "constructor" would break a regular {}.
Step 2: Publish
Iterate the channel's subscriber map and call each handler. Return the count of notified subscribers — useful for debugging and tests.
Publish
// Add to PubSub class
publish(channel, data) {
const subs = this._channels[channel];
if (!subs) return 0;
let count = 0;
const tokensToRemove = [];
for (const [token, sub] of Object.entries(subs)) {
sub.handler(data, channel);
count++;
if (sub.once) tokensToRemove.push(token);
}
// Remove once-subscriptions after the loop — not during it
tokensToRemove.forEach(t => this.unsubscribe(t));
return count;
}The tokensToRemove array is important. Mutating subs inside a for...of Object.entries(subs) loop won't corrupt the iteration, because Object.entries builds an array up front rather than iterating live. Deferring the deletes until after the loop is still cleaner.
That snapshot has a consequence worth knowing before it surprises you: if handler A calls unsubscribe(tokenB) mid-publish, B still runs for this publish — it was already in the snapshot. It's gone from the next one. Node's EventEmitter behaves the same way, and the alternative (checking liveness before each call) mostly buys you ordering bugs. Just don't write handlers that assume an unsubscribe takes effect immediately.
Step 3: Unsubscribe
The _tokenMap makes this O(1). Without it, unsubscribing by handler reference would require scanning every channel until finding the matching function — O(n) across all subscribers.
Unsubscribe
// Add to PubSub class
unsubscribe(token) {
const channel = this._tokenMap[token];
if (!channel) return false;
delete this._channels[channel][token];
delete this._tokenMap[token];
// Clean up empty channels so they don't pile up
if (Object.keys(this._channels[channel]).length === 0) {
delete this._channels[channel];
}
return true;
}Returning false for an unknown token lets callers detect double-unsubscribe without throwing. Double-unsubscribe is common in cleanup paths — useEffect teardowns, modal close handlers — so silent no-ops are the right call.
Step 4: once()
A once subscription auto-removes itself after the first event. The publish method already checks sub.once and collects those tokens for removal. once() just needs to set the flag.
once() — single-fire subscription
// Add to PubSub class
once(channel, handler) {
if (typeof handler !== 'function') {
throw new TypeError(`handler must be a function, got ${typeof handler}`);
}
if (!this._channels[channel]) {
this._channels[channel] = Object.create(null);
}
const token = this._nextToken();
this._channels[channel][token] = { handler, once: true };
this._tokenMap[token] = channel;
return token;
}once is useful for things that only happen once per lifecycle: a database connection opening, a feature flag resolving, a file finishing a download. You can also use it to implement request-response over pub/sub — publish a request, once-subscribe the reply channel, get the response, done.
let requestId = 0;
function request(pubsub, channel, data) {
return new Promise((resolve) => {
// A counter, not Date.now() — two requests in the same millisecond
// would otherwise share a reply channel and cross their wires.
const replyChannel = `${channel}:reply:${++requestId}`;
pubsub.once(replyChannel, resolve);
pubsub.publish(channel, { ...data, replyChannel });
});
}Step 5: Namespace Wildcards
This is where a pub/sub system earns its keep over a plain event emitter. Publishing user:signup should also notify anyone listening on user:* (all user events) or * (all events). An audit log or analytics service can subscribe once and catch everything.
The rule: when publishing to a:b:c, also check a:b:*, then a:*, then *.
Wildcard matching in publish
// Replace the publish method with this version
publish(channel, data) {
const tokensToRemove = [];
let count = 0;
const notify = (subs) => {
for (const [token, sub] of Object.entries(subs)) {
sub.handler(data, channel);
count++;
if (sub.once) tokensToRemove.push(token);
}
};
// 1. Exact channel match
if (this._channels[channel]) {
notify(this._channels[channel]);
}
// 2. Namespace wildcards — 'cart:item:added' checks 'cart:item:*', then 'cart:*'
const parts = channel.split(':');
for (let i = parts.length - 1; i > 0; i--) {
const wildcard = parts.slice(0, i).join(':') + ':*';
if (wildcard !== channel && this._channels[wildcard]) {
notify(this._channels[wildcard]);
}
}
// 3. Global wildcard — '*' catches everything
if (channel !== '*' && this._channels['*']) {
notify(this._channels['*']);
}
tokensToRemove.forEach(t => this.unsubscribe(t));
return count;
}The wildcard !== channel guard handles the edge case where someone publishes to a channel that literally ends in :*. Without it, a publish to cart:* would double-notify cart:* subscribers. The channel !== '*' guard on the global wildcard prevents infinite self-notification if someone publishes to '*' directly.
Subscribers receive the original channel name as the second argument to their handler: handler(data, channel). When a logger subscribes to user:* and gets called by user:signup, it knows the specific channel that fired — useful for routing or filtering inside a catch-all subscriber.
Step 6: Error Isolation
Right now, if one handler throws, publish stops and subsequent handlers never run. The fix is a try/catch inside notify. Each handler is isolated — a broken email service shouldn't prevent the workspace from being created.
Error isolation
// Update the notify function inside publish
const notify = (subs) => {
for (const [token, sub] of Object.entries(subs)) {
try {
sub.handler(data, channel);
} catch (err) {
console.error(`PubSub: handler threw on channel "${channel}"`, err);
}
count++;
if (sub.once) tokensToRemove.push(token);
}
};Notice count++ and the once check are outside the try block. A handler that throws should still be counted and removed if it was a once subscription — you don't want a crashing handler to fire again on the next publish.
This try/catch only isolates synchronous throws. An async handler that rejects returns a promise, and try/catch can't see it — so the rejection escapes publish entirely. On Node 15+ an unhandled rejection terminates the process with exit code 1, which is the opposite of isolation. If you register async handlers, use publishAsync from Step 7, or attach a .catch() inside the handler itself.
If you need to surface errors to the publisher (not just log them), collect them and throw after all handlers run:
const errors = [];
try {
sub.handler(data, channel);
} catch (err) {
errors.push(err);
}
// after loop...
if (errors.length > 0) throw new AggregateError(errors, `${errors.length} handler(s) threw`);Step 7: Async Publish
Sync publish fires handlers and moves on — if any of them are async, their returned Promises are dropped on the floor. You don't get to know when they finished, and as the previous callout showed, a rejection takes the process down with it. Sometimes you need to wait for all handlers to settle: sending an email, writing to a database, making an API call. That's publishAsync.
Async publish with Promise.allSettled
// Add to PubSub class
async publishAsync(channel, data) {
const entries = [];
const collect = (subs) => {
for (const [token, sub] of Object.entries(subs)) {
entries.push({ token, once: sub.once, handler: sub.handler });
}
};
if (this._channels[channel]) collect(this._channels[channel]);
const parts = channel.split(':');
for (let i = parts.length - 1; i > 0; i--) {
const wildcard = parts.slice(0, i).join(':') + ':*';
if (wildcard !== channel && this._channels[wildcard]) {
collect(this._channels[wildcard]);
}
}
if (channel !== '*' && this._channels['*']) {
collect(this._channels['*']);
}
// All handlers run concurrently; errors don't cancel others
const results = await Promise.allSettled(
entries.map(({ handler }) =>
Promise.resolve().then(() => handler(data, channel))
)
);
// Clean up once-subscriptions after all settle
entries.forEach((entry) => {
if (entry.once) this.unsubscribe(entry.token);
});
// Log failures but don't swallow them entirely
results.forEach((result) => {
if (result.status === 'rejected') {
console.error(`PubSub: async handler threw on "${channel}"`, result.reason);
}
});
return results; // caller can inspect per-handler outcomes
}Promise.allSettled is the right primitive here, not Promise.all. If one handler rejects with Promise.all, every other handler's result is discarded. Promise.allSettled waits for all of them and returns an array of { status, value/reason } objects — the caller can iterate and inspect each outcome.
Promise.resolve().then(() => handler(...)) wraps sync handlers too. A sync function that throws would turn into a rejected promise, treated identically to an async handler that rejects.
The Full Implementation
// pubsub.js
class PubSub {
constructor() {
this._channels = Object.create(null);
this._tokenMap = Object.create(null);
this._counter = 0;
}
_nextToken() {
return `sub_${++this._counter}`;
}
_register(channel, handler, once) {
if (typeof handler !== 'function') {
throw new TypeError(`handler must be a function, got ${typeof handler}`);
}
if (!this._channels[channel]) {
this._channels[channel] = Object.create(null);
}
const token = this._nextToken();
this._channels[channel][token] = { handler, once };
this._tokenMap[token] = channel;
return token;
}
subscribe(channel, handler) {
return this._register(channel, handler, false);
}
once(channel, handler) {
return this._register(channel, handler, true);
}
unsubscribe(token) {
const channel = this._tokenMap[token];
if (!channel) return false;
delete this._channels[channel][token];
delete this._tokenMap[token];
if (Object.keys(this._channels[channel]).length === 0) {
delete this._channels[channel];
}
return true;
}
publish(channel, data) {
const tokensToRemove = [];
let count = 0;
const notify = (subs) => {
for (const [token, sub] of Object.entries(subs)) {
try {
sub.handler(data, channel);
} catch (err) {
console.error(`PubSub: handler threw on "${channel}"`, err);
}
count++;
if (sub.once) tokensToRemove.push(token);
}
};
if (this._channels[channel]) notify(this._channels[channel]);
const parts = channel.split(':');
for (let i = parts.length - 1; i > 0; i--) {
const wildcard = parts.slice(0, i).join(':') + ':*';
if (wildcard !== channel && this._channels[wildcard]) {
notify(this._channels[wildcard]);
}
}
if (channel !== '*' && this._channels['*']) {
notify(this._channels['*']);
}
tokensToRemove.forEach(t => this.unsubscribe(t));
return count;
}
async publishAsync(channel, data) {
const entries = [];
const collect = (subs) => {
for (const [token, sub] of Object.entries(subs)) {
entries.push({ token, once: sub.once, handler: sub.handler });
}
};
if (this._channels[channel]) collect(this._channels[channel]);
const parts = channel.split(':');
for (let i = parts.length - 1; i > 0; i--) {
const wildcard = parts.slice(0, i).join(':') + ':*';
if (wildcard !== channel && this._channels[wildcard]) collect(this._channels[wildcard]);
}
if (channel !== '*' && this._channels['*']) collect(this._channels['*']);
const results = await Promise.allSettled(
entries.map(({ handler }) => Promise.resolve().then(() => handler(data, channel)))
);
entries.forEach((entry) => {
if (entry.once) this.unsubscribe(entry.token);
});
results.forEach((result) => {
if (result.status === 'rejected') {
console.error(`PubSub: async handler threw on "${channel}"`, result.reason);
}
});
return results;
}
channels() {
return Object.keys(this._channels);
}
subscriberCount(channel) {
return this._channels[channel] ? Object.keys(this._channels[channel]).length : 0;
}
}
module.exports = { PubSub };Demo: Shopping Cart Events
Wire up a shopping cart that publishes events and three independent services that react.
// demo.js
const { PubSub } = require('./pubsub');
const bus = new PubSub();
// Analytics listens to all cart events
bus.subscribe('cart:*', (data, channel) => {
console.log(`[analytics] ${channel}`, data);
});
// Logger listens to everything
bus.subscribe('*', (data, channel) => {
console.log(`[audit] ${channel} at ${new Date().toISOString()}`);
});
// Inventory only cares about cart:item-added
bus.subscribe('cart:item-added', ({ item, quantity }) => {
console.log(`[inventory] reserve ${quantity}x ${item.sku}`);
});
// Price recalculation on add or remove
const recalcToken = bus.subscribe('cart:item-added', ({ cart }) => {
const total = cart.items.reduce((sum, i) => sum + i.price * i.qty, 0);
console.log(`[pricing] cart total: $${total.toFixed(2)}`);
});
bus.subscribe('cart:item-removed', ({ cart }) => {
const total = cart.items.reduce((sum, i) => sum + i.price * i.qty, 0);
console.log(`[pricing] cart total: $${total.toFixed(2)}`);
});
// --- Simulate cart actions ---
const cart = { items: [] };
const apple = { sku: 'APPLE-001', price: 1.5, qty: 3 };
cart.items.push(apple);
bus.publish('cart:item-added', { item: apple, quantity: 3, cart });
const orange = { sku: 'ORG-002', price: 0.8, qty: 2 };
cart.items.push(orange);
bus.publish('cart:item-added', { item: orange, quantity: 2, cart });
// Remove first item
cart.items = cart.items.filter(i => i.sku !== 'APPLE-001');
bus.publish('cart:item-removed', { item: apple, cart });
// Stop recalculating (e.g. cart is locked at checkout)
bus.unsubscribe(recalcToken);
bus.publish('cart:checkout-started', { cart });
// ↑ audit logger and analytics still fire; pricing does notOutput:
[inventory] reserve 3x APPLE-001
[pricing] cart total: $4.50
[analytics] cart:item-added { item: ..., quantity: 3, ... }
[audit] cart:item-added at 2026-07-22T...
[inventory] reserve 2x ORG-002
[pricing] cart total: $6.10
[analytics] cart:item-added { item: ..., quantity: 2, ... }
[audit] cart:item-added at 2026-07-22T...
[pricing] cart total: $1.60
[analytics] cart:item-removed { item: ..., ... }
[audit] cart:item-removed at 2026-07-22T...
[analytics] cart:checkout-started { cart: ... }
[audit] cart:checkout-started at 2026-07-22T...
Testing
// test.js — node test.js
const { PubSub } = require('./pubsub');
function assert(condition, message) {
if (!condition) throw new Error(`FAIL: ${message}`);
console.log(`PASS: ${message}`);
}
async function run() {
// Basic subscribe + publish
const bus = new PubSub();
const received = [];
bus.subscribe('test', (data) => received.push(data));
bus.publish('test', 'hello');
assert(received[0] === 'hello', 'basic subscribe + publish');
// Unsubscribe stops delivery
const bus2 = new PubSub();
let count = 0;
const tok = bus2.subscribe('x', () => count++);
bus2.publish('x', null);
bus2.unsubscribe(tok);
bus2.publish('x', null);
assert(count === 1, 'unsubscribe stops delivery');
// Double unsubscribe returns false, doesn't throw
const bus3 = new PubSub();
const t = bus3.subscribe('y', () => {});
bus3.unsubscribe(t);
assert(bus3.unsubscribe(t) === false, 'double unsubscribe returns false');
// once() fires once then auto-removes
const bus4 = new PubSub();
let fires = 0;
bus4.once('ping', () => fires++);
bus4.publish('ping', null);
bus4.publish('ping', null);
assert(fires === 1, 'once() fires exactly once');
// Namespace wildcard — user:* catches user:login
const bus5 = new PubSub();
const events = [];
bus5.subscribe('user:*', (_, ch) => events.push(ch));
bus5.publish('user:login', {});
bus5.publish('user:signup', {});
bus5.publish('order:placed', {});
assert(events.length === 2, 'user:* catches only user: events');
assert(events[0] === 'user:login' && events[1] === 'user:signup', 'wildcard passes channel name');
// Global wildcard '*' catches all
const bus6 = new PubSub();
let globalCount = 0;
bus6.subscribe('*', () => globalCount++);
bus6.publish('a', null);
bus6.publish('b:c', null);
bus6.publish('x:y:z', null);
assert(globalCount === 3, 'global * catches all channels');
// Error isolation — bad handler doesn't block others
const bus7 = new PubSub();
let goodRan = false;
bus7.subscribe('ev', () => { throw new Error('boom'); });
bus7.subscribe('ev', () => { goodRan = true; });
bus7.publish('ev', null);
assert(goodRan, 'error isolation: good handler runs after bad one throws');
// publishAsync waits for async handlers
const bus8 = new PubSub();
let asyncDone = false;
bus8.subscribe('slow', async () => {
await new Promise(r => setTimeout(r, 50));
asyncDone = true;
});
await bus8.publishAsync('slow', null);
assert(asyncDone, 'publishAsync awaits async handlers');
// subscriberCount
const bus9 = new PubSub();
bus9.subscribe('ch', () => {});
bus9.subscribe('ch', () => {});
assert(bus9.subscriberCount('ch') === 2, 'subscriberCount returns correct count');
console.log('\nAll tests passed.');
}
run().catch(console.error);What We Skipped
Priority ordering — some pub/sub systems let subscribers declare a priority so high-priority handlers run first. The implementation above fires subscribers in insertion order, which the language guarantees: Object.entries returns non-integer string keys in insertion order, and our tokens are sub_1, sub_2, … Name your tokens 1, 2, 3 instead and they'd be sorted numerically — harmless here, a nasty surprise elsewhere. Priority would require sorting entries by a priority field before iterating. Practical need is rare; if you find yourself wanting it, that's usually a sign the handlers have implicit ordering dependencies that should be made explicit.
Async error aggregation — publishAsync logs rejections but the return value is Promise.allSettled results, so callers can inspect them. A production system might want to emit a separate error event: bus.publish('pubsub:error', { channel, err }). This lets a central error monitor subscribe without callers having to check the results array.
Cross-process pub/sub — everything here is in-memory. Multiple Node.js processes (or multiple machines) can't share this bus. Redis Pub/Sub and NATS are the common replacements for that use case. The programming model is identical — channel strings, subscribe/publish — but the transport is a network socket instead of a local Map.
Message replay / persistence — if a service restarts, it misses events published while it was down. Solving this requires durable messaging: Kafka, RabbitMQ, or BullMQ. The in-memory bus is fire-and-forget.
removeAllListeners(channel) — handy for teardown in tests: Object.keys(this._channels[channel] || {}).forEach(t => this.unsubscribe(t)). Snapshot the keys with Object.keys first, as shown, rather than deleting while iterating the live object.
The core is 60 lines. The wildcard matching loop, the token map, the once flag — those are the three ideas that make this more than a glorified callback array. The small emitter libraries (mitt, eventemitter3, nanoevents) start from the same map-of-channels-to-handlers core and then diverge: they unsubscribe by handler reference or by returned closure rather than by token, and only mitt ships a wildcard, and only the global one.
Comments (0)
No comments yet. Be the first to share your thoughts!