MQTT in Dart (Server-Side) with mqtt_client

Most mqtt_client examples are Flutter apps, but the package is plain Dart: MqttServerClient runs in any Dart VM program. That makes Dart a good fit for MQTT command-line tools, small backend services and device gateways that compile to a single native binary.

This tutorial builds two tools — a long-running subscriber with auto-reconnect and a one-shot publisher that waits for the broker's acknowledgement — and then adds TLS, WebSockets and a native build. Everything targets broker.connectmq.com, so you can cross-check each step in the online MQTT client. If you are building a mobile app instead, see MQTT in Flutter.

Project setup

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 ships two client classes: MqttServerClient (sockets via dart:io, for the VM) and MqttBrowserClient (WebSockets, for Dart on the web). On the server you always want MqttServerClient. The package speaks MQTT 3.1.1; for MQTT 5, the companion package mqtt5_client has an almost identical API — see MQTT 5 vs 3.1.1 to decide whether you need it.

Export your broker credentials so they never end up in source control:

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

A long-running 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);
  });
}

A few details that matter in production:

• Client ID. pid makes the ID unique per process on one machine. If you run the same tool on several hosts, include the hostname too — two clients with one ID disconnect each other in a loop.
• updates delivers lists. Each event is a List<MqttReceivedMessage>, usually with one entry; always iterate.
• Decode with utf8. The package's MqttPublishPayload.bytesToStringAsString maps bytes to characters one by one, which mangles non-ASCII text such as °C. utf8.decode handles it correctly.
• Auto-reconnect. With autoReconnect and resubscribeOnAutoReconnect, a dropped connection is re-established and your subscriptions are restored. Reconnect strategy in general is covered in reconnect best practices.

Run it with dart run bin/subscriber.dart, then publish a message to testmqtt/dart/hello from the online client. It should print within a fraction of a second.

Publish to the Dart subscriber from your browser

A one-shot publisher that waits for PUBACK

Command-line publishers have a classic bug: they call publishMessage and exit immediately, before the bytes are even on the wire. With QoS 1 you can wait for the broker's acknowledgement on the published stream instead of sleeping:

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

The exit codes make the tool scriptable: 0 means the broker confirmed the message, 1 a connection problem, 2 a missing acknowledgement. A PUBACK means the broker took responsibility for the message — not that a subscriber processed it. For what each QoS level actually guarantees, see MQTT QoS levels.

TLS on port 8883

Encrypting the connection takes two properties. SecurityContext.defaultContext uses the operating system's trusted CAs, which is what you want for a broker with a public certificate:

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

For a private CA, create a SecurityContext(withTrustedRoots: false) and call setTrustedCertificates('ca.pem') on it; for mutual TLS add useCertificateChain and usePrivateKey. Resist the onBadCertificate = (_) => true shortcut you will find in forum answers — it disables verification entirely. If the handshake fails, follow SSL certificate verify failed instead.

WebSockets when only 443-style ports are open

Some networks block 1883 and 8883 but allow HTTPS-like traffic. MqttServerClient can tunnel MQTT over WebSockets; put the scheme and path in the server string and the port separately:

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

Use ws:// with port 8083 for an unencrypted test. If the broker rejects the WebSocket upgrade, try ..websocketProtocols = MqttClientConstants.protocolsSingleDefault, which offers only the mqtt subprotocol instead of a list. More background in MQTT over WebSockets.

Build a native binary

dart compile exe produces a self-contained executable that starts instantly and needs no Dart SDK on the target machine:

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"

The binary is built for the OS and CPU you compile on, so build inside a Linux container (the official dart image works well) when you deploy to Linux servers. A multi-stage Dockerfile that copies the binary plus /runtime/ from the build stage into a scratch image keeps the final image very small.

Running it as a service

A subscriber that runs for weeks needs a little more care than a demo:

• Handle SIGTERM as well as SIGINT. Docker, Kubernetes and systemd stop processes with SIGTERM. Listen on ProcessSignal.sigterm.watch() too (it is not available on Windows) and call client.disconnect() so the broker sees a clean DISCONNECT instead of a timed-out connection.
• Keep handlers non-blocking. The updates listener runs on the main isolate. Long CPU work there delays keep-alive pings; move it to Isolate.run or a queue.
• Log state changes, not every message. onDisconnected, onAutoReconnect and onAutoReconnected are the events that tell you something is wrong at 3 a.m.
• Read configuration from the environment. Host, port, topic prefix and credentials as environment variables let the same binary run in test, staging and production.

How to test your tools

Before you put the binaries in a cron job or a container, run three quick checks. Start the subscriber, publish from the online client and confirm the message prints. Then stop your network (or block the port with a firewall rule) for half a minute and confirm reconnected appears and messages flow again without a restart. Finally, run the publisher with a wrong password and confirm it exits with code 1 — scripts that call it depend on that.

Troubleshooting

Turning on client.logging(on: true) prints every packet the client sends and receives, which quickly shows whether CONNACK, SUBACK and PUBACK arrive.

Server-side Dart gets you a typed, async-first MQTT client and native binaries with no runtime to install. Keep the connection logic in one place, wait for acknowledgements instead of sleeping, use TLS outside your laptop, and verify every tool against a real broker before it goes into a cron job or a container.

Frequently asked questions