pulse_mqtt 1.0.2

Production-ready Flutter MQTT client with command-based API, retry policies, health monitoring, and auto-reconnect. Dart port of Zomato pulse-droid.

pulse_mqtt

A robust, production-ready Flutter MQTT client library — a faithful pure-Dart port of Zomato's pulse-droid Kotlin library. Built on top of mqtt_client, it provides a command-based architecture, retry policies, health monitoring, and automatic reconnection — the same API you know from pulse-droid, just in Dart.

This is the MQTT engine used for live-tracking-style workloads (rider/order location streams, live updates), now usable directly from Flutter with no native code.

Live tracking demo

https://github.com/Arnoldaditya17/pulse-mqtt-livetracking/raw/main/example/media/live-tracking-demo.mp4

Watch the demo · Full guide

---

Getting started

Install

Add pulse_mqtt from pub.dev:

dependencies:
  pulse_mqtt: ^1.0.2

Or from GitHub:

dependencies:
  pulse_mqtt:
    git:
      url: https://github.com/Arnoldaditya17/pulse-mqtt-livetracking.git

For local development:

dependencies:
  pulse_mqtt:
    path: ../pulse-mqtt-livetracking

Then fetch dependencies:

flutter pub get

Platform setup

connectivity_plus (used for network monitoring) needs the standard platform permissions:

• Android: the plugin adds ACCESS_NETWORK_STATE automatically — no action needed for most apps.
• iOS/macOS: no extra setup for connectivity; if you connect over plain tcp:// (non-TLS) you may need to allow it via App Transport Security in Info.plist.

Import

import 'package:pulse_mqtt/pulse_mqtt.dart';

Complete example (connect → subscribe → receive → publish)

import 'package:pulse_mqtt/pulse_mqtt.dart';

/// App-specific bridge (logger, health, network config).
class MyMqttBridge implements PulseMqttKitBridge {
  @override
  Logger? getLogger() => null;

  @override
  bool get enableJsonDeserialization => true;

  @override
  HealthMonitoringConfig? getHealthMonitoringConfig() => null;

  @override
  NetworkMonitoringConfig getNetworkConfig() =>
      const NetworkMonitoringConfig(enabled: true);
}

/// Receives command results and incoming MQTT messages.
class MyListener with MqttUpdatesListener {
  @override
  void onCommandSuccess(MqttCommand command, Success result) {
    print('OK: ${command.type}');
  }

  @override
  void onCommandFailure(MqttCommand command, Failure result) {
    print('FAIL: ${command.type} → ${result.error}');
  }

  @override
  void onMqttMessageReceived(
    String? topic,
    String? payload, [
    TopicMessage? topicMessage,
  ]) {
    print('MSG on $topic: $payload');
  }

  @override
  void onMqttConnectionLost(Object? cause) {
    print('Connection lost: $cause');
  }
}

Future<void> startMqtt() async {
  final kit = PulseMqttKit();
  kit.initialize(MyMqttBridge());
  kit.addListener(MyListener());

  // 1. Connect (public HiveMQ demo broker — no auth)
  final connect = ConnectCommand(
    connectionOptions: const ConnectionOptions(
      serverUri: 'tcp://broker.hivemq.com:1883',
      clientId: 'pulse-flutter-demo',
      automaticReconnect: true,
      keepAliveIntervalSeconds: 60,
      connectionTimeoutSeconds: 30,
    ),
    retryPolicy: RetryPolicy.exponential(
      maxRetries: 3,
      baseDelayMillis: 2000,
      excludedExceptionCodes: {
        MqttExceptionCode.reasonCodeNotAuthorized,
        MqttExceptionCode.connectAlreadyInProgress,
      },
    ),
  );
  kit.submitCommand(connect);

  // 2. Subscribe — runs only after connect succeeds
  const topic = 'pulse/demo/location';
  kit.submitCommand(
    SubscribeCommand(
      topicConfigs: {
        topic: TopicTypeConfig<String>(qosLevel: QOSLevel.qos0),
      },
      dependencies: [connect],
    ),
  );

  // 3. Publish a test message (also waits for connect)
  kit.submitCommand(
    PublishCommand(
      message: ZMqttMessage(topic: topic, payload: 'hello from pulse_mqtt'),
      qos: QOSLevel.qos0,
      dependencies: [connect],
    ),
  );

  // Incoming messages arrive in MyListener.onMqttMessageReceived
  // ...

  // 4. Clean up when the app is done
  // await kit.shutDown();
}

