MQTT

The most flexible of the bundled transports, and the only one whose channel is a genuine routing key. It subscribes to topics on a broker and forwards each JSON message as an event on its topic.

When to use it

• Your data already flows through a broker — which, on a plant floor, it usually does.
• You want the transport to run in the browser. MQTT is web-native over WebSocket, so a page can subscribe directly, with no server-side component at all.
• Your source is OPC UA but the plant already bridges it to MQTT (OPC UA PubSub over MQTT, Telegraf's inputs.opcua, Kepware, Ignition). Point this transport at that broker rather than holding an OPC UA session next to the scene: no session to keep alive, and it runs anywhere.

Install

npm install mqtt

The dependency is optional and imported lazily — only a configuration that actually uses kind: "mqtt" needs it.

Configuration

sources: [
    {
        kind: "mqtt",
        config: {
            broker_url: "mqtt://broker.example.com:1883",
            topics: ["plant/+/+/motor", "plant/+/+/status"],
        },
    },
],

Option	Type	Default	Description
broker_url	string	—	Required. mqtt://, mqtts://, ws:// or wss://. May embed user:pass@
topics	Array<string>	none	Topics to subscribe to. MQTT wildcards + and # are supported

Subscribe to the topic families your mappings actually select on rather than to # — everything else living on the broker then never reaches your process at all.

What arrives

Each JSON message becomes an event on its topic:

{
    channel: "plant/line-a/3/motor",         // the topic
    payload: { rpm: 900, temp_c: 62.5 },     // the parsed JSON
    received_at: Date,
    metadata: { transport: "mqtt", qos: 0, retain: false },
}

The topic is a real routing key, so mappings select on it with a channel pattern, and — more usefully — the entity's identity is usually spelled out in one of its segments.

Messages that are not valid JSON are logged and skipped.

Entity identity from the topic

This is what MQTT does that a flat stream cannot: one mapping drives any number of entities, because the topic says which one each message is about.

import type { EventMapping } from "@3dverse/livelink-agent";

// `<site>/plant/line-a/3/motor` is cell `line-a-3`. Counting back from the end
// rather than forward keeps this working whatever the site prefix is.
function cellId(channel: string): string | null {
    const segments = channel.split("/");
    const line = segments[segments.length - 3];
    const cell = segments[segments.length - 2];
    return line && cell ? `${line}-${cell}` : null;
}

const motorMapping: EventMapping = {
    channel: "livelink-demo/plant/+/+/motor",

    schema: {
        type: "object",
        properties: { rpm: { type: "number" }, temp_c: { type: "number" } },
        required: ["rpm", "temp_c"],
    },

    // The stream defines the population: one entity per cell that publishes.
    entities: {
        spawn: {
            name: "cell-{id}",
            components: ({ id }) => ({
                debug_name: { value: `cell-${id}` },
                mesh_ref: { value: MESH_UUID },
                local_transform: { position: cellPosition(id) },
            }),
            options: { delete_on_client_disconnection: true },
        },
    },

    updates: event => {
        const id = cellId(event.channel);
        const { temp_c } = event.payload as { temp_c: number };
        if (id === null) {
            return null;
        }
        return { id, update: { local_transform: { scale: [1, heightFor(temp_c), 1] } } };
    },
};

Add a seventh cell to the plant and a seventh entity appears in the scene, with nothing to change in the mapping.

Running it in the browser

The transport works unchanged in a page — as long as the URL is a WebSocket one:

sources: [{ kind: "mqtt", config: { broker_url: "ws://localhost:8000/mqtt", topics: ["plant/#"] } }],

A browser can only speak MQTT over WebSocket

This is the trap everyone hits. A broker's default listener is plain TCP on 1883, which a page cannot open. The broker needs a second listener speaking websockets — and both listeners serve the same topics, so publishers can keep using 1883 while your page subscribes over 8000.

With Mosquitto:

mosquitto.conf

listener 1883
protocol mqtt

listener 8000
protocol websockets

Use wss:// rather than ws:// for anything that is not on your own machine.

Credentials

broker_url accepts them inline:

broker_url: `mqtts://ingestion:${process.env.MQTT_PASSWORD}@broker.example.com:8883`,

The URL is redacted before it reaches the logs. Keep the password in the environment rather than in the source, and use mqtts:// or wss:// for anything crossing a network you do not control.

Retained messages

Brokers replay retained messages to every subscriber on connect. That is useful — a status topic tells you the state of every machine immediately instead of after the next publication — but it means those messages can arrive before whatever spawns the entities they refer to.

Those events resolve to nothing and are counted as unresolved_entity drops, then resolve on their own as soon as the entities appear. A burst of them at startup is expected, not a bug. See monitoring an agent.

Try it

The MQTT Ingestion sample agent drives a plant floor of six machine cells from a live broker, with the broker and a simulator publishing into it both started by one docker compose command. Its mappings are the ones on this page, against a real stream rather than a snippet — and they run in a page unchanged, over the WebSocket listener the same broker exposes.