@studnicky/event-bus
Typed publish/subscribe with per-subscriber backpressure isolation and AbortSignal lifecycle.
Install
bash
pnpm add @studnicky/event-busRequires @studnicky:registry=https://npm.pkg.github.com in .npmrc.
Usage
Subscribe to a topic, publish a payload, and drain the queue. The subscriber receives every published item:
ts
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
All subscribers on the same topic receive each published payload independently. Calling the returned unsubscribe function removes that subscriber:
ts
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) => {
MultiSubscriberFixture.receivedA.push(payload.orderId);
});
bus.subscribe('order:placed', (payload) => {
MultiSubscriberFixture.receivedB.push(payload.orderId);
});
// Both handlers receive the first event
await bus.publish('order:placed', { 'orderId': 'order-1' });
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': 'order-2' });
await bus.drain();
console.log('After unsubscribe + second publish — A:', MultiSubscriberFixture.receivedA, 'B:', MultiSubscriberFixture.receivedB);AbortSignal-based lifecycle
Pass a signal option to bind a subscriber's lifetime to an AbortController. When the signal aborts the subscriber is removed and stops receiving events. The handler also receives the subscription's own AbortSignal as a second argument; it aborts on unsubscribe, on caller-signal abort, or on bus close. Use it to cancel in-flight async work:
ts
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);Observability hooks
Subclass EventBus or BusQueue and override any of the protected hook methods to instrument lifecycle events without modifying the base class.
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 | — |
BusQueue hooks
| Hook | When it fires | Args |
|---|---|---|
onEnqueue(depth) | Admission gate after the item is added; completes before handler delivery | depth: number |
onDequeue(depth) | Item removed from queue for processing | depth: number |
onDrop() | Enqueue called on aborted queue | — |
onOverflow(depth) | Admission gate when queue depth reaches highWaterMark; completes before handler delivery | depth: number |
onHandlerError(error) | Handler threw | error: unknown |
ts
import { EventBus } from '../src/index.js';
class TracedBus extends EventBus<OrderStatusEventMapEntity.Type> {
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.create();
// 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.
For standalone BusQueue subclasses, a rejection from onEnqueue or onOverflow cancels only that item. The queue skips the cancelled entry and continues later enqueues in FIFO order.
Try it
The pub/sub demo constructs a typed EventBus directly with EventBus.create<AppEvents>(), publishes an event, and drains the subscriber queue before closing.
Loading example…
The hooks demo subclasses EventBus and overrides seven protected lifecycle methods. Watch the full fan-out trace: subscribe fires once per handler registration; publish fires once per bus.publish() call; enqueue and dequeue fire once per subscriber per publish; and deliver fires after each handler invocation. unsubscribe fires for the explicit unsub1() call, and dispose fires on bus.close().
Loading example…
API
| Export | Type | Description |
|---|---|---|
EventBus<TTopicMap> | class | Multi-topic pub/sub; created via EventBus.create<T>(config?) |
BusQueue<T> | class | Bounded FIFO queue with backpressure; created via BusQueue.create(options) |
EventHandlerInterface<T> | interface | Callable handler contract: (payload: T, signal: AbortSignal) => Promise<void> | void |
UnsubscribeInterface | interface | Callable unsubscribe contract returned by subscribe: () => void |
BusQueueCreateOptionsInterface<T> | interface | Queue construction contract: { handler, highWaterMark?, onError?, signal? } |
BusQueueOptionsEntity | entity | Schema-backed bus and subscriber queue options |
EventBus<TTopicMap>
| Member | Signature | Description |
|---|---|---|
create | static create<T>(config?: BusQueueOptionsEntity.Type) => 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 |
BusQueue<T>
| Member | Signature | Description |
|---|---|---|
create | static create<T>(options: BusQueueCreateOptionsInterface<T>) => BusQueue<T> | Factory; constructor is protected |
enqueue | (item: T) => Promise<void> | Adds item; awaits admission hooks and blocks caller when at highWaterMark |
drain | () => Promise<void> | Resolves when queue is empty or aborted |
size | number | Current queue depth |
Entities
@studnicky/event-bus/entities exports every schema namespace in src/entities.
typescript
import { BusQueueOptionsEntity } from '@studnicky/event-bus/entities';Exports
| Symbol | Purpose | Import path |
|---|---|---|
BusQueue | Provides bus queue functionality. | @studnicky/event-bus |
BusQueueCreateOptionsInterface | Defines the bus queue create options contract. | @studnicky/event-bus |
BusQueueConfigError | Represents bus queue config failures. | @studnicky/event-bus |
EventBus | Provides event bus functionality. | @studnicky/event-bus |
EventBusError | Represents event bus failures. | @studnicky/event-bus |
EventHandlerInterface | Defines the event handler contract. | @studnicky/event-bus |
UnsubscribeInterface | Defines the unsubscribe contract. | @studnicky/event-bus |