mqtt-topic-router

TypeScript-first router for MQTT topics. Use named route parameters like devices/:mac/status, decode payloads once, and handle messages from any mqtt.js-compatible client or directly in memory.

Why Use It

• Readable routes instead of manual topic.split('/') and raw MQTT wildcards.
• Automatic subscription to compiled MQTT filters such as devices/+/status.
• In-memory dispatch() for tests or brokerless workflows.
• Payload decoding with per-route TypeScript types.
• Explicit error hooks and optional lifecycle events for logs and metrics.
• Zero runtime dependencies; ships ESM, CommonJS, and TypeScript declarations.

Installation

npm install mqtt-topic-router

Node.js 18.18 or newer is required. mqtt is not bundled; pass your own client when you want broker integration.

Quick Start

import { Buffer } from 'node:buffer';
import mqtt from 'mqtt';
import { createRouter } from 'mqtt-topic-router';

const client = mqtt.connect('mqtt://localhost:1883');
const router = createRouter(client);

router.on('devices/:mac/status', ({ topic, payload, params }) => {
  // payload is the raw transport payload, usually Buffer/Uint8Array
  console.log(topic, params.mac, Buffer.from(payload).toString('utf8'));
});

Use createStringRouter(client) or createJsonRouter(client) when handlers should receive decoded payloads directly.

The same router can run without a transport. This is useful for tests and local message pipelines:

import { createRouter } from 'mqtt-topic-router';

const router = createRouter();

router.on<string>('devices/:mac/status', ({ payload, params }) => {
  console.log(params.mac, payload);
});

const matched = await router.dispatch('devices/AA-BB/status', 'online');
// matched === true

Route Syntax

Route patterns use named parameters and are compiled to MQTT topic filters.

| Route | MQTT filter | Meaning | | --------------------- | ------------------ | ---------------------------- | | :name | + | one topic level | | :name# | # | zero or more trailing levels | | devices/:mac/status | devices/+/status | one device status topic | | devices/:mac/:path# | devices/+/# | any trailing path per device |

router.on('devices/:mac/:path#', ({ params }) => {
  console.log(params.mac); // 'AA-BB'
  console.log(params.path); // 'sensors/temp'
});

Invalid routes throw InvalidRouteError during registration or compilation.

• Use :name and :name#; raw + and # are rejected.
• Parameters must occupy the whole segment: devices/dev-:id is invalid.
• :name# must be the final segment.
• Parameter names must be unique and cannot be __proto__, constructor, or prototype.
• Routes that start with a parameter do not match $ topics, following MQTT wildcard rules.

Payloads And Types

createRouter(client) does not decode payloads. Handlers receive the raw transport payload, usually a Buffer/Uint8Array with mqtt.js.

For common formats, use the included factories:

import { createJsonRouter, createStringRouter } from 'mqtt-topic-router';

const stringRouter = createStringRouter(client); // handler payload: string
const jsonRouter = createJsonRouter(client); // handler payload: unknown

createJsonRouter parses JSON but does not validate its shape. Type each route with on<T>() or onAsync<T>() when you know what that topic carries:

type Status = { online: boolean };
type Reading = { value: number; unit: string };

const router = createJsonRouter(client);

router.on<Status>('devices/:mac/status', ({ payload }) => {
  console.log(payload.online);
});

router.on<Reading>('devices/:mac/sensors/:sensor', ({ payload }) => {
  console.log(payload.value, payload.unit);
});

The type argument is a TypeScript assertion, not runtime validation. Keep untrusted JSON as unknown until a validator or type guard accepts it.

router.on('devices/:mac/status', ({ payload }) => {
  if (!isStatus(payload)) return;
  console.log(payload.online);
});

For custom formats, pass decodePayload. It can be async and receives the raw payload, topic, and transport packet/context.

const router = createRouter(client, {
  decodePayload: (payload, topic, packet) =>
    decodeFrame(payload, topic, packet),
});

createJsonRouter also accepts maxPayloadBytes to reject oversized payloads before parsing.

const router = createJsonRouter(client, { maxPayloadBytes: 64 * 1024 });

dispatch(topic, payload, context?) expects an already decoded payload; decodePayload is only used for transport messages.

Subscriptions

on() registers a route, subscribes to its MQTT filter when a transport is present, and returns a handle.

const handle = router.on('devices/:mac/status', onStatus);

handle.route; // 'devices/:mac/status'
handle.topicFilter; // 'devices/+/status'

handle.unsubscribe();
// or: router.off(handle) -> boolean

Subscriptions are reference-counted by topic filter, so equivalent filters are subscribed once. If autoUnsubscribe: true, the router unsubscribes a filter only after the last route using it is removed.

Use onAsync() when the route should become active only after the broker accepts the subscription.

const handle = await router.onAsync('devices/:mac/status', onStatus);

await handle.unsubscribeAsync();
// or: await router.offAsync(handle)

onAsync() prefers subscribeAsync() / unsubscribeAsync() when present and falls back to promise-returning subscribe() / unsubscribe().

Route lifetime can be tied to an AbortSignal:

