Strongbus

A strongly typed, battle tested event emitter built for complex event-driven applications.

Strongbus gives you a fully type-safe event bus: events and their payloads are described by a single EventMap, and every method (on, once, emit, any, pipe, next, scan, hook) is checked against it. It also ships with subscription lifecycle hooks, bus-to-bus piping, promise-based event awaiting, and built-in memory-leak detection.

API Documentation v3 (current)
API Documentation v2
Migrating from v2 -> v3

Features

Installation

npm install strongbus
# or
yarn add strongbus

Quick start

Describe your events as a map of event name -> payload type, then parameterize a Bus with it. Use void for events that carry no payload.

import {Bus} from 'strongbus';

interface Events {
  message: string;
  count: number;
  connected: void;
}

const bus = new Bus<Events>();

const subscription = bus.on('message', (payload) => {
  // payload is inferred as `string`
  console.log(payload.toUpperCase());
});

bus.emit('message', 'hello');   // logs "HELLO"
bus.emit('connected');          // no payload required

subscription();                 // unsubscribe

Type safety is enforced against the event map:

bus.on('mesage', () => {});        // compile error: 'mesage' is not a key of Events
bus.on('count', (n: string) => {}); // compile error: 'count' carries a number
bus.emit('message');                // compile error: 'message' requires a string payload

Emitting events

emit(event, payload) returns a boolean indicating whether the event was handled (by an own listener, a wildcard sink, or a downstream bus).

const handled = bus.emit('message', 'hello');

Events mapped to a void payload can be emitted with no second argument (or null/undefined):

bus.emit('connected');
bus.emit('connected', null);

By default, emitting an event with no listeners is a no-op (onUnhandledEvent: 'ignore'). Set onUnhandledEvent: 'throw' to throw, or pass a (event, payload) => void callback to handle unhandled events yourself.

Subscriptions

Every subscriber returns a Subscription — a function that releases the subscription. It can be disposed by invoking it directly or via .unsubscribe(); both are idempotent. For Node EventEmitter familiarity, off(event, handler) removes a handler by the same function reference passed to on (returns void).

const sub = bus.on('message', handler);

sub();              // release
sub.unsubscribe();  // equivalent; safe to call again

// or, with the original handler reference:
bus.off('message', handler);

Prefer keeping the Subscription when you can; off is useful when the caller never stored it. It only removes handlers registered with on — not wrappers from once, any, or pipe.

Subscribing to events

on(event, handler, options?)

Subscribe a handler to a single event. The handler receives the event’s payload. Duplicate registrations for the same event and handler reference are governed by the bus’s duplicateSubscriptionStrategy (default: collapse — same Subscription, one invoke, warn on duplicate). Optional SubscribeOptions (see Incognito subscriptions).

bus.on('count', (n) => console.log(n)); // n: number

once(event, handler, options?)

Like on, but automatically unsubscribes after the first time the event fires. once honors options.duplicateSubscriptionStrategy.{observability, invocation, logLevel}, but disposal is always frame-based and kind-isolated from on for the same handler (off never removes once intent).

bus.once('connected', () => console.log('connected exactly once'));

any(events, handler, options?)

Subscribe one handler to several events. The handler receives the raised event as its first argument and the payload as its second.

bus.any(['message', 'count'], (event, payload) => {
  // event: 'message' | 'count'
  // payload: string | number
});

Piping

pipe attaches a graph edge into another Bus. Use tap to observe every event without linking. Use unpipe to detach a bus edge. Multi-hop graphs need an explicit call-site filter — see docs/pipe_limitations.md. For a worked analogy of directed multi-hop delivery (capture / bubble style), see DOM-style capture and bubble example.

pipe(bus, options?) — downstream piping

Pipe into another Bus, counting the downstream’s listeners as handlers on this bus (unless {incognito: true} — see Incognito subscriptions). Returns the downstream bus (so pipes chain and subclasses are preserved). The downstream must be a real Bus instance.

Shared events must be payload-compatible: identical types, or a one-way widen within the same primitive family ('a'|'b' → string, true → boolean, 1|2 → number). Object payloads still require an exact match.

const producer = new Bus<Events>();
const leaf = new Bus<Events>();

producer.pipe(leaf);         // events emitted on `producer` are observed by `leaf`'s listeners
leaf.on('message', handle);

producer.emit('message', 'hi'); // `handle` is invoked

producer.unpipe(leaf);       // detach

tap(handler, options?) — observe without linking

