@hile/message-modem

Implement request/response, push, abort, and stream protocol behavior independent of transport.

@hile/message-modem

Implement request/response, push, abort, and stream protocol behavior independent of transport.

Choose This Package When

Use @hile/message-modem for: Implement request/response, push, abort, and stream protocol behavior independent of transport.

Use When

Use the message packages for request/response messaging over WebSocket, process IPC, worker threads, file-system message handlers, service discovery, streaming RPC, and registry-backed pub/sub.

Do Not Use When

• Do not use stream() for normal single-result calls.
• Do not enable retries for a streamed request input. Input streams are consumed once and cannot be replayed safely.
• Do not rely on message IDs for business idempotency. They are transport IDs.
• Use defineMicroMessage() for Micro business handlers; reserve generic defineMessage() for transport-neutral loaders.
• Do not pass zero, fractional, non-finite, or oversized message timeouts. Explicit timeout values must be safe integers from 1 through 2_147_483_647 milliseconds.

Install

pnpm add @hile/context @hile/micro @hile/message-loader @hile/message-ws

Use transport-specific packages only when you need to build custom IPC or worker-thread bridges.

Imports

import { defineMessage, MessageLoader } from '@hile/message-loader'
import { createExecutionContext } from '@hile/context'
import { Application, defineMicroMessage, Registry, Server } from '@hile/micro'
import { MessageWs } from '@hile/message-ws'
import { MessageIpc } from '@hile/message-ipc'
import { MessageWorkerThread } from '@hile/message-worker-thread'

Copy-Paste Example

Message handler file:

// src/messages/ping.msg.ts
import { defineMicroMessage } from '@hile/micro'

export default defineMicroMessage(async ({ data, params, invocation }) => {
  return {
    type: 'pong',
    data,
    params,
    requestId: invocation.context.values.requestId,
    timestamp: Date.now(),
  }
})

Microservice boot file:

// src/services/app.boot.ts
import { defineService } from '@hile/core'
import { Application } from '@hile/micro'

export default defineService('micro.app', async (shutdown) => {
  const app = new Application({
    namespace: process.env.MICRO_NAMESPACE ?? 'example.service',
    registry: {
      host: process.env.REGISTRY_HOST ?? '127.0.0.1',
      port: Number(process.env.REGISTRY_PORT ?? 9876),
    },
    advertiseHost: process.env.HILE_ADVERTISE_HOST ?? '127.0.0.1',
  })

  await app.load(new URL('../messages', import.meta.url).pathname)
  const stop = await app.listen(Number(process.env.MICRO_PORT ?? 9877))
  shutdown(stop)
  return app
})

Caller:

import { randomUUID } from 'node:crypto'
import { createExecutionContext } from '@hile/context'

const context = createExecutionContext({ requestId: randomUUID() })
const result = await app.call('example.service', '/ping', { hello: 'world' }, { context })

More Examples

Typed fixed-path operations

Use defineMicroContract() and createMicroClient() from @hile/micro-contract to share stable JSON schemas and typed remote calls. Implement each operation in its matching message file with defineMicroMessage(operation, handler), then call loadMicroContract(app, contract, messagesDirectory) from @hile/micro to obtain { local, activate, close }.

The binding validates filesystem paths and complete implementation coverage before activation. Explicit local calls are same-service unary calls through the provider executor, not a namespace-based optimization; normal calls still use RPC. Typed remote calls default to zero retries. Native dynamic and streaming APIs remain supported. See Typed Micro contracts for the full example, schema constraints, and shutdown ordering.

Generic loader protocol registration

import { defineMessage, MessageLoader } from '@hile/message-loader'

const loader = new MessageLoader({})
const definition = defineMessage(({ data }) => data, { protocol: 'example.protocol' })
const release = loader.register('/echo', definition.fn, { protocol: definition.protocol })
const value = await loader.dispatch('/echo', { hello: 'world' }, {}, { protocol: 'example.protocol' })
release()

