MQTT aus Edge Functions (Cloudflare Workers, Vercel)

Die meisten Edge-Runtimes haben keine rohen TCP-Sockets und keinen Node.js-Netzwerkstack – die üblichen MQTT-Client-Bibliotheken laufen dort also nicht. Was sie haben, ist die WebSocket-API, und MQTT läuft über WebSockets einwandfrei. Ein Cloudflare Worker kann wss://broker.connectmq.com:8084/mqtt öffnen, ein CONNECT senden, publishen und schließen – alles innerhalb eines Requests. Auf Vercel läuft in der Node.js-Runtime ein normaler MQTT-Client; der Edge-Trick ist also nur für reine Edge-Runtimes nötig.

Dieses Tutorial baut einen abhängigkeitsfreien MQTT-Publisher für Cloudflare Workers, erklärt ctx.waitUntil, listet die relevanten Plattformlimits auf und zeigt, wie du jede Nachricht im Browser prüfst.

Was Edge-Runtimes können und was nicht

Die praktische Regel: Edge Functions sind Publisher, keine Subscriber. Sie reagieren auf einen HTTP-Request, senden eine oder wenige MQTT-Nachrichten und sind fertig. Alles, was dauerhaft zuhören muss, gehört in einen langlebigen Service.

Auf Vercel ist die einfachste Lösung, für die Route, die MQTT spricht, die Node.js-Runtime zu nehmen und sie wie jede andere Serverless-Funktion zu behandeln – die Muster zur Verbindungswiederverwendung aus Publishen aus AWS Lambda gelten direkt. Vercel selbst lenkt die meisten Workloads heute ohnehin auf die Node.js-Runtime.

Ein minimaler MQTT-Client über WebSocket

Für ein Publish mit QoS 0 braucht es nur drei Pakete: CONNECT, PUBLISH und DISCONNECT, dazu das Lesen des CONNACK. Das Byte-Layout ist einfach – ein Fixed Header, ein Feld „Remaining Length“ mit variabler Länge und UTF-8-Strings mit Längenpräfix (Details unter MQTT-Paketstruktur).

const enc = new TextEncoder();

// UTF-8-String mit 2-Byte-Big-Endian-Längenpräfix
const str = (s) => { const b = enc.encode(s); return [b.length >> 8, b.length & 0xff, ...b]; };

// MQTT-"Remaining Length" mit variabler Länge
function remLen(n) {
  const out = [];
  do {
    let digit = n % 128;
    n = Math.floor(n / 128);
    if (n > 0) digit |= 0x80;
    out.push(digit);
  } while (n > 0);
  return out;
}

const packet = (header, body) => new Uint8Array([header, ...remLen(body.length), ...body]);

function connectPacket(clientId, username, password, keepAlive = 30) {
  let flags = 0x02;                       // Clean Session
  if (username) flags |= 0x80;
  if (password) flags |= 0x40;
  const body = [...str('MQTT'), 4, flags, keepAlive >> 8, keepAlive & 0xff, ...str(clientId)];
  if (username) body.push(...str(username));
  if (password) body.push(...str(password));
  return packet(0x10, body);              // MQTT-3.1.1-CONNECT
}

const publishPacket = (topic, payload) => packet(0x30, [...str(topic), ...enc.encode(payload)]);
const DISCONNECT = new Uint8Array([0xe0, 0x00]);

export async function publishMqtt(env, topic, payload) {
  const resp = await fetch(env.MQTT_WS_URL, {
    headers: { Upgrade: 'websocket', 'Sec-WebSocket-Protocol': 'mqtt' },
  });
  const ws = resp.webSocket;
  if (!ws) throw new Error(`WebSocket upgrade failed: HTTP ${resp.status}`);
  ws.accept();

  const connack = new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error('CONNACK timeout')), 5000);
    ws.addEventListener('message', (ev) => {
      const b = new Uint8Array(ev.data);
      if (b[0] !== 0x20) return;
      clearTimeout(timer);
      b[3] === 0 ? resolve() : reject(new Error(`CONNACK return code ${b[3]}`));
    });
    ws.addEventListener('close', () => reject(new Error('socket closed before CONNACK')));
  });

  ws.send(connectPacket(`edge-${crypto.randomUUID().slice(0, 8)}`, env.MQTT_USER, env.MQTT_PASS));
  await connack;
  ws.send(publishPacket(topic, payload));
  ws.send(DISCONNECT);
  ws.close(1000, 'done');
}

Ein paar Details gehen leicht schief:

• Der Subprotocol-Header. Die meisten Broker akzeptieren das Upgrade nur, wenn der Client Sec-WebSocket-Protocol: mqtt anbietet. Ohne ihn bekommst du meist ein HTTP 400, oder der Socket schließt sofort.
• Der Pfad. Der WebSocket-Listener liegt bei ConnectMQ unter /mqtt; ein falscher Pfad liefert HTTP 404 (WebSocket-Pfad 404).
• Zufällige Client-IDs. Hunderte Edge-Standorte können deinen Worker im selben Moment ausführen; eine feste ID lässt sie sich gegenseitig trennen.
• Binäre Frames. MQTT über WebSockets nutzt binäre Nachrichten, deshalb sendet der Code Uint8Arrays.