Observe every raised event as a correlated {event, payload} message. Does not create a graph edge. The returned Subscription removes it (duplicate tap(sameHandler) follows duplicateSubscriptionStrategy). Accepts the same SubscribeOptions as on.

const stop = bus.tap((piped) => {
  console.log(`${String(piped.event)}:`, piped.payload);
  if (piped.event === 'didRemoveItem') {
    cache.delete(piped.payload.id); // payload narrowed to this event's type
  }
});

stop(); // stop receiving all events

pipe(pred).pipe(bus) — filtered multi-hop

When a bus is both a pipe target and a pipe source, unfiltered passthrough is blocked (with a warning per unique source → bridge → dest path). Allow selected relayed events with a predicate on the outbound edge:

a.pipe(b);
b.pipe((msg) => msg.event === 'foo').pipe(c);
// or, when you assert the path is sound:
// b.pipe(ASSUMED_SOUND_EDGE).pipe(c);

Local raises on b still reach c without consulting the predicate; the filter gates passthrough only.

Hubs and linear chains

root.pipe(leaf) — downstream piping. Reach for this when you need root to know when listeners for specific events are added or removed through leaf — its willAddListener / didAddListener / willRemoveListener / didRemoveListener hooks fire for the downstream’s listeners — or when you have a linear chain of buses. The downstream’s listeners count toward root’s listener count, and pipes chain (node1.pipe(node2).pipe(node3)). For multi-hop chains with different maps, filter on bridge edges (mid.pipe(pred).pipe(leaf)).

type Events = {foo: number};
const producer = new Bus<Events>();

let producing = false;
const produce = (payload: number) => {
  if (producing) {
    producer.emit('foo', payload);
  }
};

// `producer` reacts to demand for 'foo' anywhere downstream
producer.hook('didAddListener', (event) => { if (event === 'foo') producing = true; });
producer.hook('didRemoveListener', (event) => { if (event === 'foo') producing = false; });

const node1 = new Bus<Events>();
node1.on('foo', (payload) => console.log(`node 1 received event=foo, payload=${payload}`));
const node2 = new Bus<Events>();
node2.on('foo', (payload) => console.log(`node 2 received event=foo, payload=${payload}`));
const node3 = new Bus<Events>();
node3.on('foo', (payload) => console.log(`node 3 received event=foo, payload=${payload}`));

produce(1);           // producing === false, nothing emitted

producer.pipe(node1); // events flow producer -> node1; 'foo' now has a downstream listener
produce(1);           // logs "node 1 received event=foo, payload=1"

// same EventMap: allow passthrough with an explicit filter on bridge edges
node1.pipe(() => true).pipe(node2);
node2.pipe(() => true).pipe(node3);
produce(6);          // logs on node1, node2, and node3

node2.unpipe(node3);
node1.unpipe(node2);
producer.unpipe(node1); // 'foo' has no downstream listeners again; `producing` flips back to false
produce(5); // logs nothing

// alternatively, using `hasListenersFor` and no `.hook`s
const produce = (payload: number) => {
  if(producer.hasListenersFor('foo')) {
    producer.emit('foo', payload);
  }
}

Hubs — feeder.pipe(hub). Many feeders into one hub:

const hub = new Bus<Events>();
hub.on('foo', handleFoo);

const feederA = new Bus<Events>();
const feederB = new Bus<Events>();

feederA.pipe(hub);
feederB.pipe(hub);

feederA.emit('foo', payload);  // handleFoo called
feederB.emit('foo', payload);  // handleFoo called

Incognito subscriptions

Pass {incognito: true} as trailing SubscribeOptions on on, once, any, tap, pipe(bus), next, or scan when the registration should still receive or forward events but must not count toward this bus’s monitoring subsystem:

bus.on('message', logMessage, {incognito: true});
bus.tap(telemetrySink, {incognito: true});
src.pipe(target, {incognito: true}); // forward events; target's listeners do not activate src
await bus.next('connected', {incognito: true});

pipe(bus, {incognito: true}) still forwards events through the target (and any further chain that allows passthrough). It does not couple the target’s listener tree into the source’s monitoring; the target’s own active / hooks are unchanged. A second pipe of the same bus is a no-op (the first link’s mode sticks), independent of duplicateSubscriptionStrategy.

off(event, handler) still removes an incognito on registration by handler reference (no options). The same behavior applies when releasing the returned Subscription or unpipe for once / any / tap / pipe(bus).

Memory-leak logger thresholds still count own incognito handlers (real retention). To include incognito interest in queries, pass {includeIncognito: true} to introspection methods — that never flips active.