File loading preserves definition metadata automatically. Raw registration must pass the metadata explicitly. Ordinary definitions/dispatch omit protocol on both sides. Protocol tags are nonempty printable ASCII strings, at most 128 characters; registration snapshots the value. Neither payload fields nor Context/extras can select the dispatch protocol.

After one route match, a different protocol throws MessageProtocolMismatchError (HILE_MESSAGE_PROTOCOL_MISMATCH) before the handler runs, with no fallback to another matching route. A tag prevents accidental protocol misdelivery, not unauthorized invocation by a caller able to set the tag.

Native streaming and custom transports

Streaming handler:

// src/messages/events.msg.ts
import { defineMicroMessage } from '@hile/micro'

export default defineMicroMessage(async function* ({ invocation }) {
  for (let i = 0; i < 3; i++) {
    yield { seq: i, requestId: invocation.context.values.requestId }
  }
})

Streaming caller:

const stream = await app.stream('example.service', '/events', {}, { context })
for await (const chunk of stream) {
  console.log(chunk)
}

Streaming request body with a normal response:

// src/messages/upload.msg.ts
import { defineMicroMessage } from '@hile/micro'

export default defineMicroMessage(async ({ data, input, invocation }) => {
  if (!input) throw new Error('upload body is required')

  let bytes = 0
  for await (const chunk of input) bytes += Buffer.byteLength(chunk)

  return {
    filename: data.filename,
    bytes,
    requestId: invocation.context.values.requestId,
  }
})

import { createReadStream } from 'node:fs'

const result = await app.call(
  'example.service',
  '/upload',
  { filename: 'archive.tar' },
  { context, input: createReadStream('/tmp/archive.tar') },
)

If no structured metadata is needed, pass an AsyncIterable, Uint8Array, or ArrayBuffer directly as data; Hile sends it as the request input stream and the handler receives data === undefined. app.stream() accepts the same request input forms when both request and response need to stream.

Custom WebSocket modem:

import { MessageWs } from '@hile/message-ws'
import type WebSocket from 'ws'

class RpcWs extends MessageWs {
  constructor(ws: WebSocket, private readonly dispatch: (url: string, data: unknown) => Promise<unknown>) {
    super(ws)
  }

  protected exec(data: { url: string; data: unknown }) {
    return this.dispatch(data.url, data.data)
  }

  request<T>(url: string, data: unknown, timeout = 30_000) {
    return this._send<T>({ url, data }, { timeout })
  }
}

Notice that request() returns a Promise<T>. Await it directly.

Runtime And Lifecycle Notes

