mqtt_client ile Dart'ta MQTT (Sunucu Tarafı)

mqtt_client örneklerinin çoğu Flutter uygulamasıdır, ama paket saf Dart'tır: MqttServerClient herhangi bir Dart VM programında çalışır. Bu da Dart'ı MQTT komut satırı araçları, küçük backend servisleri ve tek bir yerel binary'ye derlenen cihaz gateway'leri için iyi bir seçenek yapar.

Bu rehberde iki araç geliştireceğiz — otomatik yeniden bağlanan, sürekli çalışan bir subscriber ve broker'ın onayını bekleyen tek seferlik bir publisher — ardından TLS, WebSocket ve yerel derleme ekleyeceğiz. Her şey broker.connectmq.com adresini hedefliyor; böylece her adımı online MQTT istemcisinde doğrulayabilirsiniz. Bunun yerine bir mobil uygulama geliştiriyorsanız Flutter'da MQTT yazısına bakın.

Projeyi hazırlama

dart create -t console mqtt_cli
cd mqtt_cli
dart pub add mqtt_client

environment:
  sdk: ^3.4.0

dependencies:
  mqtt_client: ^10.2.0

mqtt_client iki istemci sınıfıyla gelir: MqttServerClient (VM için, dart:io üzerinden soketler) ve MqttBrowserClient (web'deki Dart için WebSocket). Sunucuda her zaman MqttServerClient kullanırsınız. Paket MQTT 3.1.1 konuşur; MQTT 5 için kardeş paket mqtt5_client neredeyse aynı API'ye sahiptir — ihtiyacınız olup olmadığına karar vermek için MQTT 5 ile 3.1.1 karşılaştırmasına bakın.

Broker kimlik bilgilerinizi export edin ki asla sürüm kontrolüne girmesin:

export MQTT_USER="username"
export MQTT_PASS="password"

Sürekli çalışan bir subscriber

import 'dart:convert';
import 'dart:io';

import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

Future<void> main() async {
  final clientId = 'testmqtt-dart-sub-$pid';
  final client = MqttServerClient.withPort('broker.connectmq.com', clientId, 1883)
    ..logging(on: false)
    ..setProtocolV311()
    ..keepAlivePeriod = 30
    ..autoReconnect = true
    ..resubscribeOnAutoReconnect = true
    ..onConnected = (() => print('connected'))
    ..onDisconnected = (() => print('disconnected'))
    ..onAutoReconnect = (() => print('connection lost, reconnecting...'))
    ..onAutoReconnected = (() => print('reconnected'))
    ..onSubscribed = ((topic) => print('subscribed to $topic'));

  client.connectionMessage = MqttConnectMessage()
      .withClientIdentifier(clientId)
      .startClean()
      .authenticateAs(Platform.environment['MQTT_USER'], Platform.environment['MQTT_PASS']);

  try {
    await client.connect();
  } on NoConnectionException catch (e) {
    stderr.writeln('broker rejected or ignored CONNECT: $e');
    exit(1);
  } on SocketException catch (e) {
    stderr.writeln('network error: $e');
    exit(1);
  }

  if (client.connectionStatus!.state != MqttConnectionState.connected) {
    stderr.writeln('connect failed: ${client.connectionStatus!.returnCode}');
    exit(1);
  }

  client.subscribe('testmqtt/dart/#', MqttQos.atLeastOnce);

  client.updates!.listen((messages) {
    for (final m in messages) {
      final publish = m.payload as MqttPublishMessage;
      final text = utf8.decode(publish.payload.message, allowMalformed: true);
      print('${m.topic}  $text');
    }
  });

  ProcessSignal.sigint.watch().listen((_) {
    client.disconnect();
    exit(0);
  });
}

Production'da önemli olan birkaç ayrıntı:

• Client ID. pid, ID'yi bir makinede süreç başına benzersiz yapar. Aynı aracı birden fazla host'ta çalıştırıyorsanız hostname'i de ekleyin — aynı ID'ye sahip iki istemci birbirini döngü halinde düşürür.
• updates listeler teslim eder. Her olay genellikle tek elemanlı bir List<MqttReceivedMessage>'dır; her zaman üzerinde döngü kurun.
• utf8 ile çözümleyin. Paketin MqttPublishPayload.bytesToStringAsString metodu baytları tek tek karaktere çevirir ve °C gibi ASCII dışı metni bozar. utf8.decode bunu doğru işler.
• Otomatik yeniden bağlanma. autoReconnect ve resubscribeOnAutoReconnect ile kopan bağlantı yeniden kurulur ve abonelikleriniz geri yüklenir. Genel yeniden bağlanma stratejisi yeniden bağlanma için en iyi uygulamalar yazısında ele alınıyor.

dart run bin/subscriber.dart ile çalıştırın, ardından online istemciden testmqtt/dart/hello topic'ine bir mesaj publish edin. Mesaj saniyenin çok küçük bir kesrinde ekrana basılmalı.

Tarayıcıdan Dart subscriber'ına publish edin

PUBACK bekleyen tek seferlik publisher

Komut satırı publisher'larının klasik bir hatası vardır: publishMessage çağırır ve baytlar daha ağa çıkmadan hemen sonlanırlar. QoS 1 ile sleep kullanmak yerine broker'ın onayını published stream'inde bekleyebilirsiniz:

import 'dart:io';

import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

Future<void> main(List<String> args) async {
  if (args.length != 2) {
    stderr.writeln('usage: publish <topic> <message>');
    exit(64);
  }

  final clientId = 'testmqtt-dart-pub-$pid';
  final client = MqttServerClient.withPort('broker.connectmq.com', clientId, 1883)
    ..logging(on: false)
    ..setProtocolV311();
  client.connectionMessage = MqttConnectMessage()
      .withClientIdentifier(clientId)
      .startClean()
      .authenticateAs(Platform.environment['MQTT_USER'], Platform.environment['MQTT_PASS']);

  try {
    await client.connect();
  } on Exception catch (e) {
    stderr.writeln('connect failed: $e');
    exit(1);
  }

  final payload = (MqttClientPayloadBuilder()..addUTF8String(args[1])).payload!;
  final messageId = client.publishMessage(args[0], MqttQos.atLeastOnce, payload);

  try {
    await client.published!
        .firstWhere((m) => m.variableHeader?.messageIdentifier == messageId)
        .timeout(const Duration(seconds: 10));
    print('delivered to broker (id $messageId)');
  } on Exception {
    stderr.writeln('no PUBACK within 10 s');
    exitCode = 2;
  } finally {
    client.disconnect();
  }
  exit(exitCode);
}

dart run bin/publish.dart testmqtt/dart/hello "21.5 °C"
echo "exit code: $?"

Çıkış kodları aracı script'lerde kullanılabilir kılar: 0 broker'ın mesajı onayladığı, 1 bağlantı sorunu, 2 gelmeyen bir onay anlamına gelir. PUBACK, broker'ın mesajın sorumluluğunu üstlendiği anlamına gelir — bir subscriber'ın onu işlediği anlamına gelmez. Her QoS seviyesinin gerçekte neyi garanti ettiği için MQTT QoS seviyeleri yazısına bakın.

8883 portunda TLS

Bağlantıyı şifrelemek iki özellik gerektirir. SecurityContext.defaultContext işletim sisteminin güvenilir CA'larını kullanır; açık bir sertifikaya sahip broker için istediğiniz budur:

final client = MqttServerClient.withPort('broker.connectmq.com', clientId, 8883)
  ..secure = true
  ..securityContext = SecurityContext.defaultContext
  ..logging(on: false)
  ..setProtocolV311();

Özel bir CA için SecurityContext(withTrustedRoots: false) oluşturup üzerinde setTrustedCertificates('ca.pem') çağırın; karşılıklı TLS için useCertificateChain ve usePrivateKey ekleyin. Forum cevaplarında göreceğiniz onBadCertificate = (_) => true kısayoluna direnin — doğrulamayı tamamen kapatır. Handshake başarısız olursa bunun yerine SSL certificate verify failed yazısını izleyin.

Yalnızca 443 benzeri portlar açıkken WebSocket

Bazı ağlar 1883 ve 8883'ü engeller ama HTTPS benzeri trafiğe izin verir. MqttServerClient, MQTT'yi WebSocket üzerinden tünelleyebilir; şemayı ve yolu sunucu string'ine, portu ayrıca yazın:

final client = MqttServerClient.withPort('wss://broker.connectmq.com/mqtt', clientId, 8084)
  ..useWebSocket = true
  ..logging(on: false)
  ..setProtocolV311();

Şifresiz bir test için ws:// ve 8083 portunu kullanın. Broker WebSocket upgrade'ini reddederse, liste yerine yalnızca mqtt alt protokolünü sunan ..websocketProtocols = MqttClientConstants.protocolsSingleDefault ayarını deneyin. Daha fazla bilgi için WebSocket üzerinden MQTT yazısına bakın.

Yerel binary derleme

dart compile exe, anında başlayan ve hedef makinede Dart SDK gerektirmeyen bağımsız bir çalıştırılabilir dosya üretir:

dart compile exe bin/subscriber.dart -o mqtt-sub
dart compile exe bin/publish.dart -o mqtt-pub
./mqtt-pub testmqtt/dart/hello "from a native binary"

Binary, derlediğiniz işletim sistemi ve CPU için üretilir; Linux sunuculara dağıtacaksanız bir Linux container'ı içinde derleyin (resmi dart imajı iyi iş görür). Binary'yi ve /runtime/ klasörünü build aşamasından bir scratch imajına kopyalayan multi-stage bir Dockerfile, son imajı çok küçük tutar.

Servis olarak çalıştırma

Haftalarca çalışan bir subscriber, bir demodan biraz daha fazla özen ister:

• SIGINT'in yanında SIGTERM'i de yakalayın. Docker, Kubernetes ve systemd süreçleri SIGTERM ile durdurur. ProcessSignal.sigterm.watch() ile de dinleyin (Windows'ta yoktur) ve client.disconnect() çağırın; böylece broker zaman aşımına uğramış bir bağlantı yerine temiz bir DISCONNECT görür.
• Handler'ları bloklamayın. updates dinleyicisi ana isolate'ta çalışır. Orada uzun süren CPU işleri keep alive ping'lerini geciktirir; bunları Isolate.run'a veya bir kuyruğa taşıyın.
• Her mesajı değil, durum değişikliklerini loglayın. onDisconnected, onAutoReconnect ve onAutoReconnected, gece 3'te bir şeylerin ters gittiğini size söyleyen olaylardır.
• Yapılandırmayı ortamdan okuyun. Host, port, topic öneki ve kimlik bilgilerini ortam değişkeni olarak tutmak, aynı binary'nin test, staging ve production'da çalışmasını sağlar.

Araçlarınızı nasıl test edersiniz

Binary'leri bir cron job'a veya container'a koymadan önce üç hızlı kontrol yapın. Subscriber'ı başlatın, online istemciden publish edin ve mesajın ekrana basıldığını doğrulayın. Ardından ağı yarım dakikalığına kesin (veya portu bir güvenlik duvarı kuralıyla engelleyin) ve reconnected satırının göründüğünü, mesajların yeniden başlatmaya gerek kalmadan tekrar aktığını doğrulayın. Son olarak publisher'ı yanlış bir parolayla çalıştırın ve 1 koduyla çıktığını doğrulayın — onu çağıran script'ler buna güvenir.

Sorun giderme

client.logging(on: true) açıldığında istemcinin gönderdiği ve aldığı her paket yazdırılır; böylece CONNACK, SUBACK ve PUBACK'in gelip gelmediğini hızla görürsünüz.

Sunucu tarafında Dart, tipli ve baştan asenkron bir MQTT istemcisi ile kurulum gerektirmeyen yerel binary'ler sunar. Bağlantı mantığını tek bir yerde tutun, sleep yerine onayları bekleyin, dizüstü bilgisayarınızın dışında TLS kullanın ve her aracı bir cron job'a veya container'a koymadan önce gerçek bir broker'a karşı doğrulayın.

Sıkça sorulan sorular