[How To] Share the same MQTT connection between multiple classes/widgets?

[pranjal-joshi]

Hello, I am having multiple screens/pages and a lot of widgets in my app. Currently, I created a class as follow to create a MQTT connection which takes onReceive method from its constructor.

class MyMqtt {
  final String ip;
  final int port;
  MqttClient client;
  MqttConnectionState connectionState;
  StreamSubscription subscription;
  final Function onDisconnected;
  Function onReceive;
  Function onSubscribed;
  Function onConnected;
  Function onAutoReconnect;

  MyMqtt({
    this.ip = "My_Ip",
    this.port = My_port,
    @required this.onDisconnected,
    @required this.onReceive,
    @required this.onConnected,
    this.onSubscribed,
    this.onAutoReconnect,
  });
   ....
  // Some other stuff here to create and manage the connection
   ....
}

Problem:

However, the problem is this will create multiple connections that are simultaneously opened to the broker. All I want to do is use the same connection state with all widgets but with the different onReceive methods on each of them so that they can behave properly as per the incoming messages.

Potential Solution?? (It didn't work)

Should I create static MqttClient where my app begins and then pass this client to the above class in a constructor along with other arguments?? Update: Making MqttClient static makes all the callback methods also static! So it simply executes the latest callback method assigned which is not desirable

[shamblett]

The client wasn't really designed for this usage, you can use multiple clients to connect to multiple different brokers or use multiple clients to connect to a single broker with different clients id's in each of these cases the callbacks are attributes of the client, not attributes of the received data as you seem to want.

I would make your class a singleton so there is only one instantiation and one receive callback, then make some kind of data distribution class that distributes the received data to anyone who wants it. I can't see why you would want to instantiate more than one client for the use case you describe.

Note I don't use flutter so I don't know how you would do this but there are plenty of flutter users who use the client so I'm assuming it must be possible.

[pranjal-joshi]

Consider I am having two classes that build certain UI. So when both these classes are instantiated, they will also initiate there own mqtt client to communicate with the broker.

Sharing same connection but different callbacks between multiple classes saves resources as well as reduce the latency introduced while making a new connection.

[blisssan]

What I did was, use a provider in root of my app & passed a singleton instance of a custom class(I called it MQTTService) similar to what you wrote. So in whichever widget you want to access the instance of your MQTT service class, you can just use Provider.of(context, listen:false); The way I handled onDisconnected, onConnected, onPublished is by using a Broadcast Stream in my service class that broadcasts events & data just similar to the "updates" stream in the MqttClient. This way you dont have to pass callbacks to your service class. You can just listen for Events & Data from the Service class as required.

[pranjal-joshi]

@blisssan Thank you very much for guiding me in the correct direction! I am attaching a code snippet here just in case if anyone needs to achieve similar functionality.

Solution

I implemented my SmartMqtt class like:

class SmartMqtt {
  String ip;
  int port;

  MqttClient client;
  MqttConnectionState connectionState;
  StreamSubscription subscription;
  bool isConnected = false;

  // <<--------------- Important --------------->>
  final StreamController _controller = StreamController<dynamic>.broadcast();
  Stream<dynamic> get stream => _controller.stream;

  // <<--------------- Important --------------->> 
  final StreamController _subscriptionController = StreamController<String>();
  final StreamController _unsubscriptionController = StreamController<String>();

  // <<--------------- Important --------------->> Make this class a 'Singleton'
  static final SmartMqtt _instance = SmartMqtt._internal();
  SmartMqtt._internal();

  factory SmartMqtt({
    String ip = BROKER_IP,
    int port = BROKER_PORT,
  }) {
    _instance.ip = ip;
    _instance.port = port;
    return _instance;
  }
.. Other Methods goes here ..
}

Then, I added implemented streams for subscription and un-subscription in the onConnected method as follow:

        // Setup Client before connecting, feel free to add additional params like keepAlive  etc.
        client.onConnected = () {
        _subscriptionController.stream.listen((topic) {      //<---- Created a Subscribe stream listener
          client.subscribe(topic, MqttQos.exactlyOnce);
          if(debug)
            print('[SmartMqtt] Subscribed -> $topic');
        });
        _unsubscriptionController.stream.listen((topic) {  // <---- Created a Unsubscribe stream listener
          client.unsubscribe(topic);
          if(debug)
            print('[SmartMqtt] Unsubscribed -> $topic');
        });
      };

  void subscribe(String topic) {
    _subscriptionController.sink.add(topic);
  }

  void unsubscribe(String topic) {
    _unsubscriptionController.sink.add(topic);
  }

Similarly, inside a connect function, I added a BroadcastStream for broadcasting the received messages:

....
          subscription = client.updates.listen((event) {
          final MqttPublishMessage recMessage =
              event[0].payload as MqttPublishMessage;
          final String msg = MqttPublishPayload.bytesToStringAsString(
              recMessage.payload.message);

           // <<--------------- Important --------------->>
          _controller.sink.add(msg);
        });
....

With this setup, Now I can share a single MQTT connection from any class as the SmartMqtt class is Singleton. And the received messages can be accessed as:

// Stream Listener callback that handles received message
mqtt.stream.asBroadcastStream().listen((msg) {
      // Do something nice here with received message
   }
});

// Subscribe and Unsubscribe to the topics on-the-fly like this
mqtt.subscribe(topic);
mqtt.unsubscribe(topic);

I Hope this will help other people! Happy Fluttering :) @shamblett As of now, issue is resolved hence closing the topic.

[adrianburkard]

Hi

Thanks for your solution. I just can't really get my head around it. Could you post your full code or link to a github repo with the full code. Maybe I can better understand it then.

Thanks and best regards