mqttkit

简体中文

Build MQTT applications with an Elysia-like API: compose broker adapters, ordered middleware, topic routers, authentication, lifecycle events, and business services with new MqttApp().use(...).use(router(...)).

mqttkit does not reimplement the MQTT protocol. A broker such as Aedes owns CONNECT, SUBSCRIBE, PUBLISH, QoS, retain, sessions, persistence, and MQTT-over-WebSocket. mqttkit adds an application framework layer on top.

Full documentation: https://mqttkit.keyp.dev (简体中文)

Features

• Ordered use() middleware for auth, logging, audit, validation, and interception.
• router().topic() declares MQTT topic routes with publish / subscribe policies.
• Topic params (devices/:uid/events), payload validation via any Standard Schema validator, and injected services on ctx.
• MQTT 5 RPC with app.request() and ctx.reply().
• MQTT 5 shared subscriptions ($share/<group>/<filter>) for multi-instance fan-out.
• Per-route timeout and concurrency guards surface as onError phases.
• app.onMetric() emits structured per-dispatch / per-publish events for Prometheus / OpenTelemetry.
• app.on() observes broker lifecycle events; app.publish() lets workers push messages.
• Adapters: @mqttkit/aedes (TCP + WebSocket) and @mqttkit/asyncapi (AsyncAPI 3.0 docs).
• In-memory TestBroker for unit tests. Built for Bun and TypeScript.

Installation

bun add @mqttkit/core @mqttkit/aedes aedes

Quick Start

import { aedes } from '@mqttkit/aedes'
import { MqttApp, router } from '@mqttkit/core'

const app = new MqttApp<{ principal?: { uid: string } }>()
  .use(
    aedes({
      tcp: { port: 1883 },
      ws: { port: 8888, path: '/mqtt' },
      authenticate({ clientId, username }) {
        if (!username) return false
        return { uid: username || clientId }
      },
    }),
  )
  .use(
    router<{ principal?: { uid: string } }>()
      .topic('devices/:uid/events', {
        publish: ({ params, principal }) => params.uid === principal?.uid,
        async onMessage(ctx) {
          await ctx.publish(`server/${ctx.params.uid}/echo`, ctx.payload, { qos: 0 })
        },
      })
      .topic('server/:uid/echo', {
        subscribe: ({ params, principal }) => params.uid === principal?.uid,
      }),
  )

await app.listen()

App API at a glance

The MqttApp builder methods covered in the quick start are listed below. Each returns this, so they chain.

Method	Purpose
use(plugin | middleware)	Register a plugin (router(), broker adapter, …) or a global middleware.
decorate(key, value)	Inject a service into ctx.services with type inference on the App generic.
on(eventName, handler)	Observe broker lifecycle events (client.connect, client.subscribe, …).
onStart(hook) / onStop(hook)	Run code after listen() succeeds or before stop() finishes.
onError(handler)	App-level error sink — receives { error, topic, phase, route, ctx }.
onMetric(handler)	Structured per-dispatch / per-publish events for Prometheus / OTel.
onBeforePublish(hook)	Mutate outbound { topic, payload, options } (e.g. inject traceparent).
logger(logger)	Forward mqttkit's internal warnings into pino / OpenTelemetry / Sentry.
addSchemaProvider(provider)	Plug in non-Standard-Schema validators (e.g. raw TypeBox).
publish(topic, payload, opts?)	Server-side outbound publish (also funnels through onBeforePublish).
request(topic, payload, opts?)	MQTT 5 RPC with optional retries / retryDelay.
stop({ drain?, timeout? })	Graceful shutdown — defaults to draining in-flight handlers.

Error handling sketch

import { MqttApp, router } from '@mqttkit/core'

const app = new MqttApp()
  .use(adapter)
  .onError(({ error, phase, topic }) => {
    // phases: middleware | handler | validation | policy | publish | timeout | overload
    sentry.captureException(error, { tags: { topic, phase } })
  })
  .use(
    router().topic('devices/:uid/events', {
      timeout: 1_000,
      concurrency: 100,
      onError: ({ error }) => metrics.routeFailures.inc(), // route-scoped, runs first
      async onMessage(ctx) {
        await doWork(ctx)
      },
    }),
  )

Custom logger

import { MqttApp, type MqttLogger } from '@mqttkit/core'
import { pino } from 'pino'

const log = pino({ name: 'mqttkit' })
const logger: MqttLogger = {
  debug: (msg, meta) => log.debug(meta, msg),
  info: (msg, meta) => log.info(meta, msg),
  warn: (msg, meta) => log.warn(meta, msg),
  error: (msg, meta) => log.error(meta, msg),
}

new MqttApp().logger(logger)

Schema Validation

topic({ schema }) accepts any Standard Schema validator (zod, valibot, arktype, …). The validated payload is exposed on ctx.body with full type inference.

import { z } from 'zod'

router().topic('devices/:uid/events', {
  schema: { body: z.object({ temperature: z.number() }) },
  async onMessage(ctx) {
    ctx.body.temperature // typed as number
  },
})

Use @mqttkit/typebox for raw TypeBox or @mqttkit/zod to attach a JSON Schema so @mqttkit/asyncapi can emit the full payload.

See the Getting Started guide for routers, middleware, events, RPC, Kafka bridging, and more.

Packages

• @mqttkit/core: application, router, middleware, context, schema validation, RPC, event types, broker adapter interface, and @mqttkit/core/testing in-memory broker.
• @mqttkit/aedes: Aedes adapter for TCP MQTT and MQTT-over-WebSocket (forwards MQTT 5 properties for RPC).
• @mqttkit/asyncapi: AsyncAPI 3.0 generator and HTTP plugin for browsable docs.
• @mqttkit/typebox: TypeBox schema provider — register once, then pass raw Type.X(...) schemas directly.
• @mqttkit/zod: zod helper that attaches a JSON Schema representation so AsyncAPI documents the full payload.

Examples

Runnable examples under examples/ cover the basic TCP / WebSocket broker, lifecycle events, service push, Kafka bridge, schema validation, MQTT 5 RPC (including app.request({ retries, retryDelay })), AsyncAPI docs (standalone HTTP or shared Elysia port), Prometheus metrics, and a custom JSON logger (examples/custom-logger).

bun install
bun run --cwd examples/aedes-basic dev

Development

bun run test
bun run typecheck
bun run build

Releasing

scripts/publish.mjs compares each package's local version against the npm registry and only publishes packages whose local version is ahead (or not on npm yet). Packages already in sync are silently skipped.

bun run publish:status      # report local vs npm without publishing
bun run publish:dry-run     # dry-run the actual publish flow
bun run publish:packages    # publish whatever needs publishing

# Single package by name (still subject to the version-ahead check)
bun run publish:core

# Bypass the check (rarely needed)
node scripts/publish.mjs --force