MQTT in Vue 3 mit MQTT.js: ein Composable für Echtzeitdaten

Der saubere Weg zu MQTT in Vue 3 ist ein Composable: ein gemeinsamer MQTT.js-Client über sichere WebSockets, eine Funktion useMqtt(), die den Verbindungsstatus reaktiv bereitstellt und eine publish-Methode anbietet, und eine Funktion useTopic(), die abonniert, solange eine Komponente lebt, und beim Zerstören automatisch abbestellt. Nachrichten landen in shallowRef-State, damit Vue nicht jeden Payload tief proxyt.

Dieses Tutorial baut das Composable, nutzt es in einer <script setup>-Komponente und zeigt, wie du es gegen einen echten Broker testest.

Installation und Verbindungseinstellungen

npm i mqtt

MQTT.js 5 enthält TypeScript-Typen und einen Browser-Build, der ohne Konfiguration mit Vite funktioniert. Browser sprechen nur MQTT über WebSockets, die URL ist also immer ws:// oder wss://:

Nutze wss:// auf jeder Seite, die über HTTPS ausgeliefert wird; ws:// wird als Mixed Content blockiert. MQTT über WebSockets erklärt den Transport, und eine WebSocket-Verbindung testen hilft, wenn der Handshake scheitert.

Das Composable

Der Client lebt im Modul-Scope, also teilen sich alle Komponenten, die useMqtt() aufrufen, eine Verbindung. Ein einfacher Zähler schließt die Verbindung, wenn der letzte Nutzer verschwindet:

import { onScopeDispose, readonly, ref, shallowRef, toValue, watch, type MaybeRefOrGetter } from 'vue';
import mqtt, { type MqttClient } from 'mqtt';

type Status = 'idle' | 'connecting' | 'connected' | 'reconnecting' | 'offline' | 'error';
type Handler = (topic: string, payload: Uint8Array) => void;

const URL = import.meta.env.VITE_MQTT_URL ?? 'wss://broker.connectmq.com:8084/mqtt';

let client: MqttClient | null = null;
let users = 0;
const status = ref<Status>('idle');
const lastError = ref<string | null>(null);
const handlers = new Map<string, Set<Handler>>();

export function matches(filter: string, topic: string): boolean {
  const f = filter.split('/');
  const t = topic.split('/');
  for (let i = 0; i < f.length; i++) {
    if (f[i] === '#') return true;
    if (i >= t.length || (f[i] !== '+' && f[i] !== t[i])) return false;
  }
  return f.length === t.length;
}

function getClient(): MqttClient {
  if (client) return client;
  status.value = 'connecting';
  client = mqtt.connect(URL, {
    username: import.meta.env.VITE_MQTT_USER,
    password: import.meta.env.VITE_MQTT_PASS,
    clientId: `vue-${crypto.randomUUID().slice(0, 8)}`,
    reconnectPeriod: 2000,
    connectTimeout: 10_000,
  });
  client.on('connect', () => { status.value = 'connected'; lastError.value = null; });
  client.on('reconnect', () => { status.value = 'reconnecting'; });
  client.on('offline', () => { status.value = 'offline'; });
  client.on('error', (err) => { status.value = 'error'; lastError.value = err.message; });
  client.on('message', (topic, payload) => {
    handlers.forEach((set, filter) => {
      if (matches(filter, topic)) set.forEach((h) => h(topic, payload));
    });
  });
  return client;
}

export function useMqtt() {
  const c = getClient();
  users++;

  onScopeDispose(() => {
    if (--users === 0) {
      c.end(true);
      client = null;
      status.value = 'idle';
    }
  });

  function subscribe(filter: string, handler: Handler): () => void {
    let set = handlers.get(filter);
    if (!set) {
      set = new Set();
      handlers.set(filter, set);
      c.subscribe(filter, { qos: 1 });
    }
    set.add(handler);
    return () => {
      set!.delete(handler);
      if (set!.size === 0) {
        handlers.delete(filter);
        c.unsubscribe(filter);
      }
    };
  }

  function publish(topic: string, message: string, retain = false) {
    return c.publishAsync(topic, message, { qos: 1, retain });
  }

  return { status: readonly(status), error: readonly(lastError), subscribe, publish };
}

export interface TopicMessage<T> { topic: string; data: T; at: number }

export function useTopic<T = unknown>(filter: MaybeRefOrGetter<string>, max = 50) {
  const { subscribe } = useMqtt();
  const messages = shallowRef<TopicMessage<T>[]>([]);
  const latest = shallowRef<Record<string, T>>({});

  watch(
    () => toValue(filter),
    (f, _old, onCleanup) => {
      messages.value = [];
      latest.value = {};
      const off = subscribe(f, (topic, payload) => {
        const text = new TextDecoder().decode(payload);
        let data: T;
        try { data = JSON.parse(text) as T; } catch { data = text as T; }
        messages.value = [{ topic, data, at: Date.now() }, ...messages.value].slice(0, max);
        latest.value = { ...latest.value, [topic]: data };
      });
      onCleanup(off);
    },
    { immediate: true },
  );

  return { messages, latest };
}

Warum es so geschrieben ist:

