Subscribe conditionally/later

[swalkner]

Hello,

All the samples I see seem pretty much equal:
Connect - then - subscribe (to '#' or one topic) - then - consume and process messages.

Is there a mechanism where I can subscribe later (e.g. when some event/condition is triggered) to a new topic?
One method would be to subscribe to '#' and consume everything received, then filter topics, but I assume that's not what 'publish/subscribe' is intended for...

Pseudo-Code, what I would like to achieve:

async with Client(...) as client:
    async with client.messages() as messages:
        await client.subscribe("humidity/#")
        async for message in messages:
            print(message.payload)
            if message.topic.endswith("specialsensor"):
                   --> await client.subscribe("specialsensor/topic")

I hope it's clear what my problem is.
Thanks so much in advance,
stefan

[frederikaalund]

Hi Stefan, thanks for opening this discussion. Let me chip in. :)

You can subscribe (and unsubscribe) at any point within the Client context. 👍 In other words, the pseudocode that you posted should work. Did you try it out yet? I have yet to see a real-world use case for it but it should work. :)

Let me know if something is unclear. :)

Best regards,
~Frederik

[swalkner]

Hello,

Thanks for the fast reply.
Okay - my question was a bit unspecific... :-/

Where do I consume messages that are received on "specialsensor/topic"?

In the following example both subscriptions will be issued and the message-loop will receive messages from both topics.

async with Client(....) as client:
    async with client.messages() as messages:
        await client.subscribe("/my/test1")
        await client.subscribe("/my/test2")
        async for m in messages:
            print(f'a: {m.payload.decode()}')

Now - within the message-loop I'd have to filter, which topic received the given message. Is that correct?
How could I consume topics within different methods? Something like:

async def handle_topic(self, client, topic):
    await client.subscribe(f"my/{topic}")
    async with client.messages() as messages:
        async for m in messages:
            print(f'{topic}: {m.topic} {m.payload.decode()}')

async def run_events(self):

    async with Client(...) as client:
        await self.handle_topic(client, 'test1')
        await self.handle_topic(client, 'test2')

Within this trivial example I could create tasks and asyncio.gather them, to have both 'handle_topics' getting dispatched.

async with Client(...) as client:
    tasks = set()
    tasks.add(asyncio.create_task(self.handle_topic(client, 'test1')))
    tasks.add(asyncio.create_task(self.handle_topic(client, 'test2')))

    await asyncio.gather(*tasks)

But both handle_topics - again - would receive all messages that have been subscribed too and another filtering would be required.

Is my assumption correct, that you can have multiple loops around 'client.messages()' - but you'll have to filter within these loops to only handle messages you're interested in?

I hope my intentions are clearer now...
Thanks so much in advance,
stefan

[frederikaalund]

You're already well on your way to a good solution with that last code example that uses asyncio.gather. 👍

Is my assumption correct, that you can have multiple loops around 'client.messages()' - but you'll have to filter within these loops to only handle messages you're interested in?

Yes, it's correct. 👍

Under the hood, the client just receives a stream of messages that matches the subscribed topic(s). If you want further filtering on top of that, you'll have to explicitly do that. Basically, you're implementing a "virtual client" on top of the real client. The real client gets all the messages. The "virtual clients" get messages based on your chosen topic. Another way too look at it is that you're multiplexing the various topics through a single client. Again, this is completely fine but you'll have to do the multiplexing (the filtering) on top of the client. I hope these examples make sense. 😅

Try to have a look at https://sbtinstruments.github.io/asyncio-mqtt/publishing-and-subscribing.html#filtering-messages

This shows how you can use if message.topic.matches(topic): inside your handle_topic function. This is the official way on how to filter based on topic. :)

[swalkner]

Thanks for the detailed explanation.
I kind of tried to find the link between subscription and consuming the subscribed messages.
Now it makes sense and I can find a way around handling the right messages in the right places :-)
thanks,
stefan