Accessing CONNACK properties

[airtower-luna]

I'm looking to access MQTT v5 properties sent by the server in CONNACK, primarily Response Information, but access to others may be useful, too. The only way I found is hooking into aiomqtt.Client._on_connect(), e.g.:

class PropClient(aiomqtt.Client):
    def _on_connect(
            self,
            client: mqtt.Client,
            userdata: Any,
            flags: mqtt.ConnectFlags,
            reason_code: ReasonCode,
            properties: Properties | None = None) -> None:
        super()._on_connect(
            client, userdata, flags, reason_code, properties)
        self.connack_properties = properties

This is obviously undesirable because it's using (and modifying) internal API. Am I missing something, or is this currently missing from the API? This seems like a key piece for MQTT v5 request/response.

[empicano]

Hi there! Yes, we're currently ignoring the CONNACK properties completely. I'd be very happy to see request/response fully functional in aiomqtt 🙂

For PUBLISH messages we receive from the broker, we currently expose all properties via aiomqtt.Message.properties, which simply returns paho's Properties type. I don't think that's the best way to do it. Something like the subscription identifier property is probably not interesting to the user and should thus not be unnecessarily exposed.

I'd like to only expose CONNACK properties that are useful to the user. What CONNACK properties apart from response information do you think might be useful to access?

---

Some unsorted thoughts:

• CONNACK properties will only be available inside the context manager, defining them as standard properties of Client might make the typing cumbersome, because it's going to be a union with None
• CONNACK properties might change on reconnection (?)
• Can we make the interface to send requests (and receive responses -- see also Improve message routing #304) or the interface to send responses more straightforward than it is now without loosing functionality?

I'd be super happy to hear what you think!

[airtower-luna]

Because adding handlers dynamically is actually not a problem. Inside the client, we can then keep a mapping of correlation data to handlers. What do you think about that? Do you see that lacking somewhere? 🤔

The main issue I see there is: How long should response handlers stay enabled? If they're never removed the mapping will grow indefinitely as requests are sent. On the other hand as you pointed out before we might want to handle multiple responses per request, so removing a handler after the first response wouldn't be correct either, and if it was, we'd still need to handle the possibility that we never get any response. That's what the appeal of iterating using async for like for client.messages would be to me: It'd be easy to wrap that in a timeout, or break on some condition (though we'd still need to trigger cleanup on leaving the loop).