const controller = new AbortController();

router.on('devices/:mac/status', onStatus, { signal: controller.signal });

controller.abort();

Useful transport options:

const router = createRouter(client, {
  autoSubscribe: true,
  autoUnsubscribe: true,
  subscribeOptions: { qos: 1 },
  unsubscribeOptions: {
    properties: { userProperties: { source: 'mqtt-topic-router' } },
  },
});

Set autoSubscribe: false if your client already receives the topics and you want to manage MQTT subscriptions yourself. autoUnsubscribe only touches filters the router subscribed to.

The transport type is structural. The library does not import mqtt.js types; it works with compatible on, off, and subscribe methods. unsubscribe or unsubscribeAsync is only needed for automatic unsubscription.

Errors, Ordering, And Shutdown

Use onError for route handler failures and onDecodeError for decode failures. Decode failures are not routed to handlers.

const router = createJsonRouter(client, {
  onError: (error, message) => {
    log.error({ err: error, topic: message.topic, route: message.route });
  },
  onDecodeError: (error, info) => {
    log.warn({ err: error, topic: info.topic }, 'bad payload');
  },
  logger: (message, error) => {
    log.error({ err: error }, message);
  },
});

Transport message errors are caught, logged, and do not become unhandled promise rejections. Direct dispatch() calls still reject when a handler fails and no onError is configured.

Transport messages run concurrently by default. Pass ordered: true to process them one at a time in arrival order.

const router = createRouter(client, { ordered: true });

Use close() for graceful shutdown and dispose() for immediate shutdown.

await router.close(); // drains accepted messages, removes routes/resources
router.dispose(); // drops in-flight work immediately

The router also implements Symbol.asyncDispose and Symbol.dispose:

await using router = createRouter(client);

Explicit resource management needs TypeScript 5.2+ and a runtime that defines those symbols (Node.js 20.5+, or a polyfill on older versions). close() and dispose() work on every supported Node.js version.

Observer Events

Pass observer when you need synchronous lifecycle events for logs, metrics, tracing, profiling, audits, or tests.

const router = createRouter(client, {
  observer: {
    'handler.failed': (event) => {
      log.error({ err: event.error, route: event.route });
    },
    'message.completed': (event) => {
      metrics.increment('mqtt.message', { outcome: event.outcome });
    },
  },
});

Use '*' to receive every event, or pass a single function:

const router = createRouter(client, {
  observer: (event) => trace(event),
});

Event types:

• message.received, message.started, message.completed
• decode.started, decode.succeeded, decode.failed
• match.started, match.succeeded, match.failed
• handler.started, handler.succeeded, handler.failed
• route.registered, route.removed
• subscription.requested, unsubscription.requested
• router.closing, router.disposed

Observer errors are sent to logger and never affect routing. message.completed.outcome is success, decode-error, handler-error, or disposed.

match.* events are emitted per evaluated route, so they measure matching cost without handler execution. With an observer map they are only built when at least one match.* key (or '*') is subscribed, so observers that ignore them add no per-route cost.

API Reference

Factories and utilities:

• createRouter(options?)
• createRouter(transport, options?)
• createStringRouter(transport, options?)
• createJsonRouter(transport, options?)
• MqttRouter
• compileRoute(route)
• matchRoute(compiledRoute, topic)
• InvalidRouteError
• decodeString(payload)
• decodeJson(payload)

Router methods:

| Method | Returns | | -------------------------------------- | --------------------------- | | on<T>(route, handler, options?) | RouteHandle | | onAsync<T>(route, handler, options?) | Promise<AsyncRouteHandle> | | off(handle) | boolean | | offAsync(handle) | Promise<boolean> | | dispatch(topic, payload, context?) | Promise<boolean> | | routes() | RouteInfo[] | | close() | Promise<void> | | dispose() | void |

Router options:

| Option | Default | Description | | -------------------- | --------------- | ------------------------------------------------ | | autoSubscribe | true | Subscribe to each route's MQTT filter. | | autoUnsubscribe | false | Unsubscribe owned filters when no routes remain. | | ordered | false | Process transport messages sequentially. | | subscribeOptions | - | Passed to transport subscribe(). | | unsubscribeOptions | - | Passed to transport unsubscribe(). | | decodePayload | - | Decode raw transport payloads before routing. | | onError | - | Handles route handler errors. | | onDecodeError | - | Handles decoder errors. | | logger | console.error | Receives fallback and observer errors. | | observer | - | Receives lifecycle events. |

The package also exports TypeScript types for transports, router options, handlers, handles, decoded messages, packet context, route matching, and router events.

CommonJS is supported:

const { createRouter } = require('mqtt-topic-router');

Notes

• Every matching route runs; there is no route priority or first-match stop.
• Overlapping MQTT subscriptions can make brokers deliver duplicate messages; each delivery is routed independently.
• There is no built-in backpressure. Keep handlers fast or add your own queue when bursts can outpace processing.

Development

npm run typecheck
npm test
npm run lint
npm run format
npm run build

License

MIT (c) Diego Vergara