Flutter MQTT Client Wrapper

mqtt_client_wrapper.dart

import 'dart:developer';
import 'dart:math' as math;

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

enum MqttCurrentConnectionState {
  idle,
  connecting,
  connected,
  disconnected,
  errorWhenConnecting
}

enum MqttSubscriptionState {
  idle,
  subscribed
}

class MQTTClientWrapper {

  MQTTClientWrapper(
    this.onConnectedCallback,
    this.onReceivedCallback, {
      required this.topics,
      this.username = '',
      this.password = '',
      required this.serverAddress
    }) : super();

  final Function() onConnectedCallback; // VoidCallback
  final Function(String) onReceivedCallback;
  final List<String> topics;
  final String username;
  final String password;
  final String serverAddress;

  late MqttServerClient client;

  MqttCurrentConnectionState connectionState = MqttCurrentConnectionState.idle;
  MqttSubscriptionState subscriptionState = MqttSubscriptionState.idle;

  void prepareMqttClient() async {
    _setupMqttClient();
    await _connectClient();
    _subscribeToTopic(topics);
  }

  Future<void> _connectClient() async {
    try {
      log('MQTTClientWrapper:: client connecting....');
      connectionState = MqttCurrentConnectionState.connecting;
      await client.connect(username, password);
    } on Exception catch (e) {
      log('MQTTClientWrapper::client exception - $e');
      connectionState = MqttCurrentConnectionState.errorWhenConnecting;
      client.disconnect();
    }

    if (client.connectionStatus!.state == MqttConnectionState.connected) {
      connectionState = MqttCurrentConnectionState.connected;
      log('MQTTClientWrapper:: client connected');
    } else {
      log(
          'MQTTClientWrapper::ERROR client connection failed - disconnecting, status is ${client.connectionStatus}');
      connectionState = MqttCurrentConnectionState.errorWhenConnecting;
      client.disconnect();
    }
  }

  void _setupMqttClient() {
    client = MqttServerClient.withPort(serverAddress, '${DateTime.now().millisecondsSinceEpoch ~/ 1000}-${generateRandomString(len: 12)}', 1883);
    client.autoReconnect = true;
    client.logging(on: false);
    client.keepAlivePeriod = 60;
    client.onDisconnected = _onDisconnected;
    client.onConnected = _onConnected;
    client.onSubscribed = _onSubscribed;
  }

  void _subscribeToTopic(List<String> topics) {
    for (var topic in topics) {
      log('MQTTClientWrapper::Subscribing to the $topic topic');
      client.subscribe(topic, MqttQos.exactlyOnce);
    }

    client.updates!.listen((List<MqttReceivedMessage<MqttMessage>> c) {
      final recMess = c[0].payload as MqttPublishMessage;
      final pt = MqttPublishPayload.bytesToStringAsString(recMess.payload.message);

      print('MQTTClientWrapper::GOT A NEW MESSAGE $pt');
      onReceivedCallback(pt);
    });
  }

  void publishMessage(String message, String topicName) {
    final MqttClientPayloadBuilder builder = MqttClientPayloadBuilder();
    builder.addString(message);
  
    print('MQTTClientWrapper::Publishing message $message to topic ${topicName}');
    client.publishMessage(topicName, MqttQos.exactlyOnce, builder.payload!);
  }

  void _onSubscribed(String topic) {
    log('MQTTClientWrapper::Subscription confirmed for topic $topic');
    subscriptionState = MqttSubscriptionState.subscribed;
  }

  void _onDisconnected() {
    log('MQTTClientWrapper::OnDisconnected client callback - Client disconnection');
    // if (client.connectionStatus!.returnCode == MqttConnectReturnCode.solicited) {
    //   print('MQTTClientWrapper::OnDisconnected callback is solicited, this is correct');
    // }
    connectionState = MqttCurrentConnectionState.disconnected;
    onConnectedCallback();
  }

  void _onConnected() {
    connectionState = MqttCurrentConnectionState.connected;
    log('MQTTClientWrapper::OnConnected client callback - Client connection was successful');
    onConnectedCallback();
  }

  void disconnect() {
    client.disconnect();
    log('MQTT Disconnected');
  }

}

String generateRandomString({required int len, String chars = 'AaBbCcDdEeFfGgHhIiJjKkLlMmNnOoPpQqRrSsTtUuVvWwXxYyZz1234567890'}) {
  var r = math.Random();
  return List.generate(len, (index) => chars[r.nextInt(chars.length)]).join();
}

Usage

  late MQTTClientWrapper mqttClientWrapper;

  @override
  void initState() {
    super.initState();
    mqttClientWrapper = MQTTClientWrapper(
            () => log(mqttClientWrapper.connectionState.toString()),
        print, // Incoming message from topics
        topics: ['topic/#'],
        username: 'username',
        password: 'password',
        serverAddress: 'mqtt.example.com'
    );
    mqttClientWrapper.prepareMqttClient();
  }

disposing

  @override
  void dispose() {
    mqttClientWrapper.disconnect();
    super.dispose();
  }

Publish message

mqttClientWrapper.publishMessage('Hello', 'topic/1');

Original post on Medium: https://medium.com/swlh/using-mqtt-with-flutter-to-build-a-location-sharing-app-24e7307b21d3