Skip to content

feat(message-bus)!: rewrite around class-based events and per-broker queues - #878

Open
johngeorgewright wants to merge 36 commits into
masterfrom
dmg-bus
Open

johngeorgewright wants to merge 36 commits into
masterfrom
dmg-bus

Conversation

@johngeorgewright

@johngeorgewright johngeorgewright commented Sep 17, 2026 •

Copy link
Copy Markdown
Collaborator

Replace the old MessageBus API (string event names, tuple args, generators and
RPC-style invokables) with a class-based bus modelled on Enterprise Integration
Patterns. Messages are plain class instances routed by constructor identity, so
EventDescriptions are no longer needed. The invokable and generator APIs collapse
into a single streaming register/invoke pair, and the RPC-style request/response
model is removed.

Each bus member is a Participant with its own queue, so members can be paused
and resumed independently. A participant has two faces onto the same kernel: a
MessageGateway (outbound — what the application holds, to subscribe, emit and
invoke) and a MessageDispatcher (inbound — what the bus drives to deliver
messages to the registered handlers). The handler roles are named after their EIP
counterparts: Subscribers (on/once/until), Responders (register) that
stream values back for a CommandMessage, and Interceptors (intercept) that form
a Message Translator / Pipes-and-Filters chain able to transform or CANCEL a
message before subscribers see it. Filters are a Message Filter (whole-message
pass/discard).

Changes since this description was first written

The branch has moved on considerably since the PR was opened:

  • Enterprise Integration Patterns rename: Bus → MessageBus, broker →
    participant (facade: MessageGateway), Event → Message, Invocation →
    CommandMessage, and handlers are now Subscriber / Responder / Interceptor.
  • Externally abortable participants: bus.gateway(name, signal) ties a
    participant's lifetime to an AbortSignal; aborting tears down its handlers and
    frees its name for reuse.
  • Serializable payloads: emit/invoke arguments are constrained at compile
    time to JSON-safe data. Because a message can sit in a paused queue and be
    delivered later, a payload holding a live reference (e.g. an HTMLElement)
    would no longer be a faithful snapshot; the constraint guarantees a subscriber
    receives what was emitted, whenever it is emitted. It is a type-level rule
    only — nothing is cloned at runtime.

Bug fixes found while reviewing the rewrite

Each is covered by tests:

  • filters now AND their keys instead of OR
  • each emit() call resolves to its own result rather than a shared promise
  • invoke completion is the settling of each responder, not a finish() count, so a
    filtered-out or negligent responder can no longer hang or truncate the stream
  • invoke gains an error channel: fail-loud by default, isolate with { onError }
  • interceptor routing is cleaned up when a participant aborts
  • queueMethod no longer resets a running queue

BREAKING CHANGE: the entire public API of @plugola/message-bus has changed.

  • MessageBus replaces the old bus; bus.gateway(name) returns a MessageGateway,
    optionally scoped to an AbortSignal. EventDescriptions are removed.
  • Messages are classes implementing Message (or extending CommandMessage) and
    are subscribed to by class, not string name.
  • Invokables and generators are unified into a streaming API: register a responder
    with gateway.register(Command, (command, { send, signal }) => ...) and call
    gateway.invoke(new Command()).collect() / .iterate(). There is no finish() — a
    responder completes when it returns or its promise settles. The RPC-style
    request/response model is removed.
  • invoke errors are fail-loud by default; pass { onError } to invoke() to isolate
    responders and keep the stream running.
  • Message and command payloads must be Serializable (JSON-safe), enforced at the
    type level on emit/invoke.
  • Lifecycle methods are renamed: start/stop are now resume/pause on both
    MessageBus and MessageGateway, with abort for permanent teardown.
  • CancelEvent is renamed to CANCEL.

🤖 Generated with Claude Code

johngeorgewright and others added 30 commits September 4, 2026 17:24
We only need the one.
…queues

Replace the MessageBus API (string event names, tuple args, generators and
RPC-style invokables) with a class-based bus. Events are plain class instances
routed by constructor identity, so EventDescriptions are no longer needed. The
invokable and generator APIs collapse into a single streaming register/invoke
pair, and the RPC-style request/response model is removed. Each broker owns its
own queue, so brokers can be paused and resumed independently.

Also fixes flaws found while reviewing the rewrite, each covered by tests:
- filters now AND their keys instead of OR
- each emit() call resolves to its own result rather than a shared promise
- invoke completion is the settling of each handler, not a finish() count, so a
  filtered-out or negligent handler can no longer hang or truncate the stream
- invoke gains an error channel: fail-loud by default, isolate with { onError }
- interceptor routing is cleaned up when a broker aborts
- queueMethod no longer resets a running queue

BREAKING CHANGE: the entire public API of @plugola/message-bus has changed.

- `MessageBus` is replaced by `Bus`; `bus.broker(name)` returns a `PluginBroker`.
- Events are classes implementing `Event` (or extending `Invocation<T>`) and are
  subscribed to by class, not string name. `EventDescription`s are removed.
- Invokables and generators are unified into a streaming API: register a handler
  with `broker.register(Invocation, (event, { send, signal }) => ...)` and call
  `broker.invoke(new Invocation()).collect()` / `.iterate()`. There is no
  `finish()` — a handler completes when it returns or its promise settles. The
  RPC-style request/response model is removed.
- Invocation errors are fail-loud by default; pass `{ onError }` to `invoke()`
  to isolate handlers and keep the stream running.
- Lifecycle methods are renamed: `start`/`stop` are now `resume`/`pause` on both
  `Bus` and `PluginBroker`.
