Implement a convenience interface

[bencleary]

This might not be on topic or have any specific outcome but i am looking for any guidance. I have currently implemented a simple class using the standard paho-mqtt library that effectively allows me to build MQTT applications similarly to flask like so:

@subscribe("some/topic/#")
def handler_one(payload: Payload):
    # do something

It is working fine, but i am curious on how i can swap out the standard paho-mqtt library with yours to take advantage of the asyncio, so that i can effectively do the following:

@subscribe("some/topic/#")
async def handler_one(payload: Payload):
    # do something
   await asyncio.sleep(1)

I am wondering if you could give me any guidance on using your library within a class. For example my wrapper looks like this (slimmed down):

class MQTTWrapper:
    client: mqtt.Client
    topics: Dict

    def __init__(self, connection: MQTTConnection):
        self.topics = defaultdict(list)
        self.client = mqtt.Client(clean_session=True)
        # do other things
        self.client.on_connect = self.on_connection
        self.client.on_message = self.handle_message
        self.client.connect(connection.host, connection.port)

    def handle_message(self, client, userdata, msg):
        payload: Payload = Payload(client, userdata, msg.payload, msg)
        try:
            for topic, handlers in [
                (topic, handlers)
                for topic, handlers in self.topics.items()
                if mqtt.topic_matches_sub(topic, msg.topic)
            ]:
                for handler in handlers:
                    handler(payload)
        except Exception:
            raise TopicNotRegisteredException()

    # -------------------------
    # other methods
    # -------------------------

    def subscribe(self, topic):
        @wraps(topic)
        def wrapper(handler):
            if not self._is_method(handler):
                self.client.subscribe(topic)
                self.topics[topic].append(handler)
            handler.subscription = True
            handler.topic = topic
            return handler

        return wrapper

    def start(self, background_thread: bool = False):
        try:
            self._log_subscriptions()
            if background_thread:
                self.client.loop_start()
            else:
                self.client.loop_forever()
        except KeyboardInterrupt:
            self.client.loop_stop()
            self.client.disconnect()

    def stop(self):
        self.client.loop_stop()
        self.client.disconnect()

It might just be a flat no, which is fine, but any advice or guidance would be useful, thanks for your library!

TIA.

[frederikaalund]

Hi Ben.

Thanks for opening this discussion. Let me have a look. :) I don't see the "discussions" page that often, btw. I mostly attend the "issues" page. Hence the delay.

It's definitely possible to make an "application framework" wrapper around asyncio-mqtt. Of course, everything is `async' instead.

Something like this (not tested, there may very well be syntax/semantic errors):

#################################################
### The `MqttApp` library
#################################################
import asyncio

from asyncio_mqtt import Client


class MqttApp:
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        self._client = Client(*args, **kwargs)
        self._handlers: list[...] = []

    def run(self) -> None:
        asyncio.run(self._run())

    async def _run(self) -> None:
        async with self._client:
            await asyncio.gather(*self._handlers)

    def subscribe(self, topic: str) -> ...:
        def _decorator(coro: ...) -> ...:
            async def _handler() -> None:
                await self._client.subscribe(topic)
                async for message in messages:
                    await coro(message)
            self._handlers.append(coro)
            return coro
        return _decorator



#################################################
### Use case example
#################################################
app = MqttApp()


@app.subscribe("some/topic/#")
async def handler_one(message: mqtt.Message) -> None:
    # do something
   await asyncio.sleep(1)


app.run()

You're welcome to do a PR with a code example of this approach. 👍

~Frederik