@actdim/msgmesh - A type-safe, modular message mesh for scalable async communication in TypeScript

Table of Contents

• Quick Start
• Installation
• Overview
  • The Challenge
  • Analysis of Existing Solutions
  • The Solution
  • Implementation Foundation
  • Key Design Goals
• Architecture
  • Message Structure
  • Type Definition Example
• Usage Patterns
  • Global vs Local Usage
  • Creating a Message Bus
  • Type Utilities
• API Reference
  • Configuration
  • send()
  • on()
  • once()
  • stream()
  • provide()
    • Provider-Side Cancellation
  • request()
    • Request Cancellation
  • requestStream()
• Advanced Features
  • Message Replay
  • Throttling and Debouncing
  • Chain of Responsibility
  • Error Handling
  • Headers and Metadata
  • Service Adapters
• Comparison

Quick Start

Try @actdim/msgmesh instantly in your browser without any installation:

Once the project loads, run the tests to see the message bus in action:

pnpm run test

Installation

npm install @actdim/msgmesh

Peer Dependencies

This package requires the following peer dependencies:

npm install @actdim/utico rxjs

Or with pnpm:

pnpm add @actdim/utico rxjs

Overview

The Challenge

Modern client-side TypeScript applications require robust event handling mechanisms. Events may be needed within a single component or for communication between components, serving as a decoupling layer independent of component hierarchy. As applications grow in complexity and scale, the convenience, performance, and flexibility of the event system become critical factors. A well-designed messaging system enables extensibility, maintainability, and scalability without losing control over component interactions or system observability. Such a system becomes one of the pillars of high-quality application architecture.

In our case, this message bus serves as the foundation of the @actdim/dynstruct architectural framework.

Analysis of Existing Solutions

When examining popular messaging systems in the frontend ecosystem, particularly for React-based applications, several categories emerge:

Event Emitters

• Pros: Simple to understand, typically local in scope
• Cons:
  • Limited capabilities and scalability
  • Weak support for interaction structures and declarative approaches
  • Poor type safety (fictitious typing, manual implementation required)
  • Incomplete Promise integration
  • Lack of abstraction levels

Message Buses

• Pros: Reduce component coupling, beneficial for development and testing
• Cons:
  • Underdeveloped type system despite TypeScript's power
  • Often feel like academic experiments porting backend message buses to frontend
  • Poor integration with common development patterns (limited adapters for rate limiting, throttling, debouncing, retry logic)
  • More complex to maintain

Reactive Event Streams & Observer Pattern

• Pros: Powerful for compositions and complex data flows
• Cons:
  • Complex to understand, maintain, and debug
  • Strong architectural influence requiring paradigm shift (similar to procedural-to-functional programming transition)
  • Often tightly embedded throughout the system as an integral part
  • Creates hard dependencies across types, code style, tests, DI, error handling, and even team thinking
  • Essentially becomes the "language" of the application

React State Management Systems

• Pros: Purpose-built for React ecosystem
• Cons:
  • Tight coupling with React (hooks, lifecycle), making usage outside components difficult
  • Significant boilerplate code slowing development and complicating maintenance
  • Often enforce immutability paradigm, which looks elegant on paper but creates more problems and wrapper code than value in practice
  • Rarely provide configuration for event/stream connections (possibly due to weak or inconvenient payload typing)

The Solution: @actdim/msgmesh

@actdim/msgmesh addresses these shortcomings by providing a message bus that is:

• Flexible and extensible: Adapts to various use cases without imposing rigid patterns
• Scalable: Grows with your application without losing manageability
• Minimally opinionated: Doesn't force a specific paradigm
• Simple to understand: Clear mental model and API
• Local in impact: Doesn't permeate every aspect of your codebase

Implementation Foundation

@actdim/msgmesh is built on top of RxJS, leveraging the power and quality of this battle-tested library while hiding its complexity and architectural influence (see the comparison section above). This approach provides the best of both worlds: robust reactive stream processing under the hood with a simple, intuitive API on the surface.

Key RxJS components utilized:

• Subjects/Observables: Power the queue management system and state control, implementing the publish-subscribe (pub/sub) pattern efficiently
• Async Scheduler: Ensures the message bus operates independently from individual message handlers, preventing blocking and maintaining system responsiveness
• Pipe Operators: Enable flexible message flow behaviors within channels (throttling, debouncing, filtering, etc.) without exposing reactive programming complexity

By abstracting RxJS behind a clean API, @actdim/msgmesh delivers enterprise-grade stream processing capabilities without requiring developers to adopt reactive programming paradigms or deal with the steep learning curve typically associated with RxJS.

Key Design Goals

Observability

• Comprehensive logging and tracing capabilities
• Ability to subscribe to any event at any time
• Minimal system complexity and coupling
• Maintained control and visibility

Lifecycle Management

• Convenient subscription and unsubscription with various configuration options
• Automatic cleanup
• Integration with React lifecycle (when needed)
• Support for AbortSignal and AbortController patterns

Architecture

Message Structure

The message bus is defined by a type structure consisting of three levels:

1. Channels

Channels organize messages by task class, domain, event type, or any other logical grouping. Channels use string identifiers with dot notation recommended for namespacing.

Reserved System Channel: MSGBUS.ERROR - for system-level errors