• MessageLoader maps *.msg.* files to routes using @hile/loader.
• Message files use the same portable DSL as controllers: [id].msg.ts declares one segment and [...paths].msg.ts declares one or more final segments. A catch-all handler reads the slash-joined value from params.paths; Hile compiles it to rou3’s named catch-all syntax.
• Optional catch-all, mixed or non-terminal dynamic segments, multiple catch-alls, duplicate dynamic names, unsafe or object-meta names, and router-native syntax in message file names fail during loading. The configured prefix may use rou3-native or bracket-style parameters, is compiled separately from file segments, and cannot duplicate a file parameter name; manual register() calls continue to accept rou3-native paths.
• defineMessage(fn, options?) and MessageLoader.register(path, fn, options?) accept MessageProtocolOptions; MessageLoader.dispatch(path, data, extras?, options?) checks that protocol before invoking the matched handler.
• Raw registrations and file-loaded messages share route ownership. Duplicate normalized paths and equal-specificity overlaps fail with MessageRouteConflictError (HILE_MESSAGE_ROUTE_CONFLICT). This includes /a/:x versus /:x/b; regex parameters are conservatively treated as unrestricted, so disjoint regexes with the same route shape also conflict. Distinct static branches and different-specificity static/parameter/catch-all precedence remain supported by rou3.
• A MessageLoader.load() batch is invisible until it commits. Import, bind, or final commit failure rolls back only that batch. Unload is idempotent and owner-specific; an old release cannot remove a replacement. Concurrent loads or raw registration during a load fail with MessageLoadInProgressError (HILE_MESSAGE_LOAD_IN_PROGRESS); releasing existing owners remains allowed and cannot resurrect them at commit.
• protected bind(file, metadata) remains the extension point. An override that calls super.bind must return/compose its cleanup and unwind its own side effects on failure. The generic loader does not parse business payloads or inspect Context.
• MessageModem._send() returns a Promise.
• _send() and _stream() accept options.input as an AsyncIterable, Uint8Array, or ArrayBuffer. Passing one of those values directly as data automatically selects request streaming and leaves the handler’s structured data undefined.
• MessageModem._send() and _push() use a 30_000 ms timeout when none is provided. An explicit timeout must be a safe integer from 1 through 2_147_483_647; invalid values throw TypeError before a message is sent.
• MessageModem._stream() returns a Node Readable in object mode.
• Stream timeout and idleTimeout values use the same 1 through 2_147_483_647 ms range. Idle timeout is refreshed by valid input or output stream activity. The response stream window must be a safe integer from 1 through 64 and defaults to 1; request input uses receiver-issued credit with a window of 1.
• Each modem schedules request, total-stream, and idle-stream deadlines through one internal deadline scheduler. This reduces active Node.js timers without changing timeout, cancellation, ordering, or error semantics.
• Request and response chunks share STREAM_DATA frames and carry an explicit direction: 'input' | 'output'; each direction has independent sequence and credit state. STREAM_CANCEL stops one direction. A cancel carrying status fails the owning request immediately; ABORT cancels the whole invocation.
• This frame model replaces the former response-only stream, streamVersion, and streamWindow fields. All peers on one transport connection must use compatible @hile/message-modem and transport package versions; do not mix old and new peers during a rolling deployment.
• @hile/message-ws sends Uint8Array and ArrayBuffer request and response chunks as native binary frames rather than JSON/base64. Public decodeMessageFrame() payloads remain isolated from caller-owned input by default; the owned WebSocket RawData path uses a zero-copy binary view internally.
• @hile/message-ipc transparently Base64-wraps only binary STREAM_DATA payloads so request and response streams survive Node child-process IPC’s default JSON serialization. Other IPC frames keep their ordinary object representation.
• A stream request requires exec() to return an async iterable.
• MessageModem.exec(data, signal?, input?, execution?) has an optional fourth MessageExecutionOptions argument. Its responseStream flag comes from the decoded request frame, not payload metadata or the presence of an input stream. Existing three-argument transport subclasses remain compatible; custom subclasses that forward execution to another dispatcher must preserve this argument when the receiver needs to enforce unary-only operations.
• An execution whose response cannot be delivered (including a unary request returning an async iterable, or a stream failing during iteration) aborts its request scope before sending the original error. Cleanup is cooperative and does not delay the failure frame; provider shutdown still owns its bounded drain deadline.
• defineMicroMessage() handlers receive request streams as input: Readable | undefined, separately from structured data and invocation.
• Application.call(namespace, url, data, options) requires options.context and returns a promise. It accepts options.input for a streamed request with a normal response.
• Application.stream(namespace, url, data, options) requires options.context and returns a readable response stream; it can carry a request input stream at the same time.
• Application.call() and Application.stream() may target the application’s own namespace. These self RPCs use normal Registry-discovered WebSocket/modem transport, not an automatic local branch; replica selection and network behavior remain unchanged. The separate loadMicroContract() binding provides explicit local execution only for its typed fixed-path unary operations, through the same provider executor. See Typed Micro contracts for its distinct failure and lifecycle semantics.
• A peer address (host:port) is the routing and reuse identity, not an individual WebSocket identity. If both peers dial each other concurrently, including an application dialing itself, the server preserves both physical connections while they are active instead of replacing a connection that may carry an in-flight request. Later calls still reuse the cached peer connection.
• This connection bookkeeping does not add or change wire frames; request, response, stream, credit, cancel, and abort protocol fields remain unchanged.
• Application.call() and Application.stream() default retries to 0 when request input is streamed. Explicit nonzero retries fail before service discovery because streamed input is non-replayable.
• Native nonstream Application.call()/stream() retain the existing default of one transport retry. createMicroClient() and callHttpOverMicro() explicitly default to zero for both reads and writes. Opt-in retries remain subject to the underlying eligible-error policy and the caller’s end-to-end attempt/idempotency budget; provider contract/protocol rejection, HILE_MICRO_EXECUTION_FAILED, and unavailable admission do not trigger automatic retries. Safe typed execution errors must not replay a whole orchestration after a downstream validation failure; native unknown-error handling is unchanged.
• Micro adapters can pass options.protocol; it travels in existing request metadata.protocol and is checked by the generic message loader. There is no new modem frame or operation dispatcher. All selectable receivers must support this check before relying on protocol isolation: older receivers may ignore the marker.
• A caller-owned request input source failure is surfaced as MessageInputError, aborts the peer invocation, and is not counted against that peer’s circuit-breaker health.
• Application.publish(topic, payload) returns an object with update() and unpublish().
• Application.subscribe(topic, callback) returns an unsubscribe function.
• Registry stores service addresses and retained config/topic state under ~/.registry.