Brauchst du eine Zustellbestätigung, erweitere publishPacket auf QoS 1 (Header 0x32 plus zwei Byte Packet Identifier nach dem Topic) und warte vor dem Schließen auf das PUBACK (0x40). Halte Payloads klein: Der Paketgrößen-Rechner zeigt die exakten Bytes auf der Leitung.

Der Worker

import { publishMqtt } from './mqtt.js';

export default {
  async fetch(request, env, ctx) {
    const url = new URL(request.url);
    const device = url.searchParams.get('device') ?? 'demo';
    const message = JSON.stringify({ device, path: url.pathname, colo: request.cf?.colo, ts: Date.now() });

    // Sofort antworten; den MQTT-Publish im Hintergrund abschließen
    ctx.waitUntil(
      publishMqtt(env, `testmqtt/edge/${device}/hits`, message)
        .catch((err) => console.error('mqtt publish failed:', err.message)),
    );
    return new Response('queued\n');
  },
};

name = "edge-mqtt"
main = "src/index.js"
compatibility_date = "2026-10-01"

[vars]
MQTT_WS_URL = "https://broker.connectmq.com:8084/mqtt"

Speichere die Credentials als Secrets, nie in wrangler.toml: npx wrangler secret put MQTT_USER und npx wrangler secret put MQTT_PASS. Beachte, dass die URL https:// verwendet – das fetch der Workers führt das WebSocket-Upgrade durch, sobald es den Upgrade-Header sieht.

ctx.waitUntil hält den Worker nach dem Senden der Antwort am Leben, bis das Promise erledigt ist; der Besucher wartet also nicht auf den Broker-Handshake. Muss der Aufrufer wissen, ob der Publish geklappt hat, nutze vor der Antwort await publishMqtt(...) und gib bei einem Fehler 502 zurück.

Rohes TCP mit cloudflare:sockets

Cloudflare Workers können mit connect() aus cloudflare:sockets auch ausgehende TCP-Verbindungen öffnen, inklusive TLS über secureTransport: "on". Die Paketfunktionen oben funktionieren unverändert – schreib sie in socket.writable statt mit ws.send und lies das CONNACK aus socket.readable. So kannst du Port 8883 statt WebSockets nutzen. WebSockets bleiben trotzdem die portablere Wahl, weil derselbe Code auf anderen Edge-Plattformen und im Browser läuft (MQTT über WebSockets).

Limits, die das Design bestimmen

Genaue Werte hängen von Plattform und Tarif ab und ändern sich mit der Zeit – prüfe die aktuelle Doku. Für MQTT zählen diese Kategorien:

• CPU- und Wall-Clock-Zeit pro Request. Ein TLS- + WebSocket- + MQTT-Handshake kostet mehrere Netzwerk-Roundtrips; publishe lieber einen Batch pro Request als eine Verbindung pro Nachricht.
• Keine verlässliche Wiederverwendung von Verbindungen. Isolates werden eventuell wiederverwendet, behandle aber jeden Request als neue Verbindung.
• Limits für ausgehende Verbindungen und Subrequests. Jeder WebSocket zählt dazu.
• Ausgehende Ports. Manche Plattformen schränken ein, welche Ports ausgehende Requests nutzen dürfen. Ist 8084 nicht erlaubt, nimm einen Broker-Listener auf 443 (MQTT auf Port 443).
• Verbindungsrate am Broker. Eine Traffic-Spitze auf deiner Website wird zu einer Spitze an MQTT-Verbindungen. Bei hohem Ereignisvolumen lohnt es sich, die Events an einen aggregierenden Service zu schicken, der eine einzige Verbindung offen hält.

Häufige Fehler

Weitere WebSocket-spezifische Fehler behandelt WebSocket connection failed.

Testen

1. Öffne den Online-MQTT-Client, verbinde dich über sichere WebSockets mit broker.connectmq.com und abonniere testmqtt/edge/#.
2. Starte den Worker lokal mit npx wrangler dev und ruf ihn auf: curl "http://localhost:8787/checkout?device=kiosk-1".
3. Im Browser sollte eine Nachricht mit Gerätename und Zeitstempel erscheinen. Falls nicht, findest du die Fehlermeldung von publishMqtt in der Konsole von wrangler dev.
4. Deploye mit npx wrangler deploy und wiederhole den Test gegen die workers.dev-URL; das Feld colo zeigt jetzt, welcher Edge-Standort die Nachricht gepublisht hat.

Edge-Nachrichten live im Browser ansehen

Edge Functions und MQTT treffen sich beim WebSocket: Rund fünfzig Zeilen Paket-Encoding machen aus jeder Edge-Runtime einen MQTT-Publisher ohne Abhängigkeiten. Halte die Funktionen als kurzlebige Publisher, schütze ihre Credentials mit engen ACLs und überlass langlebige Subscriptions Services, die verbunden bleiben können.

Häufig gestellte Fragen