MQTT Wire

Server-side MQTT protocol codec and connection state machine for JavaScript and TypeScript runtimes. Accepts connections from MQTT clients, parses bytes into typed packets, encodes packets into bytes, and manages per-connection protocol state.

Features

• MqttWire — Connection state machine with lifecycle hooks for CONNECT, PUBLISH, SUBSCRIBE, etc.
• Protocol Support — MQTT 3.1.1 and 5.0 with full packet type coverage
• Codec Utilities — BinaryReader, BinaryWriter, StreamFramer for low-level protocol handling
• QoS Tracking — QoSFlowTracker and PacketIdAllocator for reliable message delivery
• Topic Aliases — TopicAliasManager for MQTT 5.0 topic alias negotiation
• Multi-runtime — Works with Bun, Node.js 20+, and Deno

Installation

bun add @qualithm/mqtt-wire
# or
npm install @qualithm/mqtt-wire

Quick Start

import * as net from "node:net"
import { MqttWire, PacketType } from "@qualithm/mqtt-wire"

const server = net.createServer((socket) => {
  const wire = new MqttWire({
    onSend: (data) => {
      socket.write(data)
    },

    onConnect: (connect) => {
      console.log(`Client connected: ${connect.clientId}`)
      return {
        type: PacketType.CONNACK,
        sessionPresent: false,
        reasonCode: 0x00
      }
    },

    onPublish: (packet) => {
      const payload = new TextDecoder().decode(packet.payload)
      console.log(`[${packet.topic}] ${payload}`)
    },

    onSubscribe: (packet) => ({
      type: PacketType.SUBACK,
      packetId: packet.packetId,
      reasonCodes: packet.subscriptions.map((s) => s.options.qos)
    }),

    onDisconnect: () => console.log("Client disconnected")
  })

  socket.on("data", (chunk: Buffer) => void wire.receive(chunk))
  socket.on("close", () => wire.reset())
})

server.listen(1883, () => console.log("MQTT server on port 1883"))

Usage

Low-Level Codec

import {
  BinaryReader,
  BinaryWriter,
  decodeVariableByteInteger,
  encodeVariableByteIntegerToArray
} from "@qualithm/mqtt-wire"

// Encode a variable byte integer
const encoded = encodeVariableByteIntegerToArray(16384)
console.log(encoded) // Uint8Array [0x80, 0x80, 0x01]

// Decode it back
const decoded = decodeVariableByteInteger(encoded, 0)
if (decoded.ok) {
  console.log(decoded.value.value) // 16384
}

// Build a packet manually
const writer = new BinaryWriter()
writer
  .writeUint8(0x10) // CONNECT packet type
  .writeVariableByteInteger(12) // Remaining length
  .writeMqttString("MQTT") // Protocol name
  .writeUint8(5) // Protocol version

const packet = writer.toUint8Array()

Acknowledge after delivery

By default a QoS 1 PUBACK is sent before onPublish runs. Set pubackAfterDelivery to acknowledge only once the hook resolves: a hook that throws leaves the message unacknowledged, so the client keeps it and resends it.

const wire = new MqttWire(hooks, { pubackAfterDelivery: true })

Error Handling

MqttWire uses lifecycle hooks for error reporting — receive() does not throw protocol errors.

import {
  decodeVariableByteInteger,
  MqttWire,
  ProtocolError,
  StateError,
  type DecodeResult,
  type VarintDecodeValue
} from "@qualithm/mqtt-wire"

// Protocol errors from receive() are reported via the onError hook
const wire = new MqttWire({
  onSend: (data) => {
    socket.write(data)
  },
  onConnect: (connect) => ({
    /* ... */
  }),

  onError: (error) => {
    // A ProtocolError carries an MQTT reason code. Errors from decoding inbound
    // bytes also carry a stable code: an oversized frame has code
    // "PACKET_TOO_LARGE" and reason 0x95; other malformed input has 0x81.
    if (error instanceof ProtocolError) {
      console.error(`protocol error: ${error.message}`, {
        reasonCode: error.reasonCode,
        code: error.code
      })
    }
    socket.destroy()
  }
})

// receive() handles protocol errors internally; guard against unexpected failures
socket.on("data", (chunk: Buffer) => {
  wire.receive(chunk).catch((err) => {
    console.error("unexpected receive error", err)
    socket.destroy()
  })
})

// StateError is thrown by outbound methods when called in the wrong state
try {
  await wire.publish("topic", payload)
} catch (error) {
  if (error instanceof StateError) {
    console.error(`state error: ${error.message}`, { state: error.state })
  }
}

// Codec functions return Result types (no exceptions)
const result: DecodeResult<VarintDecodeValue> = decodeVariableByteInteger(data, 0)
if (result.ok) {
  console.log(result.value.value)
} else {
  console.error(`[${result.error.code}] ${result.error.message}`)
}

API Reference

Full API documentation is generated with TypeDoc:

bun run docs
# Output in docs/

Examples

See the examples/ directory for runnable examples:

| Example | Description | | ----------------------------------------------------- | ----------------------------------- | | node-tcp.ts | Node.js TCP server | | bun-tcp.ts | Bun TCP server | | deno-tcp.ts | Deno TCP server | | websocket.ts | WebSocket server | | basic-usage.ts | Low-level codec utilities | | batch-processing.ts | Stream framing of chunked MQTT data | | error-handling.ts | Result type patterns |

bun run examples/node-tcp.ts

Development

Prerequisites

• Bun (recommended), Node.js 20+, or Deno
• mosquitto — provides mosquitto_pub / mosquitto_sub for conformance tests (brew install mosquitto)

Setup

bun install

Building

bun run build

Testing

bun run test              # unit tests
bun run test:coverage     # with coverage report

Linting & Formatting

bun run lint
bun run format
bun run typecheck

Benchmarks

bun run bench

License

Apache-2.0