Awaiting events

next(resolutionTrigger, rejectionTrigger?, options?)

Returns a CancelablePromise that resolves with {event, payload} when the resolution trigger fires. Optionally provide a disjoint rejection trigger, and/or trailing SubscribeOptions. Triggers are a single event key or an array of events (SubscribableListenable); '*' is not accepted.

const {event, payload} = await bus.next('message');
// event: 'message', payload: string

const result = await bus.next(['message', 'connected', 'count']);

// resolve on either event; reject if `count` fires first
const result = await bus.next(['message', 'connected'], 'count');

await bus.next('message', {incognito: true});
await bus.next('message', 'count', {incognito: true});

// resolution and rejection triggers must be disjoint (compile error otherwise)
bus.next('message', 'message'); // compile error

scan(trigger, evaluator, options?)

Resolve or reject a promise based on an evaluation run whenever a trigger fires. The evaluator is handed a resolve/reject pair and decides whether the current state settles the promise. Triggers are Listenable — a single event, an array of events, or the wildcard ('*'). When reading resolve.trigger.payload, narrow on event first.

const ready = await bus.scan<boolean>(
  ['message', 'connected'],
  (resolve) => {
    if (resolve.trigger.type === 'event' && resolve.trigger.event === 'connected') {
      resolve(true);
    }
  }
);

scan options (see the scan API docs for canonical defaults):

The <T> type argument is the resolved value’s type and flows into the resolver:

bus.scan<number>('count', (resolve) => resolve('nope')); // compile error: expected number

The object form scan({evaluator, trigger, ...options}) is deprecated but still supported.

Lifecycle hooks

hook(lifecycleEvent, handler) subscribes to meta events about the bus itself:

Event Fired when
willActivate / active the bus goes from 0 to 1 listeners
willIdle / idle the bus goes from 1 to 0 listeners
willAddListener / didAddListener a listener is added (payload: the event)
willRemoveListener / didRemoveListener a listener is removed (payload: the event)
willDestroy destroy() is called
error a listener or hook throws / rejects (payload: {error, event}) 
bus.hook('error', ({error, event}) => {
  console.error(`handler for "${String(event)}" failed`, error);
});

monitor(handler)

Shorthand for observing the active/idle transition. The handler receives true when the bus becomes active and false when it becomes idle.

bus.monitor((isActive) => console.log(isActive ? 'active' : 'idle'));

Lifecycle hook ordering

Hooks follow transition-centric bracketing: activation and idle bracket the bus going from 0→1 and 1→0 listeners.

Direct on() / unsubscribe() — one episode per listener. Adding two handlers for the same event:

willActivate → willAdd → didAdd → active → willAdd → didAdd

Removing those two handlers:

willRemove → didRemove → willIdle → willRemove → didRemove → idle

willActivate / willIdle fire only on the 0↔1 transitions. active follows the first didAdd that gives the bus demand; willIdle immediately precedes the willRemove for the last listener.

pipe() / unpipe() with pre-existing downstream listeners — the upstream bus reconciles the downstream snapshot using the same transition rules. With the default coalesceDownstreamLifecycleEvents: true, reconcile emits one willAdd/didAdd (or willRemove/didRemove) pair per event key when several handlers share that key. Example: piping a downstream that already has two listeners on foo:

willActivate → willAdd:foo → didAdd:foo → active

Unpiping that downstream when it was the bus’s only demand:

willIdle → willRemove:foo → didRemove:foo → idle

Set options.coalesceDownstreamLifecycleEvents: false to get one hook pair per listener during reconcile (matching the direct on() order above). Listener counts are the same either way.

After pipe() — incremental adds/removes on the downstream bus still emit hooks per listener; coalescing applies only to the initial pipe() / final unpipe() reconcile.

Partial unpipe() — when other downstream or own listeners remain, willIdle / idle are not emitted on the upstream bus.

Surfaces

Bus implements typed surfaces over the same event map. Compose them when a dependency needs only part of the API.

Surface Methods Use when
EmittingSurface emit Raising events without teardown
ControlSurface emit, destroy, clear Raising events, or clearing or tearing down a bus
SubscriptionSurface on, once, any, next, scan, tap, pipe, unpipe Subscribing, awaiting, or forwarding events
IntrospectionSurface hasListeners, getListeners, getListenerCount, hasListenersFor, getEventCount, getListenerCountFor, getListenersFor, forEach Inspecting listener state
MonitoringSurface monitor, hook, active Observing lifecycle and active/idle state