• Der Client ist nie reaktiv. Er liegt in einem einfachen let. Ein MqttClient in ref() oder reactive() würde einen großen Event Emitter tief proxyen – ohne Nutzen. Legst du ihn in einen Pinia-Store, umschließe ihn mit markRaw().
• shallowRef für Nachrichten. Vue verfolgt nur die Referenz, nicht jedes Feld jedes Payloads. Jedes Update weist ein neues Array oder Objekt zu – genau das löst das Re-Render aus.
• Cleanup hängt am Effect Scope. onScopeDispose und das onCleanup des Watchers laufen, wenn die Komponente unmountet – bei onCleanup auch, wenn sich der Topic-Filter ändert. So bleiben weder Handler noch Broker-Subscriptions übrig.
• Subscriptions sind pro Filter referenzgezählt. Zwei Komponenten mit demselben Topic teilen sich eine MQTT-Subscription. MQTT.js abonniert nach einem Reconnect automatisch neu, und Aufrufe vor dem Verbindungsaufbau werden gepuffert und nach dem Connect gesendet.

Da useTopic eine Ref oder einen Getter akzeptiert, kann eine Komponente das Topic reaktiv wechseln: Der Watcher bestellt den alten Filter ab und abonniert den neuen. Prüfe Filter mit Wildcards mit dem Topic Matcher.

Einsatz in einer Komponente

<script setup lang="ts">
import { computed, ref } from 'vue';
import { useMqtt, useTopic } from '@/composables/mqtt';

interface Reading { t: number; h?: number }

const room = ref('lab');
const { status, error, publish } = useMqtt();
const { latest, messages } = useTopic<Reading>(() => `testmqtt/vue/${room.value}/+`);

const rows = computed(() =>
  Object.entries(latest.value).map(([topic, r]) => ({ device: topic.split('/').pop(), t: r.t })),
);

function ping() {
  publish(`testmqtt/vue/${room.value}/cmd/ping`, String(Date.now()));
}
</script>

<template>
  <section>
    <p>Broker: {{ status }} <span v-if="error">({{ error }})</span></p>
    <select v-model="room">
      <option value="lab">lab</option>
      <option value="office">office</option>
    </select>
    <table>
      <tr v-for="row in rows" :key="row.device">
        <td>{{ row.device }}</td>
        <td>{{ row.t.toFixed(1) }} °C</td>
      </tr>
    </table>
    <button :disabled="status !== 'connected'" @click="ping">Ping</button>
    <p>{{ messages.length }} messages received</p>
  </section>
</template>

Ein Wechsel im <select> stellt die Subscription von testmqtt/vue/lab/+ auf testmqtt/vue/office/+ um, ohne die Verbindung anzufassen.

Bei sehr aktiven Topics löst weiterhin jede Nachricht ein Render aus. Treffen Dutzende Nachrichten pro Sekunde dasselbe Panel, puffere sie in einem einfachen Array und schreibe sie einmal pro requestAnimationFrame in das shallowRef, oder drossle mit einem Helper wie useThrottleFn aus VueUse.

Zugangsdaten in einer Vue-App

VITE_*-Variablen werden ins Bundle kompiliert und sind für jeden sichtbar, der die DevTools öffnet. Behandle Browser-Zugangsdaten als öffentlich:

• Lege einen eigenen Browser-Benutzer an, der die Dashboard-Topics abonnieren und bei Bedarf auf ein einziges Befehls-Topic publizieren darf.
• Für Zugriff pro Benutzer lässt du dein Backend nach dem Login kurzlebige Zugangsdaten ausstellen und übergibst sie an mqtt.connect (die Optionen aus getClient() in einen Parameter verschieben).
• Liefere niemals den Account aus, den deine Backend-Services nutzen.

Browser-Clients authentifizieren vergleicht die Varianten im Detail.

Das Composable testen

1. Starte den Dev-Server und prüfe, dass das Panel Broker: connected zeigt. Siehst du error, vergleiche die Meldung: Connection refused: Not authorized deutet auf Zugangsdaten hin, ein WebSocket-Fehler auf URL, Port 8084 oder Pfad /mqtt.
2. Öffne den Online-MQTT-Client, verbinde dich mit demselben Broker und publiziere {"t": 22.1} auf testmqtt/vue/lab/sensor-1. Die Zeile erscheint sofort.
3. Stelle das Select auf office und publiziere erneut auf testmqtt/vue/lab/sensor-1 – es sollte sich nichts ändern. Publiziere auf testmqtt/vue/office/sensor-9, und die neue Zeile erscheint. Das beweist, dass die alte Subscription aufgeräumt wurde.
4. Abonniere im Online-Client testmqtt/vue/# und klicke auf Ping, um das Publizieren zu prüfen.
5. Wechsle zu einer Route ohne das Panel und bestätige im Broker-Log oder in der Client-Liste, dass die Verbindung geschlossen wird, sobald keine Komponente sie mehr nutzt.

Testnachrichten an deine Vue-App senden

Für automatisierte Tests hältst du matches() und das Parsing als reine Funktionen und testest sie mit Vitest; für die Verbindung selbst findet ein kleiner Integrationstest gegen einen echten Broker deutlich mehr als ein gemockter Client. Verbindet sich die App in einer Schleife neu, lies Reconnect bei MQTT über WebSockets in JavaScript und die MQTT.js-Fehlerreferenz.

Fehlerbehebung

Ein Composable hält MQTT dort, wo Vue State erwartet: eine Verbindung, reaktiver Status und Subscriptions, die dem Lebenszyklus der Komponente folgen. Teste es mit einem zweiten Client, der echte Nachrichten publiziert, und derselbe Code funktioniert unverändert, wenn du ihn auf deinen Produktions-Broker richtest.

Häufig gestellte Fragen