MQTT en Dart (lado servidor) con mqtt_client

La mayoría de los ejemplos de mqtt_client son apps Flutter, pero el paquete es Dart puro: MqttServerClient funciona en cualquier programa de la Dart VM. Eso convierte a Dart en una buena opción para herramientas MQTT de línea de comandos, pequeños servicios backend y gateways de dispositivos que se compilan en un único binario nativo.

Este tutorial construye dos herramientas — un suscriptor de larga duración con reconexión automática y un publicador de un solo uso que espera la confirmación del broker — y después añade TLS, WebSockets y un build nativo. Todo apunta a broker.connectmq.com, así que puedes comprobar cada paso en el cliente MQTT online. Si lo que construyes es una app móvil, consulta MQTT en Flutter.

Preparar el proyecto

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 incluye dos clases de cliente: MqttServerClient (sockets vía dart:io, para la VM) y MqttBrowserClient (WebSockets, para Dart en la web). En el servidor siempre quieres MqttServerClient. El paquete habla MQTT 3.1.1; para MQTT 5, el paquete hermano mqtt5_client tiene una API casi idéntica — consulta MQTT 5 vs 3.1.1 para decidir si lo necesitas.

Exporta las credenciales de tu broker para que nunca acaben en el control de versiones:

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

Un suscriptor de larga duración

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

Algunos detalles que importan en producción:

• Client ID. pid hace que el ID sea único por proceso en una máquina. Si ejecutas la misma herramienta en varios hosts, incluye también el hostname: dos clientes con el mismo ID se desconectan mutuamente en bucle.
• updates entrega listas. Cada evento es una List<MqttReceivedMessage>, normalmente con una sola entrada; recórrela siempre.
• Decodifica con utf8. MqttPublishPayload.bytesToStringAsString convierte los bytes en caracteres uno a uno, lo que estropea texto no ASCII como °C. utf8.decode lo gestiona correctamente.
• Reconexión automática. Con autoReconnect y resubscribeOnAutoReconnect, una conexión caída se restablece y tus suscripciones se restauran. La estrategia de reconexión en general se trata en buenas prácticas de reconexión.

Ejecútalo con dart run bin/subscriber.dart y publica un mensaje en testmqtt/dart/hello desde el cliente online. Debería imprimirse en una fracción de segundo.

Publica al suscriptor Dart desde tu navegador

Un publicador de un solo uso que espera el PUBACK

Los publicadores de línea de comandos tienen un bug clásico: llaman a publishMessage y terminan al instante, antes de que los bytes salgan siquiera por la red. Con QoS 1 puedes esperar la confirmación del broker en el stream published en lugar de usar 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: $?"

Los códigos de salida hacen la herramienta scriptable: 0 significa que el broker confirmó el mensaje, 1 un problema de conexión y 2 que faltó la confirmación. Un PUBACK significa que el broker asumió la responsabilidad del mensaje, no que un suscriptor lo haya procesado. Para ver qué garantiza realmente cada nivel, consulta niveles de QoS en MQTT.

TLS en el puerto 8883

Cifrar la conexión requiere dos propiedades. SecurityContext.defaultContext usa las CAs de confianza del sistema operativo, que es lo que quieres para un broker con un certificado público:

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

Para una CA privada, crea un SecurityContext(withTrustedRoots: false) y llama a setTrustedCertificates('ca.pem'); para TLS mutuo añade useCertificateChain y usePrivateKey. Resiste el atajo onBadCertificate = (_) => true que verás en respuestas de foros: desactiva por completo la verificación. Si el handshake falla, sigue SSL certificate verify failed.

WebSockets cuando solo hay puertos tipo 443 abiertos

Algunas redes bloquean 1883 y 8883 pero permiten tráfico tipo HTTPS. MqttServerClient puede tunelizar MQTT sobre WebSockets; pon el esquema y la ruta en el string del servidor y el puerto por separado:

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

Usa ws:// con el puerto 8083 para una prueba sin cifrar. Si el broker rechaza el upgrade de WebSocket, prueba ..websocketProtocols = MqttClientConstants.protocolsSingleDefault, que ofrece solo el subprotocolo mqtt en lugar de una lista. Más contexto en MQTT sobre WebSockets.

Compilar un binario nativo

dart compile exe genera un ejecutable autocontenido que arranca al instante y no necesita el SDK de Dart en la máquina de destino:

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"

El binario se compila para el sistema operativo y la CPU donde lo compilas, así que hazlo dentro de un contenedor Linux (la imagen oficial dart va bien) cuando despliegues en servidores Linux. Un Dockerfile multi-stage que copia el binario y /runtime/ desde la etapa de build a una imagen scratch deja la imagen final muy pequeña.

Ejecutarlo como servicio

Un suscriptor que corre durante semanas necesita algo más de cuidado que una demo:

• Gestiona SIGTERM además de SIGINT. Docker, Kubernetes y systemd detienen los procesos con SIGTERM. Escucha también ProcessSignal.sigterm.watch() (no está disponible en Windows) y llama a client.disconnect() para que el broker vea un DISCONNECT limpio en lugar de una conexión caducada.
• Mantén los handlers sin bloqueos. El listener de updates corre en el isolate principal. El trabajo pesado de CPU ahí retrasa los pings de keep alive; muévelo a Isolate.run o a una cola.
• Registra cambios de estado, no cada mensaje. onDisconnected, onAutoReconnect y onAutoReconnected son los eventos que te dicen que algo va mal a las 3 de la mañana.
• Lee la configuración del entorno. Host, puerto, prefijo de topic y credenciales como variables de entorno permiten que el mismo binario funcione en pruebas, staging y producción.

Cómo probar tus herramientas

Antes de meter los binarios en un cron job o en un contenedor, haz tres comprobaciones rápidas. Arranca el suscriptor, publica desde el cliente online y confirma que el mensaje se imprime. Después corta la red (o bloquea el puerto con una regla de firewall) durante medio minuto y confirma que aparece reconnected y que los mensajes vuelven a fluir sin reiniciar. Por último, ejecuta el publicador con una contraseña incorrecta y confirma que termina con código 1: los scripts que lo llaman dependen de ello.

Solución de problemas

Activar client.logging(on: true) imprime cada paquete que el cliente envía y recibe, lo que muestra rápidamente si llegan CONNACK, SUBACK y PUBACK.

Dart en el servidor te da un cliente MQTT tipado y asíncrono de serie, y binarios nativos sin runtime que instalar. Mantén la lógica de conexión en un solo lugar, espera las confirmaciones en vez de usar sleeps, usa TLS fuera de tu portátil y verifica cada herramienta contra un broker real antes de meterla en un cron job o un contenedor.

Preguntas frecuentes