MQTT em Dart (lado servidor) com mqtt_client

A maioria dos exemplos de mqtt_client são apps Flutter, mas o pacote é Dart puro: o MqttServerClient roda em qualquer programa da Dart VM. Isso faz do Dart uma boa escolha para ferramentas MQTT de linha de comando, pequenos serviços de backend e gateways de dispositivos que compilam para um único binário nativo.

Este tutorial constrói duas ferramentas — um subscriber de longa duração com reconexão automática e um publisher de execução única que espera a confirmação do broker — e depois adiciona TLS, WebSockets e um build nativo. Tudo aponta para broker.connectmq.com, então você pode conferir cada passo no cliente MQTT online. Se você está construindo um app mobile, veja MQTT no Flutter.

Configurando o projeto

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

O mqtt_client traz duas classes de cliente: MqttServerClient (sockets via dart:io, para a VM) e MqttBrowserClient (WebSockets, para Dart na web). No servidor você sempre quer o MqttServerClient. O pacote fala MQTT 3.1.1; para MQTT 5, o pacote irmão mqtt5_client tem uma API quase idêntica — veja MQTT 5 vs 3.1.1 para decidir se precisa dele.

Exporte as credenciais do seu broker para que elas nunca parem no controle de versão:

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

Um subscriber de longa duração

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

Alguns detalhes que importam em produção:

• Client ID. pid deixa o ID único por processo em uma máquina. Se você roda a mesma ferramenta em vários hosts, inclua também o hostname — dois clientes com o mesmo ID se desconectam em loop.
• updates entrega listas. Cada evento é uma List<MqttReceivedMessage>, normalmente com um item; sempre itere.
• Decodifique com utf8. O MqttPublishPayload.bytesToStringAsString do pacote converte bytes em caracteres um a um, o que estraga texto não ASCII como °C. O utf8.decode trata isso corretamente.
• Reconexão automática. Com autoReconnect e resubscribeOnAutoReconnect, uma conexão perdida é restabelecida e suas assinaturas são restauradas. Estratégias de reconexão em geral estão em boas práticas de reconexão.

Rode com dart run bin/subscriber.dart e publique uma mensagem em testmqtt/dart/hello pelo cliente online. Ela deve aparecer em uma fração de segundo.

Publique para o subscriber Dart pelo navegador

Um publisher de execução única que espera o PUBACK

Publishers de linha de comando têm um bug clássico: chamam publishMessage e encerram na hora, antes mesmo de os bytes saírem pela rede. Com QoS 1 você pode esperar a confirmação do broker no stream published em vez de usar 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: $?"

Os códigos de saída tornam a ferramenta scriptável: 0 significa que o broker confirmou a mensagem, 1 um problema de conexão e 2 uma confirmação que não chegou. Um PUBACK significa que o broker assumiu a responsabilidade pela mensagem — não que um subscriber a processou. Para saber o que cada nível garante de fato, veja níveis de QoS do MQTT.

TLS na porta 8883

Criptografar a conexão exige duas propriedades. SecurityContext.defaultContext usa as CAs confiáveis do sistema operacional, que é o que você quer para um broker com certificado público:

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

Para uma CA privada, crie um SecurityContext(withTrustedRoots: false) e chame setTrustedCertificates('ca.pem') nele; para TLS mútuo adicione useCertificateChain e usePrivateKey. Resista ao atalho onBadCertificate = (_) => true que aparece em respostas de fóruns — ele desliga a verificação por completo. Se o handshake falhar, siga SSL certificate verify failed.

WebSockets quando só portas estilo 443 estão abertas

Algumas redes bloqueiam 1883 e 8883, mas permitem tráfego parecido com HTTPS. O MqttServerClient pode tunelar MQTT sobre WebSockets; coloque o esquema e o caminho na string do servidor e a porta separadamente:

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

Use ws:// com a porta 8083 para um teste sem criptografia. Se o broker rejeitar o upgrade de WebSocket, tente ..websocketProtocols = MqttClientConstants.protocolsSingleDefault, que oferece apenas o subprotocolo mqtt em vez de uma lista. Mais contexto em MQTT sobre WebSockets.

Gerando um binário nativo

dart compile exe produz um executável autocontido que inicia na hora e não precisa do SDK do Dart na 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"

O binário é gerado para o sistema operacional e a CPU em que você compila, então compile dentro de um container Linux (a imagem oficial dart funciona bem) quando for publicar em servidores Linux. Um Dockerfile multi-stage que copia o binário e /runtime/ da etapa de build para uma imagem scratch deixa a imagem final bem pequena.

Rodando como serviço

Um subscriber que roda por semanas precisa de um pouco mais de cuidado que uma demo:

• Trate SIGTERM além de SIGINT. Docker, Kubernetes e systemd param processos com SIGTERM. Escute também ProcessSignal.sigterm.watch() (não disponível no Windows) e chame client.disconnect() para que o broker veja um DISCONNECT limpo em vez de uma conexão expirada.
• Mantenha os handlers sem bloqueio. O listener de updates roda no isolate principal. Trabalho pesado de CPU ali atrasa os pings de keep alive; mova-o para Isolate.run ou para uma fila.
• Registre mudanças de estado, não cada mensagem. onDisconnected, onAutoReconnect e onAutoReconnected são os eventos que avisam que algo está errado às 3 da manhã.
• Leia a configuração do ambiente. Host, porta, prefixo de topic e credenciais como variáveis de ambiente permitem que o mesmo binário rode em teste, staging e produção.

Como testar suas ferramentas

Antes de colocar os binários em um cron job ou container, faça três verificações rápidas. Inicie o subscriber, publique pelo cliente online e confirme que a mensagem aparece. Depois derrube a rede (ou bloqueie a porta com uma regra de firewall) por meio minuto e confirme que reconnected aparece e que as mensagens voltam a fluir sem reiniciar. Por fim, rode o publisher com uma senha errada e confirme que ele sai com código 1 — os scripts que o chamam dependem disso.

Solução de problemas

Ativar client.logging(on: true) imprime cada pacote que o cliente envia e recebe, o que mostra rapidamente se CONNACK, SUBACK e PUBACK estão chegando.

Dart no servidor entrega um cliente MQTT tipado e assíncrono por natureza, além de binários nativos sem runtime para instalar. Mantenha a lógica de conexão em um só lugar, espere as confirmações em vez de usar sleep, use TLS fora do seu notebook e valide cada ferramenta contra um broker real antes de colocá-la em um cron job ou container.

Perguntas frequentes