2. Groups

Groups define message roles within a channel. There are two semantic kinds:

Input groups — any name except "out" and "error". Declare the payload type of messages entering the channel:

• "in" — the conventional name for the primary input group; used as default by send(), on(), provide(), and request() when no group is specified.
• Custom names ("in1", "in2", etc.) — define additional input payload types on the same channel. This is input type overloading: a single channel can accept different payload shapes through different groups, all producing the same out type. A handler registered with provide() must specify which input group it targets (and therefore which payload type it accepts).

Output group — always named "out". Declares the payload type of the channel's response. One out type covers all input groups on the channel:

• If out is not declared, MsgStruct<> implicitly adds out?: void — the channel produces no meaningful response.
• out: void is intentional: it means the handler ran and the caller can await confirmation, but no data is returned.
• out types must not be wrapped in Promise — async is handled at the API level.

Reserved groups: "error" (channel-specific error routing) and the MSGBUS.ERROR system channel.

send() vs request() — fire-and-forget vs awaited handling

This distinction maps directly to whether you care about the out group:

• send() — publishes to the channel's input group and returns immediately. No waiting for any handler. Use when you only care that the event was dispatched.
• request() / requestStream() — publishes and awaits the handler's response via the out group. Even when out: void, this confirms the message was processed, not just published. Use when you need to know the work completed.

A handler registered with provide() can skip individual messages by setting msgOut.status = 'skipped', leaving them for another handler on the same channel (see Chain of Responsibility). A skipped message produces no out response from that handler.

3. Message Types

Each group declares one payload type. Any valid TypeScript type is supported.

Use MsgStruct<...> to define bus structures — it augments your channels with the system error group and implicit out?: void / in?: void for channels that don't declare them explicitly:

Type Definition Example

import { MsgStruct } from '@actdim/msgmesh';

export type MyBusStruct = MsgStruct<{
    'TEST.COMPUTE_SUM': {
        in: { a: number; b: number };
        out: number;
    };
    'TEST.DO_SOME_WORK': {
        in: string;
        out: void;
    };
    'TEST.TEST_TASK_WITH_REPEAT': {
        in: string;
        out: void;
    };
    'TEST.MULTIPLEXER': {
        in1: string;
        in2: number;
        out: number;
    };
}>;

Usage Patterns

Global vs Local Usage

@actdim/msgmesh can be used in two primary ways:

Global Application-Level Bus

Maintain a system-wide type structure for messages/events, organizing them by:

• Tasks (component ownership)
• Groups (in/out, input type overloading)
• Topics (additional filtering)

Local Component/Module-Level Bus

Use within any logical grouping of components or modules.

Important: You only need one bus instance for the entire application. The bus routes messages based on keys, so as long as key uniqueness is maintained, a single instance can handle messages from any locally-defined schema.

Creating a Message Bus

import { createMsgBus, MsgBus } from '@actdim/msgmesh';
import { KeysOf } from '@actdim/utico/typeCore';

// Basic bus creation
const msgBus = createMsgBus<MyBusStruct>();

// With custom headers (if needed)
type CustomHeaders = MsgHeaders & {
    userId?: string;
    sessionId?: string;
};

const msgBusWithHeaders = createMsgBus<MyBusStruct, CustomHeaders>();

// Note: The instance can process messages from other structures too
// We only type the API for development convenience
// You can compose structures as needed, just ensure they don't overlap (unless intentional)

type AppBusStruct = ComponentBusStruct & ApiBusStruct;
const appMsgBus = createMsgBus<AppBusStruct>();

Type Utilities

// Export bus type for dependency injection or props
export type MyMsgBus = MsgBus<MyBusStruct, CustomHeaders>;

// Generic string literal type for channels - useful for component constraints
type MyMsgChannels<TChannel extends keyof MyBusStruct | Array<keyof MyBusStruct>> = KeysOf<
    MyBusStruct,
    TChannel
>;

// Example: Restricting a component to specific channels
// Helper types are necessary for IntelliSense with dynamic types
// All API checks are enforced at compile time - you cannot violate defined contracts
type Behavior = {
    messages: MyMsgChannels<'TEST.COMPUTE_SUM' | 'TEST.DO_SOME_WORK'>;
};

API Reference

| Method | Description | |--------|-------------| | send() | Publish a message to a channel | | on() | Subscribe to messages on a channel | | once() | Await a single message (Promise-based) | | stream() | Consume messages as an async iterable | | provide() | Register a request handler (auto-responds on out) | | request() | Send a request and await a single response | | requestStream() | Send a request and consume all responses as an async iterable (Fan-in) |

Configuration

You can configure channels with various options:

import { MsgBusConfig } from '@actdim/msgmesh';

const config: MsgBusConfig<MyBusStruct> = {
    // Static default config applied to all channels (channel-specific overrides win)
    "*": {
        mandatoryProvider: true,
    },
    'TEST.COMPUTE_SUM': {
        replayBufferSize: 10, // Number of messages to buffer for replay
        replayWindowTime: 5000, // Time window for replay (ms)
        delay: 100, // Delay before processing (ms)
        throttle: {
            // Throttle configuration
            duration: 1000,
            leading: true,
            trailing: true,
        },
        debounce: 500, // Debounce delay (ms)
        mandatoryProvider: false, // overrides "*" for this channel
    },
};

