Connect MQTT devices

Turn messages on an MQTT broker into readings for shadows and twins, and publish a twin's commands back to the device.

Most gateways, PLCs and sensors can publish to an MQTT broker. mqttConnection from @klorad/protocols/mqtt subscribes to the topics you name, turns each message into readings, and, for a Digital Twin, publishes the command of an authorised Action.

pnpm add @klorad/protocols mqtt

mqtt (MQTT.js 5) is an optional peer: only apps that import @klorad/protocols/mqtt need it.

site-broker.ts

import { z } from "zod";
import { connectSource } from "@klorad/api/integration";
import { createInteraction, createScene } from "@klorad/api/world";
import { jsonPayload, mqttConnection } from "@klorad/protocols/mqtt";

const scene = createScene({ coordinateSystem: { origin: { lat: 40.62637, lon: 22.94838 } } });
scene.add({
  id: "gate",
  name: "Loading gate",
  position: { east: 40, north: -12 },
  correspondence: "twin",
  binding: { source: "site-broker", entity: "gate-1" },
});

// What the gateway publishes on site/<device>/state.
const GateState = z.object({ state: z.enum(["open", "closed", "moving"]), ts: z.number() });

const broker = mqttConnection({
  id: "site-broker",
  url: "mqtts://broker.example.com:8883",
  connect: { username: "klorad", password: process.env.MQTT_PASSWORD ?? "" },
  topics: ["site/+/state"],
  parse: (topic, payload) => {
    const { state, ts } = GateState.parse(jsonPayload(payload));
    return [{ entity: topic.split("/")[1]!, quantity: "state", value: state, observedAt: ts }];
  },
  // For the twin: an authorised "open" publishes { "command": "open" } to site/gate-1/set.
  command: (command) => ({ topic: `site/${command.entity}/set`, payload: JSON.stringify({ command: command.name }) }),
  onError: (error) => console.warn("MQTT", error),
});

connectSource(scene, broker);

const interaction = createInteraction(scene, {
  entitlement: { check: () => ({ allowed: true }) }, // use an AccessPolicy from @klorad/access in an app
  actuators: [broker],
  behaviours: [{ id: "requested", run: ({ update, now }) => update({ properties: { requestedAt: now } }) }],
  actions: [{ id: "open", label: "Open", appliesTo: (o) => o.correspondence === "twin", behaviour: "requested", command: () => ({ name: "open" }) }],
});

export { broker, interaction };

What matters

Bind by the device's own name. parse reports entity: "gate-1" exactly as the topic names it, and the twin's binding.entity uses the same name. Mapping lives in bindings.

Validate every message. parse runs the schema before anything reaches the scene. A message it refuses is dropped and reported to onError; the next one is processed as usual.

Use the device's timestamp. observedAt must be when the device saw the state. A message that arrives late joins the history without overwriting a newer state (Time).

Commands only through Actions. command is called only for an Action that passed the Entitlement, and the twin's state changes only when the device reports it on its state topic. sent means the broker accepted the message, not that the gate has moved.

Credentials stay on the server. Read the broker password from your server's secret store or environment; never ship it to a browser.

Reference

URL schemes mqtt://, mqtts://, ws:// and wss://; QoS 1 by default; one client per connection, opened on first use and closed with close(). Pass client to reuse a client you already have. Sparkplug B payloads are not decoded yet. Full options: @klorad/protocols.

Connect MQTT devices

What matters

Reference

On this page