@studnicky/event-bus
Typed publish/subscribe with per-subscriber backpressure isolation and AbortSignal lifecycle.
What it is
A composable, typed publish/subscribe primitive with isolated subscriber queues, explicit lifecycle, and a router for caller-provided topic selection. It coordinates messages without defining an application's event vocabulary or business process.
What it is for
Northstar Books uses it to fan an accepted-order event to fulfilment, audit, and analytics while keeping a slow consumer from delaying the others. Consumers supply their event map, handlers, routing evidence, and queue policy; the package does not provide an order-management implementation.
Northstar Books examples
The runnable topic-router example shows Northstar Books selecting the subscribers for a stock-change topic from caller-owned evidence, proving that routing policy remains in application composition. The runnable pub/sub example publishes a typed order event and drains its queue, proving reliable handoff to a fulfilment subscriber. The runnable lifecycle-hooks example records subscription, fan-out, delivery, unsubscription, and close events, proving that Northstar can add audit instrumentation without coupling the bus to a logging product.
Public entrypoints
| Import path | Use it when |
|---|---|
@studnicky/event-bus/node | A Northstar Books server publishes or consumes typed catalogue, order, or fulfilment events. |
@studnicky/event-bus/browser | A browser client needs the same portable typed event contract for local UI coordination. |
@studnicky/event-bus/interfaces | TypeScript code accepts publishing, handling, or unsubscribe contracts without depending on a bus implementation. |
@studnicky/event-bus/router | Composition code routes selected Northstar Books topics using caller-owned matching evidence. |
@studnicky/event-bus/router/interfaces | TypeScript code shares the router selection contract between its matcher and router boundary. |
Install
pnpm add @studnicky/event-busRequires @studnicky:registry=https://npm.pkg.github.com in .npmrc.
Runtime imports
Import EventBus from @studnicky/event-bus/node in Node or @studnicky/event-bus/browser in browsers. Import event contracts from @studnicky/event-bus/interfaces. Subscriber queue configuration uses @studnicky/concurrency/queue/entities.
Northstar Books fulfilment events
When Northstar accepts an order, audit capture, fulfilment dispatch, and analytics each need the order event without one consumer slowing another. Publish a typed order event to the bus and give each concern its own subscription. EventBus guarantees typed topic payloads and independent bounded subscriber queues; a slow analytics subscriber applies backpressure only to its own queue, not to audit or fulfilment delivery.
Usage
Pub/sub starts with the smallest loop there is: one topic, one subscriber, one message. Here EventBus publishes a typed user:created event and drain() waits until the subscriber's queue has actually processed it, so by the time the example inspects received the handler has genuinely run — not merely been scheduled. That delivered-and-confirmed guarantee is what every other example on this page builds on:
import { EventBus } from '../src/index.js';
const bus = EventBus.create<UserCreatedEventMapEntity.Type>();
const received: { 'email': string; 'id': string; }[] = [];
bus.subscribe('user:created', (payload) => {
received.push(payload);
});
await bus.publish('user:created', { 'email': 'a@example.com', 'id': '1' });
await bus.drain();
console.log('Received:', received);Multiple subscribers
Northstar Books needs a checkout confirmation and a fulfilment queue to both react to the same order, with neither one aware the other exists. This example subscribes two independent handlers to order:placed, confirms both receive the first order, then calls the unsubscribe function the first subscribe() returned — after that, only the fulfilment handler sees the second order. Each subscriber owns its own delivery and its own lifecycle, so removing one never touches the other:
import { EventBus } from '../src/index.js';
import { MultiSubscriberFixture } from './fixtures/MultiSubscriberFixture.js';
const bus = EventBus.create<OrderPlacementEventMapEntity.Type>();
const unsubscribeA = bus.subscribe('order:placed', (payload) => {
// Northstar Books sends checkout confirmation after a successful order.
MultiSubscriberFixture.receivedA.push(payload.orderId);
});
bus.subscribe('order:placed', (payload) => {
// The fulfilment queue independently receives the same checkout event.
MultiSubscriberFixture.receivedB.push(payload.orderId);
});
// Confirmation and fulfilment both receive the first Northstar order event.
await bus.publish('order:placed', { 'orderId': 'northstar-order-1001' });
await bus.drain();
console.log('After first publish — A:', MultiSubscriberFixture.receivedA, 'B:', MultiSubscriberFixture.receivedB);
// Unsubscribe handler A — only B receives subsequent events
unsubscribeA();
await bus.publish('order:placed', { 'orderId': 'northstar-order-1002' });
await bus.drain();
console.log('After unsubscribe + second publish — A:', MultiSubscriberFixture.receivedA, 'B:', MultiSubscriberFixture.receivedB);AbortSignal-based lifecycle
A subscriber shouldn't always outlive the thing that created it — a request handler, a UI component, a worker that's shutting down. Pass an AbortController's signal when subscribing and the subscription tears itself down the moment that signal fires. This example confirms a ping topic delivers normally before the abort and confirms nothing arrives after it, while the handler's own subscription signal — passed in as its second argument — fires its own abort listener so any in-flight async work it started can react to teardown too:
import { EventBus } from '../src/index.js';
import { AbortSignalFixture } from './fixtures/AbortSignalFixture.js';
const bus = EventBus.create<PingEventMapEntity.Type>();
const controller = new AbortController();
bus.subscribe('ping', (payload, signal) => {
// The signal is the subscription lifecycle signal — check it to bail out of
// long-running async work early, or pass it to fetch()/setTimeout() etc.
if (signal.aborted) { return; }
AbortSignalFixture.received.push(payload);
// Register a listener so in-flight async work can react to teardown.
signal.addEventListener('abort', () => {
AbortSignalFixture.abortedDuringDelivery.push(true);
}, { 'once': true });
}, { 'signal': controller.signal });
await bus.publish('ping', 'first');
await bus.drain();
console.log('Received before abort:', AbortSignalFixture.received);
// Abort the subscriber — its signal aborts and it will no longer receive events.
controller.abort();
await bus.publish('ping', 'second');
await bus.drain();
console.log('Received after abort:', AbortSignalFixture.received);Topic routing
Northstar Books wants its audit trail to catch every order-related topic while fulfilment only cares about orders being created — two different matching rules reacting to the same bus. TopicRouter doesn't decide that matching logic itself: it accepts a caller-supplied matcher, then uses it to work out which registered subscriptions a published topic should reach. The example below registers a wildcard audit subscription and an exact-match fulfilment subscription, publishes one order-created event, and confirms both rules routed correctly off the same topic string.
Publish through a minimal sink
Components that only publish events accept EventSinkInterface instead of an EventBus. The contract contains only publish, so it also accepts a custom publisher without requiring subscription or lifecycle methods:
import type { EventSinkInterface } from "@studnicky/event-bus/interfaces";
interface RetryEventsInterface {
readonly "retry:failed": { readonly attempt: number };
}
async function recordFailure(
sink: EventSinkInterface<RetryEventsInterface>,
attempt: number,
): Promise<void> {
await sink.publish("retry:failed", { attempt });
}EventBus<RetryEventsInterface> satisfies this contract directly.
Observability hooks
Watching what a bus actually does — who subscribed, what fired, what backed up — usually means reaching for a metrics library. EventBus instead exposes ten protected hook methods you can override on a subclass, each a plain no-op until you say otherwise, so instrumentation lives next to the class it's observing instead of wrapped around it. Reaching for a queue without a bus around it? BusQueue from @studnicky/concurrency/queue/node gives standalone FIFO admission on its own.
EventBus hooks
| Hook | When it fires | Args |
|---|---|---|
onPublish(topic, payload) | Once per publish() call, before fan-out | topic: K, payload: TTopicMap[K] |
onSubscribe(topic) | When a subscriber registers | topic: K |
onUnsubscribe(topic) | When a subscriber is removed | topic: K |
onDeliver(topic, payload) | After each successful handler invocation | topic: K, payload: TTopicMap[K] |
onEnqueue(topic) | When an event enters a subscriber queue; completes before delivery | topic: K |
onDequeue(topic) | When an event is dequeued for processing | topic: K |
onDrop(topic) | When an event is dropped (queue aborted) | topic: K |
onOverflow(topic, depth) | When backpressure begins on a subscriber queue; completes before delivery | topic: K, depth: number |
onHandlerError(topic, error) | When a subscriber handler throws | topic: K, error: unknown |
onDispose() | When bus.close() is called | — |
import { EventBus } from '../src/index.js';
class TracedBus extends EventBus<OrderStatusEventMapEntity.Type> {
static createTraced(): TracedBus {
return new TracedBus();
}
readonly deliverLog: { 'payload': unknown; 'topic': string }[] = [];
readonly dequeueLog: string[] = [];
readonly disposeLog: number[] = [];
readonly enqueueLog: string[] = [];
readonly publishLog: { 'payload': unknown; 'topic': string }[] = [];
readonly subscribeLog: string[] = [];
readonly unsubscribeLog: string[] = [];
protected override onDeliver<K extends keyof OrderStatusEventMapEntity.Type>(topic: K, payload: OrderStatusEventMapEntity.Type[K]): void {
console.log(`[event-bus] deliver topic=${String(topic)} payload=${JSON.stringify(payload)}`);
this.deliverLog.push({ 'payload': payload, 'topic': String(topic) });
}
protected override onDequeue(topic: keyof OrderStatusEventMapEntity.Type): void {
console.log(`[event-bus] dequeue topic=${String(topic)}`);
this.dequeueLog.push(String(topic));
}
protected override onDispose(): void {
console.log('[event-bus] dispose');
this.disposeLog.push(1);
}
protected override onEnqueue(topic: keyof OrderStatusEventMapEntity.Type): void {
console.log(`[event-bus] enqueue topic=${String(topic)}`);
this.enqueueLog.push(String(topic));
}
protected override onPublish<K extends keyof OrderStatusEventMapEntity.Type>(topic: K, payload: OrderStatusEventMapEntity.Type[K]): void {
console.log(`[event-bus] publish topic=${String(topic)} payload=${JSON.stringify(payload)}`);
this.publishLog.push({ 'payload': payload, 'topic': String(topic) });
}
protected override onSubscribe(topic: keyof OrderStatusEventMapEntity.Type): void {
console.log(`[event-bus] subscribe topic=${String(topic)}`);
this.subscribeLog.push(String(topic));
}
protected override onUnsubscribe(topic: keyof OrderStatusEventMapEntity.Type): void {
console.log(`[event-bus] unsubscribe topic=${String(topic)}`);
this.unsubscribeLog.push(String(topic));
}
}
const bus = TracedBus.createTraced();
// Subscribe two handlers to 'order:created', one to 'order:updated'
const unsub1 = bus.subscribe('order:created', (payload) => {
console.log(`[handler-A] order:created id=${payload.id} total=${payload.total}`);
});
bus.subscribe('order:created', (payload) => {
console.log(`[handler-B] order:created id=${payload.id}`);
});
bus.subscribe('order:updated', (payload) => {
console.log(`[handler-C] order:updated id=${payload.id} status=${payload.status}`);
});
// Publish order:created twice, order:updated once
await bus.publish('order:created', { 'id': 'ord-1', 'total': 99 });
await bus.publish('order:created', { 'id': 'ord-2', 'total': 42 });
await bus.publish('order:updated', { 'id': 'ord-1', 'status': 'shipped' });
await bus.drain();
// Unsubscribe one handler
unsub1();
// Close bus
await bus.close();The base class never calls any logger or metrics library. All hooks are no-ops by default.
Try it
Run this one live to watch the whole publish-subscribe cycle happen in the browser: it builds a typed EventBus with EventBus.create<AppEvents>(), publishes a single event, drains the subscriber queue so delivery actually completes, then closes the bus cleanly.
Where the first hooks example explains what each hook means, this one shows the full fan-out trace side by side so you can see the order things actually happen in. Watch subscribe fire once per handler registration, publish fire once per bus.publish() call, enqueue/dequeue fire once per subscriber per publish, and deliver land after each individual handler runs — then unsubscribe fires once for the explicit unsub1() call and dispose fires once as bus.close() finishes.
API
| Export | Type | Description |
|---|---|---|
EventBus<TTopicMap> | class | Multi-topic pub/sub; created via EventBus.create<T>(config?) |
EventHandlerInterface<T> | interface | Callable handler contract: (payload: T, signal: AbortSignal) => Promise<void> | void |
EventSinkInterface<TTopicMap> | interface | Minimal typed publishing contract: publish(topic, payload) => Promise<void> |
UnsubscribeInterface | interface | Callable unsubscribe contract returned by subscribe: () => void |
TopicRouter | Registers topic subscriptions and publishes structural or caller-selected deliveries. | @studnicky/event-bus/router |
TopicSelectionInterface | Defines { id, origin, scores? } selection data. | @studnicky/event-bus/router/interfaces |
EventBus<TTopicMap>
| Member | Signature | Description |
|---|---|---|
create | static create<T>(config?: BusQueueOptionsEntity.InputType) => EventBus<T> | Constructs a bus; constructor is protected |
subscribe | (topic, handler, options?) => UnsubscribeInterface | Registers a subscriber; returns unsubscribe function |
publish | (topic, payload) => Promise<void> | Enqueues payload to all topic subscribers |
drain | () => Promise<void> | Waits for all subscriber queues to empty |
close | () => Promise<void> | Aborts all subscribers and drains |
Exports
| Symbol | Purpose | Import path |
|---|---|---|
EventBus | Provides event bus functionality. | @studnicky/event-bus/node |
EventBusClosedError | Abort reason for every subscriber queue when the bus closes. | @studnicky/event-bus/node |
EventBusError | Represents event bus failures. | @studnicky/event-bus/node |
EventBusUnsubscribedError | Abort reason for a subscriber queue when its subscription is removed. | @studnicky/event-bus/node |
EventSinkInterface | Defines the minimal typed event publishing contract. | @studnicky/event-bus/interfaces |
EventHandlerInterface | Defines the event handler contract. | @studnicky/event-bus/interfaces |
UnsubscribeInterface | Defines the unsubscribe contract. | @studnicky/event-bus/interfaces |