ControlSurface extends EmittingSurface. Prefer EmittingSurface when a dependency may raise events but must not destroy or clear the bus.

A Bus<Wide> is assignable to any narrower view (for example SubscriptionSurface<Narrow>) so consumers can declare only the events and capabilities they need. The same Wide→Narrow assignability holds surface-to-surface (SubscriptionSurface<Wide> → SubscriptionSurface<Narrow>), matching the flattened method picks domain objects used to expose.

Branding domain objects

When a domain object should expose a surface without re-declaring on / tap / pipe / … as public aliases, brand it and unwrap at the call site:

import {Bus, SubscriptionSurface, MonitoringSurface} from 'strongbus';

interface ServiceEvents {
  status: string;
}

class Service implements SubscriptionSurface.Branded<ServiceEvents> {
  private readonly bus = new Bus<ServiceEvents>();
  public readonly [SubscriptionSurface.BRAND] = this.bus;
}

const service = new Service();
SubscriptionSurface(service).on('status', handle);

Brand only the surfaces the object actually exposes. Separate brands (@@StrongbusSubscriptionSurface, @@StrongbusEmittingSurface, …) keep partial exposure sound — branding a subscription surface does not claim control or monitoring.

SubscriptionSurface

Type a dependency as SubscriptionSurface<Events> when it may listen, await, pipe, or scan but must not raise events.

import {Bus, type SubscriptionSurface} from 'strongbus';

interface AppEvents {
  message: string;
  count: number;
}

interface FeatureEvents {
  message: string;
}

function wireFeature(source: SubscriptionSurface<FeatureEvents>) {
  source.on('message', handle);   // ok
  source.on('count', handle);     // compile error: 'count' is not in FeatureEvents
}

const app = new Bus<AppEvents>();
wireFeature(app);                 // Bus<AppEvents> is a SubscriptionSurface<FeatureEvents>

Because event-map typing is bivariant on the surfaces (Wide→Narrow assignability, same as Bus), a bus that emits a wider map can be passed where only a subset of events is relevant. Methods such as scan, any, next, and pipe respect the declared map — listening to unknown events are compile errors on the narrowed view.

Composing event maps

Event maps are plain types, so you can compose them. An intersection (A & B) works for concrete maps, but breaks down when one side is an open generic type parameter: indexing (Fixed & TGeneric)[K] produces a deferred type (e.g. string & TGeneric[K]), so a generic base class can’t emit a fixed event with a plain literal payload without casting.

Merge<Base, Ext> is a flattening merge that avoids this — overlapping keys take Base, and every key resolves to a single payload rather than an intersection:

import {Bus, type Merge} from 'strongbus';

interface BaseEvents {
  healthChanged: string;
  ready: void;
}

// merge the fixed map as `Base` and the open generic as the sole `Ext`
abstract class Connection<TIncoming extends object> {
  protected readonly bus = new Bus<Merge<BaseEvents, TIncoming>>();

  public reportHealth(status: string): void {
    this.bus.emit('healthChanged', status); // literal payload, no cast
    this.bus.emit('ready', null);           // void event; see note below
  }
}

Position matters: a fixed key emits cleanly only when it resolves in a layer whose keyset is concrete, before any layer that folds in the open generic’s keyset. When you have several fixed maps, flatten them together first and merge the open generic last — prefer Merge<Merge<BaseEvents, MoreFixed>, TGeneric> over nesting the generic in an inner layer such as Merge<BaseEvents, Merge<MoreFixed, TGeneric>>.

Note: whenever the event map is still an open generic (whether merged or a bare type parameter like Bus<T>), emit void events with an explicit null (emit('ready', null)) rather than the no-argument form (emit('ready')). The no-argument overload gates on VoidEventKeys, which filters the whole keyset and can’t confirm a key is void while any part of that keyset is an unresolved generic. The explicit-null form rides the correlated overload (gated on keyof membership) and always resolves.

Forwarding a generic event key

emit proves that (event, payload) is a correlated pair — you can’t pass a union-typed event with a union-typed payload without first discriminating on event (so an un-narrowed pipe sink can’t forward a mismatched pair). Forwarding a single generic key emit(event, payload) still works, but only when the bus map is a naked type parameter. TypeScript preserves the [K, M[K]] correlation for Bus<M>, but drops it when indexing a computed mapped type like Merge<Fixed, TIncoming>. So make the whole map the type parameter when you need to forward:

abstract class Relay<M extends BaseEvents> {
  protected readonly bus = new Bus<M>();

