EMQX

@crvouga/mockingbird-service-emqx

Stateful emulator of the EMQX 5 MQTT broker: MQTT 5 over TCP and WebSocket, password and JWT authentication with JWT ACLs, persistent sessions with offline QoS 1 queues, and the REST v5 publish and client-kick endpoints.

Parity: MQTT 5 sessions, JWT ACLs, REST publish and client kickCommunication2 of 2 operationsIn-process fetchNode servermockingbird-emqx
$
npm install -D @crvouga/mockingbird-service-emqx

Playground

Requests go to the real emulator running in this tab. State persists across requests, and the journal shows what it received. A green dot marks operations whose sample succeeds as-is; others need ids from earlier responses.

Loads the real @crvouga/mockingbird-service-emqx into this tab on first send
http://localhost:18083

An API key as the username and its secret as the password.

Send a request to see the emulator's response.

Operations

2 of 2 operations in the vendored contract are emulated. Hand-authored from the EMQX 5.8 management API schema and handlers (https://github.com/emqx/emqx/tree/v5.8.6/apps/emqx_management/src) and https://docs.emqx.com/en/emqx/latest/admin/api.html.

MethodPathOperationStatus
POST/api/v5/publishPublishMessagePublish one message to a topic
DELETE/api/v5/clients/{clientid}KickClientKick a client out and discard its session

Documentation

The package README, the same file that ships in the npm tarball.Edit on GitHub

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.

Install

Shell
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.

TypeScript
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
Shell
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.

Packet Behaviour
CONNECT / CONNACK Clean Start and Session Expiry Interval as in the specification: clean: false resumes a stored session (sessionPresent: true), a clean start discards it. EMQX takes the client's interval as sent. The CONNACK carries Maximum Packet Size 1048576, Topic Alias Maximum 65535, Receive Maximum 32, Wildcard Subscription Available 1, Subscription Identifier Available 1, and what this emulator does not do: Maximum QoS 1, Retain Available 0, Shared Subscription Available 0. An empty client id is assigned 16 characters.
Authentication EMQX's chain: the built-in database (users), then the JWT authenticator (jwt). With neither configured anyone connects, as on a fresh EMQX. A wrong password is CONNACK 0x86; a token that verifies but is expired (exp), not yet valid (nbf), fails verify_claims or has a malformed acl is 0x86; a token that does not verify, or credentials no authenticator recognises, is 0x87.
Second CONNECT with a connected client id The old connection gets DISCONNECT 0x8E Session taken over; the new one resumes the session unless it asked for a clean start.
SUBSCRIBE / SUBACK Exact filters and + / # wildcards, No Local, Subscription Identifiers. Granted QoS is at most 1. A denied filter is SUBACK 0x87; $share/… is 0x9E.
PUBLISH / PUBACK QoS 0 and 1. PUBACK is 0x00, 0x10 when nobody subscribes (EMQX's behaviour), or 0x87 when denied (the message is dropped; a denied QoS 0 publish is dropped silently). Properties (content type, response topic, correlation data, user properties, payload format indicator, message expiry interval) are forwarded. Inbound topic aliases work.
Offline A session whose connection is gone keeps its subscriptions and queues messages (QoS 0 too, as mqtt.mqueue_store_qos0; at most 1000, oldest dropped, as mqtt.max_mqueue_len). Messages sent and not acknowledged are resent with DUP on resume. At most 32 QoS 1 messages are in flight (mqtt.max_inflight), fewer if the client's Receive Maximum is lower.
Session expiry Session Expiry Interval seconds after the connection ends, on the emulator clock. 0 (or absent) ends the session with the connection.
JWT expiry With disconnectAfterExpire (EMQX's default) a connected client is sent DISCONNECT 0x87 when its token's exp passes; with it off the connection stays and its ACL denies everything.

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.

Route Behaviour
POST /api/v5/publish JSON {topic, payload, qos?: 0|1|2, retain?, payload_encoding?: "plain"|"base64", properties?}. 200 {"id": "<32 hex>"} when at least one session matched (a session that is offline counts: the message is queued); 202 {"reason_code": 16, "message": "no_matching_subscribers"} when none did. 400 {"code": "BAD_REQUEST", "message"} for a body that fails the schema (missing topic, non-string payload, qos out of range, unknown field, invalid JSON); 400 {"reason_code": 144, "message": "topic_name_invalid"} for an empty topic or one with wildcards; 400 {"reason_code": 151, "message": "packet_too_large"} above 1 MB; 415 when the body is not application/json. QoS 2 is accepted and delivered at the subscription's granted QoS.
DELETE /api/v5/clients/{clientid} Kicks the client out: a connected client gets DISCONNECT 0x98 Administrative action, and the session, its subscriptions and its queue are discarded (an offline session too). 204, or 404 {"code": "CLIENTID_NOT_FOUND", "message": "Client ID not found"}.
any, without credentials 401 {"code": "AUTHORIZATION_HEADER_ERROR", "message": "Support authorization: basic/bearer "}; a wrong key or secret is 401 {"code": "BAD_API_KEY_OR_SECRET", "message": "Check api_key/api_secret"}; a bearer token is 401 BAD_TOKEN. All carry WWW-Authenticate: Basic Realm="emqx-dashboard".

Admin (beyond the standard contract)

Route Effect
GET /__admin/clients Every session: clientId, username, connected, transport, sessionExpiryInterval, connectedAt, disconnectedAt, expiresAt, and counts of subscriptions, queued and inflight messages.
GET /__admin/clients/:clientId One session, with its subscriptions under topics.
GET /__admin/subscriptions?clientId= Subscriptions: topicFilter, granted qos, options.
GET /__admin/publishes {total, publishes}: the last 1000 publishes as metadata (topic, qos, bytes, source: "mqtt" | "http" | "admin", clientId, matched, at). Payloads are never kept.
POST /__admin/inject {topic, payload | payloadBase64, qos?: 0 | 1}: deliver raw bytes to the topic's subscribers, bypassing the REST schema and every ACL. Use it to hand a consumer a payload its parser must refuse. Answers {matched}.
POST /__admin/transport/cut {clientId?}: a network outage for one client, or for all. Live connections die with no DISCONNECT and new CONNECTs are dropped unanswered. Sessions are untouched and expire on their own schedule.
POST /__admin/transport/restore {clientId?}: end the outage. GET /__admin/transport lists what is cut.
POST /__admin/clients/:clientId/disconnect {reasonCode?}: send DISCONNECT (default 0x98) and close. Unlike the vendor's kick, the session stays.
GET / PUT /__admin/settings {apiKeys?, users?, jwt?: {secret, from?, aclClaimName?, verifyClaims?, secretBase64Encoded?, disconnectAfterExpire?} | null, authorization?: {noMatch?, denyAction?}} for the calling namespace. Reset returns to the values the runtime was created with.
POST /__admin/clock Standard. Moving the clock settles expiries at once: sessions past their interval are discarded and clients with an expired token are disconnected.
POST /__admin/reset, checkpoints Standard. Reset drops every connection and session. Restoring a checkpoint restores sessions, subscriptions and queues, and drops live connections (a checkpoint cannot hold a socket).

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

Export Kind Description
createRuntime function The emulator with the full service contract. Options: settings, clock, seed, adminPrefix, adminKey, onLog, sqlite. Adds attach(transport, info?, namespace?) for an MQTT connection, webSocketPath(pathname), endpoints, and stop().
EmqxAPI class One namespace's broker: fetch(request) for the REST API, broker for MQTT, state, reset(). Options: sqlite, now, namespace, settings, connectFault.
EMQX_PRESETS object Every named fault preset.
EMQX_NAMESPACE string The service name, "emqx".
MQTT_CONNECT_OPERATION string "MqttConnect", the operation id a fault rule uses for a CONNECT.
MQTT_WEBSOCKET_PATH string "/mqtt".
MAX_PACKET_SIZE number 1048576, EMQX's mqtt.max_packet_size.
DEFAULT_SETTINGS, DEFAULT_JWT_AUTHENTICATOR objects A fresh EMQX: no authenticator, no_match = allow; and the JWT authenticator's defaults.
apiKeyCredential function The API key of a request (how credentials map to namespaces).
parseAcl, checkAcl functions Parse an acl claim, and ask it about one publish or subscribe (allow, deny or nomatch).
ReasonCode object MQTT 5 reason codes by name.
document, operationIds, supportedOperationIds values The REST contract and its operation ids.
createServer, connectStream, serveTarget, DEFAULT_PORT, DEFAULT_MQTT_PORT (./server) Node Serve HTTP, WebSocket and TCP ({url, mqttUrl, mqttPort, wsUrl, runtime, close()}; options add port, mqttPort, host, sweepMs); an in-process MQTT stream; the serve CLI target; ports 8850 and 1883.

Part of mockingbird.

↑↓ to navigate↵ to open