pulse_flutter

A production-grade MQTT orchestration layer for Flutter, modeled on Zomato's pulse-droid — built on top of mqtt_client, but adding everything real apps end up hand-rolling around it: a true Android background transport (Eclipse Paho in a foreground service), a sequential command queue, pluggable retry policies, auto-resubscription, an in-memory offline outbox, and connection health monitoring.

If you've ever shipped a driver-tracking, live-chat, or ride-hailing app in Flutter, you've probably discovered that mqtt_client alone stops publishing the moment the user backgrounds the app or locks the screen. pulse_flutter exists to fix exactly that, plus the half-dozen other things a "just use mqtt_client" approach quietly leaves you to build yourself.

Why not just mqtt_client?
Features
Installation
Android setup
Quick start
Choosing a transport
TLS is not optional in the background
Retry policies
Events
Offline outbox
Auto-resubscription
Connection health monitoring
PulseMqttKit — the pulse-droid-parity API
Testing your app's MQTT logic
Platform support
FAQ / troubleshooting
License

Why not just `mqtt_client`?

mqtt_client is a solid low-level MQTT implementation, and pulse_flutter uses it directly for its default transport. The problem isn't the protocol layer — it's everything a real app needs around it, which mqtt_client intentionally leaves out of scope:

Need	`mqtt_client` alone	`pulse_flutter`
Keeps publishing with the app backgrounded / screen off (Android)	❌ Dart isolate gets deprioritized/killed by the OS; socket dies silently	✅ `NativeForegroundTransport` — a real Android foreground service running Eclipse Paho, independent of the Flutter isolate's lifecycle
Retry on connect/publish/subscribe failure	❌ You write your own retry loop, every time	✅ `RetryPolicy` — sequential, exponential backoff, or jittered, with per-error-code exclusions
"Fire commands before connect completes" bugs	❌ Nothing stops you from calling `publish()` before `connect()` resolves — throws a raw `MqttException`	✅ Sequential command queue — every command runs in submission order, and dependent commands can declare what they depend on
Structured success/failure results	❌ Exceptions only, no attempt count, no execution time	✅ `CommandResult<T>` / `MqttCommandResult` with attempts, execution time, and typed errors
Re-subscribing after a reconnect	❌ You track subscribed topics yourself and re-fire them on every reconnect	✅ `AutoSubscriptionManager` replays every active subscription automatically
Publishing while briefly offline	❌ Throws immediately	✅ In-memory offline outbox — queues and flushes in order once reconnected
Detecting a silently-dead connection	❌ You wire up `connectivity_plus` and a timer yourself	✅ `ConnectionHealthMonitor` — periodic liveness check + immediate reconnect on network-change events, plus an isolate-independent AlarmManager/WorkManager backstop on the native transport
Testability	❌ `MqttServerClient` is difficult to fake in unit tests	✅ `PulseTransport` is an interface — inject a fake transport, test your app's MQTT logic with zero real sockets

In short: mqtt_client gives you a wire protocol. pulse_flutter gives you the orchestration layer a shipping app actually needs on top of it — the same layer Zomato built in Kotlin for pulse-droid, ported faithfully to Flutter.

Use plain mqtt_client if: you're prototyping, your app never needs to publish/subscribe while backgrounded, and you're happy hand-rolling retries.

Use pulse_flutter if: you're shipping something like live location tracking, order/delivery status, or chat that has to survive the app being backgrounded, and you don't want to rebuild retry/resubscribe/health-check logic from scratch.

Features

🔌 Two transports — MqttClientTransport (pure Dart, cross-platform, simplest option) and NativeForegroundTransport (Android-only, Eclipse Paho in a real foreground service, survives backgrounding)
🔁 Retry policies — RetryPolicy.sequential, .exponential, .jitter, .none, each with optional error-code exclusions
📬 Sequential command queue — no more "subscribed before connect finished" races
🗂️ In-memory offline outbox — publishes attempted while disconnected are queued and flushed on reconnect, in order
🔄 Auto-resubscription — every active subscription is replayed automatically after a reconnect
❤️ Connection health monitoring — periodic liveness checks + instant reaction to network changes, with a native AlarmManager/WorkManager backstop for when the Dart isolate itself gets trimmed
📊 Structured results — CommandResult<T> instead of raw exceptions, with attempt counts and stack traces
📡 Typed event stream — PulseConnected, PulseDisconnected, PulseReconnecting, PulseMessageReceived, PulseMessageQueued, PulseOutboxFlushed, PulseSubscribed, PulseUnsubscribed, PulseHealthCheckFailed, PulseError
🧪 Testable by design — PulseTransport is an interface; swap in a fake for unit tests with zero real sockets
🧬 Two API shapes — the ergonomic PulseClient (connect()/publish()/subscribe()), or PulseMqttKit for teams porting from pulse-droid who want 1:1 API parity

Installation

dependencies:
  pulse_flutter: ^1.0.0

flutter pub get

Android setup

