#MQTT

@alepha/mqtt is an MQTT 5 client: byte payloads, a Last Will you can set by code, reconnection, and shared subscriptions. Devices, gateways and brokers speak MQTT; this is how an Alepha server talks to them.

1import { AlephaMqtt, MqttClientProvider } from "@alepha/mqtt";

#Connecting

Register the module and set MQTT_URL:

MQTT_URL=mqtt://broker.example.com:1883

With MQTT_URL set (Node or Bun), the client is MqttJsClientProvider, on mqtt.js. Without it, and always in the browser and on Cloudflare Workers, it is MemoryMqttClientProvider, an in-process broker: there is no default URL, so an app never points at a broker that is not there. mqtts://, ws:// and wss:// work as they do in mqtt.js. A substitution of your own for MqttClientProvider wins over both.

The client connects on start and disconnects on stop, gracefully, so the broker does not publish the will. Everything else is the server-only mqttOptions atom:

 1import { AlephaMqtt, mqttOptions } from "@alepha/mqtt"; 2import { Alepha } from "alepha"; 3  4const alepha = Alepha.create().with(AlephaMqtt); 5  6alepha.set(mqttOptions, { 7  clientId: "gateway-7", 8  protocolVersion: 5, 9  keepalive: 60,10  lazy: false,11  reconnectPeriod: 5_000,12  username: "gateway",13  password: "s3cret",14  // Anything else mqtt.js accepts (`ca`, `cert`, `key` for mutual TLS),15  // merged last.16  connect: { rejectUnauthorized: true },17});

protocolVersion: 4 speaks MQTT 3.1.1, for a broker that has nothing newer. lazy: true defers the connection to the first publish or subscribe. After a reconnect the client restores every subscription itself before the next publish or subscribe goes through.

#Publishing and subscribing

 1import { MqttClientProvider } from "@alepha/mqtt"; 2import { $hook, $inject } from "alepha"; 3  4class Gateway { 5  mqtt = $inject(MqttClientProvider); 6  7  onStart = $hook({ 8    on: "start", 9    handler: async () => {10      await this.mqtt.subscribe(11        "plant/+/temperature",12        (message) => {13          const celsius = Number(new TextDecoder().decode(message.payload));14          console.log(message.topic, celsius);15        },16        { qos: 1 },17      );18 19      await this.mqtt.publish("plant/line-1/command", "start", { qos: 1 });20    },21  });22}

• Payloads are bytes. publish takes a string or a Uint8Array, and a callback receives payload as a Uint8Array: decode JSON yourself, or hand protobuf bytes straight to their decoder.
• Awaits mean something. subscribe resolves on the broker's SUBACK, so a message published after it cannot be missed. At QoS 1 or 2, publish resolves on the broker's acknowledgement; at QoS 0 (the default) only on the write.
• Wildcards: + matches one level, # the rest. message.topic is the topic the message was published to.
• Retained: { retain: true } stores the message on the broker for every later subscriber (message.retain is true on that delivery), and an empty retained payload clears it.
• subscribe resolves to a function that removes that one callback. Callbacks on the same filter share one broker subscription, which goes with the last of them. unsubscribe(filter) drops every callback on it at once. A later subscribe on the same filter with a higher QoS raises the subscription's QoS for all of them.
• Errors: a published topic holding + or # throws. A subscription the broker refuses (an ACL, an invalid filter) rejects subscribe. A callback that throws is logged, and the other callbacks still run.

#Shared subscriptions

Several instances of one app, each subscribed to orders/+, all receive every order. Subscribe to $share/<group>/orders/+ instead and the broker hands each message to one member of the group:

1await this.mqtt.subscribe("$share/billing/orders/+", handle, { qos: 1 });

A member is a client connection: every callback this instance holds on that filter runs for the copy the broker gives it. Shared subscriptions need MQTT 5 and are refused under protocolVersion: 4.

Each subscription carries its own MQTT 5 Subscription Identifier, and a message reaches only the subscriptions it was delivered for. So an instance holding $share/billing/orders/+ and a plain orders/+ runs each callback once per message it was meant to get, never the group's twice.

#The Last Will

A will is what the broker publishes when the client disappears without a DISCONNECT: a crash, a cut cable, a lost keepalive. The simple case is the atom:

1alepha.set(mqttOptions, {2  ...alepha.get(mqttOptions),3  will: {4    topic: "gateways/gateway-7/online",5    payload: "false",6    qos: 1,7    retain: true,8  },9});

When the will is bytes, or changes between connections, set it by code:

1this.mqtt.setWill({2  topic: "gateways/gateway-7/death",3  payload: encodeDeath({ sequence: this.sequence }),4  qos: 1,5});

The broker reads the will from the CONNECT packet, so the open connection keeps the will it connected with: setWill() applies to the next connect or reconnect. Call it before start for the first connection. setWill(undefined) clears it. Once called, it wins over the atom.

#Device topics

Business code does not hold topic strings: it declares $topics, and @alepha/mqtt/topic carries them over MQTT. A device topic names MqttTopicProvider:

 1import { AlephaTopicMqtt, MqttTopicProvider } from "@alepha/mqtt/topic"; 2import { z } from "alepha"; 3import { $subscriber, $topic } from "alepha/topic"; 4  5class DeviceStatus { 6  status = $topic({ 7    name: "mcs/v1/{site}/{device}/status", 8    provider: MqttTopicProvider, 9    schema: {10      params: z.object({ site: z.text(), device: z.text() }),11      payload: z.object({ state: z.enum(["ok", "degraded", "down"]) }),12    },13    mqtt: { qos: 1, share: "mcs" },14  });15 16  onStatus = $subscriber({17    topic: this.status,18    handler: async ({ params, payload }) => {19      // params.site, params.device: from the topic the message arrived on.20      console.log(params.device, payload.state);21    },22  });23}

Register AlephaTopicMqtt (it brings AlephaMqtt and alepha/topic).

• Opt in per topic. MqttTopicProvider is never the app's TopicProvider: the framework's own topics (WebSocket broadcasts, parameter:sync) stay on memory or Redis, off the device broker.
• Exact names. The topic name is the contract with the devices, so nothing is added to it. The mqttTopicOptions atom's prefix (empty by default) is joined with / when an app does want one.
• Params are + wildcards. A subscriber on mcs/v1/{site}/{device}/status subscribes to mcs/v1/+/+/status and receives the params of each message's topic. publish({ params, payload }) fills them in. wait() works on a templated name too, and its filter sees the params.
• The wire format is the payload alone, as UTF-8 JSON: {"state":"ok"}, which is what a device publishes. Memory and Redis wrap it in { "payload": ... }; MQTT does not. A message that does not decode, or does not match the schema, is logged and dropped.
• QoS 1 by default, so an awaited publish() means the broker has the message. mqtt: { qos } changes it for both publishing and subscribing. retain: true on the $topic publishes retained.

#Shared groups and idempotent handlers

Without share, every instance of the app receives every device message, which is right for a notification fan-out and wrong for ingestion: three instances write each status three times. mqtt: { share: "mcs" } makes the instances one group, and the broker hands each message to one of them. On that instance, the topic's handler and every $subscriber run. wait() never joins the group, so a one-shot listener does not steal a message from the group's handlers.

QoS 1 is at least once: after a reconnect, a broker may deliver a message again, shared group or not. Handlers on device topics must be idempotent: key a write on the device and a sequence number or timestamp the device sends, not on the arrival.

#Commands: request and response

A command to a device needs an answer: did the reboot happen, was the configuration applied. Give the topic a response schema and call request():

 1import { MqttTopicProvider } from "@alepha/mqtt/topic"; 2import { z } from "alepha"; 3import { $topic } from "alepha/topic"; 4  5class DeviceCommands { 6  command = $topic({ 7    name: "mcs/v1/{site}/{device}/cmd", 8    provider: MqttTopicProvider, 9    schema: {10      params: z.object({ site: z.text(), device: z.text() }),11      payload: z.object({ action: z.enum(["reboot", "out-of-service"]) }),12      response: z.object({13        accepted: z.boolean(),14        reason: z.text().optional(),15      }),16    },17  });18 19  async reboot(site: string, device: string) {20    const result = await this.command.request(21      { params: { site, device }, payload: { action: "reboot" } },22      { timeout: [10, "seconds"] },23    );24    return result.accepted;25  }26}

request() takes what publish() takes, and resolves with the typed reply. On a topic without response it does not compile.

On MQTT it is plain MQTT 5. The request carries a Response Topic and Correlation Data; the device publishes its answer, bare JSON like every message, to that Response Topic with the same Correlation Data:

Any MQTT 5 device can answer without knowing Alepha is on the other end. The reply topic is one per instance, <app name>/reply/<instance id>, subscribed on the first request. It is never prefixed, since it is an address on the wire: set replyTopic in the mqttTopicOptions atom to a topic your broker ACL lets devices publish to. Requests need MQTT 5, and are refused under protocolVersion: 4.

The responder is any handler of the topic. On a topic with a response schema, what the handler or a $subscriber returns is the reply:

1command = $topic({2  name: "mcs/v1/{site}/{device}/cmd",3  provider: MqttTopicProvider,4  schema: { params, payload, response },5  handler: async ({ params, payload }) => {6    await this.device.apply(params.device, payload.action);7    return { accepted: true };8  },9});

A message that came from a plain publish() has no reply address and gets no answer. A handler that throws, or returns what the response schema refuses, sends nothing: the requester times out. With several responders, the first reply settles the request and the others are dropped.

Timeouts and cancellation. Past its timeout (10 seconds by default) a request rejects with the 503 TimeoutError every Alepha timeout throws, and travel() fires it in tests. Inside a request handler, a client that disconnects aborts the pending request too. Either way nothing is left waiting.

The same contract works on the memory and Redis topic providers, emulated in their envelope, so a command topic's business logic tests without a broker.

#Testing on the memory broker

Without MQTT_URL, tests run on MemoryMqttClientProvider: no broker, no socket, no timing. publish resolves once every subscriber's callback has run, so a spec asserts right after it.

Two containers, a device and the app under test, share a broker with useBroker(), before they start:

 1import { 2  AlephaMqtt, 3  MemoryMqttBroker, 4  MemoryMqttClientProvider, 5} from "@alepha/mqtt"; 6import { Alepha } from "alepha"; 7  8const device = Alepha.create().with(AlephaMqtt); 9const app = Alepha.create().with(AlephaMqtt);10 11app.inject(MemoryMqttClientProvider).useBroker(device.inject(MemoryMqttBroker));12 13await device.start();14await app.start();

simulateDisconnect() drops the connection as a network cut would: the broker publishes the will, then the client reconnects and restores its subscriptions (unless reconnectPeriod is 0). broker.retainedMessage(topic) reads what the broker has retained.

The memory broker keeps the semantics a real one has (wildcards, $ topics, retained messages, the delivered QoS, shared groups round robin, the will read at connect), and the framework proves it: one contract suite runs on the memory broker and on Mosquitto, with the same assertions.

#See also

• @alepha/mqtt and @alepha/mqtt/topic
• alepha/topic