const msgBus = createMsgBus<MyBusStruct>(config);

You can also pass a function as "*" to compute per-channel defaults dynamically:

const config: MsgBusConfig<MyBusStruct> = {
    "*": (channel) => ({
        mandatoryProvider: channel.startsWith('Api.'),
    }),
    'TEST.COMPUTE_SUM': {
        mandatoryProvider: false, // overrides the function result for this channel
    },
};

Note: "*" is reserved and cannot be used as a channel name in your bus struct.

Sending Messages: send()

Send a message to the bus for a specific channel and group (default is in). The payload type is enforced according to the bus structure.

// Basic send
await msgBus.send({
    channel: 'TEST.COMPUTE_SUM',
    payload: { a: 10, b: 20 }, // Typed and validated
});

// With group specification
await msgBus.send({
    channel: 'TEST.MULTIPLEXER',
    group: 'in1',
    payload: 'hello', // Typed as string for 'in1' group
});

await msgBus.send({
    channel: 'TEST.MULTIPLEXER',
    group: 'in2',
    payload: 42, // Typed as number for 'in2' group
});

// With topic
await msgBus.send({
    channel: 'TEST.DO_SOME_WORK',
    topic: 'priority-high',
    payload: 'urgent task',
});

// With custom headers
await msgBus.send({
    channel: 'TEST.COMPUTE_SUM',
    payload: { a: 5, b: 15 },
    headers: {
        correlationId: 'task-123',
        priority: 1,
    },
});

Note: You can specify a topic when sending to enable fine-grained filtering by subscribers.

Subscribing to Messages: on()

Subscribe to messages on a specific channel and group with optional topic filtering.

// Basic subscription
msgBus.on({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg) => {
        // msg.payload is typed as { a: number; b: number }
        console.log('Received:', msg.payload);
    },
});

// Subscribe to specific group
msgBus.on({
    channel: 'TEST.COMPUTE_SUM',
    group: 'out', // Listen for responses
    callback: (msg) => {
        // msg.payload is typed as number
        console.log('Result:', msg.payload);
    },
});

// With topic filtering (regex pattern)
msgBus.on({
    channel: 'TEST.DO_SOME_WORK',
    topic: '/^task-.*/', // Match topics starting with "task-"
    callback: (msg) => {
        console.log('Task message:', msg.payload);
    },
});

// With options
msgBus.on({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg) => {
        console.log('Message:', msg.payload);
    },
    options: {
        fetchCount: 5, // Auto-unsubscribe after 5 messages
        throttle: {
            // Throttle the callback
            duration: 1000,
            leading: true,
            trailing: false,
        },
    },
});

Automatic Unsubscription

Limit message count: Use fetchCount to automatically unsubscribe after receiving a specific number of messages.

msgBus.on({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg) => {
        console.log(msg.payload);
    },
    options: {
        fetchCount: 10, // Unsubscribe after 10 messages
    },
});

Manual Unsubscription with AbortSignal

Use AbortSignal for controlled unsubscription. This allows combining abort signals from multiple AbortController instances.

const abortController = new AbortController();

msgBus.on({
    channel: "TEST.COMPUTE_SUM",
    callback: (msg) => {
        console.log(msg.payload);
    },
    options: {
        abortSignal: abortController.signal
    }
});

// Later: unsubscribe
abortController.abort();

// Combining multiple abort signals
const controller1 = new AbortController();
const controller2 = new AbortController();

const combinedSignal = AbortSignal.any([
    controller1.signal,
    controller2.signal
]);

msgBus.on({
    channel: "TEST.COMPUTE_SUM",
    options: {
        abortSignal: combinedSignal
    },
    callback: (msg) => {
        console.log(msg.payload);
    }
});

// React integration example - cleanup on unmount
import { useEffect } from 'react';

function MyComponent() {
    useEffect(() => {
        const controller = new AbortController();

        msgBus.on({
            channel: "TEST.EVENTS",
            callback: handleEvent,
            options: {
                abortSignal: controller.signal
            }
        });

        // Clean up when component unmounts
        return () => {
            controller.abort();
        };
    }, []);

    return <div>Component content</div>;
}

Awaiting a Single Message: once()

Subscribe and await the first (next) message on a specific channel and group, similar to on() but returns a Promise.

// Wait for one message
const msg = await msgBus.once({
    channel: 'TEST.COMPUTE_SUM',
});

console.log('Received:', msg.payload); // Typed as { a: number; b: number }

// With group specification
const response = await msgBus.once({
    channel: 'TEST.COMPUTE_SUM',
    group: 'out',
});

console.log('Result:', response.payload); // Typed as number

// With topic filtering
const taskMsg = await msgBus.once({
    channel: 'TEST.DO_SOME_WORK',
    topic: '/^priority-.*/', // Match topics starting with "priority-"
});

Timeout Configuration

Configure timeout duration via the timeout option (default: defaultPromiseTimeout = 5 seconds, configurable globally). The abortSignal option also works with once().

try {
    const msg = await msgBus.once({
        channel: 'TEST.COMPUTE_SUM',
        options: {
            timeout: 5000, // 5 second timeout
        },
    });
    console.log('Received:', msg.payload);
} catch (error) {
    if (error instanceof TimeoutError) {
        console.error('Timeout waiting for message');
    }
}