NativeForegroundTransport requires a few things in the host app (not the package — these are already merged into your manifest by the plugin, except where noted):

Runtime notification permission (Android 13+). The persistent notification is what keeps the foreground service alive — Android requires an explicit runtime prompt for it:
```
import 'package:permission_handler/permission_handler.dart';

final status = await Permission.notification.request();
```
Without this, startForeground() still runs, but the notification silently never shows.
TLS-enabled broker. This is not optional — see TLS is not optional in the background below before you ship anything using NativeForegroundTransport.
Nothing else to add manually — the service declaration, wake lock, and foreground-service permissions are merged into your manifest automatically by the plugin.

Quick start

import 'package:pulse_flutter/pulse_flutter.dart';

final config = PulseConfig(
  host: 'your-broker.example.com',
  port: 8883,          // use 8883 + secure:true — see TLS section below
  secure: true,
  clientId: 'my-client-${DateTime.now().millisecondsSinceEpoch}',
  username: 'optional-username',
  password: 'optional-password',
  keepAlive: const Duration(seconds: 30),
);

final client = PulseClient(config: config);

client.events.listen((event) {
  switch (event) {
    case PulseConnected():
      print('connected');
    case PulseDisconnected(:final reason):
      print('disconnected: $reason');
    case PulseMessageReceived(:final topic, :final payload):
      print('$topic -> $payload');
    default:
      break;
  }
});

final result = await client.connect();
result.when(
  success: (_) => print('connected!'),
  failure: (error, stackTrace, attempts) =>
      print('failed after $attempts attempts: $error'),
);

await client.subscribe('some/topic');
await client.publish('some/topic', 'hello');

// When you're done:
client.dispose();

Choosing a transport

// Foreground-only (e.g. a customer watching a live map while the app is
// open) — simplest option, works on every platform mqtt_client supports:
final client = PulseClient(config: config);

// Needs to keep publishing/subscribing with the app backgrounded or the
// screen off (e.g. a driver app pushing GPS updates) — Android only:
final client = PulseClient(
  config: config,
  transport: NativeForegroundTransport(config),
);

Don't reach for NativeForegroundTransport by default — it starts a real foreground service with a persistent notification. Use it only for the role that genuinely needs to survive backgrounding.

TLS is not optional in the background

