MQTT avec Vue 3 et MQTT.js : un composable pour les données temps réel

La façon propre d'utiliser MQTT dans Vue 3 est un composable : un seul client MQTT.js partagé sur des WebSockets sécurisés, une fonction useMqtt() qui expose l'état de connexion réactif et une méthode publish, et une fonction useTopic() qui s'abonne tant que le composant existe et se désabonne automatiquement à sa destruction. Les messages vont dans un état shallowRef, pour que Vue ne crée pas de proxy profond sur chaque payload.

Ce tutoriel construit ce composable, l'utilise dans un composant <script setup> et montre comment le tester avec un vrai broker.

Installation et paramètres de connexion

npm i mqtt

MQTT.js 5 inclut des types TypeScript et un build navigateur qui fonctionne avec Vite sans configuration. Les navigateurs ne parlent que MQTT sur WebSockets : l'URL est donc toujours ws:// ou wss:// :

Utilisez wss:// sur toute page servie en HTTPS ; ws:// est bloqué comme contenu mixte. MQTT sur WebSockets explique le transport, et tester une connexion WebSocket aide quand le handshake échoue.

Le composable

Le client vit dans la portée du module : tous les composants qui appellent useMqtt() partagent une seule connexion. Un simple compteur ferme la connexion quand le dernier utilisateur disparaît :

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

Pourquoi c'est écrit ainsi :

• Le client n'est jamais réactif. Il est dans un simple let. Envelopper un MqttClient dans ref() ou reactive() créerait un proxy profond sur un gros event emitter sans aucun bénéfice. Si vous le mettez dans un store Pinia, entourez-le de markRaw().
• shallowRef pour les messages. Vue ne suit que la référence, pas chaque champ de chaque payload. Chaque mise à jour assigne un nouveau tableau ou objet, ce qui déclenche justement le re-render.
• Le nettoyage est lié à l'effect scope. onScopeDispose et le onCleanup du watcher s'exécutent quand le composant est démonté (et, pour onCleanup, quand le filtre de topic change) : ni les handlers ni les abonnements broker ne fuient.
• Les abonnements sont comptés par filtre. Deux composants affichant le même topic partagent un abonnement MQTT. MQTT.js se réabonne automatiquement après une reconnexion, et les appels faits avant la connexion sont mis en file puis envoyés dès qu'elle est établie.

Comme useTopic accepte un ref ou un getter, un composant peut changer de topic de manière réactive : le watcher se désabonne de l'ancien filtre et s'abonne au nouveau. Vérifiez vos filtres avec wildcards dans le topic matcher.

Utilisation dans un composant

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

Changer le <select> fait passer l'abonnement de testmqtt/vue/lab/+ à testmqtt/vue/office/+ sans toucher à la connexion.

Sur des topics très actifs, chaque message provoque toujours un render. Si des dizaines de messages par seconde arrivent sur le même panneau, accumulez-les dans un tableau simple et versez-les dans le shallowRef une fois par requestAnimationFrame, ou limitez le débit avec un helper comme useThrottleFn de VueUse.

Identifiants dans une app Vue

Les variables VITE_* sont compilées dans le bundle et visibles par quiconque ouvre les DevTools. Considérez les identifiants du navigateur comme publics :

• Créez un utilisateur dédié au navigateur qui peut s'abonner aux topics du dashboard et, si besoin, publier sur un seul topic de commande.
• Pour un accès par utilisateur, faites émettre par votre backend des identifiants de courte durée après la connexion et passez-les à mqtt.connect (sortez les options de getClient() vers un paramètre).
• Ne livrez jamais le compte utilisé par vos services backend.

Authentifier les clients navigateur compare les options en détail.

Tester le composable

1. Lancez le serveur de développement et vérifiez que le panneau affiche Broker: connected. Si vous voyez error, comparez le message : Connection refused: Not authorized désigne les identifiants ; un échec WebSocket, l'URL, le port 8084 ou le chemin /mqtt.
2. Ouvrez le client MQTT en ligne, connectez-vous au même broker et publiez {"t": 22.1} sur testmqtt/vue/lab/sensor-1. La ligne apparaît immédiatement.
3. Passez le select sur office et publiez de nouveau sur testmqtt/vue/lab/sensor-1 : rien ne doit changer. Publiez sur testmqtt/vue/office/sensor-9 et la nouvelle ligne apparaît. Cela prouve que l'ancien abonnement a été nettoyé.
4. Abonnez-vous à testmqtt/vue/# dans le client en ligne et cliquez sur Ping pour vérifier la publication.
5. Allez sur une route sans le panneau et confirmez dans le log du broker ou la liste des clients que la connexion se ferme quand plus aucun composant ne l'utilise.

Envoyer des messages de test à votre app Vue

Pour les tests automatisés, gardez matches() et le parsing sous forme de fonctions pures et testez-les avec Vitest ; pour la connexion elle-même, un petit test d'intégration contre un vrai broker détecte bien plus de choses qu'un client mocké. Si l'app se reconnecte en boucle, lisez la reconnexion MQTT sur WebSockets en JavaScript et la référence des erreurs MQTT.js.

Dépannage

Un composable place MQTT là où Vue attend l'état : une connexion, un statut réactif et des abonnements qui suivent le cycle de vie du composant. Testez-le avec un second client qui publie de vrais messages, et le même code fonctionnera sans changement quand vous le pointerez vers votre broker de production.

Questions fréquentes