- `CancelEvent` is renamed to `CANCEL`.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Optionally pass an abort signal in to broker construction.
Realign the package's vocabulary with the EIP glossary:

- Bus            -> MessageBus
- PluginBroker   -> MessageGateway   (a participant's endpoint facade)
- Broker         -> MessageDispatcher (drives local performers)
- Event          -> Message
- Invocation     -> CommandMessage
- EventListener        -> Subscriber
- InvocationListener   -> Responder
- InterceptionListener -> Interceptor
- InvocationType -> ResponseType; $invocationType -> $responseType
- bus.broker(name) -> bus.gateway(name)

The dispatcher's inbound methods now say what they do rather than
mirroring the gateway's outbound names: emit -> dispatch, invoke ->
dispatchCommand, intercept -> runInterceptors.

Group modules by concept: Message/ (Message, CommandMessage), Roles/
(Subscriber, Responder, Interceptor), Gateway/ (MessageGateway,
MessageDispatcher, Handler). Document the EIP pattern each concept
maps to (Message Filter, Selective Consumer, Pipes-and-Filters).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Constrain emit/invoke arguments to a JSON-safe Serializable type so a
message that holds a live reference (e.g. an HTMLElement) is a compile
error. Messages can sit in a paused queue and be delivered later, so a
payload must be a faithful snapshot rather than a reference that may be
detached or mutated by then.

The check is type-level only; $-prefixed metadata keys ($name,
$responseType) are exempt since the class wrapper is never serialized.

Add tsc-gated type tests (test:types) covering the mapping and the
emit/invoke call sites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The gateway and dispatcher were coupled parent/child: the gateway reached
through `dispatcher.bus` to publish and into the dispatcher's registries to
wire up performers, while the dispatcher held `bus`/`name` purely for the
gateway's benefit.

Introduce a `Participant` that owns the shared kernel — name, queue, abort
lifecycle, and a `PerformerRegistry` — with two thin faces onto it:
`MessageGateway` (outbound) writes the registry and publishes on the bus;
`MessageDispatcher` (inbound) reads the registry and is gated on the queue.
Neither half owns the other, and the ownership graph now matches the runtime
flow (gateway.emit -> bus -> dispatcher). The dispatcher no longer holds the
bus at all.

Also rename `Handler` -> `SelectivePerformer` (`handle` -> `perform`) and the
`Gateway/` directory -> `Participant/`, aligning the whole folder with EIP
endpoint vocabulary (Messaging Gateway, Message Dispatcher, Selective
Consumer).

Note: on abort the registry now clears interceptors too, not just subscribers
and responders. This is unobservable — the bus drops the participant from all
routing and from its map at the same time — but is called out for review.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Let two MessageBus instances in separate runtimes (window, web worker,
server) exchange messages and commands over a pluggable Channel, modelled
on EIP's Messaging Bridge. The bus core is unchanged; the bridge joins a
bus as an ordinary participant.

- message()/command() factories define self-describing message classes
  (name + codec); MessageClass/CommandMessageClass now carry the
  $name/$encode/$decode statics, and Codec lives in Message/Codec.ts
- MessageRegistry: define-and-register in one act — the shared name->class
  contract and the bridge's forward set (registerMessage/registerCommand)
- Channel + Frame + LoopbackChannel; MessagingBridge relays messages and
  commands with correlation, response streaming, echo suppression, error
  envelopes, cancellation, and teardown
- end-to-end test over a real node:worker_threads Worker running the TS
  source natively (rewriteRelativeImportExtensions + .ts specifiers);
  type-only imports made verbatimModuleSyntax-clean
- README rewritten around the factory API with a transports section

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
MessagingBridge's correlationId is now optional, defaulting to
() => crypto.randomUUID() (a global in every target runtime), and stays
overridable for a custom scheme or deterministic tests. Callers no longer
have to supply one.

Also simplify the worker test fixtures: a createBridge() helper builds the
bus/registry/bridge, and PortLike is imported as a type (native transform
rejects value-importing a type).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The PortChannel can except most _port_ style messages.
Messages defined with message()/command() now carry identity only ($name).
The transport codec ($encode/$decode) is layered on by a Channel's
MessageRegistry via TransportableMixin, so only classes that cross a bridge
acquire a codec and in-process-only messages never touch one.

Update the bridge and registry types to the transportable class type, move
the codec tests from the factories to the registry, and refresh the README
and transport plan.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Messages are now created by calling a factory function (`TestMessage({...})`)
rather than `new`. `Message`/`CommandMessage` are plain interfaces, and each
message is a plain object carrying its payload plus `$name` and a `$factory`
back-reference — the factory is the routing identity (replacing
`message.constructor` in the bus/dispatcher).

- Rename `MessageClass`/`CommandMessageClass` -> `MessageFactory`/
  `CommandMessageFactory`; add `MessageOf<F>` to replace `InstanceType<F>`.
- Rename `TransportableMixin` -> `makeTransportable` (mutates and returns the
  same factory); drop `NamedMixin` (keep the `Named` interface).
- `Message` is payload-agnostic (no `Message<T>`; payload rides along as `& T`),
  avoiding the contravariance friction a payload type parameter reintroduced.
- Split the registry into message/command buckets: `messageFactories` /
  `commandFactories` (kind-agnostic `factoryFor` kept for decode), so the bridge
  needs no `$isCommand` flag and no cast.
- Fix `ResponseType` inference through the registry's `& Transportable<…>`
  intersection (it collapsed to `unknown` on a responder's `send`); infer the
  response type from the factory's return type. Add regression type tests.
- Update README, transport-plan doc, and all tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant