Cancel listen handler in client.updates(Stream) when unsubscribing

[boborua]

In this example https://github.com/shamblett/mqtt_client/blob/master/example/mqtt_server_client.dart, when we need to subsribe to topic test/a, we do:

  // client initilization
  String topic = 'test/a';
  client.subscirbe(topic, MqttQos.atMostOnce);
  /// The client has a change notifier object(see the Observable class) which we then listen to to get
  /// notifications of published updates to each subscribed topic.
  client.updates!.listen((List<MqttReceivedMessage<MqttMessage?>>? c) {
    final recMess = c![0].payload as MqttPublishMessage;
    final pt =
        MqttPublishPayload.bytesToStringAsString(recMess.payload.message);

    /// The above may seem a little convoluted for users only interested in the
    /// payload, some users however may be interested in the received publish message,
    /// lets not constrain ourselves yet until the package has been in the wild
    /// for a while.
    /// The payload is a byte buffer, this will be specific to the topic
    print(
        'EXAMPLE::Change notification:: topic is <${c[0].topic}>, payload is <-- $pt -->');
    print('');
  });

When we need to unsubscribe, just do:

client.unsubscribe(topic);

Since client.updates is an instance of broadcast stream, and method subscribe and unsubscribe is just modifiy the pendingSubscription map and subscriptions map in subscriptionsManager. So, in case we need to subscribe to topic/a, then topic/b, afterwards we unsubscribe topic/b, threre are still two callback functions in stream listners. It took some time to figure this out, when I was trying to fix some bugs.

I'm a flutter developer so I have same use case with #319. It's important for application developers to do clean up actions according to App lifecycle.

So, is it suitable for a PR to do this when unsubcibing toipcs?

[shamblett]

Yes by all means submit a PR thanks

[shamblett]

Issue #510 addressed this issue, closing.