In-memory transport

InMemoryTransport is the SDK's in-process implementation of ITransportAdapter. It uses an InMemoryBroker (a Map<topic, Set<handler>>) for fan-out. No NATS process. No serialization. No network. Two InMemoryTransport instances connected to the same broker can talk to each other as if they were on a real bus.

The implementation is small enough to read cover-to-cover:

• projects/matrix-3/packages/core/src/transport/InMemoryBroker.ts — 38 lines.
• projects/matrix-3/packages/core/src/transport/InMemoryTransport.ts — 190 lines.

When to use it

• Unit tests. No setup, no teardown, no port allocation. Tests run in milliseconds.
• Examples and demos. Code that needs to compile and run without a Host Service.
• MatrixRuntime defaults. If you call new MatrixRuntime() with no transport, the runtime constructs new InMemoryTransport(new InMemoryBroker()) for you (MatrixRuntime.ts:146-152).
• Headless DOM tests. Run a browser shell against an in-process actor system.

Do not use it for:

• Anything that needs to cross processes (use NatsTransport).
• Anything that needs to persist messages across restarts (InMemoryBroker is volatile).
• Production runtimes (the bundled NATS sibling is what production runs on).

Constructor

new InMemoryTransport(broker: InMemoryBroker, options?: InMemoryTransportOptions)

Where InMemoryTransportOptions (InMemoryTransport.ts:7-15) is:

interface InMemoryTransportOptions {
  queueMessages?: boolean;     // queue messages published before any subscriber exists
  latencyMs?: number;          // simulated network latency
  name?: string;               // used in error messages
  root?: string;               // for parity with NatsTransport.root
  realm?: string;              // deprecated alias for root
  enableTimingLogs?: boolean;  // log [TRANSPORT-PUB] timings to console
}

Defaults: queueMessages: false, latencyMs: 0, name: 'unnamed', enableTimingLogs: false.

API surface

InMemoryTransport implements ITransportAdapter:

get topicPrefix(): string         // always '' for in-memory
get root(): string                // options.root ?? options.realm ?? ''

subscribe(topic, handler): () => void
publish(topic, payload?): void

getWireTopicInfo(topic): IWireTopicInfo
disconnect(): Promise<void>
getUnderlyingClient(): unknown    // returns the broker

getWireTopicInfo returns { appTopic: topic, wireTopic: topic, prefix: '', root: undefined, realm: undefined, sessionId: undefined } — for in-memory, application-level topics ARE the wire topics.

Delivery semantics

// InMemoryTransport.ts:160-189
publish(topic, payload?) {
  if (queueMessages && noSubscribers) → queue;
  if (latencyMs > 0) → setTimeout, then publish;
  else → broker.publish(topic, payload) synchronously;
}

subscribe(topic, handler) {
  // first subscriber for the topic → register with broker
  // delivery: synchronous if latencyMs === 0; setTimeout otherwise
}

Synchronous delivery (when latencyMs === 0) is the default. This makes tests deterministic — transport.publish('foo', x) and a subsequent assertion on a subscriber both run in the same microtask.

Caution: Synchronous delivery is not what NATS does. If your code happens to work in tests because of synchronous semantics but fails over real NATS, you have an ordering bug. Asynchronous expectations (await new Promise(r => setTimeout(r, 0))) are good practice even with InMemoryTransport.

Queueing messages before subscription

When queueMessages: true, publishes to a topic with no subscribers are queued (InMemoryTransport.ts:160-172). The first subscriber receives the queued payloads in order. Useful for tests where the publisher races the subscriber.

const broker = new InMemoryBroker();
const transport = new InMemoryTransport(broker, { queueMessages: true });

transport.publish('hello', { msg: 'first' });
transport.publish('hello', { msg: 'second' });

transport.subscribe('hello', (payload) => console.log(payload));
// → { msg: 'first' }
// → { msg: 'second' }

By default (queueMessages: false), publishes to topics with no subscribers are dropped, exactly like NATS pub/sub.

Simulated latency

new InMemoryTransport(broker, { latencyMs: 50 });

Every publish is wrapped in a setTimeout(deliver, latencyMs). Useful for testing time-sensitive flows (timeouts, retries) without actually waiting for network round-trips.

Error isolation

When a subscriber's handler throws or returns a rejected promise, the transport catches it and logs [InMemoryTransport:<name>] Async handler error for <topic>: (InMemoryTransport.ts:99-110). Other subscribers continue to receive the payload. This mirrors the NATS adapter's isolation contract.

Connecting two transports through one broker

The whole point of having a separate InMemoryBroker is so multiple InMemoryTransport instances can share it:

const broker = new InMemoryBroker();
const t1 = new InMemoryTransport(broker, { name: 'producer' });
const t2 = new InMemoryTransport(broker, { name: 'consumer' });

t2.subscribe('foo', (p) => console.log('consumer got', p));
t1.publish('foo', { x: 1 });
// → consumer got { x: 1 }

This is the pattern integration tests use to model multiple actor systems talking to each other in one Node process.

Cleanup

transport.disconnect() clears all subscriptions and the queued-messages map. The underlying broker is not affected — you can reuse it with a fresh transport.

await transport.disconnect();

A test typically:

1. Creates a broker.
2. Creates one or more transports against it.
3. Constructs runtimes/actors against the transports.
4. Runs assertions.
5. Awaits runtime.shutdown() for each runtime (which calls transport.disconnect() for transports it owns).

Use in MatrixRuntime

The minimum viable example (already shown in Define an actor):

import {
  MatrixRuntime, InMemoryTransport, InMemoryBroker, MatrixActor,
} from '@open-matrix/core';

const broker = new InMemoryBroker();
const transport = new InMemoryTransport(broker, { name: 'demo' });
const runtime = new MatrixRuntime({ transport });

class Greeter extends MatrixActor {
  static accepts = { 'hello': { description: 'Say hi', schema: {}, returns: {} } };
  async onHello() { return { msg: 'hi' }; }
}

await runtime.create(Greeter, 'greeter');

See also

• NATS transport — the production-grade transport.
• Transport testing — writing transport-level fakes.
• Actor testing — testing actors over InMemoryTransport.

Source: projects/matrix-3/packages/core/src/transport/InMemoryBroker.ts (38 lines, full broker), projects/matrix-3/packages/core/src/transport/InMemoryTransport.ts:1-190 (full transport).