I guess we'd need some kind of handle that can be used to remove the request handler, possibly to be used as async context manager that does that on __aexit__? Or maybe for each request create a task that'll remove the response handler after a timeout, or earlier if cancelled (that'd require returning the task object from the request function)? Timeout seems suitable here because it allows making requests without worrying about explicit cleanup, as long as the timeout is short enough that the handler mapping won't grow to an unreasonable size (whatever that means in a given context), and I assume most people would want to use one for waiting on responses anyway.

Another thing is that people might want to pass additional data to handlers and use the results elsewhere, but that could be solved with closures, functools.partial(), or by passing a method of an object that holds the necessary state, so it doesn't necessarily have to be part of the API.

Makes sense 👍 If we go with the above, I think we could implement both request and response by adding some optional arguments to Client.publish(), but I think you convinced me that it's a good idea to have proper request() and respond() convenience functions. I think that I'd like to have the respond() function as member of the client, not as member of the message, though, somehow it doesn't seem clean to me to pass the client around as part of each message.

Sounds good to me! 🙂

[empicano]

The main issue I see there is: How long should response handlers stay enabled?

That's a good point, and you're right that it would be easier to do that with an async for loop. I don't see a way how we can could have that and not use additional queues, and I think the single queue is really important 🙁

I'm thinking that the reasons we would want to remove a response handler should be relatively few. I can think of:

• Timeout (e.g. stop handling responses after 10s)
• Limited number of messages (e.g. stop after the first message)
• Certain type of message (e.g. stop after we get a stop=True message)

We could implement the first two as optional arguments to Client.request() e.g. Client.request("example", response_topic="response/topic", response_handler=handle_bar, timeout=10), for the third one, we could make the message router of the client directly accessible, so that the user can call Client.router.remove("response/topic") from inside the message handler.

I feel like that somehow leads us down a bad rabbit hole, though. Maybe the optional handlers on a request-basis that I proposed are already one step too far, and we should keep it simpler. How about:

• Add Client.Router.insert() and Client.Router.remove() functions to add/remove message handlers together with Improve message routing #304
• Add Client.request() and Client.respond/reply() or Request.respond/reply() convenience functions
• Expose CONNACK properties as properties of the Client

That's kind of chickening out from tracking correlation data and automating handler expiry and letting the user handle that instead, I admit that I don't have a good solution at the moment that still let's us keep the single queue. But it gives a few tools to make request/reponse easier, and shouldn't abstract so much that we're loosing functionality. The last point also addresses your original problem with regarding Response Information.

Another thing is that people might want to pass additional data to handlers and use the results elsewhere

You already mention a few options, can you give an example of how your preferred solution would look like? You already proposed to attach the client to the message object instead of passing it as the second argument to the handlers. We could take this one step further and do something like Starlette's lifespan state to attach a user-defined "state" object to each message.

[airtower-luna]

I feel like that somehow leads us down a bad rabbit hole, though. Maybe the optional handlers on a request-basis that I proposed are already one step too far, and we should keep it simpler. How about:

• Add Client.Router.insert() and Client.Router.remove() functions to add/remove message handlers together with Improve message routing #304
• Add Client.request() and Client.respond/reply() or Request.respond/reply() convenience functions
• Expose CONNACK properties as properties of the Client

That seems reasonable to me. Together with V5 content type support (which should be fairly trivial) that'd simplify my code a lot already.

The one thing that seems odd to me is that I'd have to subscribe and set up routing for the response topic separately on the client that sends the request, unless I want to use the exact same handler for each request.

You already mention a few options, can you give an example of how your preferred solution would look like? You already proposed to attach the client to the message object instead of passing it as the second argument to the handlers. We could take this one step further and do something like Starlette's lifespan state to attach a user-defined "state" object to each message.

What I want is the same thing I can do with any thread, pass arguments to the handler function. Let's say I want to request some status report, and then process all the results, the code might end up roughly like this:

async def handle_report(message, client, corr: bytes, q: asyncio.Queue):
    if not hasattr(message.properties, 'CorrelationData') \
            or message.properties.CorrelationData != corr:
        # not for us
        return
    data = json.loads(message.payload)
    await q.put(data)


async def process_reports(client, q, r_topic):
    try:
        async with asyncio.timeout(10):
            data = await q.get()
            # do stuff with the data
    except TimeoutError:
        # okay, done.
        pass
    await client.unsubscribe(r_topic)
    client.router.remove(r_topic)
    return ...


async def keep_routing(client):
    async for message in client.messages:
        await client.route(message)


async with aiomqtt.Client('test.mosquitto.org') as client:
    q = asyncio.Queue()
    corr = secrets.token_bytes(8)
    # might use Response Information from CONNACK
    r_topic = f'response/status/{secrets.token_hex(8)}'
    client.router.insert(r_topic, handle_message, args=(corr, q))
    await client.subscribe(r_topic)
    await client.request(
        'device/send-status', response_topic=r_topic, correlation_data=corr)
    task1 = asyncio.create_task(process_reports(client, q, r_topic))
    task2 = asyncio.create_task(keep_routing(client))
    stuff = await task1
    task2.cancel()

Using kwargs might make more sense, but point is, I need to be able to pass data around, including to check the correlation data. Maybe I could limit the subscription to response/status/+ and leave it active, but I think you can see my point here.

[empicano]

The one thing that seems odd to me is that I'd have to subscribe and set up routing for the response topic separately on the client that sends the request, unless I want to use the exact same handler for each request.

We basically have three options for routing, that differ in how we handle overlapping subscriptions:

• We can route a message to the most specific handler that matches
• We can route it to all handlers that match
• We could assign each subscription a specific handler. Most brokers support subscription IDs, which would let us then match incoming messages to the exact subscription(s) -- and thus handler(s) -- that caused it

For reference, the MQTT spec states:

When Clients make subscriptions with Topic Filters that include wildcards, it is possible for a Client’s subscriptions to overlap so that a published message might match multiple filters. In this case the Server MUST deliver the message to the Client respecting the maximum QoS of all the matching subscriptions [MQTT-3.3.4-2]. In addition, the Server MAY deliver further copies of the message, one for each additional matching subscription and respecting the subscription’s QoS in each case.

Leveraging subscription IDs would probably be the cleanest way to go. What I like about the first option is the simple interface:

async for message in client.messages:
    await client.route(message)

Otherwise, we would have to deal with potentially multiple handlers:

async for message in client.messages:
    for handler in client.route(message):
        await handler(message)

Although, maybe we might be able to duplicate incoming messages that have multiple subscription IDs assigned? That would standardize how aiomqtt works, even with different brokers.

We're really touching all details here, I like it 😄

[airtower-luna]

• We can route a message to the most specific handler that matches

That seems more intuitive than "all matching handlers" to me.

• We could assign each subscription a specific handler. Most brokers support subscription IDs, which would let us then match incoming messages to the exact subscription(s) -- and thus handler(s) -- that caused it

I agree, that seems like the most elegant solution, because it avoids having to set up subscription and routing separately when making a request. It'll require a slight change to Client.unsubscribe() to allow cancelling subscriptions by ID (instead of topic). I'm thinking of making a request, and canceling the response subscription (and thus handler) when done processing responses by whatever logic fits in the given context. This would solve the issue of being able to parametrize response handlers.

I'm not sure how it'd integrate with your routing concept, though. Would a subscription-specific handler override the default routing?

Although, maybe we might be able to duplicate incoming messages that have multiple subscription IDs assigned? That would standardize how aiomqtt works, even with different brokers.

That seems useful, otherwise code might simply break depending on the broker it's used with, e.g. if it expects to receive messages for all matching handlers, but the broker sends the message only once. Either way I think it'd be better to handle that internally in the routing code, and not require user code to add boilerplate for calling handlers.