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

The client connects on start and disconnects on stop. Everything else is the 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.

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

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