diff --git a/package-lock.json b/package-lock.json
index a73409a24..2255d7083 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 8a13b2b1c..e84d99497 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..6e962874e 100644
--- a/packages/message-bus/src/Broker.ts
+++ b/packages/message-bus/src/Broker.ts
@@ -1,29 +1,26 @@
-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'
-
-export default class Broker<
-  Events extends EventsT = EventsT,
-  EventGens extends EventGeneratorsT = EventGeneratorsT,
-  Invokables extends InvokablesDict = InvokablesDict
-> {
+import {
+  ErrorHandler,
+  MessageBusContext,
+  Unsubscriber,
+} from './types/MessageBus.js'
+import { StreamReaderArgs } from './types/streams.js'
+
+export default class Broker<$ extends MessageBusContext = MessageBusContext> {
   constructor(
-    messageBus: MessageBus<Events, EventGens, Invokables>,
+    messageBus: MessageBus<$>,
     id: string,
-    abortController: AbortController
+    abortController: AbortController,
   ) {
     this.#abortController = abortController
     this.#id = id
@@ -40,7 +37,7 @@ export default class Broker<
     return this.abortSignal.aborted
   }
 
-  get abortSignal() {
+  get abortSignal(): AbortSignal {
     return this.#abortController.signal
   }
 
@@ -54,7 +51,7 @@ export default class Broker<
     return this.#id
   }
 
-  readonly #messageBus: MessageBus<Events, EventGens, Invokables>
+  readonly #messageBus: MessageBus<$>
   get messageBus() {
     return this.#messageBus
   }
@@ -75,147 +72,164 @@ export default class Broker<
     })
   }
 
