Advanced Example w/o filtered_messages available?

[avollkopf]

I am maintaining a repo that is using asyncio-mqtt. I just noticed, that the filtered_messages is deprecated and will be removed from future versions. I changed one part of the code successfully. However, it uses also the advanced example of the old readme to cancel tasks on exit.

This example has now been removed recently from the readme and I cannot find an example in the new documentation. Will there be something similar like the advanced example in upcoming versions of the documentation?

I was not able to find any similar discussion.

[frederikaalund]

Hi Alexander, thanks for opening this discussion.

However, it uses also the advanced example of the old readme to cancel tasks on exit.

Will there be something similar like the advanced example in upcoming versions of the documentation?

Yes, in fact it is already there: https://sbtinstruments.github.io/asyncio-mqtt/listening-without-blocking.html#listening-without-blocking
asyncio.TaskGroup replaces asyncio.gather along with all the manual tracking and housekeeping of individual tasks. Basically, asyncio.TaskGroup maintains the tasks list from the advanced example and ensures proper cancellation (I'm simplifying a lot here―there is a lot more to asyncio.TaskGroup behind the scenes).

Note that asyncio.TaskGroup first appeared in python 3.11. If you're on an older version, you can use anyio.create_task_group instead.

I understand your confusion. The advanced example suddenly vanished from out documentation/readme/examples without a migration guide. If you are up for it, I'd gladly review a PR that adds a migration guide to the documentation that shows how to move from asyncio.gather to asyncio.TaskGroup. 👍

[avollkopf]

H Frederik, thank you for your prompt response.

I have seen this example, but was a bit confused :-). I will look into it and give feedback, once I have adapted it to my needs. Likewise, I just need some time as I am working on the brewing software just in my free-time.

Alex

[avollkopf]

Hi @frederikaalund , I have found a solution that seems to be working. It is based on your fire and forget example as the mqtt code is only part of a sever with different routines in parallel. I am using now different topic filters and call corresponding methods.

...........................
        self.topic_filters = [
            ("cbpi/actor/+/on", self._actor_on),
            ("cbpi/actor/+/off", self._actor_off),
            ("cbpi/actor/+/power", self._actor_power),
            ("cbpi/updateactor", self._actorupdate),
            ("cbpi/updatekettle", self._kettleupdate),
            ("cbpi/updatesensor", self._sensorupdate),
            ("cbpi/updatefermenter", self._fermenterupdate),
        
        ]
        self.tasks = set()

    async def init(self):

        
        self.client = Client(self.host, port=self.port, username=self.username, password=self.password, will=Will(topic="cbpi/disconnect", payload="CBPi Server Disconnected"),client_id=self.client_id)
        self. Loop = asyncio.get_event_loop()
        ## Listen for mqtt messages in an (unawaited) asyncio task
        task = self.loop.create_task(self.listen())
        ## Save a reference to the task so it doesn't get garbage collected
        self.tasks.add(task)
        task.add_done_callback(self.tasks.remove)

        self.logger.info("MQTT Connected to {}:{}".format(self.host, self.port))

    async def listen(self):
        while True:
            try:
                async with self.client as client:
                    async with client.messages() as messages:
                        await client.subscribe("#")
                        async for message in messages:
                            for topic_filter in self.topic_filters:
                                topic = topic_filter[0]
                                method = topic_filter[1]
                                if message.topic.matches(topic):
                                    await (method(message))
            except MqttError as e:
                self.logger.error("MQTT Exception: {}".format(e))
            except Exception as e:
                self.logger.error("MQTT General Exception: {}".format(e))
            await asyncio.sleep(5)
.........................................