// With abort signal
const abortController = new AbortController();

const messagePromise = msgBus.once({
    channel: 'TEST.COMPUTE_SUM',
    options: {
        timeout: 10000,
        abortSignal: abortController.signal,
    },
});

// Can cancel from elsewhere
setTimeout(() => abortController.abort('User canceled'), 2000);

try {
    const msg = await messagePromise;
} catch (error) {
    if (error instanceof AbortError) {
        console.error('Aborted:', error.cause);
    }
}

Streaming Messages: stream()

Create an async iterable iterator for consuming messages as a stream.

// Basic streaming
const messageStream = msgBus.stream({
    channel: 'TEST.COMPUTE_SUM',
});

for await (const msg of messageStream) {
    console.log('Received:', msg.payload);
    // Process messages as they arrive
}

// With topic filtering
const taskStream = msgBus.stream({
    channel: 'TEST.DO_SOME_WORK',
    topic: '/^task-.*/',
});

for await (const msg of taskStream) {
    await processTask(msg.payload);
}

Timeout and Cancellation

The timeout option is an inactivity timeout — the timer resets on each received message. If no message arrives within the timeout window, the stream ends with a TimeoutError. This is useful for detecting when the producer has stopped sending.

For a hard time limit on the stream's total duration, use AbortSignal.timeout().

// Inactivity timeout: end stream if no messages for 5s
const stream1 = msgBus.stream({
    channel: 'TEST.EVENTS',
    options: {
        timeout: 5000,
    },
});

// Total duration limit: end stream after 60s regardless of activity
const stream2 = msgBus.stream({
    channel: 'TEST.EVENTS',
    options: {
        abortSignal: AbortSignal.timeout(60000),
    },
});

// Both: inactivity 5s + hard limit 60s
const stream3 = msgBus.stream({
    channel: 'TEST.EVENTS',
    options: {
        timeout: 5000,
        abortSignal: AbortSignal.timeout(60000),
    },
});

Providing Response Handlers: provide()

Register a handler for messages on a selected channel and group (typically in), which generates a response message for the out group of the same channel. This is essentially a subscription with automatic response handling.

The callback can be asynchronous and its result is automatically used to form the response.

// Simple provider
msgBus.provide({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg) => {
        // msg.payload is typed as { a: number; b: number }
        // Return type is inferred as number (from 'out' type)
        return msg.payload.a + msg.payload.b;
    },
});

// Async provider
msgBus.provide({
    channel: 'TEST.DO_SOME_WORK',
    callback: async (msg) => {
        // msg.payload is typed as string
        await performWork(msg.payload);
        // Return type is void (from 'out' type)
    },
});

// With topic filtering
msgBus.provide({
    channel: 'TEST.COMPUTE_SUM',
    topic: '/^calc-.*/',
    callback: (msg) => {
        return msg.payload.a + msg.payload.b;
    },
});

// With options
msgBus.provide({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg) => {
        return msg.payload.a + msg.payload.b;
    },
    options: {
        fetchCount: 100, // Handle 100 requests then unsubscribe
        abortSignal: someController.signal,
    },
});

Cancellation Handling

The provider callback receives msgIn (the incoming request) and msgOut (the pre-initialized outgoing message for the out group). When a request is canceled by the caller (via AbortSignal in request()), a cancel message with msg.status === 'canceled' is delivered to the provider callback. The provider should check msg.status and handle cancellation accordingly. The bus awaits the callback even for cancel messages (so async cleanup completes), then skips publishing the out response.

A provider can also initiate cancellation itself by setting msgOut.status = 'canceled' before returning — the bus will skip the out publish and request() will reject with OperationCanceledError.

To skip handling and let another provider respond (chain of responsibility), set msgOut.status = 'skipped' and return.

For providers that don't need cancellation support, simply check that msg.status !== 'canceled' before doing work:

msgBus.provide({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg, msgOut) => {
        if (msg.status === 'canceled') return;
        return msg.payload.a + msg.payload.b;
    },
});

For providers with long-running or cancelable operations (e.g. fetch), track active requests by requestId and abort them when a cancel message arrives:

const activeRequests = new Map<string, AbortController>();

msgBus.provide({
    channel: 'API.FETCH_DATA',
    callback: async (msg, msgOut) => {
        const { requestId } = msg.headers;

        // Cancel message — abort the in-flight request
        if (msg.status === 'canceled') {
            activeRequests.get(requestId)?.abort();
            activeRequests.delete(requestId);
            return;
        }

        // Normal request — create AbortController and track it
        const controller = new AbortController();
        activeRequests.set(requestId, controller);

        try {
            const response = await fetch(msg.payload.url, {
                signal: controller.signal,
            });
            return await response.json();
        } finally {
            activeRequests.delete(requestId);
        }
    },
});

Request-Response Pattern: request()

Send a message and automatically await a response from a handler (registered via provide()) on the same channel's out group. Returns a Promise that resolves with the response message.

// Basic request
const response = await msgBus.request({
    channel: 'TEST.COMPUTE_SUM',
    payload: { a: 10, b: 20 },
});

console.log('Result:', response.payload); // Typed as number

