Listen to specific topic.

[uwejan]

This mighty sound silly, but how can i listen as stream for specific topic? just tried the pub example listener block of code, seemed i got it to missed up with my implementation, where have multiple sub with different payloads of json. Edit: while waiting have tried multiple things without luck, really stuck on this appreciate response.

    client.updates.listen((dynamic c) {
      final MqttReceivedMessage mm = MqttReceivedMessage('dummy/topic/third', c[0].payload);
      final String tt = MqttPublishPayload.bytesToStringAsString(mm.payload);
      //tt no luck
      //final String gg  = MqttByteBuffer.readMqttString(mm.payload);
      //gg no luck

      print(
          'EXAMPLE::Change notification:: topic is <${c[0].topic}>, payload is <-- $tt -->');
    });

[shamblett]

At the mo you can't do this, the listen interface will just pump out messages for everything you have subscribed to, you have to do the topic filtering yourself. I am going to add a topic filter class that you will be able to attach to the listener, set topics in which you are interested and listen on it rather than the client just not got round to it yet. You can of course listen multiple times so you can of course get everything and/or filter as you wish.

[uwejan]

That would be great. Can you show me an example on how to filter a topic please.

[shamblett]

Thinking about this I could use this ticket to implement topic filtering as I've been intending for a while but this would take a week or so, if you want an example quickly I can probably provide one in the next few days

[uwejan]

Yes. I will wait for this.

[shamblett]

OK this where I'm going with this, from my development branch

/// This class allows specific topics to be listened for. It essentially /// acts as a bandpass filter for the topics you are interested in if /// you subscribe to more than one topic or use wildcard topics. /// Simply construct it, and listen to its message stream rather than /// that of the client. class MqttClientTopicFilter { /// Construction MqttClientTopicFilter(this._topic, this._clientUpdates) { _subscriptionTopic = SubscriptionTopic(_topic); _clientUpdates.listen(_topicIn); _updates = StreamController<List<MqttReceivedMessage>>.broadcast( sync: true); }

String _topic;

SubscriptionTopic _subscriptionTopic;

/// The topic on which to filter String get topic => _topic;

Stream<List<MqttReceivedMessage>> _clientUpdates;

StreamController<List<MqttReceivedMessage>> _updates;

/// The stream on which all matching topic updates are published to Stream<List<MqttReceivedMessage>> get updates => _updates.stream;

void _topicIn(List<MqttReceivedMessage> c) { // Pass through if we have a match if (_subscriptionTopic.matches(PublicationTopic(c[0].topic))) { _updates.add(c); } } }

So you just construct one of these and listen on its update interface rather than that of the client, then you get filtered topics.

[uwejan]

Great!

1 how can i link flutter pubspec to dev?

2 have not tested it yet, but i will right away, i was running client on my main, and importing it on other screens, with this way of filtering, no need for the client anymore to be imported yes? just subscribe on main, and run the filter on wanted screen, yes?

[shamblett]

For 1 you can add this to your pubspec, replacing your existing mqtt_client entry

mqtt_client: git: url: https://github.com/shamblett/mqtt_client.git ref: development

see here

Can't really help with point 2 as I don't use flutter

[uwejan]

Pretty sure you do not expect such questions; bare with me please. is this how am supposed to get the filter to work. Because its not working this way, i get many errors. String topic 'dummy/topic/third'; var updates MqttClientTopicFilter clientTopicFilter = MqttClientTopicFilter(topic, updates); // then i do what i want to do with updates of that specific filter.

[shamblett]

The 'updates' variable is the one on the client, so do something like this

String topic 'dummy/topic/third'; var client = MqttClient('test.mosquitto.org', ''); //etc etc MqttClientTopicFilter clientTopicFilter = MqttClientTopicFilter(topic, client); // Then instead of listening on the client for updates listen on your filter clientTopicFilter .updates.listen((List<MqttReceivedMessage> c) {.........}

Note this is early code, I've not written any unit tests yet or done any other testing so I'm not saying its working, its just where I am.

[shamblett]

Topic filtering now added in version 5.5.0 and re-published