@hile/micro

Run registry-backed service discovery, RPC, streaming RPC, and pub/sub between Node services.

@hile/micro

Run registry-backed service discovery, RPC, streaming RPC, and pub/sub between Node services.

Choose This Package When

User asks for	Use	Also read
Build service discovery or RPC	@hile/micro	packages/messaging-micro.md, recipes/micro-rpc-message-loader.md
Push runtime config without restarts	@hile/micro-dynamic-configs	packages/messaging-micro.md, recipes/runtime-config.md

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 rely on message IDs for business idempotency. They are transport IDs.
• Do not bypass defineMessage() for file-loaded handlers.

Install

pnpm add @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 { Application, 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 { defineMessage } from '@hile/message-loader'

export default defineMessage(async ({ data, params }) => {
  return {
    type: 'pong',
    data,
    params,
    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 ?? 0))
  shutdown(stop)
  return app
})

Caller:

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

More Examples

Streaming handler:

// src/messages/events.msg.ts
import { defineMessage } from '@hile/message-loader'

export default defineMessage(async function* () {
  for (let i = 0; i < 3; i++) {
    yield { seq: i }
  }
})

Streaming caller:

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

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.
• MessageLoader.dispatch(path, data, extras?) invokes the matched handler.
• MessageModem._send() returns a Promise.
• MessageModem._stream() returns a Node Readable in object mode.
• A stream request requires exec() to return an async iterable.
• Application.call(namespace, url, data, options?) returns a promise.
• Application.stream(namespace, url, data, options?) returns a readable stream.
• 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.

Registry Read APIs

Registry exposes read-only routes for diagnostics and admin tooling. These routes return snapshots of current registry state; they do not change service discovery, topic declarations, subscriptions, or retained config data.

• /-/namespaces returns all registered namespaces with peer counts and addresses.
• /-/namespace/peers accepts { namespace, exclude? } and returns every peer for one namespace.
• /-/registry/status returns registry uptime and counts for clients, namespaces, topics, and config namespaces.
• /-/topics accepts { prefix? } and returns topic summaries with publisher/subscriber counts and retained/data flags.
• /-/topic/get accepts { topic } and returns the current topic payload snapshot when the topic exists.
• /-/configs returns loaded config namespaces and keys.
• /-/config/get accepts { namespace, key? } and returns either a config snapshot or one config value.

Anti-Patterns

• Appending a secondary response getter to client.request('/x', data)
• Returning a plain object from a handler called through stream().
• Using pub/sub as a durable queue.
• Forgetting to register shutdown(await app.listen(...)).

Verification Checklist

• Message files default-export defineMessage(...).
• RPC callers use await app.call(...).
• Streaming handlers are async generators.
• 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 { defineMessage } from '@hile/message-loader'

export default defineMessage(async ({ data }) => {
  return { charged: true, input: data }
})

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:

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

Runtime Dynamic Config

Complete Example

import { z } from 'zod'
import { MicroDynamicConfigsServer } from '@hile/micro-dynamic-configs'

const schema = z.object({
  featureCheckout: z.boolean().default(false),
  maxRetries: z.number().int().min(1).max(10).default(3),
})

const configs = new MicroDynamicConfigsServer({
  app,
  redis,
  schema,
  redis_key: 'configs:checkout',
})

const cleanup = await configs.initialize()
shutdown(cleanup)

configs.on('change:featureCheckout', (next, previous) => {
  logger.info({ previous, next }, 'featureCheckout changed')
})

await configs.save({ featureCheckout: true })

Stable Runtime Reload

Complete Example

import { defineService, loadService } from '@hile/core'
import { Application } from '@hile/micro'
import { Http } from '@hile/http'
import { createConfigAggregator, createRuntimeReloader } from '@hile/reloader'
import appService from './micro.app.boot'
import httpService from './http.boot'

type RuntimeConfig = {
  mysql: { host: string; port: number }
  redis: { host: string; port: number }
  flags: Record<string, boolean>
}

export default defineService('runtime.reload', async (shutdown) => {
  const app = await loadService<Application>(appService)
  const http = await loadService<Http>(httpService)

  const reloader = createRuntimeReloader<RuntimeConfig, AppRuntime>({
    debounceMs: 100,
    normalize: config => ({
      mysql: config.mysql,
      redis: config.redis,
      flags: config.flags,
    }),
    create: config => createAppRuntime(config),
    dispose: runtime => runtime.close(),
    onError: (error, context) => {
      logger.error({ error, stage: context.stage }, 'runtime reload failed')
    },
  })

  const configs = createConfigAggregator<RuntimeConfig>({
    required: ['mysql', 'redis'],
    defaults: { flags: {} },
    debounceMs: 100,
    onError: (error) => {
      logger.error({ error }, 'runtime config emit failed')
    },
  })

  configs.onChange(config => reloader.update(config))

  const cleanupMysql = await app.subscribe('config:mysql', value => configs.set('mysql', value))
  const cleanupRedis = await app.subscribe('config:redis', value => configs.set('redis', value))
  const cleanupFlags = await app.subscribe('config:flags', value => configs.set('flags', value))

  http.use(async (ctx, next) => {
    const runtime = reloader.current()
    if (!runtime) {
      ctx.status = 503
      ctx.body = { error: 'runtime is not ready' }
      return
    }
    await runtime.handle(ctx, next)
  })

  shutdown(async () => {
    await cleanupMysql()
    await cleanupRedis()
    await cleanupFlags()
    configs.dispose()
    await reloader.stop()
  })

  return reloader
})

Package-Local AI Guide

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