From 9c765d32131bcac5566af848c7d6217b044e9791 Mon Sep 17 00:00:00 2001 From: John Wright Date: Wed, 10 Jun 2026 15:16:13 +0100 Subject: [PATCH 1/5] feat: add broker streaming --- package-lock.json | 29 ++-- packages/message-bus/package.json | 1 + packages/message-bus/src/Broker.ts | 33 ++++- packages/message-bus/src/MessageBus.ts | 131 +++++++++++++++--- packages/message-bus/src/array.ts | 46 +++++- .../src/errors/InvokableNotRegisteredError.ts | 10 +- packages/message-bus/src/types/streams.ts | 57 ++++++++ packages/message-bus/test/MessageBus.test.ts | 91 ++++++++++++ packages/message-bus/tsconfig.json | 2 +- 9 files changed, 356 insertions(+), 44 deletions(-) create mode 100644 packages/message-bus/src/types/streams.ts diff --git a/package-lock.json b/package-lock.json index 5706410d4..de4a99205 100644 --- a/package-lock.json +++ b/package-lock.json @@ -452,11 +452,11 @@ } }, "node_modules/@johngw/stream": { - "version": "4.2.0", - "resolved": "https://registry.npmjs.org/@johngw/stream/-/stream-4.2.0.tgz", - "integrity": "sha512-+t1pX6KZM6CtIFZS5MIIh8rR1tmwnhnZrGWA5cGXQBhIRAWFPpgcYcUpksHgv9tt5WuYCPSwhUcdG8L2oyYvUw==", + "version": "4.4.1", + "resolved": "https://registry.npmjs.org/@johngw/stream/-/stream-4.4.1.tgz", + "integrity": "sha512-cohmqSHNbjLAqSSV0eosXcLJZ9ZYZPdX3lpPOxkDlXFq/VbiHhyvUTmW5+jAHUkxGVm+Azyd9pOjNp91QT58vQ==", "dependencies": { - "@johngw/stream-common": "^2.1.1", + "@johngw/stream-common": "^2.3.1", "@types/node": "^25.0.0", "assert-never": "1.4.0", "ts-toolbelt": "9.6.0", @@ -464,9 +464,9 @@ } }, "node_modules/@johngw/stream-common": { - "version": "2.1.1", - "resolved": "https://registry.npmjs.org/@johngw/stream-common/-/stream-common-2.1.1.tgz", - "integrity": "sha512-KwzN2V3fhJnGH7sgLlakDCNnVOhsk8shzTYSyYeRRi2lIxAWhrr/MW87uSBeVyTHRs66btvWly/fg3J8uqg5wg==", + "version": "2.3.1", + "resolved": "https://registry.npmjs.org/@johngw/stream-common/-/stream-common-2.3.1.tgz", + "integrity": "sha512-QQf12kHlzjlTWmg0Q4N360up5HEQRHn0R/882o/gtmRcKX0n9RDrFy6Dr418vPBJgD2gT3IvYTM5T+R+4GPF2w==", "license": "MIT", "dependencies": { "@types/node": "^25.0.0", @@ -532,14 +532,14 @@ } }, "node_modules/@napi-rs/wasm-runtime": { - "version": "1.1.4", - "resolved": "https://registry.npmjs.org/@napi-rs/wasm-runtime/-/wasm-runtime-1.1.4.tgz", - "integrity": "sha512-3NQNNgA1YSlJb/kMH1ildASP9HW7/7kYnRI2szWJaofaS1hWmbGI4H+d3+22aGzXXN9IJ+n+GiFVcGipJP18ow==", + "version": "1.1.5", + "resolved": "https://registry.npmjs.org/@napi-rs/wasm-runtime/-/wasm-runtime-1.1.5.tgz", + "integrity": "sha512-AWPoBRJ9tsnVhor4sjO7rkni+7p+2IAEFj6cx06UgP10jkQHqay/36uRV/bFkgrh18D9vb4cr8Q0Pthskgzy+Q==", "dev": true, "license": "MIT", "optional": true, "dependencies": { - "@tybys/wasm-util": "^0.10.1" + "@tybys/wasm-util": "^0.10.2" }, "funding": { "type": "github", @@ -2676,9 +2676,9 @@ } }, "node_modules/semver": { - "version": "7.8.3", - "resolved": "https://registry.npmjs.org/semver/-/semver-7.8.3.tgz", - "integrity": "sha512-wnilbGyMxzbY7dNOl7jpKbLSjcfeweJWU5j4+u5qW+6/wuGD9KzIGOyZnQVSBM9E7DtWaaH3CyHkppYrKYoxwg==", + "version": "7.8.4", + "resolved": "https://registry.npmjs.org/semver/-/semver-7.8.4.tgz", + "integrity": "sha512-rUCObTnP32Q08R2uuIrt7r9PlEonuTmtuXYcW6s5kjdlj3xbnwe+21yXptAUYcMAABLkYYTtnmzb3w3EDZfueA==", "dev": true, "license": "ISC", "bin": { @@ -3269,6 +3269,7 @@ "dependencies": { "@johngw/async": "^5.0.1", "@johngw/async-iterator": "^4.0.1", + "@johngw/stream": "^4.2.0", "@plugola/invoke": "^2.0.3", "ts-toolbelt": "^9.6.0", "tslib": "^2.8.1" diff --git a/packages/message-bus/package.json b/packages/message-bus/package.json index 13cb26baa..8d06c9c91 100644 --- a/packages/message-bus/package.json +++ b/packages/message-bus/package.json @@ -33,6 +33,7 @@ "dependencies": { "@johngw/async": "^5.0.1", "@johngw/async-iterator": "^4.0.1", + "@johngw/stream": "^4.2.0", "@plugola/invoke": "^2.0.3", "ts-toolbelt": "^9.6.0", "tslib": "^2.8.1" diff --git a/packages/message-bus/src/Broker.ts b/packages/message-bus/src/Broker.ts index c95a7650e..00c2faecd 100644 --- a/packages/message-bus/src/Broker.ts +++ b/packages/message-bus/src/Broker.ts @@ -14,16 +14,18 @@ import type { } from './types/generators.js' import type { InvokerInterceptorArgs } from './types/invokables.js' import { ErrorHandler, Unsubscriber } from './types/MessageBus.js' +import { StreamReaderArgs, StreamsDict } from './types/streams.js' export default class Broker< Events extends EventsT = EventsT, EventGens extends EventGeneratorsT = EventGeneratorsT, - Invokables extends InvokablesDict = InvokablesDict + Invokables extends InvokablesDict = InvokablesDict, + Streamables extends StreamsDict = StreamsDict, > { constructor( - messageBus: MessageBus, + messageBus: MessageBus, id: string, - abortController: AbortController + abortController: AbortController, ) { this.#abortController = abortController this.#id = id @@ -40,7 +42,7 @@ export default class Broker< return this.abortSignal.aborted } - get abortSignal() { + get abortSignal(): AbortSignal { return this.#abortController.signal } @@ -54,7 +56,7 @@ export default class Broker< return this.#id } - readonly #messageBus: MessageBus + readonly #messageBus: MessageBus get messageBus() { return this.#messageBus } @@ -117,7 +119,7 @@ export default class Broker< async until< EventName extends keyof Events, - Args extends UntilArgs + Args extends UntilArgs, >( eventName: EventName, ...args: Args @@ -127,7 +129,7 @@ export default class Broker< async untilSignal< EventName extends keyof Events, - Args extends UntilArgs + Args extends UntilArgs, >( eventName: EventName, abortSignal: AbortSignal, @@ -218,4 +220,21 @@ export default class Broker< ): Unsubscriber { return this.#messageBus.interceptInvoker(this, invokableName, args) } + + reader( + streamName: StreamName, + ...args: StreamReaderArgs< + Streamables[StreamName]['args'], + Streamables[StreamName]['item'] + > + ): Unsubscriber { + return this.#messageBus.reader(this, streamName, args) + } + + stream( + streamName: StreamName, + ...args: Streamables[StreamName]['args'] + ): ReadableStream { + return this.#messageBus.stream(this, streamName, args) + } } diff --git a/packages/message-bus/src/MessageBus.ts b/packages/message-bus/src/MessageBus.ts index 5d57383dc..622685488 100644 --- a/packages/message-bus/src/MessageBus.ts +++ b/packages/message-bus/src/MessageBus.ts @@ -4,7 +4,7 @@ import { combineIterators, iteratorRace, } from '@johngw/async-iterator' -import { init, last, removeItem, replaceLastItem } from './array.js' +import { filterMap, init, last, removeItem, replaceLastItem } from './array.js' import Broker from './Broker.js' import MessageBusError from './MessageBusError.js' import { amend } from './object.js' @@ -30,7 +30,11 @@ import { Invokers, } from './types/invokables.js' import { Stringable, UnpackResolvableValue } from './types/util.js' -import { ErrorHandler, Unsubscriber } from './types/MessageBus.js' +import { + AddAbortSignal, + ErrorHandler, + Unsubscriber, +} from './types/MessageBus.js' import { anySignal, fromSignal } from './AbortController.js' import { InvokableNotRegisteredError } from './errors/InvokableNotRegisteredError.js' import { @@ -38,11 +42,21 @@ import { InvokerFn, InvokerRegistrationArgs, } from '@plugola/invoke' +import { + ReaderFn, + StreamReader, + StreamReaderArgs, + Streams, + StreamsDict, +} from './types/streams.js' +import { WritableReadablePair } from '@johngw/stream/transformers/WritableReadablePair' +import { mergeUnderlyingSource } from '@johngw/stream' export default class MessageBus< Events extends EventsT = EventsT, EventGens extends EventGeneratorsT = EventGeneratorsT, Invokables extends InvokablesDict = InvokablesDict, + Streamables extends StreamsDict = StreamsDict, > { #errorHandlers: ErrorHandler[] = [] #eventInterceptors: EventInterceptors = {} @@ -51,6 +65,7 @@ export default class MessageBus< #invokerInterceptors: InvokerInterceptors = {} #queued: Array<() => unknown> = [] #started = false + #streams: Streams = {} #subscribers: Subscribers = {} onError(errorHandler: ErrorHandler) { @@ -72,7 +87,11 @@ export default class MessageBus< ? abort : fromSignal(abort) - return new Broker(this, id, abortController) + return new Broker( + this, + id, + abortController, + ) } async start() { @@ -81,7 +100,7 @@ export default class MessageBus< } emit( - broker: Broker, + broker: Broker, eventName: EventName, args: Events[EventName], abortSignal?: AbortSignal, @@ -117,7 +136,7 @@ export default class MessageBus< } interceptEvent( - broker: Broker, + broker: Broker, eventName: EventName, args: EventInterceptorArgs, ): Unsubscriber { @@ -142,7 +161,7 @@ export default class MessageBus< } interceptInvoker( - broker: Broker, + broker: Broker, invokableName: InvokableName, args: InvokerInterceptorArgs< Invokables[InvokableName]['args'], @@ -170,7 +189,7 @@ export default class MessageBus< } on( - broker: Broker, + broker: Broker, eventName: EventName, args: SubscriberArgs, ): Unsubscriber { @@ -201,7 +220,7 @@ export default class MessageBus< } once( - broker: Broker, + broker: Broker, eventName: EventName, args: SubscriberArgs, ): Unsubscriber { @@ -222,7 +241,7 @@ export default class MessageBus< EventName extends keyof Events, Args extends UntilArgs, >( - broker: Broker, + broker: Broker, eventName: EventName, args: Args, abortSignal?: AbortSignal, @@ -250,7 +269,7 @@ export default class MessageBus< } generator( - broker: Broker, + broker: Broker, eventName: EventName, args: EventGeneratorArgs< EventGens[EventName]['args'], @@ -284,7 +303,7 @@ export default class MessageBus< } async *iterate( - broker: Broker, + broker: Broker, eventName: EventName, args: EventGens[EventName]['args'], abortSignal?: AbortSignal, @@ -304,7 +323,7 @@ export default class MessageBus< } iterateWithin( - broker: Broker, + broker: Broker, within: number, eventName: EventName, args: EventGens[EventName]['args'], @@ -318,7 +337,7 @@ export default class MessageBus< } async accumulate( - broker: Broker, + broker: Broker, eventName: EventName, args: EventGens[EventName]['args'], abortSignal?: AbortSignal, @@ -327,7 +346,7 @@ export default class MessageBus< } async accumulateWithin( - broker: Broker, + broker: Broker, within: number, eventName: EventName, args: EventGens[EventName]['args'], @@ -339,7 +358,7 @@ export default class MessageBus< } register( - broker: Broker, + broker: Broker, invokableName: InvokableName, allArgs: InvokerRegistrationArgs< Invokables[InvokableName]['args'], @@ -389,7 +408,7 @@ export default class MessageBus< } async invoke( - broker: Broker, + broker: Broker, invokableName: InvokableName, args: Invokables[InvokableName]['args'], abortSignal?: AbortSignal, @@ -408,6 +427,84 @@ export default class MessageBus< return this.#started ? handle() : this.#queue(broker, handle) } + reader( + broker: Broker, + streamName: StreamName, + allArgs: StreamReaderArgs< + Streamables[StreamName]['args'], + Streamables[StreamName]['item'] + >, + ): Unsubscriber { + const args = init(allArgs) as Streamables[StreamName]['args'] + + const fn = last(allArgs) as ReaderFn< + Streamables[StreamName]['args'], + Streamables[StreamName]['item'] + > + + const streamer: StreamReader = { + broker, + args, + fn, + } + + this.#streams[streamName] ??= [] + + this.#streams[streamName].push(streamer) + + const cancel = () => { + this.#streams[streamName] = removeItem( + streamer, + this.#streams[streamName] as any, + ) + } + + broker.onAbort(cancel) + + return cancel + } + + stream( + broker: Broker, + streamName: StreamName, + args: Streamables[StreamName]['args'], + abortSignal?: AbortSignal, + ): ReadableStream { + type Item = Streamables[StreamName]['item'] + + const streamers = this.#streams[streamName] ?? [] + + const abortSignalComposite = anySignal(abortSignal, broker.abortSignal) + + const streams = () => + filterMap( + streamers, + (streamer) => this.#argumentIndex(streamer.args, args), + (_streamer, argumentIndex) => argumentIndex !== -1, + (streamer, argumentIndex) => + new ReadableStream( + streamer.fn( + ...([ + ...streamer.args.slice(0, argumentIndex), + ...args.slice(argumentIndex), + abortSignalComposite, + ] as AddAbortSignal), + ), + ), + ) + + return new ReadableStream({ + start: async (controller) => { + if (!this.#started) await this.#queue(broker, () => {}) + const abort = () => controller.error(abortSignalComposite.reason) + if (abortSignalComposite.aborted) return abort() + abortSignalComposite.addEventListener('abort', abort) + }, + }).pipeThrough( + new WritableReadablePair({}, mergeUnderlyingSource(streams)), + ) + } + #callEventInterceptors( eventName: EventName, args: Events[EventName], @@ -527,7 +624,7 @@ export default class MessageBus< } async #queue( - broker: Broker, + broker: Broker, handler: () => T, ) { return new Promise>((resolve, reject) => { diff --git a/packages/message-bus/src/array.ts b/packages/message-bus/src/array.ts index 509015436..854a9f5b7 100644 --- a/packages/message-bus/src/array.ts +++ b/packages/message-bus/src/array.ts @@ -20,8 +20,52 @@ export function removeIndex(index: number, array: T[]) { export function replaceLastItem(array: never[], item: unknown): never[] export function replaceLastItem( array: Ts, - item: T + item: T, ): L.Append, T> export function replaceLastItem(array: Ts, item: T) { return array.length === 0 ? array : [...init(array), item] } + +export function filterMap( + array: I[], + context: (input: I) => C, + filter: (input: I, context: C) => boolean, + map: (input: I, context: C) => O, +): O[] +export function filterMap( + array: I[], + filter: (input: I) => I, + map: (input: I) => O, +): O[] +export function filterMap( + array: I[], + contextOrFilter: ((input: I) => C) | ((input: I, context: C) => boolean), + filterOrMap: + | ((input: I, context: C) => boolean) + | ((input: I, context: C) => O), + maybeMap?: (input: I, context: C) => O, +): O[] { + type GetContext = (input: I) => C + type Filter = (input: I, context: C) => boolean + type Map = (input: I, context: C) => O + + let context: undefined | GetContext + let filter: Filter + let map: Map + + if (maybeMap) { + context = contextOrFilter as GetContext + filter = filterOrMap as Filter + map = maybeMap as Map + } else { + filter = contextOrFilter as Filter + map = filterOrMap as Map + } + + const output: O[] = [] + for (const item of array) { + const c = context?.(item) + if (filter(item, c as any)) output.push(map(item, c as any)) + } + return output +} diff --git a/packages/message-bus/src/errors/InvokableNotRegisteredError.ts b/packages/message-bus/src/errors/InvokableNotRegisteredError.ts index b1caa4935..d65197764 100644 --- a/packages/message-bus/src/errors/InvokableNotRegisteredError.ts +++ b/packages/message-bus/src/errors/InvokableNotRegisteredError.ts @@ -2,18 +2,20 @@ import { InvokablesDict } from '@plugola/invoke' import MessageBus from '../MessageBus.js' import { EventsT } from '../types/events.js' import { EventGeneratorsT } from '../types/generators.js' +import { StreamsDict } from '../types/streams.js' export class InvokableNotRegisteredError< Events extends EventsT, EventGens extends EventGeneratorsT, - Invokables extends InvokablesDict + Invokables extends InvokablesDict, + Streamables extends StreamsDict = StreamsDict, > extends Error { - #messageBus: MessageBus + #messageBus: MessageBus #invokableName: string constructor( - messageBus: MessageBus, - invokableName: string + messageBus: MessageBus, + invokableName: string, ) { super(`Cannot find matching invoker for "${invokableName}".`) this.#messageBus = messageBus diff --git a/packages/message-bus/src/types/streams.ts b/packages/message-bus/src/types/streams.ts new file mode 100644 index 000000000..7f609588e --- /dev/null +++ b/packages/message-bus/src/types/streams.ts @@ -0,0 +1,57 @@ +import { L } from 'ts-toolbelt' +import Broker from '../Broker.js' +import { AddAbortSignal } from './MessageBus.js' +import { UnderlyingDefaultSource } from 'node:stream/web' + +export type StreamsDict = Record + +export type CreateStreamsDict = T + +export type ReaderFn = ( + ...args: AddAbortSignal +) => UnderlyingDefaultSource + +export interface StreamReader { + broker: B + args: Args + fn: ReaderFn +} + +export type Streams = Partial<{ + [StreamName in keyof Streamables]: StreamReader< + Broker, + Streamables[StreamName]['args'], + Streamables[StreamName]['item'] + >[] +}> + +/** + * Create a `.reader` argument union from a list of arguments and a return type. + * + * @example + * type Args = StreamReaderArgs<[string, number], string> + * // | [string, number, ReaderFn<[], string>] + * // | [string, ReaderFn<[number], string>] + * // | [ReaderFn<[string, number], string>] + */ +export type StreamReaderArgs = _StreamReaderArgs< + A, + Item, + [], + [ReaderFn] +> + +type _StreamReaderArgs< + A extends unknown[], + Item, + B extends unknown[], + Acc extends unknown[], +> = + L.Length extends 0 + ? Acc + : _StreamReaderArgs< + L.Pop, + Item, + L.Prepend>, + L.Append> | Acc + > diff --git a/packages/message-bus/test/MessageBus.test.ts b/packages/message-bus/test/MessageBus.test.ts index 03f2db76f..1936caf67 100644 --- a/packages/message-bus/test/MessageBus.test.ts +++ b/packages/message-bus/test/MessageBus.test.ts @@ -8,6 +8,7 @@ import MessageBusError from '../src/MessageBusError.js' import { CreateEvents } from '../src/types/events.js' import { CreateInvokablesDict } from '../src/types/invokables.js' import { CreateEventGenerators } from '../src/types/generators.js' +import { write } from '@johngw/stream' describe('events', () => { type Events = CreateEvents<{ foo: []; bar: [string]; mung: [string, number] }> @@ -321,3 +322,93 @@ describe('error handling', () => { messageBus.start() })) }) + +describe('streams', () => { + type Streamables = { foo: { args: [number]; item: string } } + let messageBus: MessageBus<{}, {}, {}, Streamables> + let broker: Broker<{}, {}, {}, Streamables> + + beforeEach(() => { + messageBus = new MessageBus() + messageBus.start() + broker = messageBus.broker('test') + }) + + test('single reader', async () => { + const fn = vi.fn() + + broker.reader('foo', (x) => ({ + start(controller) { + controller.enqueue(x.toString()) + controller.close() + }, + })) + + await broker.stream('foo', 100).pipeTo(write(fn)) + + expect(fn).toHaveBeenCalledTimes(1) + expect(fn.mock.calls[0][0]).toBe('100') + }) + + test('multiple readers', async () => { + const fn = vi.fn() + + broker.reader('foo', (x) => ({ + start(controller) { + controller.enqueue(x.toString()) + controller.close() + }, + })) + + broker.reader('foo', (x) => ({ + start(controller) { + controller.enqueue(x.toString()) + controller.close() + }, + })) + + await broker.stream('foo', 100).pipeTo(write(fn)) + + expect(fn).toHaveBeenCalledTimes(2) + expect(fn.mock.calls[0][0]).toBe('100') + expect(fn.mock.calls[1][0]).toBe('100') + }) + + test('aborting', async () => { + broker.reader('foo', (x) => ({ + start(controller) { + controller.enqueue(x.toString()) + controller.close() + }, + })) + + const promise = broker.stream('foo', 100).pipeTo(write()) + + broker.abort() + + await expect(promise).rejects.toThrow() + }) + + test('specification filtering', async () => { + const fn = vi.fn() + + broker.reader('foo', 10, () => ({ + start(controller) { + controller.enqueue('10') + controller.close() + }, + })) + + broker.reader('foo', (x) => ({ + start(controller) { + controller.enqueue(x.toString()) + controller.close() + }, + })) + + await broker.stream('foo', 100).pipeTo(write(fn)) + + expect(fn).toHaveBeenCalledTimes(1) + expect(fn.mock.calls[0][0]).toBe('100') + }) +}) diff --git a/packages/message-bus/tsconfig.json b/packages/message-bus/tsconfig.json index 7f075e9b4..f61bf9eaa 100644 --- a/packages/message-bus/tsconfig.json +++ b/packages/message-bus/tsconfig.json @@ -3,7 +3,7 @@ "compilerOptions": { "composite": true, "rootDir": "./src", - "lib": ["es2022", "dom"], + "lib": ["ESNext"], "outDir": "dist" }, "include": ["src"], From 529ee5bda1652549dcbc2e2f13933b268c0980e0 Mon Sep 17 00:00:00 2001 From: John Wright Date: Wed, 10 Jun 2026 15:58:54 +0100 Subject: [PATCH 2/5] feat(message-bus)!: collapse generics into a single MessageBusContext MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the four positional generics on `MessageBus`, `Broker`, and `InvokableNotRegisteredError` with a single `$ extends MessageBusContext` bag. Adding a new feature now means adding a property to `MessageBusContext` rather than threading a new positional generic through every reference. A `CreateMessageBusContext` helper fills unspecified slots with their defaults, so consumers only specify the slots they use. The collection types (`Subscribers`, `Invokers`, `Streams`, …) and their record interfaces now take `$` directly, which also threads the context down into the stored `Broker<$>`. BREAKING CHANGE: `MessageBus`, `Broker`, and `InvokableNotRegisteredError` now take a single `MessageBusContext` type parameter instead of four positional generics. Replace `MessageBus` with `MessageBus>`. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/message-bus/src/Broker.ts | 135 +++++----- packages/message-bus/src/MessageBus.ts | 237 ++++++++---------- .../src/errors/InvokableNotRegisteredError.ts | 17 +- packages/message-bus/src/types/MessageBus.ts | 65 +++-- packages/message-bus/src/types/events.ts | 76 +++--- packages/message-bus/src/types/generators.ts | 29 +-- packages/message-bus/src/types/invokables.ts | 52 ++-- packages/message-bus/src/types/streams.ts | 20 +- packages/message-bus/test/MessageBus.test.ts | 47 +++- 9 files changed, 348 insertions(+), 330 deletions(-) diff --git a/packages/message-bus/src/Broker.ts b/packages/message-bus/src/Broker.ts index 00c2faecd..6e962874e 100644 --- a/packages/message-bus/src/Broker.ts +++ b/packages/message-bus/src/Broker.ts @@ -1,29 +1,24 @@ -import { InvokablesDict, InvokerRegistrationArgs } from '@plugola/invoke' +import { InvokerRegistrationArgs } from '@plugola/invoke' import type MessageBus from './MessageBus.js' import SubscriptionDisposer from './SubscriptionDisposer.js' import type { EventInterceptorArgs, - EventsT, SubscriberArgs, UntilArgs, UntilRtn, } from './types/events.js' -import type { - EventGeneratorArgs, - EventGeneratorsT, -} from './types/generators.js' +import type { EventGeneratorArgs } from './types/generators.js' import type { InvokerInterceptorArgs } from './types/invokables.js' -import { ErrorHandler, Unsubscriber } from './types/MessageBus.js' -import { StreamReaderArgs, StreamsDict } from './types/streams.js' - -export default class Broker< - Events extends EventsT = EventsT, - EventGens extends EventGeneratorsT = EventGeneratorsT, - Invokables extends InvokablesDict = InvokablesDict, - Streamables extends StreamsDict = StreamsDict, -> { +import { + ErrorHandler, + MessageBusContext, + Unsubscriber, +} from './types/MessageBus.js' +import { StreamReaderArgs } from './types/streams.js' + +export default class Broker<$ extends MessageBusContext = MessageBusContext> { constructor( - messageBus: MessageBus, + messageBus: MessageBus<$>, id: string, abortController: AbortController, ) { @@ -56,7 +51,7 @@ export default class Broker< return this.#id } - readonly #messageBus: MessageBus + readonly #messageBus: MessageBus<$> get messageBus() { return this.#messageBus } @@ -77,164 +72,164 @@ export default class Broker< }) } - emit( + emit( eventName: EventName, - ...args: Events[EventName] + ...args: $['events'][EventName] ): void | Promise { return this.#messageBus.emit(this, eventName, args) } - emitSignal( + emitSignal( eventName: EventName, signal: AbortSignal, - ...args: Events[EventName] + ...args: $['events'][EventName] ): void | Promise { return this.#messageBus.emit(this, eventName, args, signal) } - interceptEvent( + interceptEvent( eventName: EventName, - ...args: EventInterceptorArgs + ...args: EventInterceptorArgs<$['events'][EventName]> ): Unsubscriber { return this.#messageBus.interceptEvent(this as any, eventName, args) } - on( + on( eventName: EventName, - ...args: SubscriberArgs + ...args: SubscriberArgs<$['events'][EventName]> ): Unsubscriber { return this.#messageBus.on(this, eventName, args) } - once( + once( eventName: EventName, - ...args: SubscriberArgs + ...args: SubscriberArgs<$['events'][EventName]> ): Unsubscriber { return this.#messageBus.once(this, eventName, args) } - hasSubscriber(eventName: keyof Events) { + hasSubscriber(eventName: keyof $['events']) { return this.#messageBus.hasSubscriber(eventName) } async until< - EventName extends keyof Events, - Args extends UntilArgs, + EventName extends keyof $['events'], + Args extends UntilArgs<$['events'][EventName]>, >( eventName: EventName, ...args: Args - ): Promise> { + ): Promise> { return this.#messageBus.until(this, eventName, args) as any } async untilSignal< - EventName extends keyof Events, - Args extends UntilArgs, + EventName extends keyof $['events'], + Args extends UntilArgs<$['events'][EventName]>, >( eventName: EventName, abortSignal: AbortSignal, ...args: Args - ): Promise> { + ): Promise> { return this.#messageBus.until(this, eventName, args, abortSignal) as any } - generator( + generator( eventName: EventName, ...args: EventGeneratorArgs< - EventGens[EventName]['args'], - EventGens[EventName]['yield'] + $['generators'][EventName]['args'], + $['generators'][EventName]['yield'] > ): Unsubscriber { return this.#messageBus.generator(this, eventName, args) } - iterate( + iterate( eventName: EventName, - ...args: EventGens[EventName]['args'] - ): AsyncIterable { + ...args: $['generators'][EventName]['args'] + ): AsyncIterable<$['generators'][EventName]['yield']> { return this.#messageBus.iterate(this, eventName, args) } - iterateSignal( + iterateSignal( eventName: EventName, abortSignal: AbortSignal, - ...args: EventGens[EventName]['args'] - ): AsyncIterable { + ...args: $['generators'][EventName]['args'] + ): AsyncIterable<$['generators'][EventName]['yield']> { return this.#messageBus.iterate(this, eventName, args, abortSignal) } - iterateWithin( + iterateWithin( within: number, eventName: EventName, - ...args: EventGens[EventName]['args'] - ): AsyncIterable { + ...args: $['generators'][EventName]['args'] + ): AsyncIterable<$['generators'][EventName]['yield']> { return this.#messageBus.iterateWithin(this, within, eventName, args) } - accumulate( + accumulate( eventName: EventName, - ...args: EventGens[EventName]['args'] - ): Promise { + ...args: $['generators'][EventName]['args'] + ): Promise<$['generators'][EventName]['yield'][]> { return this.#messageBus.accumulate(this, eventName, args) } - accumulateWithin( + accumulateWithin( within: number, eventName: EventName, - ...args: EventGens[EventName]['args'] - ): Promise { + ...args: $['generators'][EventName]['args'] + ): Promise<$['generators'][EventName]['yield'][]> { return this.#messageBus.accumulateWithin(this, within, eventName, args) } - register( + register( invokableName: InvokableName, ...args: InvokerRegistrationArgs< - Invokables[InvokableName]['args'], - Invokables[InvokableName]['return'] + $['invokables'][InvokableName]['args'], + $['invokables'][InvokableName]['return'] > ): Unsubscriber { return this.#messageBus.register(this, invokableName, args) } - invoke( + invoke( invokableName: InvokableName, - ...args: Invokables[InvokableName]['args'] - ): Promise { + ...args: $['invokables'][InvokableName]['args'] + ): Promise<$['invokables'][InvokableName]['return']> { return this.#messageBus.invoke(this, invokableName, args) } - invokeSignal( + invokeSignal( invokableName: InvokableName, abortSignal: AbortSignal, - ...args: Invokables[InvokableName]['args'] - ): Promise { + ...args: $['invokables'][InvokableName]['args'] + ): Promise<$['invokables'][InvokableName]['return']> { return this.#messageBus.invoke(this, invokableName, args, abortSignal) } - interceptInvoker( + interceptInvoker( invokableName: InvokableName, ...args: InvokerInterceptorArgs< - Invokables[InvokableName]['args'], - Invokables[InvokableName]['return'] + $['invokables'][InvokableName]['args'], + $['invokables'][InvokableName]['return'] > ): Unsubscriber { return this.#messageBus.interceptInvoker(this, invokableName, args) } - reader( + reader( streamName: StreamName, ...args: StreamReaderArgs< - Streamables[StreamName]['args'], - Streamables[StreamName]['item'] + $['streams'][StreamName]['args'], + $['streams'][StreamName]['item'] > ): Unsubscriber { return this.#messageBus.reader(this, streamName, args) } - stream( + stream( streamName: StreamName, - ...args: Streamables[StreamName]['args'] - ): ReadableStream { + ...args: $['streams'][StreamName]['args'] + ): ReadableStream<$['streams'][StreamName]['item']> { return this.#messageBus.stream(this, streamName, args) } } diff --git a/packages/message-bus/src/MessageBus.ts b/packages/message-bus/src/MessageBus.ts index 622685488..b16b367c9 100644 --- a/packages/message-bus/src/MessageBus.ts +++ b/packages/message-bus/src/MessageBus.ts @@ -12,18 +12,13 @@ import { CancelEvent } from './symbols.js' import { EventInterceptorArgs, EventInterceptors, - EventsT, SubscriberArgs, SubscriberFn, Subscribers, UntilArgs, UntilRtn, } from './types/events.js' -import { - EventGeneratorArgs, - EventGenerators, - EventGeneratorsT, -} from './types/generators.js' +import { EventGeneratorArgs, EventGenerators } from './types/generators.js' import { InvokerInterceptorArgs, InvokerInterceptors, @@ -33,40 +28,33 @@ import { Stringable, UnpackResolvableValue } from './types/util.js' import { AddAbortSignal, ErrorHandler, + MessageBusContext, Unsubscriber, } from './types/MessageBus.js' import { anySignal, fromSignal } from './AbortController.js' import { InvokableNotRegisteredError } from './errors/InvokableNotRegisteredError.js' -import { - InvokablesDict, - InvokerFn, - InvokerRegistrationArgs, -} from '@plugola/invoke' +import { InvokerFn, InvokerRegistrationArgs } from '@plugola/invoke' import { ReaderFn, StreamReader, StreamReaderArgs, Streams, - StreamsDict, } from './types/streams.js' import { WritableReadablePair } from '@johngw/stream/transformers/WritableReadablePair' import { mergeUnderlyingSource } from '@johngw/stream' export default class MessageBus< - Events extends EventsT = EventsT, - EventGens extends EventGeneratorsT = EventGeneratorsT, - Invokables extends InvokablesDict = InvokablesDict, - Streamables extends StreamsDict = StreamsDict, + $ extends MessageBusContext = MessageBusContext, > { #errorHandlers: ErrorHandler[] = [] - #eventInterceptors: EventInterceptors = {} - #eventGenerators: EventGenerators = {} - #invokers: Invokers = {} - #invokerInterceptors: InvokerInterceptors = {} + #eventInterceptors: EventInterceptors<$> = {} + #eventGenerators: EventGenerators<$> = {} + #invokers: Invokers<$> = {} + #invokerInterceptors: InvokerInterceptors<$> = {} #queued: Array<() => unknown> = [] #started = false - #streams: Streams = {} - #subscribers: Subscribers = {} + #streams: Streams<$> = {} + #subscribers: Subscribers<$> = {} onError(errorHandler: ErrorHandler) { this.#errorHandlers.push(errorHandler) @@ -87,11 +75,7 @@ export default class MessageBus< ? abort : fromSignal(abort) - return new Broker( - this, - id, - abortController, - ) + return new Broker<$>(this, id, abortController) } async start() { @@ -99,10 +83,10 @@ export default class MessageBus< return Promise.all(this.#queued.map((handle) => handle())) } - emit( - broker: Broker, + emit( + broker: Broker<$>, eventName: EventName, - args: Events[EventName], + args: $['events'][EventName], abortSignal?: AbortSignal, ): void | Promise { const handle = () => { @@ -135,10 +119,10 @@ export default class MessageBus< ) } - interceptEvent( - broker: Broker, + interceptEvent( + broker: Broker<$>, eventName: EventName, - args: EventInterceptorArgs, + args: EventInterceptorArgs<$['events'][EventName]>, ): Unsubscriber { const interceptor = { broker, @@ -160,12 +144,12 @@ export default class MessageBus< } } - interceptInvoker( - broker: Broker, + interceptInvoker( + broker: Broker<$>, invokableName: InvokableName, args: InvokerInterceptorArgs< - Invokables[InvokableName]['args'], - Invokables[InvokableName]['return'] + $['invokables'][InvokableName]['args'], + $['invokables'][InvokableName]['return'] >, ): Unsubscriber { const interceptor = { @@ -188,10 +172,10 @@ export default class MessageBus< } } - on( - broker: Broker, + on( + broker: Broker<$>, eventName: EventName, - args: SubscriberArgs, + args: SubscriberArgs<$['events'][EventName]>, ): Unsubscriber { if (broker.aborted) return () => {} @@ -219,61 +203,63 @@ export default class MessageBus< return cancel } - once( - broker: Broker, + once( + broker: Broker<$>, eventName: EventName, - args: SubscriberArgs, + args: SubscriberArgs<$['events'][EventName]>, ): Unsubscriber { - const fn = last(args) as SubscriberFn - const onceFn: SubscriberFn = (...args) => { + const fn = last(args) as SubscriberFn<$['events'][EventName]> + const onceFn: SubscriberFn<$['events'][EventName]> = (...args) => { cancel() return fn(...args) } const cancel = this.on( broker, eventName, - replaceLastItem(args, onceFn) as SubscriberArgs, + replaceLastItem(args, onceFn) as SubscriberArgs<$['events'][EventName]>, ) return cancel } async until< - EventName extends keyof Events, - Args extends UntilArgs, + EventName extends keyof $['events'], + Args extends UntilArgs<$['events'][EventName]>, >( - broker: Broker, + broker: Broker<$>, eventName: EventName, args: Args, abortSignal?: AbortSignal, - ): Promise> { - return new Promise>((resolve, reject) => { - const abortSignalComposite = anySignal(abortSignal, broker.abortSignal) + ): Promise> { + return new Promise>( + (resolve, reject) => { + const abortSignalComposite = anySignal(abortSignal, broker.abortSignal) - if (abortSignalComposite.aborted) return reject(new AbortError()) + if (abortSignalComposite.aborted) return reject(new AbortError()) - const subscriberArgs = [ - ...args, - (...args: any) => resolve(args), - ] as SubscriberArgs + const subscriberArgs = [ + ...args, + (...args: any) => resolve(args), + ] as SubscriberArgs<$['events'][EventName]> - this.once(broker, eventName, subscriberArgs) + this.once(broker, eventName, subscriberArgs) - abortSignalComposite.addEventListener('abort', () => { - reject(new AbortError()) - }) - }) + abortSignalComposite.addEventListener('abort', () => { + reject(new AbortError()) + }) + }, + ) } - hasSubscriber(eventName: keyof Events) { + hasSubscriber(eventName: keyof $['events']) { return !!this.#subscribers[eventName]?.length } - generator( - broker: Broker, + generator( + broker: Broker<$>, eventName: EventName, args: EventGeneratorArgs< - EventGens[EventName]['args'], - EventGens[EventName]['yield'] + $['generators'][EventName]['args'], + $['generators'][EventName]['yield'] >, ): Unsubscriber { if (broker.aborted) return () => {} @@ -302,12 +288,12 @@ export default class MessageBus< return cancel } - async *iterate( - broker: Broker, + async *iterate( + broker: Broker<$>, eventName: EventName, - args: EventGens[EventName]['args'], + args: $['generators'][EventName]['args'], abortSignal?: AbortSignal, - ): AsyncIterable { + ): AsyncIterable<$['generators'][EventName]['yield']> { if (!this.#started) await this.#queue(broker, () => {}) yield* combineIterators( @@ -322,13 +308,13 @@ export default class MessageBus< ) } - iterateWithin( - broker: Broker, + iterateWithin( + broker: Broker<$>, within: number, eventName: EventName, - args: EventGens[EventName]['args'], + args: $['generators'][EventName]['args'], abortSignal?: AbortSignal, - ): AsyncIterable { + ): AsyncIterable<$['generators'][EventName]['yield']> { return iteratorRace( this.iterate(broker, eventName, args, abortSignal), within, @@ -336,20 +322,20 @@ export default class MessageBus< ) } - async accumulate( - broker: Broker, + async accumulate( + broker: Broker<$>, eventName: EventName, - args: EventGens[EventName]['args'], + args: $['generators'][EventName]['args'], abortSignal?: AbortSignal, ) { return accumulate(this.iterate(broker, eventName, args, abortSignal)) } - async accumulateWithin( - broker: Broker, + async accumulateWithin( + broker: Broker<$>, within: number, eventName: EventName, - args: EventGens[EventName]['args'], + args: $['generators'][EventName]['args'], abortSignal?: AbortSignal, ) { return accumulate( @@ -357,20 +343,20 @@ export default class MessageBus< ) } - register( - broker: Broker, + register( + broker: Broker<$>, invokableName: InvokableName, allArgs: InvokerRegistrationArgs< - Invokables[InvokableName]['args'], - Invokables[InvokableName]['return'] + $['invokables'][InvokableName]['args'], + $['invokables'][InvokableName]['return'] >, ): Unsubscriber { if (broker.aborted) return () => {} - const args = init(allArgs) as Invokables[InvokableName]['args'] + const args = init(allArgs) as $['invokables'][InvokableName]['args'] const fn = last(allArgs) as InvokerFn< - Invokables[InvokableName]['args'], - Invokables[InvokableName]['return'] + $['invokables'][InvokableName]['args'], + $['invokables'][InvokableName]['return'] > const invokers = this.#invokers[invokableName] || [] const registeredInvoker = invokers.find( @@ -393,13 +379,13 @@ export default class MessageBus< this.#invokers[invokableName] = [ ...invokers, subscriber, - ] as unknown as Invokers[InvokableName] + ] as unknown as Invokers<$>[InvokableName] const cancel = () => { this.#invokers[invokableName] = removeItem( subscriber, this.#invokers[invokableName] as any, - ) as unknown as Invokers[InvokableName] + ) as unknown as Invokers<$>[InvokableName] } broker.onAbort(() => setTimeout(cancel, 0)) @@ -407,12 +393,12 @@ export default class MessageBus< return cancel } - async invoke( - broker: Broker, + async invoke( + broker: Broker<$>, invokableName: InvokableName, - args: Invokables[InvokableName]['args'], + args: $['invokables'][InvokableName]['args'], abortSignal?: AbortSignal, - ): Promise { + ): Promise<$['invokables'][InvokableName]['return']> { const handle = async () => new Promise((resolve, reject) => { const abortSignalComposite = anySignal(abortSignal, broker.abortSignal) @@ -427,19 +413,19 @@ export default class MessageBus< return this.#started ? handle() : this.#queue(broker, handle) } - reader( - broker: Broker, + reader( + broker: Broker<$>, streamName: StreamName, allArgs: StreamReaderArgs< - Streamables[StreamName]['args'], - Streamables[StreamName]['item'] + $['streams'][StreamName]['args'], + $['streams'][StreamName]['item'] >, ): Unsubscriber { - const args = init(allArgs) as Streamables[StreamName]['args'] + const args = init(allArgs) as $['streams'][StreamName]['args'] const fn = last(allArgs) as ReaderFn< - Streamables[StreamName]['args'], - Streamables[StreamName]['item'] + $['streams'][StreamName]['args'], + $['streams'][StreamName]['item'] > const streamer: StreamReader = { @@ -464,13 +450,13 @@ export default class MessageBus< return cancel } - stream( - broker: Broker, + stream( + broker: Broker<$>, streamName: StreamName, - args: Streamables[StreamName]['args'], + args: $['streams'][StreamName]['args'], abortSignal?: AbortSignal, - ): ReadableStream { - type Item = Streamables[StreamName]['item'] + ): ReadableStream<$['streams'][StreamName]['item']> { + type Item = $['streams'][StreamName]['item'] const streamers = this.#streams[streamName] ?? [] @@ -488,7 +474,7 @@ export default class MessageBus< ...streamer.args.slice(0, argumentIndex), ...args.slice(argumentIndex), abortSignalComposite, - ] as AddAbortSignal), + ] as AddAbortSignal<$['streams'][StreamName]['args']>), ), ), ) @@ -505,16 +491,16 @@ export default class MessageBus< ) } - #callEventInterceptors( + #callEventInterceptors( eventName: EventName, - args: Events[EventName], - ): void | Promise { + args: $['events'][EventName], + ): void | Promise<$['events'][EventName] | typeof CancelEvent> { const eventInterceptors = (this.#eventInterceptors[eventName] || [])! if (!eventInterceptors.length) return return (async () => { - let moddedArgs: Events[EventName] | typeof CancelEvent = args + let moddedArgs: $['events'][EventName] | typeof CancelEvent = args for (const interceptor of eventInterceptors) { const index = this.#argumentIndex(interceptor.args, moddedArgs) @@ -528,24 +514,24 @@ export default class MessageBus< moddedArgs = [ ...moddedArgs.slice(0, index), ...newArgs, - ] as Events[EventName] + ] as $['events'][EventName] } return moddedArgs })() } - async #invokeChain( + async #invokeChain( invokableName: InvokableName, - args: Invokables[InvokableName]['args'], + args: $['invokables'][InvokableName]['args'], signal: AbortSignal, - ): Promise { + ): Promise<$['invokables'][InvokableName]['return']> { const invokerInterceptors = this.#invokerInterceptors[invokableName] || [] const invokeChain = async ( index: number, - args: Invokables[InvokableName]['args'], - ): Promise => { + args: $['invokables'][InvokableName]['args'], + ): Promise<$['invokables'][InvokableName]['return']> => { const interceptor = invokerInterceptors[index] if (!interceptor) return this.#invoke(invokableName, args, signal) const argIndex = this.#argumentIndex(interceptor.args, args) @@ -560,9 +546,9 @@ export default class MessageBus< return invokeChain(0, args) } - #callSubscribers( + #callSubscribers( eventName: EventName, - args: Events[EventName], + args: $['events'][EventName], abortSignal?: AbortSignal, ): void | Promise { const subscribers = (this.#subscribers[eventName] || [])! @@ -592,11 +578,11 @@ export default class MessageBus< } } - async #invoke( + async #invoke( invokableName: InvokableName, - args: Invokables[InvokableName]['args'], + args: $['invokables'][InvokableName]['args'], abortSignal: AbortSignal, - ): Promise { + ): Promise<$['invokables'][InvokableName]['return']> { const invokers = this.#invokers[invokableName] const invoker = invokers && @@ -609,7 +595,7 @@ export default class MessageBus< return invoker.fn( ...(args.slice( this.#argumentIndex(invoker.args, args), - ) as Invokables[InvokableName]['args']), + ) as $['invokables'][InvokableName]['args']), abortSignal, ) } @@ -623,10 +609,7 @@ export default class MessageBus< return i } - async #queue( - broker: Broker, - handler: () => T, - ) { + async #queue(broker: Broker<$>, handler: () => T) { return new Promise>((resolve, reject) => { if (broker.aborted) return reject(new AbortError()) diff --git a/packages/message-bus/src/errors/InvokableNotRegisteredError.ts b/packages/message-bus/src/errors/InvokableNotRegisteredError.ts index d65197764..07635f16e 100644 --- a/packages/message-bus/src/errors/InvokableNotRegisteredError.ts +++ b/packages/message-bus/src/errors/InvokableNotRegisteredError.ts @@ -1,22 +1,13 @@ -import { InvokablesDict } from '@plugola/invoke' import MessageBus from '../MessageBus.js' -import { EventsT } from '../types/events.js' -import { EventGeneratorsT } from '../types/generators.js' -import { StreamsDict } from '../types/streams.js' +import { MessageBusContext } from '../types/MessageBus.js' export class InvokableNotRegisteredError< - Events extends EventsT, - EventGens extends EventGeneratorsT, - Invokables extends InvokablesDict, - Streamables extends StreamsDict = StreamsDict, + $ extends MessageBusContext = MessageBusContext, > extends Error { - #messageBus: MessageBus + #messageBus: MessageBus<$> #invokableName: string - constructor( - messageBus: MessageBus, - invokableName: string, - ) { + constructor(messageBus: MessageBus<$>, invokableName: string) { super(`Cannot find matching invoker for "${invokableName}".`) this.#messageBus = messageBus this.#invokableName = invokableName diff --git a/packages/message-bus/src/types/MessageBus.ts b/packages/message-bus/src/types/MessageBus.ts index ff4aced3a..2d182915e 100644 --- a/packages/message-bus/src/types/MessageBus.ts +++ b/packages/message-bus/src/types/MessageBus.ts @@ -1,7 +1,11 @@ import type { L } from 'ts-toolbelt' +import type { InvokablesDict } from '@plugola/invoke' import type Broker from '../Broker.js' import type MessageBus from '../MessageBus.js' import type MessageBusError from '../MessageBusError.js' +import type { EventsT } from './events.js' +import type { EventGeneratorsT } from './generators.js' +import type { StreamsDict } from './streams.js' export interface Unsubscriber { (): void @@ -9,34 +13,43 @@ export interface Unsubscriber { export type AddAbortSignal = L.Append -export type MessageBusEvents = MB extends MessageBus< - infer Events, - any, - any -> - ? Events - : never +/** + * The single "bag" of type information a {@link MessageBus} (and its + * {@link Broker}) is configured with. Adding a new feature means adding a + * property here rather than appending a positional generic to every reference. + */ +export interface MessageBusContext { + events: EventsT + generators: EventGeneratorsT + invokables: InvokablesDict + streams: StreamsDict +} + +/** + * Build a {@link MessageBusContext} from a partial one, defaulting any slot the + * consumer doesn't specify. + * + * @example + * type Ctx = CreateMessageBusContext<{ events: { foo: [string] } }> + * let bus: MessageBus + */ +export type CreateMessageBusContext> = + Omit & C + +export type MessageBusEvents = + MB extends MessageBus ? $['events'] : never export type MessageBusEventGenerators = - MB extends MessageBus - ? EventGenerators - : never - -export type MessageBusInvokers = MB extends MessageBus< - any, - any, - infer Invokers -> - ? Invokers - : never - -export type MessageBusBroker = MB extends MessageBus< - infer Events, - infer EventGenerators, - infer Invokables -> - ? Broker - : never + MB extends MessageBus ? $['generators'] : never + +export type MessageBusInvokers = + MB extends MessageBus ? $['invokables'] : never + +export type MessageBusStreams = + MB extends MessageBus ? $['streams'] : never + +export type MessageBusBroker = + MB extends MessageBus ? Broker<$> : never export interface ErrorHandler { (error: MessageBusError): any diff --git a/packages/message-bus/src/types/events.ts b/packages/message-bus/src/types/events.ts index 52777c462..77b910852 100644 --- a/packages/message-bus/src/types/events.ts +++ b/packages/message-bus/src/types/events.ts @@ -1,7 +1,7 @@ import type { L } from 'ts-toolbelt' import type Broker from '../Broker.js' import type { CancelEvent } from '../symbols.js' -import type { AddAbortSignal } from './MessageBus.js' +import type { AddAbortSignal, MessageBusContext } from './MessageBus.js' export type EventsT = Record @@ -29,14 +29,15 @@ export type SubscriberArgs = _SubscriberArgs< type _SubscriberArgs< A extends unknown[], B extends unknown[], - Acc extends unknown[] -> = L.Length extends 0 - ? Acc - : _SubscriberArgs< - L.Pop, - L.Prepend>, - L.Append> | Acc - > + Acc extends unknown[], +> = + L.Length extends 0 + ? Acc + : _SubscriberArgs< + L.Pop, + L.Prepend>, + L.Append> | Acc + > /** * Create an `.until` argument union from a list of types. @@ -49,10 +50,8 @@ type _SubscriberArgs< */ export type UntilArgs = _UntilArgs -type _UntilArgs< - A extends unknown[], - Acc extends unknown[] -> = L.Length extends 0 ? Acc : _UntilArgs, L.Pop | Acc> +type _UntilArgs = + L.Length extends 0 ? Acc : _UntilArgs, L.Pop | Acc> /** * The return value of `.until` from given arguments and @@ -68,28 +67,26 @@ type _UntilArgs< * type Rtn = UntilRtn<[string, number, string], [string, number, string]> * // [] */ -export type UntilRtn< - T extends unknown[], - Args extends unknown[] -> = L.Length extends 0 - ? T - : L.Head extends L.Head - ? UntilRtn, L.Tail> - : never +export type UntilRtn = + L.Length extends 0 + ? T + : L.Head extends L.Head + ? UntilRtn, L.Tail> + : never -export type Subscribers = Partial<{ - [EventName in keyof Events]: Subscriber[] +export type Subscribers<$ extends MessageBusContext> = Partial<{ + [EventName in keyof $['events']]: Subscriber<$>[] }> -export interface Subscriber { - broker: B +export interface Subscriber<$ extends MessageBusContext> { + broker: Broker<$> args: unknown[] fn: SubscriberFn } export type EventInterceptorFn< Args extends unknown[], - NewArgs extends unknown[] + NewArgs extends unknown[], > = ( ...args: Args ) => @@ -109,22 +106,23 @@ type _EventInterceptorArgs< A extends unknown[], B extends unknown[], C extends unknown[], - Acc extends unknown[] -> = L.Length extends 0 - ? Acc - : _EventInterceptorArgs< - L.Pop, - L.Prepend>, - C, - Acc | [...A, EventInterceptorFn] - > + Acc extends unknown[], +> = + L.Length extends 0 + ? Acc + : _EventInterceptorArgs< + L.Pop, + L.Prepend>, + C, + Acc | [...A, EventInterceptorFn] + > -export type EventInterceptors = Partial<{ - [EventName in keyof Events]: EventInterceptor[] +export type EventInterceptors<$ extends MessageBusContext> = Partial<{ + [EventName in keyof $['events']]: EventInterceptor<$>[] }> -export interface EventInterceptor { - broker: B +export interface EventInterceptor<$ extends MessageBusContext> { + broker: Broker<$> args: unknown[] fn: EventInterceptorFn } diff --git a/packages/message-bus/src/types/generators.ts b/packages/message-bus/src/types/generators.ts index cf0b5c718..1d176e7a7 100644 --- a/packages/message-bus/src/types/generators.ts +++ b/packages/message-bus/src/types/generators.ts @@ -1,6 +1,6 @@ import type { L } from 'ts-toolbelt' import type Broker from '../Broker.js' -import type { AddAbortSignal } from './MessageBus.js' +import type { AddAbortSignal, MessageBusContext } from './MessageBus.js' export type EventGeneratorsT = Record< string, @@ -13,8 +13,8 @@ export type EventGeneratorFn = ( ...args: AddAbortSignal ) => AsyncIterable -export interface EventGenerator { - broker: B +export interface EventGenerator<$ extends MessageBusContext> { + broker: Broker<$> args: unknown[] fn: EventGeneratorFn } @@ -30,16 +30,17 @@ export type _EventGeneratorArgs< A extends unknown[], R, B extends unknown[], - Acc extends unknown[] -> = L.Length extends 0 - ? Acc - : _EventGeneratorArgs< - L.Pop, - R, - L.Prepend>, - Acc | L.Append> - > + Acc extends unknown[], +> = + L.Length extends 0 + ? Acc + : _EventGeneratorArgs< + L.Pop, + R, + L.Prepend>, + Acc | L.Append> + > -export type EventGenerators = Partial<{ - [EventName in keyof EventGens]: EventGenerator[] +export type EventGenerators<$ extends MessageBusContext> = Partial<{ + [EventName in keyof $['generators']]: EventGenerator<$>[] }> diff --git a/packages/message-bus/src/types/invokables.ts b/packages/message-bus/src/types/invokables.ts index b7ce11a89..c643f8c48 100644 --- a/packages/message-bus/src/types/invokables.ts +++ b/packages/message-bus/src/types/invokables.ts @@ -1,6 +1,7 @@ -import type { InvokablesDict, InvokerFn } from '@plugola/invoke' +import type { InvokerFn } from '@plugola/invoke' import type Broker from '../Broker.js' import type { L } from 'ts-toolbelt' +import type { MessageBusContext } from './MessageBus.js' export { CreateInvokablesDict, @@ -9,17 +10,21 @@ export { InvokerRegistrationArgs, } from '@plugola/invoke' -export interface Invoker { - broker: B +export interface Invoker< + $ extends MessageBusContext, + Args extends unknown[], + Return, +> { + broker: Broker<$> args: Args fn: InvokerFn } -export type Invokers = Partial<{ - [InvokableName in keyof Invokables]: Invoker< - Broker, - Invokables[InvokableName]['args'], - Invokables[InvokableName]['return'] +export type Invokers<$ extends MessageBusContext> = Partial<{ + [InvokableName in keyof $['invokables']]: Invoker< + $, + $['invokables'][InvokableName]['args'], + $['invokables'][InvokableName]['return'] >[] }> @@ -30,29 +35,30 @@ export type InvokerInterceptorFn = ( export type InvokerInterceptorArgs< A extends unknown[], - Return + Return, > = _InvokerInterceptorArgs]> export type _InvokerInterceptorArgs< A extends unknown[], Return, B extends unknown[], - Acc extends unknown[] -> = L.Length extends 0 - ? Acc - : _InvokerInterceptorArgs< - L.Pop, - Return, - B, - Acc | L.Append> - > - -export type InvokerInterceptors = Partial<{ - [InvokableName in keyof Invokables]: InvokerInterceptor[] + Acc extends unknown[], +> = + L.Length extends 0 + ? Acc + : _InvokerInterceptorArgs< + L.Pop, + Return, + B, + Acc | L.Append> + > + +export type InvokerInterceptors<$ extends MessageBusContext> = Partial<{ + [InvokableName in keyof $['invokables']]: InvokerInterceptor<$>[] }> -export interface InvokerInterceptor { - broker: B +export interface InvokerInterceptor<$ extends MessageBusContext> { + broker: Broker<$> args: unknown[] fn: InvokerInterceptorFn } diff --git a/packages/message-bus/src/types/streams.ts b/packages/message-bus/src/types/streams.ts index 7f609588e..af2201161 100644 --- a/packages/message-bus/src/types/streams.ts +++ b/packages/message-bus/src/types/streams.ts @@ -1,6 +1,6 @@ import { L } from 'ts-toolbelt' import Broker from '../Broker.js' -import { AddAbortSignal } from './MessageBus.js' +import { AddAbortSignal, MessageBusContext } from './MessageBus.js' import { UnderlyingDefaultSource } from 'node:stream/web' export type StreamsDict = Record @@ -11,17 +11,21 @@ export type ReaderFn = ( ...args: AddAbortSignal ) => UnderlyingDefaultSource -export interface StreamReader { - broker: B +export interface StreamReader< + $ extends MessageBusContext, + Args extends unknown[], + Item, +> { + broker: Broker<$> args: Args fn: ReaderFn } -export type Streams = Partial<{ - [StreamName in keyof Streamables]: StreamReader< - Broker, - Streamables[StreamName]['args'], - Streamables[StreamName]['item'] +export type Streams<$ extends MessageBusContext> = Partial<{ + [StreamName in keyof $['streams']]: StreamReader< + $, + $['streams'][StreamName]['args'], + $['streams'][StreamName]['item'] >[] }> diff --git a/packages/message-bus/test/MessageBus.test.ts b/packages/message-bus/test/MessageBus.test.ts index 1936caf67..a7c492512 100644 --- a/packages/message-bus/test/MessageBus.test.ts +++ b/packages/message-bus/test/MessageBus.test.ts @@ -6,14 +6,15 @@ import { CancelEvent } from '../src/symbols.js' import { AbortError, timeout } from '@johngw/async' import MessageBusError from '../src/MessageBusError.js' import { CreateEvents } from '../src/types/events.js' +import { CreateMessageBusContext } from '../src/types/MessageBus.js' import { CreateInvokablesDict } from '../src/types/invokables.js' import { CreateEventGenerators } from '../src/types/generators.js' import { write } from '@johngw/stream' describe('events', () => { type Events = CreateEvents<{ foo: []; bar: [string]; mung: [string, number] }> - let messageBus: MessageBus - let broker: Broker + let messageBus: MessageBus> + let broker: Broker> let foo: Mock<() => void> let bar: Mock<(x: string) => void> @@ -138,8 +139,8 @@ describe('iterators', () => { foo: { args: []; yield: string } bar: { args: [string]; yield: string } }> - let messageBus: MessageBus<{}, Iterables, {}> - let broker: Broker<{}, Iterables, {}> + let messageBus: MessageBus> + let broker: Broker> beforeEach(() => { messageBus = new MessageBus() @@ -198,8 +199,10 @@ describe('invokables', () => { afoo: { args: [string]; return: string } never: { args: []; return: Promise } }> - let messageBus: MessageBus<{}, {}, Invokables> - let broker: Broker<{}, {}, Invokables> + let messageBus: MessageBus< + CreateMessageBusContext<{ invokables: Invokables }> + > + let broker: Broker> let foo: Mock<() => string> let bar: Mock<(x: string) => string> @@ -286,8 +289,8 @@ describe('invokables', () => { describe('error handling', () => { type Events = { foo: []; bar: [string] } - let messageBus: MessageBus - let broker: Broker + let messageBus: MessageBus> + let broker: Broker> beforeEach(() => { messageBus = new MessageBus() @@ -325,8 +328,8 @@ describe('error handling', () => { describe('streams', () => { type Streamables = { foo: { args: [number]; item: string } } - let messageBus: MessageBus<{}, {}, {}, Streamables> - let broker: Broker<{}, {}, {}, Streamables> + let messageBus: MessageBus> + let broker: Broker> beforeEach(() => { messageBus = new MessageBus() @@ -411,4 +414,28 @@ describe('streams', () => { expect(fn).toHaveBeenCalledTimes(1) expect(fn.mock.calls[0][0]).toBe('100') }) + + test('specification applying', async () => { + const fn = vi.fn() + + broker.reader('foo', 10, () => ({ + start(controller) { + controller.enqueue('10') + controller.close() + }, + })) + + broker.reader('foo', (x) => ({ + start(controller) { + controller.enqueue(x.toString()) + controller.close() + }, + })) + + await broker.stream('foo', 10).pipeTo(write(fn)) + + expect(fn).toHaveBeenCalledTimes(2) + expect(fn.mock.calls[0][0]).toBe('10') + expect(fn.mock.calls[1][0]).toBe('10') + }) }) From 364b260c7cf377c3ceefa2ca57b06ee925089e04 Mon Sep 17 00:00:00 2001 From: John Wright Date: Wed, 10 Jun 2026 16:40:58 +0100 Subject: [PATCH 3/5] refactor: simplify reader types --- packages/message-bus/src/MessageBus.ts | 22 ++++++---------------- packages/message-bus/src/types/streams.ts | 16 +++++++--------- 2 files changed, 13 insertions(+), 25 deletions(-) diff --git a/packages/message-bus/src/MessageBus.ts b/packages/message-bus/src/MessageBus.ts index b16b367c9..632b3be57 100644 --- a/packages/message-bus/src/MessageBus.ts +++ b/packages/message-bus/src/MessageBus.ts @@ -34,12 +34,7 @@ import { import { anySignal, fromSignal } from './AbortController.js' import { InvokableNotRegisteredError } from './errors/InvokableNotRegisteredError.js' import { InvokerFn, InvokerRegistrationArgs } from '@plugola/invoke' -import { - ReaderFn, - StreamReader, - StreamReaderArgs, - Streams, -} from './types/streams.js' +import { StreamReader, StreamReaderArgs, Streams } from './types/streams.js' import { WritableReadablePair } from '@johngw/stream/transformers/WritableReadablePair' import { mergeUnderlyingSource } from '@johngw/stream' @@ -421,17 +416,12 @@ export default class MessageBus< $['streams'][StreamName]['item'] >, ): Unsubscriber { - const args = init(allArgs) as $['streams'][StreamName]['args'] + type $StreamReader = StreamReader<$, StreamName> - const fn = last(allArgs) as ReaderFn< - $['streams'][StreamName]['args'], - $['streams'][StreamName]['item'] - > - - const streamer: StreamReader = { + const streamer: $StreamReader = { broker, - args, - fn, + args: init(allArgs) as $StreamReader['args'], + fn: last(allArgs) as $StreamReader['fn'], } this.#streams[streamName] ??= [] @@ -441,7 +431,7 @@ export default class MessageBus< const cancel = () => { this.#streams[streamName] = removeItem( streamer, - this.#streams[streamName] as any, + this.#streams[streamName]!, ) } diff --git a/packages/message-bus/src/types/streams.ts b/packages/message-bus/src/types/streams.ts index af2201161..4f213a906 100644 --- a/packages/message-bus/src/types/streams.ts +++ b/packages/message-bus/src/types/streams.ts @@ -13,20 +13,18 @@ export type ReaderFn = ( export interface StreamReader< $ extends MessageBusContext, - Args extends unknown[], - Item, + StreamName extends keyof $['streams'], > { broker: Broker<$> - args: Args - fn: ReaderFn + args: $['streams'][StreamName]['args'] + fn: ReaderFn< + $['streams'][StreamName]['args'], + $['streams'][StreamName]['item'] + > } export type Streams<$ extends MessageBusContext> = Partial<{ - [StreamName in keyof $['streams']]: StreamReader< - $, - $['streams'][StreamName]['args'], - $['streams'][StreamName]['item'] - >[] + [StreamName in keyof $['streams']]: StreamReader<$, StreamName>[] }> /** From 0fdbfc081db2ef22c06759cc20dfb03c55d5faeb Mon Sep 17 00:00:00 2001 From: John Wright Date: Wed, 10 Jun 2026 16:56:22 +0100 Subject: [PATCH 4/5] refactor: remove unused types --- packages/message-bus/src/types/MessageBus.ts | 17 +---------------- 1 file changed, 1 insertion(+), 16 deletions(-) diff --git a/packages/message-bus/src/types/MessageBus.ts b/packages/message-bus/src/types/MessageBus.ts index 2d182915e..05fd2718f 100644 --- a/packages/message-bus/src/types/MessageBus.ts +++ b/packages/message-bus/src/types/MessageBus.ts @@ -33,24 +33,9 @@ export interface MessageBusContext { * type Ctx = CreateMessageBusContext<{ events: { foo: [string] } }> * let bus: MessageBus */ -export type CreateMessageBusContext> = +export type CreateMessageBusContext = {}> = Omit & C -export type MessageBusEvents = - MB extends MessageBus ? $['events'] : never - -export type MessageBusEventGenerators = - MB extends MessageBus ? $['generators'] : never - -export type MessageBusInvokers = - MB extends MessageBus ? $['invokables'] : never - -export type MessageBusStreams = - MB extends MessageBus ? $['streams'] : never - -export type MessageBusBroker = - MB extends MessageBus ? Broker<$> : never - export interface ErrorHandler { (error: MessageBusError): any } From a3e8599568f67374235f33c82c610dc00bd6b67e Mon Sep 17 00:00:00 2001 From: John Wright Date: Wed, 10 Jun 2026 17:19:24 +0100 Subject: [PATCH 5/5] refactor: simplify filtermap typing --- packages/message-bus/src/array.ts | 48 +++++++++++++++++-------------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/packages/message-bus/src/array.ts b/packages/message-bus/src/array.ts index 854a9f5b7..99502f020 100644 --- a/packages/message-bus/src/array.ts +++ b/packages/message-bus/src/array.ts @@ -28,38 +28,32 @@ export function replaceLastItem(array: Ts, item: T) { export function filterMap( array: I[], - context: (input: I) => C, - filter: (input: I, context: C) => boolean, - map: (input: I, context: C) => O, + context: ContextCreator, + filter: Filterer, + map: Mapper, ): O[] export function filterMap( array: I[], - filter: (input: I) => I, - map: (input: I) => O, + filter: Filterer, + map: Mapper, ): O[] export function filterMap( array: I[], - contextOrFilter: ((input: I) => C) | ((input: I, context: C) => boolean), - filterOrMap: - | ((input: I, context: C) => boolean) - | ((input: I, context: C) => O), - maybeMap?: (input: I, context: C) => O, + contextOrFilter: ContextCreator | Filterer, + filterOrMap: Filterer | Mapper, + maybeMap?: Mapper, ): O[] { - type GetContext = (input: I) => C - type Filter = (input: I, context: C) => boolean - type Map = (input: I, context: C) => O - - let context: undefined | GetContext - let filter: Filter - let map: Map + let context: undefined | ContextCreator + let filter: Filterer + let map: Mapper if (maybeMap) { - context = contextOrFilter as GetContext - filter = filterOrMap as Filter - map = maybeMap as Map + context = contextOrFilter as ContextCreator + filter = filterOrMap as Filterer + map = maybeMap as Mapper } else { - filter = contextOrFilter as Filter - map = filterOrMap as Map + filter = contextOrFilter as Filterer + map = filterOrMap as Mapper } const output: O[] = [] @@ -69,3 +63,13 @@ export function filterMap( } return output } + +type Filterer = C extends never + ? (input: I) => boolean + : (input: I, context: C) => boolean + +type Mapper = C extends never + ? (input: I) => O + : (input: I, context: C) => O + +type ContextCreator = (input: I) => C