@crvouga/mockingbird-service-emqx

Local emulators. Real API contracts. Part of Mockingbird.

Stateful emulator of an EMQX 5 MQTT broker for test suites and local development: an MQTT 5 broker over TCP and WebSocket (/mqtt), EMQX's password and JWT authentication with JWT ACLs, persistent sessions with offline QoS 1 queues, and the two REST v5 endpoints a publish-only client needs (POST /api/v5/publish, DELETE /api/v5/clients/{clientid}). It runs in the test process on OS-assigned ports, with no installed broker, Docker, or native binary, and session expiry follows a clock the suite moves instead of waiting.

• Operation coverage (HTTP): SUPPORT.md
• No EMQX broker was available as an oracle, so behaviour follows the MQTT 5.0 specification, the EMQX documentation and the EMQX 5.8.6 source. What those do not settle is listed under Unverified.

Install

npm install -D @crvouga/mockingbird-service-emqx

ESM only. Node >= 22 or Bun >= 1.2. No native dependencies. Serve it with npx mockingbird-emqx serve or createServer from ./server (Node and Bun). createRuntime (the default entry) is portable: it serves the REST API over fetch and MQTT through runtime.attach, with no listener.

Usage

Point the app's MQTT URL at mqttUrl (or wsUrl), its EMQX API origin at url, and give it the API key, MQTT user and JWT secret you configured here.

import { createServer } from "@crvouga/mockingbird-service-emqx/server"
import mqtt from "mqtt"

const emqx = await createServer({
  settings: {
    apiKeys: [{ key: "test-key", secret: "test-secret" }],
    users: [{ username: "service", password: "service-password", superuser: true }],
    jwt: { secret: "test-jwt-secret" }, // HS256 token in the CONNECT password
    authorization: { noMatch: "deny" },
  },
})
// emqx.url      http://127.0.0.1:<port>       REST API and /__admin
// emqx.mqttUrl  mqtt://127.0.0.1:<port>       MQTT 5 over TCP
// emqx.wsUrl    ws://127.0.0.1:<port>/mqtt    MQTT 5 over WebSocket

const client = await mqtt.connectAsync(emqx.mqttUrl, {
  protocolVersion: 5,
  clientId: "device-1",
  clean: false,
  properties: { sessionExpiryInterval: 600 },
  username: "service",
  password: "service-password",
})
await client.subscribeAsync("space:example", { qos: 1 })

// EMQX publishes no JavaScript SDK for its REST API, so the publisher calls it over HTTP.
await fetch(`${emqx.url}/api/v5/publish`, {
  method: "POST",
  headers: {
    authorization: `Basic ${btoa("test-key:test-secret")}`,
    "content-type": "application/json",
  },
  body: JSON.stringify({ topic: "space:example", payload: "hello", qos: 1, retain: false }),
})

// Test control: move time instead of waiting for a session to expire.
await fetch(`${emqx.url}/__admin/clock`, { method: "POST", body: '{"advance": "600s"}' })

await client.endAsync()
await emqx.close() // closes every listener and drops every client

npx mockingbird-emqx serve --port 8850 --mqtt-port 1883 \
  --api-key test-key:test-secret --mqtt-user service:service-password \
  --jwt-secret test-jwt-secret --no-match deny

Each createServer() is its own broker with its own ports; two of them share nothing. Under Bun, mqtt's Node WebSocket transport does not start (Bun's ws has no createWebSocketStream), so connect to wsUrl with the library's browser transport (new MqttClient((c) => browserStreamBuilder(c, options), options) from mqtt/lib/connect/ws), or use connectStream(runtime) from ./server for a socket-free connection. On Node the plain mqtt.connect("ws://…/mqtt") works.

MQTT

MQTT 5.0 over TCP and over WebSocket at /mqtt (subprotocol mqtt, also mqtt-v3, mqtt-v3.1.1, mqtt-v5; an upgrade without one is answered 400, as EMQX does with fail_if_no_subprotocol). Both transports reach the same sessions.

JWT ACL. The token's acl claim (name configurable) is asked before anything else:

• Object form, {"pub": [...], "sub": [...], "all": [...]}: listed topics are allowed and everything else is denied.
• List form, [{"permission": "allow" | "deny", "action": "pub" | "sub" | "all", "topic": "…", "qos"?: [0, 1], "retain"?: false}]: the first matching rule decides; a topic no rule matches falls to authorization.noMatch (allow by default, as in EMQX 5; EMQX 6 changed its default to deny).

ACL topics are MQTT filters with ${clientid} and ${username} expanded; eq <topic> compares literally. A requested filter is matched word by word as written, so the rule a/+ covers a subscription to a/# (EMQX's emqx_topic:match/2). A superuser from users skips authorization. With authorization.denyAction: "disconnect" a denied operation ends the connection with DISCONNECT 0x87 instead.

The emulator models what the broker does with a topic and nothing else: space:example is one topic level. Mapping an application's logical names onto topics is the client's job.

Routes

Every route needs Authorization: Basic <api key>:<secret>. With apiKeys empty any pair is accepted; a missing header is always refused.

Admin (beyond the standard contract)

Fault presets (POST /__admin/faults {"preset": "<name>", "count"?: n}; GET /__admin/faults/presets): connect_bad_credentials (CONNACK 0x86), connect_not_authorized (0x87), connect_server_unavailable (0x88), publish_unavailable (REST publish answers 503 {"reason_code": 131, "message": "failed_to_dispatch"}), api_unauthorized (every REST call answers 401). A CONNECT is the fault operation MqttConnect, so any reason code can be forced: {"operationId": "MqttConnect", "effect": "connack", "params": {"reasonCode": 151}}.

Namespaces

Parallel workers isolate themselves by namespace. HTTP: x-mockingbird-namespace, a /__admin/ns/<name> prefix on the API origin, or by API key (PUT /__admin/credentials {"credentials": {"<api key>": "<namespace>"}}). MQTT: the WebSocket URL ws://…/__admin/ns/<name>/mqtt, or the same credential map keyed by the CONNECT's username, then its client id. Unmapped clients use the default namespace.

Unverified

Taken from the EMQX 5.8.6 source or left open, not observed on a broker:

• The message text of a schema 400 (EMQX prints its config library's error; the emulator prints {"kind": "validation_error", "path": "root.<field>", "reason": …}), of a bad base64 payload, and of an unknown path's 404.
• EMQX coerces an integer payload to its decimal string; the emulator refuses it.
• The CONNACK Receive Maximum is EMQX's min(client's, 32); the emulator always sends 32.
• The is_superuser token claim is not read.

Deliberately not modelled

• MQTT 3.1 and 3.1.1: such a CONNECT is refused with that protocol's return code 1.
• QoS 2: the CONNACK says Maximum QoS 1, a QoS 2 subscription is granted QoS 1, and a QoS 2 publish is answered DISCONNECT 0x9B. EMQX itself supports QoS 2.
• Retained messages: the CONNACK says Retain Available 0, a retained publish is answered DISCONNECT 0x9A, and retain: true on the REST API is 400.
• Will messages (accepted in CONNECT, never published), shared subscriptions, enhanced authentication (AUTH), keep-alive timeouts (PINGREQ is answered; use a transport cut for a dead connection), message expiry, and outbound topic aliases.
• Public-key and JWKS tokens, password hashing, and authenticators or authorizers other than the built-in database and JWT.
• Clustering, the dashboard and its login tokens, the rule engine, bridges, billing, TLS and listener configuration.

API

Part of mockingbird.