-  emit<EventName extends keyof Events>(
+  emit<EventName extends keyof $['events']>(
     eventName: EventName,
-    ...args: Events[EventName]
+    ...args: $['events'][EventName]
   ): void | Promise<void> {
     return this.#messageBus.emit(this, eventName, args)
   }
 
-  emitSignal<EventName extends keyof Events>(
+  emitSignal<EventName extends keyof $['events']>(
     eventName: EventName,
     signal: AbortSignal,
-    ...args: Events[EventName]
+    ...args: $['events'][EventName]
   ): void | Promise<void> {
     return this.#messageBus.emit(this, eventName, args, signal)
   }
 
-  interceptEvent<EventName extends keyof Events>(
+  interceptEvent<EventName extends keyof $['events']>(
     eventName: EventName,
-    ...args: EventInterceptorArgs<Events[EventName]>
+    ...args: EventInterceptorArgs<$['events'][EventName]>
   ): Unsubscriber {
     return this.#messageBus.interceptEvent(this as any, eventName, args)
   }
 
-  on<EventName extends keyof Events>(
+  on<EventName extends keyof $['events']>(
     eventName: EventName,
-    ...args: SubscriberArgs<Events[EventName]>
+    ...args: SubscriberArgs<$['events'][EventName]>
   ): Unsubscriber {
     return this.#messageBus.on(this, eventName, args)
   }
 
-  once<EventName extends keyof Events>(
+  once<EventName extends keyof $['events']>(
     eventName: EventName,
-    ...args: SubscriberArgs<Events[EventName]>
+    ...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<Events[EventName]>
+    EventName extends keyof $['events'],
+    Args extends UntilArgs<$['events'][EventName]>,
   >(
     eventName: EventName,
     ...args: Args
-  ): Promise<UntilRtn<Events[EventName], Args>> {
+  ): Promise<UntilRtn<$['events'][EventName], Args>> {
     return this.#messageBus.until(this, eventName, args) as any
   }
 
   async untilSignal<
-    EventName extends keyof Events,
-    Args extends UntilArgs<Events[EventName]>
+    EventName extends keyof $['events'],
+    Args extends UntilArgs<$['events'][EventName]>,
   >(
     eventName: EventName,
     abortSignal: AbortSignal,
     ...args: Args
-  ): Promise<UntilRtn<Events[EventName], Args>> {
+  ): Promise<UntilRtn<$['events'][EventName], Args>> {
     return this.#messageBus.until(this, eventName, args, abortSignal) as any
   }
 
-  generator<EventName extends keyof EventGens>(
+  generator<EventName extends keyof $['generators']>(
     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<EventName extends keyof EventGens>(
+  iterate<EventName extends keyof $['generators']>(
     eventName: EventName,
-    ...args: EventGens[EventName]['args']
-  ): AsyncIterable<EventGens[EventName]['yield']> {
+    ...args: $['generators'][EventName]['args']
+  ): AsyncIterable<$['generators'][EventName]['yield']> {
     return this.#messageBus.iterate(this, eventName, args)
   }
 
-  iterateSignal<EventName extends keyof EventGens>(
+  iterateSignal<EventName extends keyof $['generators']>(
     eventName: EventName,
     abortSignal: AbortSignal,
-    ...args: EventGens[EventName]['args']
-  ): AsyncIterable<EventGens[EventName]['yield']> {
+    ...args: $['generators'][EventName]['args']
+  ): AsyncIterable<$['generators'][EventName]['yield']> {
     return this.#messageBus.iterate(this, eventName, args, abortSignal)
   }
 
-  iterateWithin<EventName extends keyof EventGens>(
+  iterateWithin<EventName extends keyof $['generators']>(
     within: number,
     eventName: EventName,
-    ...args: EventGens[EventName]['args']
-  ): AsyncIterable<EventGens[EventName]['yield']> {
+    ...args: $['generators'][EventName]['args']
+  ): AsyncIterable<$['generators'][EventName]['yield']> {
     return this.#messageBus.iterateWithin(this, within, eventName, args)
   }
 
-  accumulate<EventName extends keyof EventGens>(
+  accumulate<EventName extends keyof $['generators']>(
     eventName: EventName,
-    ...args: EventGens[EventName]['args']
-  ): Promise<EventGens[EventName]['yield'][]> {
+    ...args: $['generators'][EventName]['args']
+  ): Promise<$['generators'][EventName]['yield'][]> {
     return this.#messageBus.accumulate(this, eventName, args)
   }
 
-  accumulateWithin<EventName extends keyof EventGens>(
+  accumulateWithin<EventName extends keyof $['generators']>(
     within: number,
     eventName: EventName,
-    ...args: EventGens[EventName]['args']
-  ): Promise<EventGens[EventName]['yield'][]> {
+    ...args: $['generators'][EventName]['args']
+  ): Promise<$['generators'][EventName]['yield'][]> {
     return this.#messageBus.accumulateWithin(this, within, eventName, args)
   }
 
-  register<InvokableName extends keyof Invokables>(
+  register<InvokableName extends keyof $['invokables']>(
     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<InvokableName extends keyof Invokables>(
+  invoke<InvokableName extends keyof $['invokables']>(
     invokableName: InvokableName,
-    ...args: Invokables[InvokableName]['args']
-  ): Promise<Invokables[InvokableName]['return']> {
+    ...args: $['invokables'][InvokableName]['args']
+  ): Promise<$['invokables'][InvokableName]['return']> {
     return this.#messageBus.invoke(this, invokableName, args)
   }
 
-  invokeSignal<InvokableName extends keyof Invokables>(
+  invokeSignal<InvokableName extends keyof $['invokables']>(
     invokableName: InvokableName,
     abortSignal: AbortSignal,
-    ...args: Invokables[InvokableName]['args']
-  ): Promise<Invokables[InvokableName]['return']> {
+    ...args: $['invokables'][InvokableName]['args']
+  ): Promise<$['invokables'][InvokableName]['return']> {
     return this.#messageBus.invoke(this, invokableName, args, abortSignal)
   }
 
-  interceptInvoker<InvokableName extends keyof Invokables>(
+  interceptInvoker<InvokableName extends keyof $['invokables']>(
     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<StreamName extends keyof $['streams']>(
+    streamName: StreamName,
+    ...args: StreamReaderArgs<
+      $['streams'][StreamName]['args'],
+      $['streams'][StreamName]['item']
+    >
+  ): Unsubscriber {
+    return this.#messageBus.reader(this, streamName, args)
+  }
+
+  stream<StreamName extends keyof $['streams']>(
+    streamName: StreamName,
+    ...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 5d57383dc..632b3be57 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'
@@ -12,46 +12,44 @@ 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,
   Invokers,
 } from './types/invokables.js'
 import { Stringable, UnpackResolvableValue } from './types/util.js'
-import { ErrorHandler, Unsubscriber } from './types/MessageBus.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 { StreamReader, StreamReaderArgs, Streams } 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,
+  $ extends MessageBusContext = MessageBusContext,
 > {
   #errorHandlers: ErrorHandler[] = []
-  #eventInterceptors: EventInterceptors<Events> = {}
-  #eventGenerators: EventGenerators<EventGens> = {}
-  #invokers: Invokers<Invokables> = {}
-  #invokerInterceptors: InvokerInterceptors<Invokables> = {}
+  #eventInterceptors: EventInterceptors<$> = {}
+  #eventGenerators: EventGenerators<$> = {}
+  #invokers: Invokers<$> = {}
+  #invokerInterceptors: InvokerInterceptors<$> = {}
   #queued: Array<() => unknown> = []
   #started = false
-  #subscribers: Subscribers<Events> = {}
+  #streams: Streams<$> = {}
+  #subscribers: Subscribers<$> = {}
 
   onError(errorHandler: ErrorHandler) {
     this.#errorHandlers.push(errorHandler)
@@ -72,7 +70,7 @@ export default class MessageBus<
         ? abort
         : fromSignal(abort)
 
-    return new Broker<Events, EventGens, Invokables>(this, id, abortController)
+    return new Broker<$>(this, id, abortController)
   }
 
   async start() {
@@ -80,10 +78,10 @@ export default class MessageBus<
     return Promise.all(this.#queued.map((handle) => handle()))
   }
 
-  emit<EventName extends keyof Events>(
-    broker: Broker<Events, EventGens, Invokables>,
+  emit<EventName extends keyof $['events']>(
+    broker: Broker<$>,
     eventName: EventName,
-    args: Events[EventName],
+    args: $['events'][EventName],
     abortSignal?: AbortSignal,
   ): void | Promise<void> {
     const handle = () => {
@@ -116,10 +114,10 @@ export default class MessageBus<
         )
   }
 
-  interceptEvent<EventName extends keyof Events>(
-    broker: Broker<Events, EventGens, Invokables>,
+  interceptEvent<EventName extends keyof $['events']>(
+    broker: Broker<$>,
     eventName: EventName,
-    args: EventInterceptorArgs<Events[EventName]>,
+    args: EventInterceptorArgs<$['events'][EventName]>,
   ): Unsubscriber {
     const interceptor = {
       broker,
@@ -141,12 +139,12 @@ export default class MessageBus<
     }
   }
 
-  interceptInvoker<InvokableName extends keyof Invokables>(
-    broker: Broker<Events, EventGens, Invokables>,
+  interceptInvoker<InvokableName extends keyof $['invokables']>(
+    broker: Broker<$>,
     invokableName: InvokableName,
     args: InvokerInterceptorArgs<
-      Invokables[InvokableName]['args'],
-      Invokables[InvokableName]['return']
+      $['invokables'][InvokableName]['args'],
+      $['invokables'][InvokableName]['return']
     >,
   ): Unsubscriber {
     const interceptor = {
@@ -169,10 +167,10 @@ export default class MessageBus<
     }
   }
 
-  on<EventName extends keyof Events>(
-    broker: Broker<Events, EventGens, Invokables>,
+  on<EventName extends keyof $['events']>(
+    broker: Broker<$>,
     eventName: EventName,
-    args: SubscriberArgs<Events[EventName]>,
+    args: SubscriberArgs<$['events'][EventName]>,
   ): Unsubscriber {
     if (broker.aborted) return () => {}
 
@@ -200,61 +198,63 @@ export default class MessageBus<
     return cancel
   }
 
-  once<EventName extends keyof Events>(
-    broker: Broker<Events, EventGens, Invokables>,
+  once<EventName extends keyof $['events']>(
+    broker: Broker<$>,
     eventName: EventName,
-    args: SubscriberArgs<Events[EventName]>,
+    args: SubscriberArgs<$['events'][EventName]>,
   ): Unsubscriber {
-    const fn = last(args) as SubscriberFn<Events[EventName]>
-    const onceFn: SubscriberFn<Events[EventName]> = (...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<Events[EventName]>,
+      replaceLastItem(args, onceFn) as SubscriberArgs<$['events'][EventName]>,
     )
     return cancel
   }
 
   async until<
-    EventName extends keyof Events,
-    Args extends UntilArgs<Events[EventName]>,
+    EventName extends keyof $['events'],
+    Args extends UntilArgs<$['events'][EventName]>,
   >(
-    broker: Broker<Events, EventGens, Invokables>,
+    broker: Broker<$>,
     eventName: EventName,
     args: Args,
     abortSignal?: AbortSignal,
-  ): Promise<UntilRtn<Events[EventName], Args>> {
-    return new Promise<UntilRtn<Events[EventName], Args>>((resolve, reject) => {
-      const abortSignalComposite = anySignal(abortSignal, broker.abortSignal)
+  ): Promise<UntilRtn<$['events'][EventName], Args>> {
+    return new Promise<UntilRtn<$['events'][EventName], Args>>(
+      (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<Events[EventName]>
+        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<EventName extends keyof EventGens>(
-    broker: Broker<Events, EventGens, Invokables>,
+  generator<EventName extends keyof $['generators']>(
+    broker: Broker<$>,
     eventName: EventName,
     args: EventGeneratorArgs<
-      EventGens[EventName]['args'],
-      EventGens[EventName]['yield']
+      $['generators'][EventName]['args'],
+      $['generators'][EventName]['yield']
     >,
   ): Unsubscriber {
     if (broker.aborted) return () => {}
@@ -283,12 +283,12 @@ export default class MessageBus<
     return cancel
   }
 
-  async *iterate<EventName extends keyof EventGens>(
-    broker: Broker<Events, EventGens, Invokables>,
+  async *iterate<EventName extends keyof $['generators']>(
+    broker: Broker<$>,
     eventName: EventName,
-    args: EventGens[EventName]['args'],
+    args: $['generators'][EventName]['args'],
     abortSignal?: AbortSignal,
-  ): AsyncIterable<EventGens[EventName]['yield']> {
+  ): AsyncIterable<$['generators'][EventName]['yield']> {
     if (!this.#started) await this.#queue(broker, () => {})
 
     yield* combineIterators(
@@ -303,13 +303,13 @@ export default class MessageBus<
     )
   }
 
-  iterateWithin<EventName extends keyof EventGens>(
-    broker: Broker<Events, EventGens, Invokables>,
+  iterateWithin<EventName extends keyof $['generators']>(
+    broker: Broker<$>,
     within: number,
     eventName: EventName,
-    args: EventGens[EventName]['args'],
+    args: $['generators'][EventName]['args'],
     abortSignal?: AbortSignal,
-  ): AsyncIterable<EventGens[EventName]['yield']> {
+  ): AsyncIterable<$['generators'][EventName]['yield']> {
     return iteratorRace(
       this.iterate(broker, eventName, args, abortSignal),
       within,
@@ -317,20 +317,20 @@ export default class MessageBus<
     )
   }
 
-  async accumulate<EventName extends keyof EventGens>(
-    broker: Broker<Events, EventGens, Invokables>,
+  async accumulate<EventName extends keyof $['generators']>(
+    broker: Broker<$>,
     eventName: EventName,
-    args: EventGens[EventName]['args'],
+    args: $['generators'][EventName]['args'],
     abortSignal?: AbortSignal,
   ) {
     return accumulate(this.iterate(broker, eventName, args, abortSignal))
   }
 
-  async accumulateWithin<EventName extends keyof EventGens>(
-    broker: Broker<Events, EventGens, Invokables>,
+  async accumulateWithin<EventName extends keyof $['generators']>(
+    broker: Broker<$>,
     within: number,
     eventName: EventName,
-    args: EventGens[EventName]['args'],
+    args: $['generators'][EventName]['args'],
     abortSignal?: AbortSignal,
   ) {
     return accumulate(
@@ -338,20 +338,20 @@ export default class MessageBus<
     )
   }
 
-  register<InvokableName extends keyof Invokables>(
-    broker: Broker<Events, EventGens, Invokables>,
+  register<InvokableName extends keyof $['invokables']>(
+    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(
@@ -374,13 +374,13 @@ export default class MessageBus<
     this.#invokers[invokableName] = [
       ...invokers,
       subscriber,
-    ] as unknown as Invokers<Invokables>[InvokableName]
+    ] as unknown as Invokers<$>[InvokableName]
 
     const cancel = () => {
       this.#invokers[invokableName] = removeItem(
         subscriber,
         this.#invokers[invokableName] as any,
-      ) as unknown as Invokers<Invokables>[InvokableName]
+      ) as unknown as Invokers<$>[InvokableName]
     }
 
     broker.onAbort(() => setTimeout(cancel, 0))
@@ -388,12 +388,12 @@ export default class MessageBus<
     return cancel
   }
 
-  async invoke<InvokableName extends keyof Invokables>(
-    broker: Broker<Events, EventGens, Invokables>,
+  async invoke<InvokableName extends keyof $['invokables']>(
+    broker: Broker<$>,
     invokableName: InvokableName,
-    args: Invokables[InvokableName]['args'],
+    args: $['invokables'][InvokableName]['args'],
     abortSignal?: AbortSignal,
-  ): Promise<Invokables[InvokableName]['return']> {
+  ): Promise<$['invokables'][InvokableName]['return']> {
     const handle = async () =>
       new Promise((resolve, reject) => {
         const abortSignalComposite = anySignal(abortSignal, broker.abortSignal)
@@ -408,16 +408,89 @@ export default class MessageBus<
     return this.#started ? handle() : this.#queue(broker, handle)
   }
 
-  #callEventInterceptors<EventName extends keyof Events>(
+  reader<StreamName extends keyof $['streams']>(
+    broker: Broker<$>,
+    streamName: StreamName,
+    allArgs: StreamReaderArgs<
+      $['streams'][StreamName]['args'],
+      $['streams'][StreamName]['item']
+    >,
+  ): Unsubscriber {
+    type $StreamReader = StreamReader<$, StreamName>
+
+    const streamer: $StreamReader = {
+      broker,
+      args: init(allArgs) as $StreamReader['args'],
+      fn: last(allArgs) as $StreamReader['fn'],
+    }
+
+    this.#streams[streamName] ??= []
+
+    this.#streams[streamName].push(streamer)
+
+    const cancel = () => {
+      this.#streams[streamName] = removeItem(
+        streamer,
+        this.#streams[streamName]!,
+      )
+    }
+
+    broker.onAbort(cancel)
+
+    return cancel
+  }
+
+  stream<StreamName extends keyof $['streams']>(
+    broker: Broker<$>,
+    streamName: StreamName,
+    args: $['streams'][StreamName]['args'],
+    abortSignal?: AbortSignal,
+  ): ReadableStream<$['streams'][StreamName]['item']> {
+    type Item = $['streams'][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<$['streams'][StreamName]['args']>),
+            ),
+          ),
+      )
+
+    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<never, Item>({}, mergeUnderlyingSource(streams)),
+    )
+  }
+
+  #callEventInterceptors<EventName extends keyof $['events']>(
     eventName: EventName,
-    args: Events[EventName],
-  ): void | Promise<Events[EventName] | typeof CancelEvent> {
+    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)
@@ -431,24 +504,24 @@ export default class MessageBus<
           moddedArgs = [
             ...moddedArgs.slice(0, index),
             ...newArgs,
-          ] as Events[EventName]
+          ] as $['events'][EventName]
       }
 
       return moddedArgs
     })()
   }
 
-  async #invokeChain<InvokableName extends keyof Invokables>(
+  async #invokeChain<InvokableName extends keyof $['invokables']>(
     invokableName: InvokableName,
-    args: Invokables[InvokableName]['args'],
+    args: $['invokables'][InvokableName]['args'],
     signal: AbortSignal,
-  ): Promise<Invokables[InvokableName]['return']> {
+  ): Promise<$['invokables'][InvokableName]['return']> {
     const invokerInterceptors = this.#invokerInterceptors[invokableName] || []
 
     const invokeChain = async (
       index: number,
-      args: Invokables[InvokableName]['args'],
-    ): Promise<Invokables[InvokableName]['return']> => {
+      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)
@@ -463,9 +536,9 @@ export default class MessageBus<
     return invokeChain(0, args)
   }
 
-  #callSubscribers<EventName extends keyof Events>(
+  #callSubscribers<EventName extends keyof $['events']>(
     eventName: EventName,
-    args: Events[EventName],
+    args: $['events'][EventName],
     abortSignal?: AbortSignal,
   ): void | Promise<void> {
     const subscribers = (this.#subscribers[eventName] || [])!
@@ -495,11 +568,11 @@ export default class MessageBus<
     }
   }
 
-  async #invoke<InvokableName extends keyof Invokables>(
+  async #invoke<InvokableName extends keyof $['invokables']>(
     invokableName: InvokableName,
-    args: Invokables[InvokableName]['args'],
+    args: $['invokables'][InvokableName]['args'],
     abortSignal: AbortSignal,
-  ): Promise<Invokables[InvokableName]['return']> {
+  ): Promise<$['invokables'][InvokableName]['return']> {
     const invokers = this.#invokers[invokableName]
     const invoker =
       invokers &&
@@ -512,7 +585,7 @@ export default class MessageBus<
     return invoker.fn(
       ...(args.slice(
         this.#argumentIndex(invoker.args, args),
-      ) as Invokables[InvokableName]['args']),
+      ) as $['invokables'][InvokableName]['args']),
       abortSignal,
     )
   }
@@ -526,10 +599,7 @@ export default class MessageBus<
     return i
   }
 
-  async #queue<T>(
-    broker: Broker<Events, EventGens, Invokables>,
-    handler: () => T,
-  ) {
+  async #queue<T>(broker: Broker<$>, handler: () => T) {
     return new Promise<UnpackResolvableValue<T>>((resolve, reject) => {
       if (broker.aborted) return reject(new AbortError())
 
diff --git a/packages/message-bus/src/array.ts b/packages/message-bus/src/array.ts
index 509015436..99502f020 100644
--- a/packages/message-bus/src/array.ts
+++ b/packages/message-bus/src/array.ts
@@ -20,8 +20,56 @@ export function removeIndex<T>(index: number, array: T[]) {
 export function replaceLastItem(array: never[], item: unknown): never[]
 export function replaceLastItem<Ts extends unknown[], T>(
   array: Ts,
-  item: T
+  item: T,
 ): L.Append<L.Pop<Ts>, T>
 export function replaceLastItem<Ts extends unknown[], T>(array: Ts, item: T) {
   return array.length === 0 ? array : [...init(array), item]
 }
+
+export function filterMap<I, O, C>(
+  array: I[],
+  context: ContextCreator<I, C>,
+  filter: Filterer<I, C>,
+  map: Mapper<I, O, C>,
+): O[]
+export function filterMap<I, O>(
+  array: I[],
+  filter: Filterer<I, never>,
+  map: Mapper<I, O, never>,
+): O[]
+export function filterMap<I, O, C>(
+  array: I[],
+  contextOrFilter: ContextCreator<I, C> | Filterer<I, C>,
+  filterOrMap: Filterer<I, C> | Mapper<I, O, C>,
+  maybeMap?: Mapper<I, O, C>,
+): O[] {
+  let context: undefined | ContextCreator<I, C>
+  let filter: Filterer<I, C>
+  let map: Mapper<I, O, C>
+
+  if (maybeMap) {
+    context = contextOrFilter as ContextCreator<I, C>
+    filter = filterOrMap as Filterer<I, C>
+    map = maybeMap as Mapper<I, O, C>
+  } else {
+    filter = contextOrFilter as Filterer<I, C>
+    map = filterOrMap as Mapper<I, O, C>
+  }
+
+  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
+}
+
+type Filterer<I, C> = C extends never
+  ? (input: I) => boolean
+  : (input: I, context: C) => boolean
+
+type Mapper<I, O, C> = C extends never
+  ? (input: I) => O
+  : (input: I, context: C) => O
+
+type ContextCreator<I, C> = (input: I) => C
diff --git a/packages/message-bus/src/errors/InvokableNotRegisteredError.ts b/packages/message-bus/src/errors/InvokableNotRegisteredError.ts
index b1caa4935..07635f16e 100644
--- a/packages/message-bus/src/errors/InvokableNotRegisteredError.ts
+++ b/packages/message-bus/src/errors/InvokableNotRegisteredError.ts
@@ -1,20 +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 { MessageBusContext } from '../types/MessageBus.js'
 
 export class InvokableNotRegisteredError<
-  Events extends EventsT,
-  EventGens extends EventGeneratorsT,
-  Invokables extends InvokablesDict
+  $ extends MessageBusContext = MessageBusContext,
 > extends Error {
-  #messageBus: MessageBus<Events, EventGens, Invokables>
+  #messageBus: MessageBus<$>
   #invokableName: string
 
-  constructor(
-    messageBus: MessageBus<Events, EventGens, Invokables>,
-    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..05fd2718f 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,28 @@ export interface Unsubscriber {
 
 export type AddAbortSignal<Args extends unknown[]> = L.Append<Args, AbortSignal>
 
-export type MessageBusEvents<MB extends MessageBus> = MB extends MessageBus<
-  infer Events,
-  any,
-  any
->
-  ? Events
-  : never
-
-export type MessageBusEventGenerators<MB extends MessageBus> =
-  MB extends MessageBus<any, infer EventGenerators, any>
-    ? EventGenerators
-    : never
-
-export type MessageBusInvokers<MB extends MessageBus> = MB extends MessageBus<
-  any,
-  any,
-  infer Invokers
->
-  ? Invokers
-  : 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
+}
 
-export type MessageBusBroker<MB extends MessageBus> = MB extends MessageBus<
-  infer Events,
-  infer EventGenerators,
-  infer Invokables
->
-  ? Broker<Events, EventGenerators, Invokables>
-  : never
+/**
+ * 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<Ctx>
+ */
+export type CreateMessageBusContext<C extends Partial<MessageBusContext> = {}> =
+  Omit<MessageBusContext, keyof C> & C
 
 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<string, unknown[]>
 
@@ -29,14 +29,15 @@ export type SubscriberArgs<A extends unknown[]> = _SubscriberArgs<
 type _SubscriberArgs<
   A extends unknown[],
   B extends unknown[],
-  Acc extends unknown[]
-> = L.Length<A> extends 0
-  ? Acc
-  : _SubscriberArgs<
-      L.Pop<A>,
-      L.Prepend<B, L.Last<A>>,
-      L.Append<A, SubscriberFn<B>> | Acc
-    >
+  Acc extends unknown[],
+> =
+  L.Length<A> extends 0
+    ? Acc
+    : _SubscriberArgs<
+        L.Pop<A>,
+        L.Prepend<B, L.Last<A>>,
+        L.Append<A, SubscriberFn<B>> | Acc
+      >
 
 /**
  * Create an `.until` argument union from a list of types.
@@ -49,10 +50,8 @@ type _SubscriberArgs<
  */
 export type UntilArgs<A extends unknown[]> = _UntilArgs<A, A>
 
-type _UntilArgs<
-  A extends unknown[],
-  Acc extends unknown[]
-> = L.Length<A> extends 0 ? Acc : _UntilArgs<L.Pop<A>, L.Pop<A> | Acc>
+type _UntilArgs<A extends unknown[], Acc extends unknown[]> =
+  L.Length<A> extends 0 ? Acc : _UntilArgs<L.Pop<A>, L.Pop<A> | 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<Args> extends 0
-  ? T
-  : L.Head<T> extends L.Head<Args>
-  ? UntilRtn<L.Tail<T>, L.Tail<Args>>
-  : never
+export type UntilRtn<T extends unknown[], Args extends unknown[]> =
+  L.Length<Args> extends 0
+    ? T
+    : L.Head<T> extends L.Head<Args>
+      ? UntilRtn<L.Tail<T>, L.Tail<Args>>
+      : never
 
-export type Subscribers<Events extends EventsT> = Partial<{
-  [EventName in keyof Events]: Subscriber<Broker>[]
+export type Subscribers<$ extends MessageBusContext> = Partial<{
+  [EventName in keyof $['events']]: Subscriber<$>[]
 }>
 
-export interface Subscriber<B extends Broker> {
-  broker: B
+export interface Subscriber<$ extends MessageBusContext> {
+  broker: Broker<$>
   args: unknown[]
   fn: SubscriberFn<unknown[]>
 }
 
 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<A> extends 0
-  ? Acc
-  : _EventInterceptorArgs<
-      L.Pop<A>,
-      L.Prepend<B, L.Last<A>>,
-      C,
-      Acc | [...A, EventInterceptorFn<B, C>]
-    >
+  Acc extends unknown[],
+> =
+  L.Length<A> extends 0
+    ? Acc
+    : _EventInterceptorArgs<
+        L.Pop<A>,
+        L.Prepend<B, L.Last<A>>,
+        C,
+        Acc | [...A, EventInterceptorFn<B, C>]
+      >
 
-export type EventInterceptors<Events extends EventsT> = Partial<{
-  [EventName in keyof Events]: EventInterceptor<Broker>[]
+export type EventInterceptors<$ extends MessageBusContext> = Partial<{
+  [EventName in keyof $['events']]: EventInterceptor<$>[]
 }>
 
-export interface EventInterceptor<B extends Broker> {
-  broker: B
+export interface EventInterceptor<$ extends MessageBusContext> {
+  broker: Broker<$>
   args: unknown[]
   fn: EventInterceptorFn<unknown[], unknown[]>
 }
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 extends unknown[], R> = (
   ...args: AddAbortSignal<Args>
 ) => AsyncIterable<R>
 
-export interface EventGenerator<B extends Broker> {
-  broker: B
+export interface EventGenerator<$ extends MessageBusContext> {
+  broker: Broker<$>
   args: unknown[]
   fn: EventGeneratorFn<unknown[], unknown>
 }
@@ -30,16 +30,17 @@ export type _EventGeneratorArgs<
   A extends unknown[],
   R,
   B extends unknown[],
-  Acc extends unknown[]
-> = L.Length<A> extends 0
-  ? Acc
-  : _EventGeneratorArgs<
-      L.Pop<A>,
-      R,
-      L.Prepend<B, L.Last<A>>,
-      Acc | L.Append<A, EventGeneratorFn<B, R>>
-    >
+  Acc extends unknown[],
+> =
+  L.Length<A> extends 0
+    ? Acc
+    : _EventGeneratorArgs<
+        L.Pop<A>,
+        R,
+        L.Prepend<B, L.Last<A>>,
+        Acc | L.Append<A, EventGeneratorFn<B, R>>
+      >
 
-export type EventGenerators<EventGens extends EventGeneratorsT> = Partial<{
-  [EventName in keyof EventGens]: EventGenerator<Broker>[]
+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<B extends Broker, Args extends unknown[], Return> {
-  broker: B
+export interface Invoker<
+  $ extends MessageBusContext,
+  Args extends unknown[],
+  Return,
+> {
+  broker: Broker<$>
   args: Args
   fn: InvokerFn<Args, Return>
 }
 
-export type Invokers<Invokables extends InvokablesDict> = 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<Args extends unknown[], Return> = (
 
 export type InvokerInterceptorArgs<
   A extends unknown[],
-  Return
+  Return,
 > = _InvokerInterceptorArgs<A, Return, A, [InvokerInterceptorFn<A, Return>]>
 
 export type _InvokerInterceptorArgs<
   A extends unknown[],
   Return,
   B extends unknown[],
-  Acc extends unknown[]
-> = L.Length<A> extends 0
-  ? Acc
-  : _InvokerInterceptorArgs<
-      L.Pop<A>,
-      Return,
-      B,
-      Acc | L.Append<A, InvokerInterceptorFn<B, Return>>
-    >
-
-export type InvokerInterceptors<Invokables extends InvokablesDict> = Partial<{
-  [InvokableName in keyof Invokables]: InvokerInterceptor<Broker>[]
+  Acc extends unknown[],
+> =
+  L.Length<A> extends 0
+    ? Acc
+    : _InvokerInterceptorArgs<
+        L.Pop<A>,
+        Return,
+        B,
+        Acc | L.Append<A, InvokerInterceptorFn<B, Return>>
+      >
+
+export type InvokerInterceptors<$ extends MessageBusContext> = Partial<{
+  [InvokableName in keyof $['invokables']]: InvokerInterceptor<$>[]
 }>
 
-export interface InvokerInterceptor<B extends Broker> {
-  broker: B
+export interface InvokerInterceptor<$ extends MessageBusContext> {
+  broker: Broker<$>
   args: unknown[]
   fn: InvokerInterceptorFn<unknown[], unknown>
 }
diff --git a/packages/message-bus/src/types/streams.ts b/packages/message-bus/src/types/streams.ts
new file mode 100644
index 000000000..4f213a906
--- /dev/null
+++ b/packages/message-bus/src/types/streams.ts
@@ -0,0 +1,59 @@
+import { L } from 'ts-toolbelt'
+import Broker from '../Broker.js'
+import { AddAbortSignal, MessageBusContext } from './MessageBus.js'
+import { UnderlyingDefaultSource } from 'node:stream/web'
+
+export type StreamsDict = Record<string, { args: unknown[]; item: unknown }>
+
+export type CreateStreamsDict<T extends StreamsDict> = T
+
+export type ReaderFn<Args extends unknown[], I> = (
+  ...args: AddAbortSignal<Args>
+) => UnderlyingDefaultSource<I>
+
+export interface StreamReader<
+  $ extends MessageBusContext,
+  StreamName extends keyof $['streams'],
+> {
+  broker: Broker<$>
+  args: $['streams'][StreamName]['args']
+  fn: ReaderFn<
+    $['streams'][StreamName]['args'],
+    $['streams'][StreamName]['item']
+  >
+}
+
+export type Streams<$ extends MessageBusContext> = Partial<{
+  [StreamName in keyof $['streams']]: StreamReader<$, StreamName>[]
+}>
+
+/**
+ * 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<A extends unknown[], Item> = _StreamReaderArgs<
+  A,
+  Item,
+  [],
+  [ReaderFn<A, Item>]
+>
+
+type _StreamReaderArgs<
+  A extends unknown[],
+  Item,
+  B extends unknown[],
+  Acc extends unknown[],
+> =
+  L.Length<A> extends 0
+    ? Acc
+    : _StreamReaderArgs<
+        L.Pop<A>,
+        Item,
+        L.Prepend<B, L.Last<A>>,
+        L.Append<A, ReaderFn<B, Item>> | Acc
+      >
diff --git a/packages/message-bus/test/MessageBus.test.ts b/packages/message-bus/test/MessageBus.test.ts
index 03f2db76f..a7c492512 100644
--- a/packages/message-bus/test/MessageBus.test.ts
+++ b/packages/message-bus/test/MessageBus.test.ts
@@ -6,13 +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<Events, {}, {}>
-  let broker: Broker<Events, {}, {}>
+  let messageBus: MessageBus<CreateMessageBusContext<{ events: Events }>>
+  let broker: Broker<CreateMessageBusContext<{ events: Events }>>
   let foo: Mock<() => void>
   let bar: Mock<(x: string) => void>
 
@@ -137,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<CreateMessageBusContext<{ generators: Iterables }>>
+  let broker: Broker<CreateMessageBusContext<{ generators: Iterables }>>
 
   beforeEach(() => {
     messageBus = new MessageBus()
@@ -197,8 +199,10 @@ describe('invokables', () => {
     afoo: { args: [string]; return: string }
     never: { args: []; return: Promise<never> }
   }>
-  let messageBus: MessageBus<{}, {}, Invokables>
-  let broker: Broker<{}, {}, Invokables>
+  let messageBus: MessageBus<
+    CreateMessageBusContext<{ invokables: Invokables }>
+  >
+  let broker: Broker<CreateMessageBusContext<{ invokables: Invokables }>>
   let foo: Mock<() => string>
   let bar: Mock<(x: string) => string>
 
@@ -285,8 +289,8 @@ describe('invokables', () => {
 
 describe('error handling', () => {
   type Events = { foo: []; bar: [string] }
-  let messageBus: MessageBus<Events, {}, {}>
-  let broker: Broker<Events, {}, {}>
+  let messageBus: MessageBus<CreateMessageBusContext<{ events: Events }>>
+  let broker: Broker<CreateMessageBusContext<{ events: Events }>>
 
   beforeEach(() => {
     messageBus = new MessageBus()
@@ -321,3 +325,117 @@ describe('error handling', () => {
       messageBus.start()
     }))
 })
+
+describe('streams', () => {
+  type Streamables = { foo: { args: [number]; item: string } }
+  let messageBus: MessageBus<CreateMessageBusContext<{ streams: Streamables }>>
+  let broker: Broker<CreateMessageBusContext<{ streams: 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')
+  })
+
+  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')
+  })
+})
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"],