Since Android 9 (API 28+), the platform blocks plaintext (tcp://) sockets by default for anything going through Android's Java networking stack. NativeForegroundTransport runs Eclipse Paho on exactly that stack — so a plaintext broker (port 1883, secure: false) will connect fine in plain Dart (MqttClientTransport bypasses this policy entirely) and then mysteriously fail the instant you switch to the native/background transport, with an error like "Unable to connect to server".

Fix: always use a TLS listener with the native transport:

PulseConfig(
  host: 'your-broker.example.com',
  port: 8883,
  secure: true,
  clientId: '...',
);

If your broker only exposes plaintext 1883 and you can't add a TLS listener, the alternative is a network_security_config.xml in the host app allowing cleartext for that specific domain — but this is a work-around, not a recommendation. Production MQTT traffic (credentials, location data, order details) should be encrypted regardless of which transport you use.

Also don't use a public test broker (e.g. broker.hivemq.com, broker.emqx.io) for production. These are explicitly documented as test/learning brokers only, with no uptime guarantees — use a self-hosted (Mosquitto/EMQX) or managed (HiveMQ Cloud, EMQX Cloud) broker for anything shipping to real users.

Retry policies

PulseClient(
  config: config,
  connectRetryPolicy: const RetryPolicy.exponential(
    maxAttempts: 8,
    initialDelay: Duration(milliseconds: 500),
    maxDelay: Duration(seconds: 20),
  ),
  commandRetryPolicy: const RetryPolicy.sequential(maxAttempts: 2),
);

Policy	Behavior
`RetryPolicy.none()`	Fail immediately, no retry
`RetryPolicy.sequential(maxAttempts, delay)`	Fixed delay between attempts
`RetryPolicy.exponential(maxAttempts, initialDelay, multiplier, maxDelay)`	Exponential backoff
`RetryPolicy.jitter(maxAttempts, baseDelay, maxJitter)`	Randomized delay — spreads out reconnect storms across many clients after a shared broker blip

Every policy accepts excludedErrorCodes to skip retrying specific failures (e.g. don't retry a connect that failed due to bad credentials).

Events

client.events.listen((event) {
  switch (event) {
    case PulseConnected():
    case PulseDisconnected(:final reason):
    case PulseReconnecting(:final attempt):
    case PulseMessageReceived(:final topic, :final payload, :final payloadBytes):
    case PulseMessageQueued(:final topic):
    case PulseOutboxFlushed(:final count):
    case PulseSubscribed(:final topic):
    case PulseUnsubscribed(:final topic):
    case PulseHealthCheckFailed(:final reason):
    case PulseError(:final error, :final stackTrace):
  }
});

Offline outbox

Enabled by default (PulseConfig.enableOfflineOutbox = true). Publishing while disconnected queues the message in memory instead of failing:

await client.publish('topic', 'message'); // succeeds even if offline —
                                           // emits PulseMessageQueued

Once reconnected, queued messages are flushed in submission order and a PulseOutboxFlushed(count) event fires. This queue is in-memory only — it does not survive the app process being killed, and it isn't a substitute for a persistent MQTT session (this client always connects with cleanSession: true).

Auto-resubscription

Every topic you subscribe() to is tracked automatically. After any reconnect — whether triggered by the health monitor, a network change, or you calling connect() again — every previously active subscription is replayed without you having to call subscribe() yourself.

Connection health monitoring

Enabled by default (enableHealthMonitor: true on PulseClient). Two independent checks:

A periodic liveness check (belt-and-suspenders on top of MQTT keep-alive) that triggers a reconnect if the transport reports itself disconnected.
A device connectivity listener (connectivity_plus) that triggers an immediate reconnect the moment the network comes back, instead of waiting for the next timer tick.

When using NativeForegroundTransport, a third, isolate-independent backstop is available via AlarmManager/WorkManager — this keeps checking connection health even if the Dart isolate itself gets trimmed by the OS:

final transport = NativeForegroundTransport(config);
await transport.configureNativeHealthCheck(
  type: HealthMonitoringType.workManager,
  freqSeconds: 60,
);

`PulseMqttKit` — the pulse-droid-parity API

If you're porting an app from Zomato's Kotlin pulse-droid and want a matching API shape, PulseMqttKit mirrors it exactly: a single submitCommand() entry point, commands that declare their own dependencies, and listener-based results with attempt counts and execution time.

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

final connect = ConnectCommand(
  connectionOptions: ConnectionOptions(
    serverUri: 'ssl://your-broker.example.com:8883',
    clientId: 'client-1',
    useNativeForegroundService: true, // background-capable
  ),
  retryPolicy: const RetryPolicy.exponential(),
);

final subscribe = SubscribeCommand(
  topics: ['orders/updates'],
  dependencies: [connect], // runs `connect` first if it hasn't run yet
);

final result = await kit.submitCommand(subscribe);
result.when(
  success: (_) => print('subscribed'),
  failure: (error, st, attempts) => print('failed after $attempts: $error'),
);

kit.startHealthMonitoring();
await kit.shutDown();

Most apps don't need this — PulseClient is the simpler, recommended default. Reach for PulseMqttKit specifically when API parity with pulse-droid matters to your team.

Testing your app's MQTT logic

PulseTransport is an interface — inject a fake implementation to test connect/publish/subscribe/retry/outbox logic without any real sockets:

class FakeTransport implements PulseTransport {
  bool connected = false;

  @override
  Future<void> connect() async => connected = true;

  @override
  bool get isConnected => connected;

  // ...implement the rest of PulseTransport as needed for your test
}

final client = PulseClient(
  config: config,
  transport: FakeTransport(),
);

This is exactly how pulse_flutter's own test suite validates command ordering, dispose-safety, and retry behavior — see the package's test/ directory for full examples.

Platform support

Platform	`MqttClientTransport`	`NativeForegroundTransport`
Android	✅	✅
iOS	✅	❌ (not available — use `MqttClientTransport`, or pair with iOS background modes / silent push for your background use case)
Web	✅	❌
macOS / Windows / Linux	✅	❌

FAQ / troubleshooting

"Unable to connect to server" only when I switch to background/native mode, but it works fine in the foreground. See TLS is not optional in the background — this is almost always a plaintext (tcp://1883) broker hitting Android's cleartext-traffic block. Switch to ssl:// on port 8883.

ECONNREFUSED even after switching to TLS. Confirm the broker itself actually has a TLS listener up — public test brokers occasionally have partial outages on their TLS listener specifically. Test with openssl s_client -connect your-host:8883 from a machine on the same network, or check the broker's status page.

Notification permission denied on Android 13+. The foreground service will still run and keep publishing — it's only the visible notification that's suppressed. Request Permission.notification before starting the service to avoid a silent, unexplained missing notification.

License

MIT

pulse_flutter

Table of contents

Why not just `mqtt_client`?

Features

Installation

Android setup

Quick start

Choosing a transport

TLS is not optional in the background

Retry policies

Events

Offline outbox

Auto-resubscription

Connection health monitoring

`PulseMqttKit` — the pulse-droid-parity API

Testing your app's MQTT logic

Platform support

FAQ / troubleshooting

License

Libraries

pulse_flutter package

pulse_flutter

Table of contents

Why not just mqtt_client?

Features

Installation

Android setup

Quick start

Choosing a transport

TLS is not optional in the background

Retry policies

Events

Offline outbox

Auto-resubscription

Connection health monitoring

PulseMqttKit — the pulse-droid-parity API

Testing your app's MQTT logic

Platform support

FAQ / troubleshooting

License

Libraries

pulse_flutter package

Why not just `mqtt_client`?

`PulseMqttKit` — the pulse-droid-parity API