MQTT en Dart (côté serveur) avec mqtt_client

La plupart des exemples mqtt_client sont des apps Flutter, mais le package est du Dart pur : MqttServerClient fonctionne dans n'importe quel programme de la Dart VM. Dart devient ainsi un bon choix pour des outils MQTT en ligne de commande, de petits services backend et des passerelles d'appareils compilés en un seul binaire natif.

Ce tutoriel construit deux outils — un subscriber qui tourne en continu avec reconnexion automatique et un publisher ponctuel qui attend l'accusé de réception du broker — puis ajoute TLS, les WebSockets et un build natif. Tout cible broker.connectmq.com, vous pouvez donc vérifier chaque étape dans le client MQTT en ligne. Si vous développez plutôt une app mobile, consultez MQTT avec Flutter.

Préparer le projet

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 fournit deux classes de client : MqttServerClient (sockets via dart:io, pour la VM) et MqttBrowserClient (WebSockets, pour Dart sur le web). Côté serveur, il vous faut toujours MqttServerClient. Le package parle MQTT 3.1.1 ; pour MQTT 5, le package compagnon mqtt5_client offre une API presque identique — voyez MQTT 5 vs 3.1.1 pour savoir si vous en avez besoin.

Exportez les identifiants de votre broker pour qu'ils ne finissent jamais dans le gestionnaire de versions :

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

Un subscriber qui tourne en continu

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

Quelques détails qui comptent en production :

• Client ID. pid rend l'ID unique par processus sur une machine. Si vous lancez le même outil sur plusieurs hôtes, ajoutez aussi le nom d'hôte — deux clients avec le même ID se déconnectent mutuellement en boucle.
• updates livre des listes. Chaque événement est une List<MqttReceivedMessage>, généralement avec un seul élément ; itérez toujours.
• Décodez avec utf8. MqttPublishPayload.bytesToStringAsString convertit les octets en caractères un par un, ce qui abîme le texte non ASCII comme °C. utf8.decode le gère correctement.
• Reconnexion automatique. Avec autoReconnect et resubscribeOnAutoReconnect, une connexion perdue est rétablie et vos abonnements sont restaurés. La stratégie de reconnexion en général est traitée dans bonnes pratiques de reconnexion.

Lancez-le avec dart run bin/subscriber.dart, puis publiez un message sur testmqtt/dart/hello depuis le client en ligne. Il doit s'afficher en une fraction de seconde.

Publier vers le subscriber Dart depuis le navigateur

Un publisher ponctuel qui attend le PUBACK

Les publishers en ligne de commande ont un bug classique : ils appellent publishMessage et se terminent aussitôt, avant même que les octets soient partis sur le réseau. En QoS 1, vous pouvez attendre l'accusé de réception du broker sur le stream published au lieu de faire un sleep :

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: $?"

Les codes de sortie rendent l'outil scriptable : 0 signifie que le broker a confirmé le message, 1 un problème de connexion, 2 un accusé de réception manquant. Un PUBACK signifie que le broker a pris la responsabilité du message — pas qu'un subscriber l'a traité. Pour savoir ce que garantit réellement chaque niveau, voyez les niveaux de QoS MQTT.

TLS sur le port 8883

Chiffrer la connexion demande deux propriétés. SecurityContext.defaultContext utilise les CA de confiance du système d'exploitation, ce qu'il vous faut pour un broker avec un certificat public :

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

Pour une CA privée, créez un SecurityContext(withTrustedRoots: false) et appelez setTrustedCertificates('ca.pem') dessus ; pour du TLS mutuel, ajoutez useCertificateChain et usePrivateKey. Résistez au raccourci onBadCertificate = (_) => true qu'on trouve dans les forums : il désactive complètement la vérification. Si le handshake échoue, suivez plutôt SSL certificate verify failed.

WebSockets quand seuls des ports type 443 sont ouverts

Certains réseaux bloquent 1883 et 8883 mais laissent passer un trafic de type HTTPS. MqttServerClient peut faire passer MQTT dans des WebSockets ; mettez le schéma et le chemin dans la chaîne du serveur, et le port à part :

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

Utilisez ws:// avec le port 8083 pour un test non chiffré. Si le broker refuse l'upgrade WebSocket, essayez ..websocketProtocols = MqttClientConstants.protocolsSingleDefault, qui ne propose que le sous-protocole mqtt au lieu d'une liste. Plus de contexte dans MQTT sur WebSockets.

Produire un binaire natif

dart compile exe produit un exécutable autonome qui démarre instantanément et ne nécessite pas le SDK Dart sur la machine cible :

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"

Le binaire est construit pour le système et le processeur sur lesquels vous compilez : compilez donc dans un conteneur Linux (l'image officielle dart convient très bien) quand vous déployez sur des serveurs Linux. Un Dockerfile multi-stage qui copie le binaire et /runtime/ depuis l'étape de build vers une image scratch garde l'image finale très légère.

L'exécuter comme service

Un subscriber qui tourne pendant des semaines demande un peu plus de soin qu'une démo :

• Gérez SIGTERM en plus de SIGINT. Docker, Kubernetes et systemd arrêtent les processus avec SIGTERM. Écoutez aussi ProcessSignal.sigterm.watch() (indisponible sous Windows) et appelez client.disconnect() pour que le broker voie un DISCONNECT propre au lieu d'une connexion expirée.
• Gardez les handlers non bloquants. Le listener updates s'exécute sur l'isolate principal. Un long calcul CPU à cet endroit retarde les pings de keep alive ; déplacez-le dans Isolate.run ou une file.
• Journalisez les changements d'état, pas chaque message. onDisconnected, onAutoReconnect et onAutoReconnected sont les événements qui vous signalent un problème à 3 heures du matin.
• Lisez la configuration dans l'environnement. Hôte, port, préfixe de topic et identifiants en variables d'environnement permettent au même binaire de tourner en test, en staging et en production.

Comment tester vos outils

Avant de placer les binaires dans un cron ou un conteneur, faites trois vérifications rapides. Démarrez le subscriber, publiez depuis le client en ligne et confirmez que le message s'affiche. Coupez ensuite le réseau (ou bloquez le port avec une règle de pare-feu) pendant une demi-minute et confirmez que reconnected apparaît et que les messages circulent de nouveau sans redémarrage. Enfin, lancez le publisher avec un mauvais mot de passe et vérifiez qu'il se termine avec le code 1 — les scripts qui l'appellent en dépendent.

Dépannage

Activer client.logging(on: true) affiche chaque paquet envoyé et reçu par le client, ce qui montre vite si CONNACK, SUBACK et PUBACK arrivent.

Dart côté serveur vous offre un client MQTT typé, asynchrone par nature, et des binaires natifs sans runtime à installer. Gardez la logique de connexion à un seul endroit, attendez les accusés de réception au lieu de faire des sleeps, utilisez TLS hors de votre portable et vérifiez chaque outil contre un vrai broker avant de le placer dans un cron ou un conteneur.

Questions fréquentes