pulse_mqtt 1.0.1

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

example/lib/main.dart

import 'package:flutter/material.dart';
import 'package:pulse_mqtt/pulse_mqtt.dart';

void main() => runApp(const PulseMqttExampleApp());

/// Bridge implementation — the Flutter equivalent of `PulseMqttKitBridgeImpl`.
class ExampleBridge implements PulseMqttKitBridge {
  @override
  Logger? getLogger() => _ConsoleLogger();

  @override
  bool get enableJsonDeserialization => true;

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

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

class _ConsoleLogger implements Logger {
  @override
  void debug(String message) => debugPrint('PulseMQTT D: $message');
  @override
  void info(String message) => debugPrint('PulseMQTT I: $message');
  @override
  void warning(String message) => debugPrint('PulseMQTT W: $message');
  @override
  void error(String message, [Object? throwable]) =>
      debugPrint('PulseMQTT E: $message $throwable');
}

class PulseMqttExampleApp extends StatelessWidget {
  const PulseMqttExampleApp({super.key});

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Pulse MQTT Live Tracking',
      theme: ThemeData(colorSchemeSeed: Colors.red, useMaterial3: true),
      home: const HomePage(),
    );
  }
}

class HomePage extends StatefulWidget {
  const HomePage({super.key});

  @override
  State<HomePage> createState() => _HomePageState();
}

/// Mirrors the Android `MainActivity`: connect / subscribe / publish + status.
class _HomePageState extends State<HomePage> with MqttUpdatesListener {
  final _kit = PulseMqttKit();

  final _serverUri =
      TextEditingController(text: 'tcp://broker.hivemq.com:1883');
  final _clientId = TextEditingController(text: 'pulse-flutter-demo');
  final _username = TextEditingController();
  final _password = TextEditingController();
  final _subscribeTopic = TextEditingController(text: 'pulse/demo/location');
  final _message = TextEditingController(text: 'hello from flutter');

  String _status = 'Status: idle';
  String _lastMessage = 'Message: -';
  String? _subscribedTopic;

  @override
  void initState() {
    super.initState();
    _kit.initialize(ExampleBridge());
    _kit.addListener(this);
  }

  @override
  void dispose() {
    _kit.shutDown();
    _kit.removeListener(this);
    for (final c in [
      _serverUri,
      _clientId,
      _username,
      _password,
      _subscribeTopic,
      _message,
    ]) {
      c.dispose();
    }
    super.dispose();
  }

  void _connect() {
    _kit.submitCommand(
      ConnectCommand(
        connectionOptions: ConnectionOptions(
          serverUri: _serverUri.text,
          clientId: _clientId.text,
          username: _username.text.isEmpty ? null : _username.text,
          password: _password.text.isEmpty ? null : _password.text,
          connectionTimeoutSeconds: 120,
          keepAliveIntervalSeconds: 60,
          automaticReconnect: true,
          autoSubscriptionConfig: const AutoSubscriptionConfig(enabled: true),
        ),
        retryPolicy: RetryPolicy.exponential(
          maxRetries: 3,
          baseDelayMillis: 2000,
          excludedExceptionCodes: {
            MqttExceptionCode.reasonCodeNotAuthorized,
            MqttExceptionCode.connectAlreadyInProgress,
          },
        ),
      ),
    );
  }

  void _subscribe() {
    final topic = _subscribeTopic.text;
    if (topic.isEmpty) return;
    _subscribedTopic = topic;
    _kit.submitCommand(
      SubscribeCommand(
        topicConfigs: {
          topic: TopicTypeConfig<String>(qosLevel: QOSLevel.qos0),
        },
      ),
    );
  }

  void _publish() {
    final topic = _subscribedTopic;
    if (topic == null || topic.isEmpty) return;
    _kit.submitCommand(
      PublishCommand(
        message: ZMqttMessage(topic: topic, payload: _message.text),
        qos: QOSLevel.qos0,
      ),
    );
  }

  // ---- MqttUpdatesListener callbacks ----

  @override
  void onCommandSuccess(MqttCommand command, Success result) {
    setState(
        () => _status = 'Status: Success of command type: ${command.type}');
  }

  @override
  void onCommandFailure(MqttCommand command, Failure result) {
    setState(() => _status =
        'Status: Failure of command type: ${command.type}, reason: ${result.error}');
  }

  @override
  void onCommandIgnored(MqttCommand command, Ignored result) {
    setState(() => _status =
        'Status: Ignored of command type: ${command.type}, reason: ${result.reason}');
  }

  @override
  void onMqttConnectionLost(Object? cause) {
    setState(() => _status = 'Status: Connection lost due to: $cause');
  }

  @override
  void onMqttMessageReceived(String? topic, String? payload,
      [TopicMessage? topicMessage]) {
    setState(() => _lastMessage = 'Message: $payload on topic: $topic');
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('Pulse MQTT Live Tracking')),
      body: SingleChildScrollView(
        padding: const EdgeInsets.all(16),
        child: Column(
          crossAxisAlignment: CrossAxisAlignment.stretch,
          children: [
            _field(_serverUri, 'Server URI'),
            _field(_clientId, 'Client ID'),
            _field(_username, 'Username (optional)'),
            _field(_password, 'Password (optional)', obscure: true),
            const SizedBox(height: 8),
            FilledButton(onPressed: _connect, child: const Text('Connect')),
            const Divider(height: 32),
            _field(_subscribeTopic, 'Subscribe topic'),
            FilledButton.tonal(
                onPressed: _subscribe, child: const Text('Subscribe')),
            const SizedBox(height: 16),
            _field(_message, 'Message payload'),
            FilledButton.tonal(
                onPressed: _publish, child: const Text('Publish')),
            const Divider(height: 32),
            Text(_status, style: Theme.of(context).textTheme.titleMedium),
            const SizedBox(height: 8),
            Text(_lastMessage),
          ],
        ),
      ),
    );
  }

  Widget _field(TextEditingController c, String label, {bool obscure = false}) {
    return Padding(
      padding: const EdgeInsets.symmetric(vertical: 4),
      child: TextField(
        controller: c,
        obscureText: obscure,
        decoration: InputDecoration(
            labelText: label, border: const OutlineInputBorder()),
      ),
    );
  }
}