// With group overloading (using different input groups)
const response1 = await msgBus.request({
    channel: 'TEST.MULTIPLEXER',
    group: 'in1',
    payload: 'hello',
});

const response2 = await msgBus.request({
    channel: 'TEST.MULTIPLEXER',
    group: 'in2',
    payload: 42,
});

// Both responses have payload with type - number ('out' group)

// With timeout
try {
    const response = await msgBus.request({
        channel: 'TEST.COMPUTE_SUM',
        payload: { a: 5, b: 15 },
        options: {
            timeout: 5000, // Overall timeout
        },
    });
} catch (error) {
    if (error instanceof TimeoutError) {
        console.error('Request timed out');
    }
}

// Fail immediately if no provider is registered
try {
    const response = await msgBus.request({
        channel: 'TEST.COMPUTE_SUM',
        payload: { a: 5, b: 15 },
        options: {
            throwIfNoProvider: true,
        },
    });
} catch (error) {
    if (error instanceof NoProviderError) {
        console.error('No provider registered for channel:', error.channel);
    }
}

// With separate send and response timeouts
const response = await msgBus.request({
    channel: 'TEST.COMPUTE_SUM',
    payload: { a: 5, b: 15 },
    options: {
        sendTimeout: 1000, // Timeout for sending the message
        responseTimeout: 5000, // Timeout for receiving the response
    },
});

// With headers for correlation
const response = await msgBus.request({
    channel: 'TEST.COMPUTE_SUM',
    payload: { a: 5, b: 15 },
    headers: {
        sourceId: 'component-123',
        correlationId: 'request-456',
    },
});

// The response will include matching headers
console.log(response.headers.requestId); // Original message ID
console.log(response.headers.correlationId); // Preserved correlation ID

Key Features

1. Input Type Overloading: Use different input groups within the same channel to support multiple request signatures while maintaining a single response type.

2. Timeout Control: Configure response timeout via the responseTimeout option to prevent indefinite waiting. Default is defaultPromiseTimeout (5 seconds), overridable globally via import { defaultPromiseTimeout } from '@actdim/msgmesh'.

3. Header Propagation: Headers like correlationId are automatically propagated from request to response for tracing.

4. Cancellation: Cancel in-flight requests with AbortSignal (see below).

5. No-Provider Guard: Use throwIfNoProvider: true in options or mandatoryProvider: true in channel config to get an immediate NoProviderError instead of waiting for the timeout when no provider is registered.

Cancellation

Request cancellation is cooperative: when request() is aborted, the provider callback receives a cancel message (headers.status === 'canceled') so it can stop in-flight work and clean up resources.

Cancel an in-flight request by passing an AbortSignal via options.abortSignal. When aborted, the bus sends a cancel message to the provider and rejects the returned Promise with an OperationCanceledError.

See provide() -> Cancellation Handling for provider-side handling details.

const abortController = new AbortController();

const responsePromise = msgBus.request({
    channel: 'API.FETCH_DATA',
    payload: { url: 'https://api.example.com/data' },
    options: {
        abortSignal: abortController.signal,
    },
});

// Cancel the request (sends cancel message to provider)
abortController.abort('user navigated away');

try {
    await responsePromise;
} catch (error) {
    // OperationCanceledError: The request was canceled by the caller
    console.error(error.message);
}

Fan-in Streaming Pattern: requestStream()

Send a request and consume all responses as an async iterable. Unlike request() which takes only the first response, requestStream() collects responses from all registered providers on the channel — enabling Fan-in patterns where multiple workers process the same request and report back.

// Multiple providers registered on the same channel
msgBus.provide({
    channel: 'DATA.FETCH',
    callback: async (msg) => fetchFromSourceA(msg.payload),
});
msgBus.provide({
    channel: 'DATA.FETCH',
    callback: async (msg) => fetchFromSourceB(msg.payload),
});

// Collect all responses
const results: DataResult[] = [];

for await (const msg of msgBus.requestStream({
    channel: 'DATA.FETCH',
    payload: { query: 'search term' },
    options: { fetchCount: 2 }, // stop after 2 responses
})) {
    results.push(msg.payload);
}

Options

requestStream() supports the same options as stream() plus throwIfNoProvider:

• fetchCount — stop after N responses (otherwise streams until timeout or abort)
• timeout — inactivity timeout in ms (resets on each received response)
• abortSignal — abort the stream externally
• throwIfNoProvider — throw NoProviderError immediately if no provider is registered (also triggered by mandatoryProvider: true in channel config)
• throttle / debounce — applied to the out subscription

// With timeout and fetchCount
for await (const msg of msgBus.requestStream({
    channel: 'SEARCH.QUERY',
    payload: { term: 'typescript' },
    options: {
        fetchCount: 10,    // at most 10 results
        timeout: 3000,     // stop if no new result for 3s
    },
})) {
    displayResult(msg.payload);
}

// With abort
const abortController = new AbortController();

const streamTask = (async () => {
    for await (const msg of msgBus.requestStream({
        channel: 'DATA.FETCH',
        payload: { query: 'all' },
        options: { abortSignal: abortController.signal },
    })) {
        if (isEnough(msg.payload)) {
            abortController.abort(); // stops iteration cleanly
            break;
        }
    }
})();

Key Differences from request()

