diff --git a/packages/message-bus/src/Broker.ts b/packages/message-bus/src/Broker.ts index 6e962874e..c80ac6dda 100644 --- a/packages/message-bus/src/Broker.ts +++ b/packages/message-bus/src/Broker.ts @@ -1,4 +1,3 @@ -import { InvokerRegistrationArgs } from '@plugola/invoke' import type MessageBus from './MessageBus.js' import SubscriptionDisposer from './SubscriptionDisposer.js' import type { @@ -8,7 +7,10 @@ import type { UntilRtn, } from './types/events.js' import type { EventGeneratorArgs } from './types/generators.js' -import type { InvokerInterceptorArgs } from './types/invokables.js' +import type { + InvokerInterceptorArgs, + MatchableInvokerRegistrationArgs, +} from './types/invokables.js' import { ErrorHandler, MessageBusContext, @@ -183,7 +185,7 @@ export default class Broker<$ extends MessageBusContext = MessageBusContext> { register( invokableName: InvokableName, - ...args: InvokerRegistrationArgs< + ...args: MatchableInvokerRegistrationArgs< $['invokables'][InvokableName]['args'], $['invokables'][InvokableName]['return'] > diff --git a/packages/message-bus/src/MessageBus.ts b/packages/message-bus/src/MessageBus.ts index 632b3be57..b409561df 100644 --- a/packages/message-bus/src/MessageBus.ts +++ b/packages/message-bus/src/MessageBus.ts @@ -23,6 +23,7 @@ import { InvokerInterceptorArgs, InvokerInterceptors, Invokers, + MatchableInvokerRegistrationArgs, } from './types/invokables.js' import { Stringable, UnpackResolvableValue } from './types/util.js' import { @@ -33,7 +34,8 @@ import { } from './types/MessageBus.js' import { anySignal, fromSignal } from './AbortController.js' import { InvokableNotRegisteredError } from './errors/InvokableNotRegisteredError.js' -import { InvokerFn, InvokerRegistrationArgs } from '@plugola/invoke' +import { InvokerFn } from '@plugola/invoke' +import { match } from './matcher.js' import { StreamReader, StreamReaderArgs, Streams } from './types/streams.js' import { WritableReadablePair } from '@johngw/stream/transformers/WritableReadablePair' import { mergeUnderlyingSource } from '@johngw/stream' @@ -341,7 +343,7 @@ export default class MessageBus< register( broker: Broker<$>, invokableName: InvokableName, - allArgs: InvokerRegistrationArgs< + allArgs: MatchableInvokerRegistrationArgs< $['invokables'][InvokableName]['args'], $['invokables'][InvokableName]['return'] >, @@ -595,7 +597,7 @@ export default class MessageBus< else if (args1.length > args2.length) return -1 let i = 0 - for (; i < args1.length; i++) if (args1[i] !== args2[i]) return -1 + for (; i < args1.length; i++) if (!match(args1[i], args2[i])) return -1 return i } diff --git a/packages/message-bus/src/index.ts b/packages/message-bus/src/index.ts index f042ddd0c..b5433b828 100644 --- a/packages/message-bus/src/index.ts +++ b/packages/message-bus/src/index.ts @@ -2,6 +2,7 @@ export { AbortError } from '@johngw/async' export { default as Broker } from './Broker.js' export { CancelEvent } from './symbols.js' export { default as MessageBus } from './MessageBus.js' +export * from './matcher.js' export * from './types/MessageBus.js' export * from './types/events.js' export * from './types/generators.js' diff --git a/packages/message-bus/src/matcher.ts b/packages/message-bus/src/matcher.ts new file mode 100644 index 000000000..10fef21ca --- /dev/null +++ b/packages/message-bus/src/matcher.ts @@ -0,0 +1,86 @@ +/** + * A {@link Matcher} stands in for a literal argument in a filter prefix. When + * one sits in a registered argument position, the bus runs its match method + * against the emitted value instead of comparing for strict equality. + * + * The contract is a single symbol-keyed method, so anything — a plain object, a + * class instance, a larger value that wants to double as a matcher — becomes a + * matcher simply by implementing it. + * + * `T` is contravariant (it only appears in the match parameter), so a narrow + * matcher such as `Matcher<{ bar: number }>` is accepted wherever a wider + * object that structurally includes it is expected. + */ +export const MatcherSymbol = Symbol.for('@plugola/message-bus matcher') + +export interface Matcher { + [MatcherSymbol](value: T): boolean +} + +export function isMatcher(value: unknown): value is Matcher { + return ( + typeof value === 'object' && + value !== null && + typeof (value as Partial)[MatcherSymbol] === 'function' + ) +} + +/** + * Compare a registered filter argument against an emitted value: run the + * predicate if it's a {@link Matcher}, otherwise compare for strict equality. + */ +export function match(matcher: unknown, value: unknown): boolean { + return isMatcher(matcher) ? matcher[MatcherSymbol](value) : matcher === value +} + +/** + * Widen a filter-prefix tuple so a {@link Matcher} is accepted in place of any + * literal argument. + * + * @example + * type A = Matchable<[string, { bar: number }]> + * // [string | Matcher, { bar: number } | Matcher<{ bar: number }>] + */ +export type Matchable = { + [K in keyof A]: A[K] | Matcher +} + +/** + * Match an argument against any boolean test. + * + * @example + * broker.on('foo', predicate((n: number) => n > 2), () => {}) + */ +export function predicate(match: (value: T) => boolean): Matcher { + return { [MatcherSymbol]: match } +} + +/** + * Match an object argument that deeply contains the given subset. Nested + * objects are compared recursively as partials; every other value is compared + * with strict equality. + * + * @example + * broker.on('foo', objectWith({ bar: 2 }), () => {}) + * // fires for emit('foo', { bar: 2, baz: 9 }) + */ +export function objectWith(subset: S): Matcher { + return { [MatcherSymbol]: (value) => objectContains(value, subset) } +} + +function objectContains(value: unknown, subset: object): boolean { + if (typeof value !== 'object' || value === null) return false + + for (const key of Reflect.ownKeys(subset)) { + const expected = (subset as Record)[key] + const actual = (value as Record)[key] + + if (expected !== null && typeof expected === 'object') { + if (!objectContains(actual, expected)) return false + } else if (actual !== expected) { + return false + } + } + + return true +} diff --git a/packages/message-bus/src/types/events.ts b/packages/message-bus/src/types/events.ts index 77b910852..187650dea 100644 --- a/packages/message-bus/src/types/events.ts +++ b/packages/message-bus/src/types/events.ts @@ -1,4 +1,5 @@ import type { L } from 'ts-toolbelt' +import type { Matcher, Matchable } from '../matcher.js' import type Broker from '../Broker.js' import type { CancelEvent } from '../symbols.js' import type { AddAbortSignal, MessageBusContext } from './MessageBus.js' @@ -36,7 +37,7 @@ type _SubscriberArgs< : _SubscriberArgs< L.Pop, L.Prepend>, - L.Append> | Acc + L.Append, SubscriberFn> | Acc > /** @@ -48,10 +49,10 @@ type _SubscriberArgs< * // | [string] * // | [string, number] */ -export type UntilArgs = _UntilArgs +export type UntilArgs = _UntilArgs> type _UntilArgs = - L.Length extends 0 ? Acc : _UntilArgs, L.Pop | Acc> + L.Length extends 0 ? Acc : _UntilArgs, Matchable> | Acc> /** * The return value of `.until` from given arguments and @@ -70,9 +71,11 @@ type _UntilArgs = export type UntilRtn = L.Length extends 0 ? T - : L.Head extends L.Head + : L.Head extends Matcher ? UntilRtn, L.Tail> - : never + : L.Head extends L.Head + ? UntilRtn, L.Tail> + : never export type Subscribers<$ extends MessageBusContext> = Partial<{ [EventName in keyof $['events']]: Subscriber<$>[] @@ -114,7 +117,7 @@ type _EventInterceptorArgs< L.Pop, L.Prepend>, C, - Acc | [...A, EventInterceptorFn] + Acc | [...Matchable, EventInterceptorFn] > export type EventInterceptors<$ extends MessageBusContext> = Partial<{ diff --git a/packages/message-bus/src/types/generators.ts b/packages/message-bus/src/types/generators.ts index 1d176e7a7..e9f9c6e5c 100644 --- a/packages/message-bus/src/types/generators.ts +++ b/packages/message-bus/src/types/generators.ts @@ -1,4 +1,5 @@ import type { L } from 'ts-toolbelt' +import type { Matchable } from '../matcher.js' import type Broker from '../Broker.js' import type { AddAbortSignal, MessageBusContext } from './MessageBus.js' @@ -38,7 +39,7 @@ export type _EventGeneratorArgs< L.Pop, R, L.Prepend>, - Acc | L.Append> + Acc | L.Append, EventGeneratorFn> > export type EventGenerators<$ extends MessageBusContext> = Partial<{ diff --git a/packages/message-bus/src/types/invokables.ts b/packages/message-bus/src/types/invokables.ts index c643f8c48..6c918af1e 100644 --- a/packages/message-bus/src/types/invokables.ts +++ b/packages/message-bus/src/types/invokables.ts @@ -1,6 +1,7 @@ -import type { InvokerFn } from '@plugola/invoke' +import type { InvokerFn, InvokerRegistrationArgs } from '@plugola/invoke' import type Broker from '../Broker.js' import type { L } from 'ts-toolbelt' +import type { Matchable } from '../matcher.js' import type { MessageBusContext } from './MessageBus.js' export { @@ -10,6 +11,23 @@ export { InvokerRegistrationArgs, } from '@plugola/invoke' +/** + * {@link InvokerRegistrationArgs} with matchers allowed in every filter-prefix + * position. Invoke owns the base type and stays matcher-agnostic; we widen each + * union member's prefix here, at the message-bus boundary. + */ +export type MatchableInvokerRegistrationArgs< + A extends unknown[], + Return, +> = WrapPrefix> + +type WrapPrefix = U extends [ + ...infer Pre extends unknown[], + infer Fn, +] + ? [...Matchable
, Fn]
+  : U
+
 export interface Invoker<
   $ extends MessageBusContext,
   Args extends unknown[],
@@ -50,7 +68,7 @@ export type _InvokerInterceptorArgs<
         L.Pop,
         Return,
         B,
-        Acc | L.Append>
+        Acc | L.Append, InvokerInterceptorFn>
       >
 
 export type InvokerInterceptors<$ extends MessageBusContext> = Partial<{
diff --git a/packages/message-bus/src/types/streams.ts b/packages/message-bus/src/types/streams.ts
index 4f213a906..b3fed29dd 100644
--- a/packages/message-bus/src/types/streams.ts
+++ b/packages/message-bus/src/types/streams.ts
@@ -1,4 +1,5 @@
 import { L } from 'ts-toolbelt'
+import type { Matchable } from '../matcher.js'
 import Broker from '../Broker.js'
 import { AddAbortSignal, MessageBusContext } from './MessageBus.js'
 import { UnderlyingDefaultSource } from 'node:stream/web'
@@ -55,5 +56,5 @@ type _StreamReaderArgs<
         L.Pop,
         Item,
         L.Prepend>,
-        L.Append> | Acc
+        L.Append, ReaderFn> | Acc
       >
diff --git a/packages/message-bus/test/MessageBus.test.ts b/packages/message-bus/test/MessageBus.test.ts
index a7c492512..4d4459cfb 100644
--- a/packages/message-bus/test/MessageBus.test.ts
+++ b/packages/message-bus/test/MessageBus.test.ts
@@ -10,6 +10,7 @@ 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'
+import { objectWith, predicate } from '../src/matcher.js'
 
 describe('events', () => {
   type Events = CreateEvents<{ foo: []; bar: [string]; mung: [string, number] }>
@@ -439,3 +440,106 @@ describe('streams', () => {
     expect(fn.mock.calls[1][0]).toBe('10')
   })
 })
+
+describe('matchers', () => {
+  type $ = CreateMessageBusContext<{
+    events: {
+      foo: [{ bar: number; baz?: string }]
+      keyed: [{ id: number }, string]
+      nested: [{ meta: { tag: string } }]
+      num: [number]
+    }
+    invokables: {
+      handle: { args: [{ id: number }]; return: string }
+    }
+    streams: { feed: { args: [{ kind: string }]; item: string } }
+    generators: {
+      gen: { args: [{ tag: string }]; yield: string }
+    }
+  }>
+
+  let messageBus: MessageBus<$>
+  let broker: Broker<$>
+
+  beforeEach(() => {
+    messageBus = new MessageBus()
+    broker = messageBus.broker('test')
+    messageBus.start()
+  })
+
+  test('objectWith fires only on matching objects', () => {
+    const fn = vi.fn()
+    broker.on('foo', objectWith({ bar: 2 }), fn)
+    broker.emit('foo', { bar: 1 })
+    broker.emit('foo', { bar: 2 })
+    broker.emit('foo', { bar: 2, baz: 'x' })
+    expect(fn).toHaveBeenCalledTimes(2)
+  })
+
+  test('the matched argument is consumed', () => {
+    const fn = vi.fn()
+    broker.on('keyed', objectWith({ id: 1 }), fn)
+    broker.emit('keyed', { id: 1 }, 'hello')
+    expect(fn).toHaveBeenCalledWith('hello', expect.any(AbortSignal))
+  })
+
+  test('objectWith matches nested objects deeply', () => {
+    const fn = vi.fn()
+    broker.on('nested', objectWith({ meta: { tag: 'a' } }), fn)
+    broker.emit('nested', { meta: { tag: 'b' } })
+    broker.emit('nested', { meta: { tag: 'a' } })
+    expect(fn).toHaveBeenCalledTimes(1)
+  })
+
+  test('predicate matches arbitrary values', () => {
+    const fn = vi.fn()
+    broker.on(
+      'num',
+      predicate((n: number) => n > 2),
+      fn,
+    )
+    broker.emit('num', 1)
+    broker.emit('num', 3)
+    expect(fn).toHaveBeenCalledTimes(1)
+    expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal))
+  })
+
+  test('matchers filter invoker registrations', async () => {
+    broker.register('handle', objectWith({ id: 1 }), () => 'one')
+    broker.register('handle', () => 'other')
+    expect(await broker.invoke('handle', { id: 1 })).toBe('one')
+    expect(await broker.invoke('handle', { id: 2 })).toBe('other')
+  })
+
+  test('matchers filter stream readers', async () => {
+    const fn = vi.fn()
+    broker.reader('feed', objectWith({ kind: 'a' }), () => ({
+      start(controller) {
+        controller.enqueue('matched')
+        controller.close()
+      },
+    }))
+    broker.reader('feed', objectWith({ kind: 'b' }), () => ({
+      start(controller) {
+        controller.enqueue('ERROR')
+        controller.close()
+      },
+    }))
+    await broker.stream('feed', { kind: 'a' }).pipeTo(write(fn))
+    expect(fn).toHaveBeenCalledTimes(1)
+    expect(fn.mock.calls[0][0]).toBe('matched')
+  })
+
+  test('matchers filter generators', async () => {
+    const results: string[] = []
+    broker.generator('gen', objectWith({ tag: 'a' }), async function* () {
+      yield 'matched'
+    })
+    broker.generator('gen', objectWith({ tag: 'b' }), async function* () {
+      yield 'ERROR'
+    })
+    for await (const result of broker.iterate('gen', { tag: 'a' }))
+      results.push(result)
+    expect(results).toEqual(['matched'])
+  })
+})