Anti-Patterns

• Appending a secondary response getter to client.request('/x', data)
• Returning a plain object from a handler called through stream().
• Retrying a consumed request stream or hiding it inside a replay-unsafe factory.
• Branching business code on whether a Micro target namespace is local or remote.
• Using pub/sub as a durable queue.
• Forgetting to register shutdown(await app.listen(...)).

Verification Checklist

• Micro message files default-export defineMicroMessage(...) and receive invocation.context.
• Native untyped Micro message files may use [...name].msg.ts; typed Micro Contract implementations remain fixed-path only and reject all parameterized file routes.
• RPC callers use await app.call(..., { context }).
• Same-namespace call() and stream() use normal Registry discovery, while streamPeer() keeps its exact-address selection; all three use the normal WebSocket/modem transport path and callers do not branch on locality.
• Streaming handlers are async generators.
• Request-stream handlers consume input and callers either pass options.input alongside metadata or pass a stream directly as data.
• Streamed request calls do not configure nonzero retries.
• Custom modem timeout values use the documented safe-integer range.
• File and raw owner conflicts, atomic batch failure, concurrent load rejection, and protocol misdelivery are tested before any handler side effect.
• Typed fixed-path services validate both actual source and compiled route trees; explicit local and real remote calls test the same provider executor, admission, cancellation, and bounded shutdown.
• Registry is started before application nodes need discovery.
• Micro apps use stable namespaces and advertise reachable hosts.

Related Recipes

Micro RPC With Message Loader

Complete Example

Provider handler:

// src/messages/charge.msg.ts
import { defineMicroMessage } from '@hile/micro'

export default defineMicroMessage(async ({ data, invocation }) => {
  return { charged: true, input: data, requestId: invocation.context.values.requestId }
})

Untyped dynamic handlers may use a required catch-all when the operation is intentionally not a typed Micro Contract:

// src/messages/files/[...paths].msg.ts
import { defineMicroMessage } from '@hile/micro'

export default defineMicroMessage(async ({ params, data, invocation }) => ({
  path: params.paths,
  input: data,
  requestId: invocation.context.values.requestId,
}))

Typed Micro Contract operation.path values stay fixed and cannot use [id] or [...paths].

Provider boot:

// src/services/app.boot.ts
import { defineService } from '@hile/core'
import { Application } from '@hile/micro'

export default defineService('billing.micro', async (shutdown) => {
  const app = new Application({
    namespace: 'billing',
    registry: { host: '127.0.0.1', port: 9876 },
    advertiseHost: '127.0.0.1',
  })

  await app.load(new URL('../messages', import.meta.url).pathname)
  const stop = await app.listen(9101)
  shutdown(stop)
  return app
})

Consumer:

import { randomUUID } from 'node:crypto'
import { createExecutionContext } from '@hile/context'

const context = createExecutionContext({ requestId: randomUUID(), tenantId: 't1' })
const result = await app.call('billing', '/charge', {
  tenantId: 't1',
  amount: 100,
}, { context })

Package-Local AI Guide

This package also ships AI.md in npm so agents can read accurate examples after installation.