@studnicky/event-bus

Typed publish/subscribe with per-subscriber backpressure isolation and AbortSignal lifecycle.

Install

bash

pnpm add @studnicky/event-bus

Requires @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:

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:

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:

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	`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	`topic: K`, `depth: number`
`onSlowConsumer(topic, depth)`	Same as onOverflow — queue reached highWaterMark	`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)`	Item added to queue	`depth: number`
`onDequeue(depth)`	Item removed from queue for processing	`depth: number`
`onDrop()`	Enqueue called on aborted queue	—
`onOverflow(depth)`	Queue depth reached highWaterMark	`depth: number`
`onSlowConsumer(depth, highWaterMark)`	Same moment as onOverflow with hwm context	`depth: number`, `highWaterMark: number`
`onHandlerError(error)`	Handler threw	`error: unknown`

import { EventBus } from '../src/index.js';

class TracedBus extends EventBus<OrderStatusEventMapEntity.Type> {
  static override create(): 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<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): 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<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): 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<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): void {
    console.log(`[event-bus] subscribe topic=${String(topic)}`);
    this.subscribeLog.push(String(topic));
  }
  protected override onUnsubscribe<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): 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.

Try it

The builder demo constructs a typed EventBus via EventBus.builder<AppEvents>().build() and publishes three events across two topics. Watch each subscriber receive only the events for its topic, and note the final drain() call ensures all handlers complete before the bus closes.

EventBus builder

/** builderEventBus — constructs an EventBus via EventBus.builder().build() and exercises subscribe/publish/drain/close. Run: npx tsx examples/builderEventBus.ts */

import assert from 'node:assert/strict';

import type { OrderLifecycleEventMapEntity } from './entities/OrderLifecycleEventMapEntity.js';

// #region usage
import { EventBus } from '../src/index.js';

// Build an EventBus using the fluent builder
const bus = EventBus.builder<OrderLifecycleEventMapEntity.Type>().build();

console.log('EventBus built.');

const received: { 'payload': unknown; 'topic': string; }[] = [];

// Subscribe to both topics
bus.subscribe('order:placed', (payload) => {
  console.log(`[handler] order:placed id=${payload.id} total=${payload.total}`);
  received.push({ 'payload': payload, 'topic': 'order:placed' });
});

bus.subscribe('order:shipped', (payload) => {
  console.log(`[handler] order:shipped id=${payload.id} carrier=${payload.carrier}`);
  received.push({ 'payload': payload, 'topic': 'order:shipped' });
});

// Publish events
await bus.publish('order:placed', { 'id': 'ORD-1', 'total': 149 });
await bus.publish('order:shipped', { 'carrier': 'FedEx', 'id': 'ORD-1' });
await bus.publish('order:placed', { 'id': 'ORD-2', 'total': 29 });

await bus.drain();

console.log('Events received:', received.length);
for (const { payload, topic } of received) {
  console.log(` - ${topic}:`, payload);
}

await bus.close();
// #endregion usage

assert.equal(received.length, 3, 'All 3 published events received');
assert.equal(received[0]?.topic, 'order:placed');
assert.equal(received[1]?.topic, 'order:shipped');
assert.equal(received[2]?.topic, 'order:placed');

console.log('builderEventBus: all assertions passed');

Output

Press Execute to run this example against the real library.

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().

EventBus lifecycle hooks

/** observedEventBus — trace every lifecycle stage with console.log overrides. Run: npx tsx examples/observedEventBus.ts */

import assert from 'node:assert/strict';

import type { OrderStatusEventMapEntity } from './entities/OrderStatusEventMapEntity.js';

// #region usage
import { EventBus } from '../src/index.js';

class TracedBus extends EventBus<OrderStatusEventMapEntity.Type> {
  static override create(): 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<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): 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<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): 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<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): void {
    console.log(`[event-bus] subscribe topic=${String(topic)}`);
    this.subscribeLog.push(String(topic));
  }
  protected override onUnsubscribe<K extends keyof OrderStatusEventMapEntity.Type>(topic: K): 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();
// #endregion usage

// Assertions
assert.equal(bus.subscribeLog.length, 3, 'Three subscribes');
assert.equal(bus.publishLog.length, 3, 'Three publishes');
// 2 handlers × 2 order:created publishes = 4, plus 1 handler × 1 order:updated = 5 delivers
assert.equal(bus.deliverLog.length, 5, 'Five delivers total');
// 2 subscribers × 2 publishes + 1 × 1 = 5 enqueues
assert.equal(bus.enqueueLog.length, 5, 'Five enqueues');
assert.equal(bus.dequeueLog.length, 5, 'Five dequeues');
assert.equal(bus.unsubscribeLog.length, 1, 'One explicit unsubscribe');
assert.equal(bus.disposeLog.length, 1, 'One dispose');

console.log('observedEventBus: all assertions passed');

Output

Press Execute to run this example against the real library.

API

Export	Type	Description
`EventBus<TTopicMap>`	class	Multi-topic pub/sub; created via `EventBus.create<T>()` or `EventBus.builder<T>().build()`
`EventBusBuilder<TTopicMap>`	class	Fluent builder for `EventBus`; obtained via `EventBus.builder<T>()`
`BusQueue<T>`	class	Bounded FIFO queue with backpressure; created via `BusQueue.create(options)` or `BusQueue.builder<T>()`
`BusQueueBuilder<T>`	class	Fluent builder for `BusQueue`; obtained via `BusQueue.builder<T>()`
`EventHandlerType<T>`	type	`(payload: T, signal: AbortSignal) => Promise<void> \| void`
`UnsubscribeType`	type	`() => void` (returned by `subscribe`)
`BusQueueCreateOptionsType<T>`	type	`{ handler, highWaterMark?, onError?, signal? }`

`EventBus<TTopicMap>`

Member	Signature	Description
`create`	`static create<T>() => EventBus<T>`	Factory; constructor is protected
`builder`	`static builder<T>() => EventBusBuilder<T>`	Returns a fluent builder
`subscribe`	`(topic, handler, options?) => UnsubscribeType`	Registers a subscriber; returns unsubscribe fn
`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: BusQueueCreateOptionsType<T>) => BusQueue<T>`	Factory; constructor is protected
`builder`	`static builder<T>() => BusQueueBuilder<T>`	Returns a fluent builder
`enqueue`	`(item: T) => Promise<void>`	Adds item; blocks caller when at `highWaterMark`
`drain`	`() => Promise<void>`	Resolves when queue is empty or aborted
`size`	`number`	Current queue depth

Source on GitHub

@studnicky/event-bus ​

Install ​

Usage ​

Multiple subscribers ​

AbortSignal-based lifecycle ​

Observability hooks ​

EventBus hooks ​

BusQueue hooks ​

Try it ​

API ​

EventBus<TTopicMap> ​

BusQueue<T> ​