  // `[K, M[K]]` stays correlated over a naked `M`, so this type-checks
  public forward<K extends keyof M>(event: K, payload: M[K]): boolean {
    return this.bus.emit(event, payload);
  }
}

If you must key the bus off a Merge, forward inside a discriminated branch (a switch/if on event) so each arm emits a literal key instead.

Introspection

The introspection methods take optional IntrospectionOptions:

ListenerScope:

bus.active / monitor / lifecycle hooks always ignore incognito interest.

import {Bus, ListenerScope} from 'strongbus';

bus.active;                                                          // boolean: monitored subscribers only
bus.hasListeners(/* options? */);                                    // any scope (default)
bus.getListenerCount(/* options? */);                                // total handlers in scope
bus.getListeners(/* options? */);                                    // union of all handlers in scope
bus.getEventCount(/* options? */);                                   // events with at least one listener in scope

bus.hasListenersFor('message', {scope: ListenerScope.OWN});
bus.getListenerCountFor('message', {scope: ListenerScope.DOWNSTREAM});
bus.getListenersFor('message', /* options? */);                        // empty set when none
bus.forEach((event, handlers) => { /* ... */ }, /* options? */);
bus.hasListeners({includeIncognito: true});

Teardown

destroy(reason?) releases all event subscribers, fires willDestroy, removes all hooks, detaches all downstreams, and detaches it from every upstream that pipes into it. Afterwards, bus.destroyed is true.

bus.destroy('session closed');

To start over without destroying the bus, clear() cancels pending next / scan promises, releases all event subscribers, and severs every pipe edge in both directions. Lifecycle hooks stay registered and fire for the removed listeners (including idle), and the bus stays usable: destroyed remains false.

bus.clear();
bus.on('foo', handler); // works as on a fresh bus

clear takes the same options shape as the introspection methods, with different defaults (scope: ListenerScope.ANY, includeIncognito: true):

bus.clear({scope: ListenerScope.DOWNSTREAM}); // unpipe everything, keep own listeners
bus.clear({includeIncognito: false});         // reset, but keep incognito plumbing

By default a destroyed bus still accepts calls, so a consumer that outlives the bus’s owner keeps subscribing without any sign of trouble. Set onUseAfterDestroy to catch this:

const bus = new Bus<Events>({onUseAfterDestroy: 'warn'});
bus.destroy('session closed');

bus.on('foo', handler);
// logs StrongbusLogCode.UseAfterDestroy:
// "<name> Bus: on() invoked after destroy (reason: session closed)"
// and returns a no-op Subscription without registering `handler`

Stubs: on, once, any, tap, hook, and monitor return a no-op Subscription; pipe(bus) returns bus without linking; pipe(predicate) returns a handle whose pipe does nothing; next and scan return a CancelablePromise that never settles unless canceled; off and unpipe return.

emit and destroy are exempt from logging and throwing: under any onUseAfterDestroy setting other than false, they quietly do nothing (emit returns false). Introspection methods (hasListeners, getListenerCount, destroyed, etc.) are never affected.

Piping a live bus into a destroyed one counts as using the destroyed bus: it is reported under the destroyed bus’s onUseAfterDestroy (as <live name>.pipe()), and no edge is created.

While destroy() is running, including inside willDestroy and the remove-listener hooks it fires, the bus is already off-limits regardless of onUseAfterDestroy: those same methods are skipped and logged at debug as StrongbusLogCode.UseDuringDestroy, returning the stubs above, except that next and scan return an already-canceled promise. A nested destroy() or clear() does nothing. emit still runs, so a willDestroy hook can send a final event to downstreams that are still attached.

Options

new Bus<Events>(options) accepts:

const bus = new Bus<Events>({
  name: 'MyBus',
  onUnhandledEvent: 'throw',
  thresholds: {warn: 50},
  // logger omitted → built-in console adapter (record.message → console)
  duplicateSubscriptionStrategy: DuplicateSubscriptionStrategy.SharedHandler
});

See the Options and ListenerThresholds API docs for the canonical defaults and types.

Defaults for all instances can be set globally with Bus.configure():

Bus.configure({
  onUnhandledEvent: 'throw',
  thresholds: {warn: 50},
  logger: myLogger
});

configure recursively merges onto static defaults.

Memory-leak detection

When the number of listeners for a single event crosses a configured threshold, Strongbus logs a message at the corresponding severity (info/warn/error). This surfaces leaks from subscriptions that are never released. With verbose: false (the default), messages are throttled to threshold boundaries and multiples rather than every listener. Set verbose: true to log on every listener added above a threshold.

License

MIT