| | request() | requestStream() | |--|-------------|-------------------| | Return | Promise<Msg> | AsyncIterableIterator<Msg> | | Responses | First one wins | All providers respond | | Completion | On first response | fetchCount / timeout / abort | | Use case | Single provider RPC | Fan-in / multi-provider aggregation |

Error Handling

If a provider responds with status: 'error', the generator throws Error. If status: 'canceled', it throws OperationCanceledError. Both stop the iteration.

try {
    for await (const msg of msgBus.requestStream({
        channel: 'DATA.FETCH',
        payload: { query: 'test' },
        options: { throwIfNoProvider: true },
    })) {
        process(msg.payload);
    }
} catch (err) {
    if (err instanceof NoProviderError) console.error('No provider:', err.channel);
    if (err instanceof TimeoutError) console.error('Timed out');
    if (err instanceof OperationCanceledError) console.error('Canceled');
}

Advanced Features

Message Replay

Configure channels to buffer and replay messages for late subscribers.

const msgBus = createMsgBus<MyBusStruct>({
    'TEST.EVENTS': {
        replayBufferSize: 50, // Keep last 50 messages
        replayWindowTime: 60000, // Keep messages for 60 seconds
    },
});

// Send messages
for (let i = 0; i < 100; i++) {
    await msgBus.send({
        channel: 'TEST.EVENTS',
        payload: `Message ${i}`,
    });
}

// Late subscriber receives last 50 messages
msgBus.on({
    channel: 'TEST.EVENTS',
    callback: (msg) => {
        console.log('Replayed:', msg.payload);
    },
});

Throttling and Debouncing

Control message processing rate at both channel and subscription levels.

// Channel-level throttling
const msgBus = createMsgBus<MyBusStruct>({
    'TEST.UPDATES': {
        throttle: {
            duration: 1000,
            leading: true,
            trailing: true,
        },
    },
});

// Subscription-level debouncing
msgBus.on({
    channel: 'TEST.UPDATES',
    callback: (msg) => {
        updateUI(msg.payload);
    },
    options: {
        debounce: 500, // Wait 500ms of inactivity before processing
    },
});

Chain of Responsibility

Multiple providers can register on the same channel. By default, all of them process every message independently. For a chain-of-responsibility pattern — where only one provider should handle each message — set msgOut.status = 'skipped' to pass the message to the next provider:

// Provider A: handles only even sums
msgBus.provide({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg, msgOut) => {
        const sum = msg.payload.a + msg.payload.b;
        if (sum % 2 !== 0) {
            msgOut.status = 'skipped'; // pass to next provider
            return;
        }
        return sum * 10;
    },
});

// Provider B: handles only odd sums
msgBus.provide({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg, msgOut) => {
        const sum = msg.payload.a + msg.payload.b;
        if (sum % 2 === 0) {
            msgOut.status = 'skipped'; // pass to next provider
            return;
        }
        return sum * 100;
    },
});

Each subscriber receives an independent copy of the message envelope (via structuredClone on the envelope, payload is shared by reference), so setting msgOut.status in one provider does not affect others.

Error Handling

The bus includes built-in error handling and a reserved error channel.

// Subscribe to errors for a specific channel
msgBus.on({
    channel: 'TEST.COMPUTE_SUM',
    group: 'error',
    callback: (msg) => {
        console.error('Error in ComputeSum:', msg.payload.error);
    },
});

// Subscribe to all system errors
msgBus.on({
    channel: 'MSGBUS.ERROR',
    callback: (msg) => {
        console.error('System error:', msg.payload);
    },
});

// Subscribe to NoProviderError (thrown synchronously, not routed to error channel)
// — use try/catch or .catch() on the request() Promise

// Errors in providers are automatically caught and routed
msgBus.provide({
    channel: 'TEST.COMPUTE_SUM',
    callback: (msg) => {
        if (msg.payload.a < 0) {
            throw new Error('Negative numbers not allowed');
        }
        return msg.payload.a + msg.payload.b;
    },
});

Headers and Metadata

Messages support rich metadata through headers.

import { MsgHeaders } from '@actdim/msgmesh';

// Standard headers
type StandardHeaders = {
    sourceId?: string; // Sender identifier
    targetId?: string; // Recipient identifier
    correlationId?: string; // Activity/trace identifier
    traceId?: string; // Distributed trace identifier
    requestId?: string; // Logical request identifier (generated by request/dispatch)
    inResponseToId?: string; // Reply reference (links response to requestId)
    publishedAt?: number; // Timestamp (Unix epoch, ms)
    priority?: number; // Message priority
    ttl?: number; // Time to live (ms)
    tags?: string | string[]; // Message tags
};

// Custom headers
type MyHeaders = MsgHeaders & {
    userId: string;
    tenantId: string;
    version: string;
};

const msgBus = createMsgBus<MyBusStruct, MyHeaders>();

await msgBus.send({
    channel: 'TEST.COMPUTE_SUM',
    payload: { a: 10, b: 20 },
    headers: {
        userId: 'user-123',
        tenantId: 'tenant-456',
        version: '1.0',
        correlationId: 'trace-789',
        priority: 1,
    },
});

Service Adapters

