// @ozaco/server

Transport

Remote transports that route broker dispatches across a message bus or worker threads.

import from@ozaco/server/transport/nats

A transport is the dispatch leaf of the broker pipeline: it decides where a resolved call actually runs. The bundled InternalTransport (in @ozaco/server/core) runs actions in-process; installing one of these transports routes the same calls elsewhere — across a NATS message bus, or into Web Workers / worker threads — with no change to the actions themselves.

Each is a Transport implementation, so they compose: registered by priority with a next(failure) fall-through predicate, several transports can coexist (e.g. a remote transport in front of the internal one). All three broker verbs — dispatch (request/reply, including streaming request and response bodies), emit (group-targeted), and broadcast (fan-out) — are carried over the wire, with values encoded through the installed Codec (or, for workers, the structured-clone algorithm).

Streaming is first-class in both: input streams are pumped to the far side over dedicated subjects/channels, and a streamed action result is subscribed back and delivered chunk-by-chunk with backpressure. Cancellation propagates — halting a dispatch tears down the remote work.

// additional entry points

@ozaco/server/transport/nats`NatsTransport` — a NATS-backed transport. Request/reply dispatch over subjects, group `emit` and `broadcast`, and streaming over per-call input/output subjects. Requires the `nats` client. Default priority `10`.
@ozaco/server/transport/worker`WorkerTransport` — a Web Worker / worker-thread transport. Dispatches to one or more worker endpoints (round-robin per service), with `structured`-clone or `codec` wire encoding. Default priority `10`.

// on this page

NATS transport Worker transport

NATS transport

From `@ozaco/server/transport/nats`.

const

NatsTransport

A `Transport` implementation backed by a NATS connection.

signature

const NatsTransport: Plugin<Nats.Context, [options?: Nats.Options], TransportDef.Actions>

interface Nats.Options extends TransportDef.Options {
  servers?: string | string[]      // default 'nats://localhost:4222'
  subjectPrefix?: string           // default 'ozaco'
  queueGroup?: string
  requestTimeoutMs?: number        // default 30_000; 0 or negative DISABLES the transport timeout
}

On setup it connects (with reconnect), subscribes to the emit/broadcast wildcards, watches the broker, and registers itself with the Transport protocol (default priority 10); teardown drains subscriptions and closes the connection. dispatch uses NATS core request/reply on a per-action subject: the params (and any input streams) are encoded through the installed Codec; a plain reply returns its value, a __stream__ reply opens an output subscription streamed back to the caller, and a __failure__ reply is re-raised. requestTimeoutMs <= 0 maps to the max timer (~24.8 days) so the action TimeoutPolicy stays the sole timeout authority while a responder fault still cannot hang forever.

example

import { NatsTransport } from '@ozaco/server/transport/nats'
import { install } from '@ozaco/std/plugin'

// Install alongside the broker; it registers itself as a transport.
yield* install(NatsTransport, {
  servers: 'nats://localhost:4222',
  subjectPrefix: 'ozaco',
  queueGroup: 'workers',
})

// Calls now route over NATS; the API is unchanged.
const user = yield* Broker.actions.call(Users.actions.get, ['42'])

const

NatsErrors

The `server:nats-transport.*` error tags NATS faults are mapped to.

signature

const NatsErrors: {
  NoResponders: 'server:nats-transport.no-responders'
  Timeout: 'server:nats-transport.timeout'
  ConnectionClosed: 'server:nats-transport.connection-closed'
  ConnectionRefused: 'server:nats-transport.connection-refused'
  AuthorizationViolation: 'server:nats-transport.authorization-violation'
  MaxPayloadExceeded: 'server:nats-transport.max-payload-exceeded'
  BadSubject: 'server:nats-transport.bad-subject'
  // ...and more
}

Built with std:shared createTags. The internal error map normalizes NATS client errors into these tags so dispatch failures pattern-match consistently.

Nats

The NATS transport types: options, context, and the request/reply wire shapes.

signature

namespace Nats {
  interface Options extends TransportDef.Options { servers?; subjectPrefix?; queueGroup?; requestTimeoutMs? }
  interface Context extends TransportDef.Context {
    connection: NatsConnection
    prefix: string
    queueGroup?: string
    requestTimeoutMs: number
    subscriptions: Map<string, NatsSubscription>
    scope: Scope
  }
  interface DispatchPayload { cid; serviceName; actionKey; params?; inputSubjects?; outputSubject?; contexts? }
  type Wire = WireSuccess | WireFailure | WireStream   // tagged by _t
}

Worker transport

From `@ozaco/server/transport/worker`.

const

WorkerTransport

A `Transport` implementation that dispatches across Web Workers / worker threads.

signature

const WorkerTransport: Plugin<WorkerDef.Context, [options?: WorkerDef.Options], TransportDef.Actions>

interface WorkerDef.Options extends TransportDef.Options {
  // spawn worker(s) from a script, or adopt existing port(s)
  script?: SpawnSpec | SpawnSpec[]   // string | URL | { script; count? }
  count?: number
  endpoint?: PortLike | PortLike[]
  wire?: 'structured' | 'codec'      // default 'structured'
}

On setup it creates its endpoints (spawning worker scripts and/or adopting supplied ports), starts a reader per endpoint, registers with the Transport protocol (default priority 10), and posts a ready handshake advertising the local services; teardown closes every endpoint. dispatch picks an endpoint hosting the target service (round-robin per service name), encodes params via the chosen wire (structured clone or the installed codec), and awaits the reply — a __stream__ reply is consumed as an inbound stream, a __failure__ reply is re-raised. Input streams are pumped to the worker on dedicated stream ids; halting a dispatch posts a cancel.

example

import { WorkerTransport } from '@ozaco/server/transport/worker'
import { install } from '@ozaco/std/plugin'

// Spawn a pool of workers, each running the given entry script.
yield* install(WorkerTransport, {
  script: { script: new URL('./worker.ts', import.meta.url), count: 4 },
  wire: 'structured',
})

// Or adopt an existing MessagePort / Worker as the endpoint.
yield* install(WorkerTransport, { endpoint: self as unknown as PortLike })

WorkerDef

The worker transport types: options, endpoints, context, envelopes, and wire shapes.

signature

namespace WorkerDef {
  type WireMode = 'structured' | 'codec'
  type SpawnSpec = string | URL | { script: string | URL; count?: number }

  interface PortLike { postMessage(msg, transfer?); addEventListener?; on?; start?; terminate?; close? }
  interface Options extends TransportDef.Options { script?; count?; endpoint?; wire? }
  interface Endpoint { wire: WireMode; services: Set<string>; post(msg): boolean; recv: Stream<unknown, void>; markReady(); close() }
  interface Context extends TransportDef.Context { wire; adoptWire; endpoints; pending; streams; handlers; scope; rr }

  type Wire = WireSuccess | WireFailure | WireStream           // tagged by _t
  type Envelope =                                              // the message protocol
    | DispatchEnvelope
    | { kind: 'ready'; wire: WireMode; services?: string[] }
    | { kind: 'reply'; cid: string; wire: Wire }
    | { kind: 'cancel'; cid: string }
    | { kind: 'emit'; req: unknown } | { kind: 'broadcast'; req: unknown }
    | { kind: 'chunk'; sid: string; data: unknown }
    | { kind: 'end'; sid: string } | { kind: 'error'; sid: string; failure: StreamErrorPayload }
}

The Envelope union is the full message protocol spoken over a port: dispatch/reply/cancel for request-reply, emit/broadcast for events, and chunk/end/error for streaming bodies. wire: structured posts values directly (structured clone); codec encodes through the installed codec for ports that cannot clone.