Reconnection example throws "Does not support reentrant"

[westbywest]

Adapting the reconnection example to throw exceptions verbosely:

import asyncio
import aiomqtt

async def main():
    client = aiomqtt.Client("testbad.mosquitto.org")
    interval = 5  # Seconds
    while True:
        try:
            async with client:
                async with client.messages() as messages:
                    await client.subscribe("humidity/#")
                    async for message in messages:
                        print(message.payload)
        except aiomqtt.MqttError as e:
            print(f"MQTT error {e}; Reconnecting in {interval} seconds ...")
            await asyncio.sleep(interval)

asyncio.run(main())

Subsequent reconnection attempts actually will always fail on "Does not support reentrant," even if the target host becomes available.

MQTT error [Errno -2] Name or service not known; Reconnecting in 5 seconds ...
MQTT error Does not support reentrant; Reconnecting in 5 seconds ...
MQTT error Does not support reentrant; Reconnecting in 5 seconds ...

For reconnection to succeed, I have to move client = ... statement to within the while loop. Is this intended usage, and the example happens to be incorrect?

Observed under python 3.10.7, asyncio 3.4.3, aiomqtt 1.1.0.

[vvanglro]

@westbywest Hi, I can't reproduce it. I use the same code. After restarting the mqtt service. It can still be reconnected normally.

[empicano]

Hi Ben,

Thanks for the minimal example, I can reproduce it on my side 👍 Moving the client statement inside the while loop is a good option in the meantime, but we do want it to work like in the example (without re-initializing the client instance), so it seems there's something broken, good catch!

For reference, we want the client to be reusable, but not reentrant, the corresponding PR is #216 (by @vvanglro, thanks again! 🚀).

This is an interesting one! I tested this a bit more thouroughly with a local Docker mosquitto broker that I shut down. With some logging on the exact exception types and on the relevant lock, this gives:

lock acquired
lock released
<class 'aiomqtt.error.MqttCodeError'>
ERROR: [code:7] The connection was lost.; Reconnecting in 5 seconds ...

lock acquired
<class 'aiomqtt.error.MqttError'>
ERROR: [Errno 61] Connection refused; Reconnecting in 5 seconds ...

<class 'aiomqtt.error.MqttReentrantError'>
ERROR: Does not support reentrant; Reconnecting in 5 seconds ...

What we see here is that the lock is released when an MqttCodeError is raised, but not when an MqttError is raised. Due to the lock not being correctly released, the code runs into this condition inside __aenter__ and raises the "Does not support reentrant" error that you see.

If I got it right the reason for the lock not being released is that the code already fails inside __aenter__ after aquiring the lock and inside the call to connect(). Due to this failure __aexit__ (where the lock is released) is not called.

@vvanglro, are you able to reproduce the error with this extra information?

[empicano]

I just pushed a script to spin up a local Mosquitto broker via Docker to make it easier to reproduce these steps. After calling the script you should be able to connect to the broker via aiomqtt.Client("localhost", port=1883) 🙂

[vvanglro]

The emqx image I use. Maybe we should catch the exception of connect method and release the lock.

[empicano]

Sounds good, do you want to make a PR with a fix and a test? 💯

[vvanglro]

Sounds good, do you want to make a PR with a fix and a test? 💯

next monday i'll take the time to do it.

[frederikaalund]

Great job to both of you (@vvanglro and @empicano) with this quick fix on this issue. :smile: :+1:

[dunielpls]

Hello, and thank you for the great library. Can you consider tagging a release with the fix?

I am running into this exact issue (the Lock not being released appropriately) and it would be really nice to avoid running the main code long term.

[empicano]

Done 🙂 Happy to hear that you like aiomqtt!