`in-memory-bus` — capture-all `PipelineEventsBus` for tests

Purpose

InMemoryPipelineEventsBus is the canonical fixture wired by createTestHost (Stage-5 keystone) and consumed by every smoke / integration test that needs to assert what happened inside the §6 11-step request pipeline. Every emitted event is appended to an internal FIFO and broadcast synchronously to every registered subscriber.

The bus is intentionally simple — capture-all (no sampling), drainable (so multi-request tests don't leak), with byRequest and byStage slicers for the common assertion patterns. It is NOT suitable for production observability (unbounded memory growth, single-process); production bridges to OTel via a separate adapter in @suaveplan/api-host-otel-pipeline (follow-up change).

Features

FIFO capture of every emitted event, accessible via events, length, byRequest(id), byStage(name).
Synchronous in-order fan-out to subscribers (registration order).
Subscriber exception isolation — a throwing subscriber does not halt fan-out or prevent FIFO append.
Idempotent subscribe return — calling the returned unsubscribe twice is a no-op.
drain() returns and clears the captured FIFO in one call for between-request reset.

Usage: Basic

import { InMemoryPipelineEventsBus } from "@suaveplan/api-host";

const bus = new InMemoryPipelineEventsBus();
bus.emit({ requestId: "r1", stage: "auth", outcome: "entered", at: 1 });
bus.emit({ requestId: "r1", stage: "auth", outcome: "completed", at: 2, durationMs: 1 });
expect(bus.byStage("auth")).toHaveLength(2);

Usage: Advanced

Drain between requests so each request's assertions stay isolated:

const bus = new InMemoryPipelineEventsBus();
host.adapters.pipelineEvents = bus; // wired automatically by createTestHost

await host.fetch(new Request("http://h/probe"));
const first = bus.drain();
expect(first.filter((e) => e.outcome === "entered")).toHaveLength(11);

await host.fetch(new Request("http://h/probe"));
const second = bus.drain();
expect(second).not.toEqual(first); // fresh requestId, fresh timestamps

Subscribe a live bridge in addition to the FIFO capture:

const bus = new InMemoryPipelineEventsBus();
const off = bus.subscribe((event) => console.log(event.stage, event.outcome));
// ...
off();
expect(bus.subscriberCount).toBe(0);

API Reference

export class InMemoryPipelineEventsBus implements PipelineEventsBus {
    emit(event: PipelineStageEvent): void;
    subscribe(listener: PipelineEventListener): PipelineEventUnsubscribe;
    readonly events: readonly PipelineStageEvent[];
    readonly length: number;
    drain(): PipelineStageEvent[];
    byRequest(requestId: string): PipelineStageEvent[];
    byStage(stage: PipelineStage): PipelineStageEvent[];
    readonly subscriberCount: number;
}

Implementation Notes

events returns a copy on every call so mutating the snapshot does not corrupt the internal FIFO. Test authors who hot-loop on the assertion path should cache the result in a local variable to avoid the O(n) copy on every read.

Subscriber exceptions are swallowed silently in the test fixture. Production bridges (e.g. the OTel bridge) MUST log via the configured boundary logger — but the in-memory fixture deliberately stays silent so tests aren't drowning in console.error noise from intentionally-throwing subscribers used to verify the isolation invariant.

The bus does NOT enforce ordering across requestIds — interleaved requests produce interleaved events in the FIFO. Use byRequest(id) to slice; the conformance harness does this for the per-request order invariant.

3.6 KiB Raw Blame History

in-memory-bus — capture-all PipelineEventsBus for tests