HiveMQ Cloud / TLS (ssl://)

final connect = ConnectCommand(
  connectionOptions: ConnectionOptions(
    serverUri: 'ssl://xxxxxxxx.s1.eu.hivemq.cloud:8883',
    clientId: 'customer-app-1',
    username: 'your-username',
    password: 'your-password',
    automaticReconnect: true,
  ),
  retryPolicy: RetryPolicy.exponential(maxRetries: 3, baseDelayMillis: 2000),
);

Disconnect

kit.submitCommand(DisconnectCommand());
// or fully tear down:
await kit.shutDown();

Run the interactive UI demo:

cd example && flutter run

• UI demo: example/lib/main.dart
• Live driver tracking (publisher + subscriber + map glide, HiveMQ Cloud): example/LIVE_TRACKING.md

---

1. Initialization

Implement the PulseMqttKitBridge to provide app-specific dependencies, then initialize the kit:

final pulseMqttKit = PulseMqttKit();

class MyMqttBridge implements PulseMqttKitBridge {
  @override
  Logger? getLogger() => MyLogger();

  @override
  bool get enableJsonDeserialization => true;

  @override
  HealthMonitoringConfig? getHealthMonitoringConfig() =>
      HealthMonitoringConfig(monitoringFreqSeconds: 30);

  @override
  NetworkMonitoringConfig getNetworkConfig() =>
      const NetworkMonitoringConfig(enabled: true);
}

pulseMqttKit.initialize(MyMqttBridge());

Note (vs. Android): the Kotlin initialize(context, bridge) took an Android Context. Flutter has no equivalent, so this takes only the bridge. The bridge also no longer supplies a Gson instance or a coroutine scope — JSON is handled with dart:convert, and the Dart event loop replaces coroutines.

2. Command-Based API

For every MQTT action, you submit a command:

• Connect: ConnectCommand(connectionOptions: ..., retryPolicy: ...)
• Publish: PublishCommand(message: ..., qos: ...)
• Subscribe: SubscribeCommand(topicConfigs: ...)
• Unsubscribe: UnsubscribeCommand(topics: ...)
• Disconnect: DisconnectCommand()

pulseMqttKit.submitCommand(command);

Command Dependencies

Each command supports a dependencies list. A dependent command runs only if all its dependencies succeed:

final connectCommand = ConnectCommand(
  connectionOptions: options,
  retryPolicy: RetryPolicy.exponential(),
);
final subscribeCommand = SubscribeCommand(
  topicConfigs: {'my/topic': TopicTypeConfig<String>(qosLevel: QOSLevel.qos1)},
  dependencies: [connectCommand],
);
pulseMqttKit.submitCommand(subscribeCommand);

Retry Policies

• Sequential: fixed retry intervals — RetryPolicy.sequential(...)
• Exponential: exponential backoff — RetryPolicy.exponential(...)
• Jitter: adds random jitter — RetryPolicy.jitter(...)
• None: no retry — RetryPolicy.none()

Exclude specific MQTT exception codes from retrying:

final retryPolicy = RetryPolicy.exponential(
  excludedExceptionCodes: {
    MqttExceptionCode.reasonCodeNotAuthorized,
    MqttExceptionCode.connectAlreadyInProgress,
  },
);

3. Auto Subscription

ConnectionOptions.autoSubscriptionConfig restores subscriptions after a reconnect automatically:

final options = ConnectionOptions(
  serverUri: 'tcp://broker.hivemq.com:1883',
  clientId: 'my-client',
  automaticReconnect: true,
  autoSubscriptionConfig: AutoSubscriptionConfig(
    enabled: true,
    subscriptionStore: {
      'driver/notification':
          TopicTypeConfig<String>(qosLevel: QOSLevel.qos2, typeLabel: 'notif'),
      'driver/alerts':
          TopicTypeConfig<String>(qosLevel: QOSLevel.qos1, typeLabel: 'alert'),
    },
  ),
);

4. Listening for Events

class MyListener with MqttUpdatesListener {
  @override
  void onCommandSuccess(MqttCommand command, Success result) { /* ... */ }

  @override
  void onMqttMessageReceived(String? topic, String? payload,
      [TopicMessage? topicMessage]) {
    if (topicMessage is Deserialized) {
      final data = topicMessage.data; // typed
    }
  }
}

pulseMqttKit.addListener(MyListener());
// ...
pulseMqttKit.removeListener(MyListener());

Each command result carries number of attempts and total execution time.

5. Type-safe Deserialization

The Kotlin library used Gson + Class<T> reflection. Dart has no runtime reflection, so you supply a deserializer function (usually a fromJson factory) and a typeLabel:

class RiderLocation {
  RiderLocation(this.lat, this.lon);
  final double lat;
  final double lon;
  factory RiderLocation.fromJson(Object? json) {
    final m = json as Map<String, dynamic>;
    return RiderLocation((m['lat'] as num).toDouble(), (m['lon'] as num).toDouble());
  }
}

final config = TopicTypeConfig<RiderLocation>(
  typeLabel: 'RiderLocation',
  deserializer: RiderLocation.fromJson,
  qosLevel: QOSLevel.qos0,
);

When no deserializer is set (or enableJsonDeserialization is false), messages arrive as Plain(String).

6. Health Monitoring

pulseMqttKit.startHealthMonitoring();
// ...
pulseMqttKit.stopHealthMonitoring();

Note (vs. Android): the Kotlin library backed health checks with WorkManager and AlarmManager. Flutter has no portable in-app equivalent, so both HealthMonitoringTypes use a periodic Timer that runs while the app process is alive. For execution when the app is killed, drive HealthMonitoringConfig.healthCheck from a background plugin such as workmanager or flutter_background_service.

7. Shutting Down

await pulseMqttKit.shutDown();

Unsubscribes from all topics, disconnects, stops monitoring, and clears listeners.

---

Porting notes (Kotlin → Dart)

pulse-droid (Kotlin)	pulse_mqtt (Dart)
Eclipse Paho	mqtt_client
Gson + Class<T>	TopicTypeConfig.deserializer + dart:convert
Kotlin coroutines / CoroutineScope	Future / async (event loop)
WorkManager / AlarmManager	periodic Timer (TimerMonitoringProvider)
Android ConnectivityManager	connectivity_plus
initialize(context, bridge)	initialize(bridge)
sealed CommandResult subclasses	Success / Failure / Ignored / None

The command model, retry semantics, dependency resolution, auto-subscription, and listener callbacks are preserved 1:1.

See example/lib/main.dart for a full connect → subscribe → publish demo.

---

Credits

This package is a Flutter/Dart port of pulse-droid, the original Kotlin MQTT library by Zomato, built on Eclipse Paho. All of the architecture — the command model, retry policies, auto-subscription, health monitoring, and listener/bridge design — originates from that project. Full credit for the design and API goes to the Zomato team and the pulse-droid authors.

• Parent project: https://github.com/Zomato/pulse-droid
• License: see the parent project's LICENSES/ directory.

This Dart port additionally builds on the excellent mqtt_client and connectivity_plus packages.