Automatically register any service object (e.g. a Swagger-generated API client) as a message bus provider. The adapter system uses TypeScript's type system to map service methods to bus channels at compile time — channel names, payload types, and return types are all derived from the service class. No manual wiring, no runtime errors.

Complete Example

import {
    ToMsgChannelPrefix, ToMsgStruct, BaseServiceSuffix,
    registerAdapters, getMsgChannelSelector, MsgProviderAdapter
} from '@actdim/msgmesh/adapters';
import { createMsgBus } from '@actdim/msgmesh/core';

// 1. Your service class (e.g. a Swagger-generated API client)
class OrderApiClient {
    static readonly name = 'OrderApiClient' as const;
    readonly name = 'OrderApiClient' as const;

    createOrder(items: Item[], priority: number): Promise<OrderResult> { /* ... */ }
    getOrder(id: string): Promise<Order> { /* ... */ }
    formatResponse() { /* internal — will be skipped */ }
}

// 2. Derive channel prefix from class name:
//    "OrderApiClient" → strip "Client", strip "Api" → "Order" → uppercase → "API.ORDER."
type OrderPrefix = ToMsgChannelPrefix<typeof OrderApiClient.name, 'API', BaseServiceSuffix>;

// 3. Map service methods to bus struct (skip internal helpers):
//    createOrder → "API.ORDER.CREATEORDER": { in: [Item[], number]; out: OrderResult }
//    getOrder    → "API.ORDER.GETORDER":    { in: [string];         out: Order }
type OrderApiStruct = ToMsgStruct<OrderApiClient, OrderPrefix, 'formatResponse'>;

// 4. Create bus and register the service as a provider
const services: Record<OrderPrefix, OrderApiClient> = {
    'API.ORDER.': new OrderApiClient(),
};

const msgBus = createMsgBus<OrderApiStruct>();
const abortController = new AbortController();

registerAdapters(
    msgBus,
    Object.entries(services).map(([_, service]) => ({
        service,
        channelSelector: getMsgChannelSelector(services),
    }) as MsgProviderAdapter),
    abortController.signal,
);

// 5. Call via the bus — fully type-safe, payloadFn matches the original method signature
const response = await msgBus.request({
    channel: 'API.ORDER.CREATEORDER',
    payloadFn: fn => fn([{ id: '1', qty: 2 }], 1),
});
console.log(response.payload); // OrderResult

// Clean up providers when done
abortController.abort();

How It Works

Given a service class:

class OrderApiClient {
    static readonly name = 'OrderApiClient' as const;
    readonly name = 'OrderApiClient' as const;

    createOrder(items: Item[], priority: number): Promise<OrderResult> { /* ... */ }
    getOrder(id: string): Promise<Order> { /* ... */ }

    // Internal helper — should not be exposed on the bus
    formatResponse() { /* ... */ }
}

The type utilities transform it into a bus structure:

import {
    ToMsgChannelPrefix,
    ToMsgStruct,
    BaseServiceSuffix,
    registerAdapters,
    getMsgChannelSelector,
    MsgProviderAdapter
} from '@actdim/msgmesh/adapters';

// 1. Generate channel prefix from class name
//    "OrderApiClient" → remove suffix "Client" → uppercase → "API.ORDER."
type ApiPrefix = 'API';
type OrderChannelPrefix = ToMsgChannelPrefix<
    typeof OrderApiClient.name, // "OrderApiClient"
    ApiPrefix,                  // "API"
    BaseServiceSuffix           // removes CLIENT, API, SERVICE, etc.
>;
// Result: "API.ORDER."

// 2. Transform service methods into bus struct (skip internal methods)
type OrderApiStruct = ToMsgStruct<
    OrderApiClient,
    OrderChannelPrefix,
    'formatResponse' // skip this method
>;
// Result type (compile-time):
// {
//     "API.ORDER.CREATEORDER": {
//         in: [items: Item[], priority: number]; // ← tuple from Parameters<>
//         out: OrderResult;                      // ← from ReturnType<>
//     };
//     "API.ORDER.GETORDER": {
//         in: [id: string];
//         out: Order;
//     };
// }

All channel names, payload types, and return types are verified at compile time. If you rename a method, add a parameter, or change a return type — the compiler catches it immediately.

Registering Adapters

const services: Record<OrderChannelPrefix, any> = {
    'API.ORDER.': new OrderApiClient(),
};

const adapters = Object.entries(services).map(([_, service]) => ({
    service,
    channelSelector: getMsgChannelSelector(services),
}) as MsgProviderAdapter);

const msgBus = createMsgBus<OrderApiStruct>();
const abortController = new AbortController();

// Register all methods as providers
registerAdapters(msgBus, adapters, abortController.signal);

// Clean up when done
abortController.abort();

registerAdapters() iterates over each method of the service prototype, resolves the channel name via channelSelector, and calls msgBus.provide() for each one. The provider callback spreads msg.payload (a tuple) as arguments to the original method: service[methodName](...msg.payload).

Calling Adapted Methods

Since method parameters are mapped to tuple types in the bus struct, use payloadFn for a natural function-call syntax:

// Type-safe call — fn signature matches createOrder(items, priority)
const response = await msgBus.request({
    channel: 'API.ORDER.CREATEORDER',
    payloadFn: fn => fn([{ id: '1', qty: 2 }], 1),
});

