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.

Getting started

Install

Add pulse_mqtt from pub.dev:

dependencies:
  pulse_mqtt: ^1.0.0

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';

Quick start (connect → subscribe → publish)

final kit = PulseMqttKit();
kit.initialize(MyMqttBridge());
kit.addListener(MyListener());

// 1. Connect
kit.submitCommand(
  ConnectCommand(
    connectionOptions: ConnectionOptions(
      serverUri: 'tcp://broker.hivemq.com:1883',
      clientId: 'pulse-flutter-demo',
      automaticReconnect: true,
    ),
    retryPolicy: RetryPolicy.exponential(maxRetries: 3, baseDelayMillis: 2000),
  ),
);

// 2. Subscribe (runs after connect via a dependency)
kit.submitCommand(
  SubscribeCommand(
    topicConfigs: {
      'pulse/demo/location': TopicTypeConfig<String>(qosLevel: QOSLevel.qos0),
    },
  ),
);

// 3. Publish
kit.submitCommand(
  PublishCommand(
    message: ZMqttMessage(topic: 'pulse/demo/location', payload: 'hi'),
    qos: QOSLevel.qos0,
  ),
);

// 4. Clean up when done
await kit.shutDown();

The full runnable version is in example/lib/main.dart.

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.

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

Libraries

pulse_mqtt
Pulse MQTT — a robust, production-ready Flutter MQTT client library.