feat(core-events): InMemoryEventBus with failFast option
TDD: red (test-only), then green. InMemoryEventBus validates payloads via descriptor.schema before fanout, delivers to all handlers via Promise.allSettled, swallows handler errors by default, and rethrows first error when failFast:true. 5 new tests (8 total passing). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
40
packages/core-events/src/in-memory-event-bus.ts
Normal file
40
packages/core-events/src/in-memory-event-bus.ts
Normal file
@@ -0,0 +1,40 @@
|
||||
import type { z } from "zod";
|
||||
import type { EventDescriptor } from "./event-descriptor";
|
||||
import type { EventHandler, IEventBus } from "./event-bus.interface";
|
||||
|
||||
export type InMemoryEventBusOptions = {
|
||||
/** When true, rethrow the first handler error (default: false — errors swallowed). */
|
||||
failFast?: boolean;
|
||||
};
|
||||
|
||||
export class InMemoryEventBus implements IEventBus {
|
||||
private readonly handlers = new Map<string, EventHandler<unknown>[]>();
|
||||
|
||||
constructor(private readonly options: InMemoryEventBusOptions = {}) {}
|
||||
|
||||
async publish<T>(
|
||||
descriptor: EventDescriptor<string, z.ZodType<T>>,
|
||||
payload: T,
|
||||
): Promise<void> {
|
||||
descriptor.schema.parse(payload);
|
||||
const subscribers = this.handlers.get(descriptor.name) ?? [];
|
||||
if (subscribers.length === 0) return;
|
||||
const settled = await Promise.allSettled(
|
||||
subscribers.map((h) => h(payload)),
|
||||
);
|
||||
if (this.options.failFast) {
|
||||
const failure = settled.find((s) => s.status === "rejected");
|
||||
if (failure && failure.status === "rejected") throw failure.reason;
|
||||
}
|
||||
}
|
||||
|
||||
subscribe<T>(
|
||||
descriptor: EventDescriptor<string, z.ZodType<T>>,
|
||||
_consumerFeature: string,
|
||||
handler: EventHandler<T>,
|
||||
): void {
|
||||
const arr = this.handlers.get(descriptor.name) ?? [];
|
||||
arr.push(handler as EventHandler<unknown>);
|
||||
this.handlers.set(descriptor.name, arr);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user