Messaging

MessageModem, message loader, micro RPC, streaming, and runtime config.

Messaging

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.

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 })

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(...)).