MQTT mit React: Echtzeit-Dashboards mit Hooks

npm i mqtt

import { createContext, useContext, useEffect, useMemo, useRef, useState, type ReactNode } from 'react';
import mqtt, { type IClientOptions, type MqttClient } from 'mqtt';

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

interface MqttApi {
  status: Status;
  error: string | null;
  subscribe: (filter: string, handler: Handler) => () => void;
  publish: (topic: string, message: string) => Promise<void>;
}

const MqttContext = createContext<MqttApi | null>(null);

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;
}

export function MqttProvider({ url, options, children }: { url: string; options?: IClientOptions; children: ReactNode }) {
  const clientRef = useRef<MqttClient | null>(null);
  const handlers = useRef(new Map<string, Set<Handler>>());
  const optionsRef = useRef(options);
  const [status, setStatus] = useState<Status>('connecting');
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    const client = mqtt.connect(url, { reconnectPeriod: 2000, connectTimeout: 10_000, ...optionsRef.current });
    clientRef.current = client;

    client.on('connect', () => { setStatus('connected'); setError(null); });
    client.on('reconnect', () => setStatus('reconnecting'));
    client.on('offline', () => setStatus('offline'));
    client.on('error', (err) => { setStatus('error'); setError(err.message); });
    client.on('message', (topic, payload) => {
      handlers.current.forEach((set, filter) => {
        if (matches(filter, topic)) set.forEach((h) => h(topic, payload));
      });
    });

    // Subscriptions wiederherstellen, die Komponenten vor dieser Verbindung registriert haben.
    for (const filter of handlers.current.keys()) client.subscribe(filter, { qos: 1 });

    return () => {
      client.end(true);
      clientRef.current = null;
    };
  }, [url]);

  const api = useMemo<MqttApi>(() => ({
    status,
    error,
    subscribe(filter, handler) {
      let set = handlers.current.get(filter);
      if (!set) {
        set = new Set();
        handlers.current.set(filter, set);
        clientRef.current?.subscribe(filter, { qos: 1 });
      }
      set.add(handler);
      return () => {
        set!.delete(handler);
        if (set!.size === 0) {
          handlers.current.delete(filter);
          clientRef.current?.unsubscribe(filter);
        }
      };
    },
    async publish(topic, message) {
      await clientRef.current?.publishAsync(topic, message, { qos: 1 });
    },
  }), [status, error]);

  return <MqttContext.Provider value={api}>{children}</MqttContext.Provider>;
}

export function useMqtt(): MqttApi {
  const ctx = useContext(MqttContext);
  if (!ctx) throw new Error('useMqtt must be used inside <MqttProvider>');
  return ctx;
}

import { useEffect, useState } from 'react';
import { useMqtt } from './MqttProvider';

export function useLatest<T>(filter: string): Record<string, T> {
  const { subscribe } = useMqtt();
  const [values, setValues] = useState<Record<string, T>>({});

  useEffect(() => {
    let pending: Record<string, T> = {};
    let frame = 0;

    const off = subscribe(filter, (topic, payload) => {
      try {
        pending[topic] = JSON.parse(new TextDecoder().decode(payload)) as T;
      } catch {
        return; // Payloads ohne JSON ignorieren
      }
      if (!frame) {
        frame = requestAnimationFrame(() => {
          const batch = pending;
          pending = {};
          frame = 0;
          setValues((prev) => ({ ...prev, ...batch }));
        });
      }
    });

    return () => {
      off();
      cancelAnimationFrame(frame);
    };
  }, [filter, subscribe]);

  return values;
}

import { MqttProvider, useMqtt } from './mqtt/MqttProvider';
import { useLatest } from './mqtt/useLatest';

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

function StatusBadge() {
  const { status, error } = useMqtt();
  return <p>Broker: {status}{error ? ` (${error})` : ''}</p>;
}

function Sensors() {
  const readings = useLatest<Reading>('testmqtt/react/sensors/+');
  return (
    <table>
      <tbody>
        {Object.entries(readings).map(([topic, r]) => (
          <tr key={topic}>
            <td>{topic.split('/').pop()}</td>
            <td>{r.t.toFixed(1)} °C</td>
          </tr>
        ))}
      </tbody>
    </table>
  );
}

function PingButton() {
  const { publish, status } = useMqtt();
  return (
    <button disabled={status !== 'connected'} onClick={() => publish('testmqtt/react/ping', String(Date.now()))}>
      Ping
    </button>
  );
}

export default function App() {
  return (
    <MqttProvider
      url="wss://broker.connectmq.com:8084/mqtt"
      options={{
        username: import.meta.env.VITE_MQTT_USER,
        password: import.meta.env.VITE_MQTT_PASS,
        clientId: `dashboard-${crypto.randomUUID().slice(0, 8)}`,
      }}
    >
      <StatusBadge />
      <Sensors />
      <PingButton />
    </MqttProvider>
  );
}

import mqtt from 'mqtt';

const client = await mqtt.connectAsync('mqtts://broker.connectmq.com:8883', {
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
});

setInterval(() => {
  const t = 20 + Math.random() * 5;
  client.publish('testmqtt/react/sensors/sim-1', JSON.stringify({ t }), { qos: 1 });
}, 1000);