Vue 3'te MQTT.js ile MQTT: gerçek zamanlı veri için bir composable

Vue 3'te MQTT kullanmanın temiz yolu bir composable'dır: güvenli WebSocket üzerinden paylaşılan tek bir MQTT.js client'ı, reaktif bağlantı durumunu ve bir publish metodunu sunan bir useMqtt() fonksiyonu ve bileşen yaşadığı sürece abone olup yok edildiğinde aboneliği otomatik iptal eden bir useTopic() fonksiyonu. Mesajlar shallowRef state'ine gider; böylece Vue her payload için derin proxy oluşturmaz.

Bu eğitim o composable'ı kuruyor, bir <script setup> bileşeninde kullanıyor ve gerçek bir broker'a karşı nasıl test edeceğinizi gösteriyor.

Kurulum ve bağlantı ayarları

npm i mqtt

MQTT.js 5, TypeScript tiplerini ve Vite ile ek ayar gerektirmeden çalışan bir tarayıcı build'ini içerir. Tarayıcılar yalnızca WebSocket üzerinden MQTT konuşur; bu yüzden URL her zaman ws:// veya wss:// olur:

HTTPS üzerinden sunulan her sayfada wss:// kullanın; ws:// mixed content olarak engellenir. WebSocket üzerinden MQTT transport'u açıklar; handshake başarısız olduğunda WebSocket bağlantısını test etmek yazısı yardımcı olur.

Composable

Client modül kapsamında yaşar; dolayısıyla useMqtt() çağıran tüm bileşenler tek bir bağlantıyı paylaşır. Basit bir sayaç, son kullanıcı ayrıldığında bağlantıyı kapatır:

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

Neden böyle yazıldı:

• Client asla reaktif değildir. Düz bir let içinde durur. Bir MqttClient'ı ref() veya reactive() ile sarmak, büyük bir event emitter'ı hiçbir fayda sağlamadan derin proxy'ye çevirir. Pinia store'una koyacaksanız markRaw() ile sarın.
• Mesajlar için shallowRef. Vue her payload'un her alanını değil, yalnızca referansı izler. Her güncelleme yeni bir dizi veya nesne atar; yeniden render'ı tetikleyen de tam olarak budur.
• Temizlik effect scope'a bağlıdır. onScopeDispose ve watcher'ın onCleanup'ı bileşen unmount olduğunda (ve onCleanup için topic filtresi değiştiğinde) çalışır; böylece ne handler'lar ne de broker abonelikleri sızıntı yapar.
• Abonelikler filtre başına referans sayımlıdır. Aynı topic'i gösteren iki bileşen tek bir MQTT aboneliğini paylaşır. MQTT.js yeniden bağlandıktan sonra otomatik olarak tekrar abone olur; bağlantı kurulmadan yapılan çağrılar kuyruğa alınır ve bağlanınca gönderilir.

useTopic bir ref veya getter kabul ettiğinden bir bileşen topic'i reaktif olarak değiştirebilir: watcher eski filtrenin aboneliğini iptal eder ve yenisine abone olur. Wildcard kullandığınızda filtrelerinizi topic matcher ile kontrol edin.

Bir bileşende kullanım

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

<select> değiştirildiğinde abonelik, bağlantıya dokunmadan testmqtt/vue/lab/+'dan testmqtt/vue/office/+'a geçer.

Çok yoğun topic'lerde her mesaj yine bir render'a yol açar. Aynı panele saniyede onlarca mesaj geliyorsa bunları düz bir dizide biriktirin ve requestAnimationFrame başına bir kez shallowRef'e aktarın ya da VueUse'un useThrottleFn gibi bir yardımcısıyla hızı sınırlayın.

Bir Vue uygulamasında kimlik bilgileri

VITE_* değişkenleri bundle'a derlenir ve DevTools'u açan herkes tarafından görülebilir. Tarayıcıdaki kimlik bilgilerini herkese açık sayın:

• Dashboard topic'lerine abone olabilen ve gerekirse tek bir komut topic'ine publish edebilen tarayıcıya özel bir kullanıcı oluşturun.
• Kullanıcı bazlı erişim için backend'inizin girişten sonra kısa ömürlü kimlik bilgileri üretmesini sağlayın ve bunları mqtt.connect'e verin (seçenekleri getClient()'tan bir parametreye taşıyın).
• Backend servislerinizin kullandığı hesabı asla tarayıcıya göndermeyin.

Tarayıcı client'larında kimlik doğrulama seçenekleri ayrıntılı olarak karşılaştırır.

Composable'ı test etmek

1. Geliştirme sunucusunu başlatın ve panelin Broker: connected gösterdiğini kontrol edin. error görüyorsanız mesaja bakın: Connection refused: Not authorized kimlik bilgilerine; bir WebSocket hatası ise URL'ye, 8084 portuna veya /mqtt path'ine işaret eder.
2. Online MQTT client'ı açın, aynı broker'a bağlanın ve testmqtt/vue/lab/sensor-1 topic'ine {"t": 22.1} publish edin. Satır anında görünür.
3. Select'i office olarak değiştirin ve tekrar testmqtt/vue/lab/sensor-1 topic'ine publish edin: hiçbir şey değişmemelidir. testmqtt/vue/office/sensor-9 topic'ine publish edin; yeni satır görünür. Bu, eski aboneliğin temizlendiğini kanıtlar.
4. Online client'ta testmqtt/vue/# topic'ine abone olun ve publish işlemini doğrulamak için Ping'e tıklayın.
5. Paneli içermeyen bir route'a geçin ve hiçbir bileşen kullanmadığında bağlantının kapandığını broker logunda veya client listesinde doğrulayın.

Vue uygulamanıza test mesajları gönderin

Otomatik testler için matches() ve ayrıştırma mantığını saf fonksiyonlar olarak tutun ve Vitest ile test edin; bağlantının kendisi için gerçek bir broker'a karşı küçük bir entegrasyon testi, mock'lanmış bir client'tan çok daha fazlasını yakalar. Uygulama döngü halinde yeniden bağlanıyorsa JavaScript'te WebSocket üzerinden MQTT yeniden bağlanma yazısını ve MQTT.js hata referansını okuyun.

Sorun giderme

Bir composable, MQTT'yi Vue'nun state'i beklediği yerde tutar: tek bağlantı, reaktif durum ve bileşen yaşam döngüsünü izleyen abonelikler. Gerçek mesajlar publish eden ikinci bir client ile test edin; aynı kod, üretim broker'ınıza yönlendirdiğinizde de değişmeden çalışacaktır.

Sıkça sorulan sorular