Broken messages generator with client context re-use

[ndf-zz]

Hi, the client context reconnect example in reconnection.html does not work as expected in my testing due to what looks like reuse of an invalidated messages generator.

To see the issue in action, add a few print statements to the example code, run it and trigger a re-connect:

async def main():
    client = aiomqtt.Client("test.mosquitto.org")
    interval = 5  # Seconds
    while True:
        try:
            print(f"Entering client context...")
            async with client:
                await client.subscribe("humidity/#")
                async for message in client.messages:
                    print(message.payload)
                print(f"Messages generator ends")
        except aiomqtt.MqttError:
            print(f"Connection lost; Reconnecting in {interval} seconds ...")
            await asyncio.sleep(interval)

This first time through, the async for loop works as expected, but when the client is re-used after a loss of connection, the messages generator ends and the code will loop connecting then immediately disconnecting.

Sorry, I'm not sure where the fix for this ought to go, nor what the proper use of the client.messages should to be - but the following change seems to work:

• in client.init() set client.messages to None
• in client.aenter() assign client.messages to client._messages()
• in client.aexit() assign client.messages to None

Alternatively, adjusting the example code for loop as follows also works:

async for message in client._messages():

Setup:

• Python 3.11.2 / Debian 12 amd64
• paho-mqtt 1.6.1

[empicano]

Hi there,

Thanks for opening this issue, good find! 👍

I'm not sure where the fix for this ought to go, nor what the proper use of the client.messages should to be

The client context manager is reusable, so we should also be able to reuse client.messages as well. Your changes to init, aenter, and aexit look fine to me. Would you like to make a PR (in the ideal case including a test case)? 🙂

---

Apart from that, I wonder if we should start allowing the use of client.messages outside the client's context:

import asyncio
import aiomqtt


async def main():
    client = aiomqtt.Client("test.mosquitto.org")
    async with client:
        await client.subscribe("humidity/#")
        await asyncio.sleep(5)
    async for message in client.messages:
        print(message.payload)


asyncio.run(main())

This is related to #48 which argues that we should allow calls to publish() outside of the context manager.

This would require that the client.messages generator does no longer throw an exception if the client is disconnected. However, I'm not sure how we would alert to connection losses in this case. @frederikaalund, what are your thoughts on this? 🙂

Edit: Our current implementation of client.messages throws the exception only after the messages queue is empty. This means that reconnection can potentially take some time and lead to message loss.

[spacemanspiff2007]

I'm hit by this issue after updating to 2.0, too.
I can also confirm that assigning client._messages() to client.messages after client.__aenter__() fixes the issue.

I'm not sure how we would alert to connection losses in this case.

It makes sense that client.messages alerts in case of a connection loss.
If that would not be the case we'd need something else, e.g. await client.await_disconnect() which seems strange.

Binding both publish() and messages to the client is something that I would intuitively expect and having e.g. publish() work outside the context manager would be rather surprising.