console.log(response.payload); // OrderResult

// Also works with payload directly (tuple)
const response2 = await msgBus.request({
    channel: 'API.ORDER.GETORDER',
    payload: ['order-123'],
});

Type Transformation Chain

Example: OrderApiClient with base prefix "API" and Skip = "formatResponse".

ToMsgChannelPrefix builds the channel prefix from the class name:

| Step | Value | |---|---| | Class name | OrderApiClient | | Strip known suffixes (Client, Api, …) | Order | | Uppercase | ORDER | | Add base prefix + dots | "API.ORDER." |

ToMsgStruct maps each method to a bus channel under that prefix:

| Method | Channel | in | out | |---|---|---|---| | createOrder(a, b) | "API.ORDER.CREATEORDER" | Parameters<createOrder> | Awaited<ReturnType<createOrder>> | | getOrder(id) | "API.ORDER.GETORDER" | Parameters<getOrder> | Awaited<ReturnType<getOrder>> | | formatResponse() | — | — | (skipped via Skip) |

Supported Service Suffixes

The following suffixes are automatically removed from class names: CLIENT, API, SERVICE, FETCHER, CONTROLLER, LOADER, REPOSITORY, PROVIDER.

Comparison with Other Solutions

| Feature | @actdim/msgmesh | Event Emitters | RxJS | | ---------------- | --------------- | -------------- | ----------- | | Type Safety | ✅ Full | ⚠️ Limited | ✅ Full | | Learning Curve | Low | Low | High | | Async Support | ✅ Native | ⚠️ Limited | ✅ Full | | Request-Response | ✅ Built-in | ❌ Manual | ⚠️ Complex | | Boilerplate | Minimal | Minimal | Medium | | Paradigm Shift | None | None | Significant | | Scalability | ✅ Excellent | ⚠️ Limited | ✅ Good |

Conclusion

@actdim/msgmesh provides a powerful, type-safe, and flexible message bus solution for TypeScript applications. It combines the simplicity of event emitters with the power of message-oriented middleware, while maintaining excellent type safety and developer experience.

Key benefits:

• Type Safety: Full TypeScript support with compile-time checks
• Flexibility: Works at any scale - from single components to entire applications
• Observability: Built-in support for logging, tracing, and debugging
• Developer Experience: Minimal boilerplate, clear API, excellent IntelliSense support
• Performance: Single-instance architecture with efficient message routing
• Integration: Works seamlessly with React, async operations, and existing patterns

The message bus serves as a solid foundation for the @actdim/dynstruct architectural framework, enabling the development of scalable, maintainable, and testable applications.

Changelog

v1.5 — Default channel config & provider context

• "*" key in MsgBusConfig sets default config for all channels; accepts a static object or a function (channel) => MsgChannelConfig for dynamic per-channel defaults. "*" is reserved and cannot be used as a channel name.
• provide() callback now receives msgOut: Msg<..., "out"> as second parameter — the pre-initialized outgoing message. Providers write msgOut.status = 'skipped' (chain of responsibility) or msgOut.status = 'canceled' (provider-side cancellation) instead of mutating the incoming message.
• msg.status moved to top-level Msg (was in headers). Renamed Outcome → MsgStatus: 'pending' | 'handled' | 'failed' | 'canceled' | 'skipped' | 'timeout'.
• Message envelope cloned per subscriber via structuredClone — mutations in one subscriber do not affect others. Payload is shared by reference.
• Chain of responsibility: multiple providers on the same channel; skip by setting msgOut.status = 'skipped'.

v1.4 — requestStream() & error propagation

• Added requestStream(): fan-in pattern — one request, all registered providers respond, consumed as an async iterable.
• Provider errors propagate immediately to request() instead of waiting for timeout.
• Default request timeout reduced to 5 seconds (defaultPromiseTimeout, globally overridable).

v1.3 — Service adapters

• Added adapters.ts: registerAdapters, ToMsgStruct, ToMsgChannelPrefix, getMsgChannelSelector — auto-wire any service class as bus providers with compile-time type safety.
• payloadFn parameter on send() / request() for tuple-typed payloads.

v1.2 — Cancellation & NoProviderError

• Cooperative request cancellation via AbortSignal — cancel message delivered to provider with msg.status === 'canceled'.
• NoProviderError on throwIfNoProvider: true or mandatoryProvider: true in channel config.
• MsgStruct<> factory enforces explicit in/out declarations; adds implicit error group to every channel.

v1.0 — Headers, errors & type system

• MsgHeaders: requestId, inResponseToId, correlationId, traceId, sourceId, priority, ttl, tags, timestamp.
• Custom error types: TimeoutError, AbortError, OperationCanceledError with type guards.
• Per-channel config: replayBufferSize, replayWindowTime, throttle, debounce, delay.

v0.9 — Initial release

• Core pub/sub: send(), on(), once(), stream(), provide(), request().
• Channel → group → topic addressing with regex topic matching.
• Replay buffer via ReplaySubject.

TODO

• rate limiting (for single channel, using signal after auto-'ack') and backpressure (for "in" and "out" channel pair), real send promise

Further Reading

• GitHub Repository
• @actdim/dynstruct Documentation
• Type Safety Best Practices
• Message-Oriented Middleware Patterns