fix: issue212

[vvanglro]

This PR fixes #212 .
If _disconnected is found to be completed after connecting, it is not correct, so we need to reset it.
If _connected is still in the completed state after disconnection, reset it.

[codecov]

Codecov Report

Merging #216 (9cb10d5) into main (ddf3f72) will increase coverage by 2.0%.
The diff coverage is 98.2%.

❗ Current head 9cb10d5 differs from pull request most recent head f6cbad1. Consider uploading reports for the commit f6cbad1 to get more accurate results

@@           Coverage Diff           @@
##            main    #216     +/-   ##
=======================================
+ Coverage   89.7%   91.7%   +2.0%     
=======================================
  Files          6       6             
  Lines        738     836     +98     
  Branches     156     177     +21     
=======================================
+ Hits         662     767    +105     
+ Misses        49      46      -3     
+ Partials      27      23      -4     

Impacted Files	Coverage Δ
asyncio_mqtt/client.py	83.5% <94.7%> (+2.0%)	⬆️
tests/test_client.py	99.7% <98.9%> (-0.3%)	⬇️
asyncio_mqtt/error.py	100.0% <100.0%> (ø)

[frederikaalund]

Thanks for opening this PR. 👍 Let me have a look. :)

I assume the overall purpose is to make asyncio_mqtt.Client reusable. I quote the linked docs to save you a click:

Distinct from both single use and reentrant context managers are “reusable” context managers (or, to be completely explicit, “reusable, but not reentrant” context managers, since reentrant context managers are also reusable). These context managers support being used multiple times, but will fail (or otherwise not work correctly) if the specific context manager instance has already been used in a containing with statement.

That's the purpose of this PR, right? Let me know if I'm wrong.

If I'm right, then this PR also relates to #48. I'll provide more feedback after we settle on the intent of this PR.

I hope that makes sense. :)

[vvanglro]

Yes, after adding the judgment, it will become a reusable context. I read #48, and I don’t know much about the publish and QOS in the question. What I can only confirm is that this PR will make aiomqtt a reusable of the context.

[vvanglro]

To be honest, I don't really like to use context in projects, such as the example used in fastapi in your documentation.
I prefer this: (Of course this is just my personal preference😅)

client = aiomqtt.Client("test.mosquitto.org")

@contextlib.asynccontextmanager
async def lifespan(app):
    await clinet.connect()
    yield
    await client.disconnect()

I found that you mentioned paho.client.reconnect in #48.
It would be great if we could integrate this method. See this method literally means reconnect. My good wish is that we can use this method to do More things, such as when the connection is disconnected when calling methods such as subscribing and getting messages, then we will automatically try to reconnect for the user, and the number of reconnections is configurable.
This can be very useful in long-running projects, such as web projects.

[frederikaalund]

Yes, after adding the judgment, it will become a reusable context. I read #48, and I don’t know much about the publish and QOS in the question. What I can only confirm is that this PR will make aiomqtt a reusable of the context.

Great, because that's what we need to test for then! :) What I need from you to merge this PR:

• Add tests that verify that asyncio_mqtt.Client is now reusable (but not reentrant!). Things to consider for the tests:
  • What happens if task A reuses the client while task B listens for messages? Will task B get an exception? Will task B get stuck?
  • What happens if we try to reenter the client? Does it raise an exception? Does it deadlock? Does it work?
  • In general, it's good to both tests for what is supposed to work and what is definitely supposed to not work (e.g., do we raise proper exceptions that the user can understand?).
• Change/add to the docs that you can now reuse the client. People will like this feature. 👍

The test part is the most important. I can do the docs if you don't feel like it. :)

---

On the topic of "automatic reconnect": Yes! That would be great to have. It is also quite difficult to get right. 😅 See #6 and #26 for some of the subtleties.
This PR (#216) is the first step towards "automatic reconnect". 👍 We can build the latter on top of the former.

[vvanglro]

OK, I'll do it.

[vvanglro]

This PR changed 6 points:

1. reusable and non reentrant context.
2. Extract the disconnect method.
3. Change message queue to class attribute.
4. Add test case.
5. update ruff version.
6. If _disconnected is found to be completed after connecting, it is not correct, so we need to reset it.
   If _connected is still in the completed state after disconnection, reset it.

[frederikaalund]

Looks great and thanks for the detailed tests! 👍 :) The more coverage we get on that front the better.

Have a look at the comments in the review. Let me know if anything is unclear.

Side note for future reference (not directly relevant to this PR)

Looking through the internals like this again makes me wish that I never exposed connect and disconnect as part of the public interface. 😅 It's quite the maintenance burden!

Also, now we have this mess of internal state (I'm the cause of that myself):

• self._connected: asyncio.Future
• self._disconnected: asyncio.Future
• self._lock: asyncio.Lock

All three variables have semantic overlap! Really, we should just have the single _lock (where "lock acquired" means "client connecting/connected"). Extending on that: Have all internal state inside a tagged union (sum type) in rust-style, concurrency-safe semantics.

In any case, all of that is just food for thought for, e.g., an anyio-based rewrite. :)

[empicano]

Hi vvanglro, this is a very cool addition, thank you! 🎉

[...] makes me wish that I never exposed connect and disconnect as part of the public interface. 😅 It's quite the maintenance burden!

Our documentation didn't document connect and disconnect for a long time, in fact, it states: "asyncio-mqtt doesn’t support manual calls to connect and disconnect."

It'd be a breaking change nonetheless, but I think it's valid to remove it as well. I'm always for simplifying things 😄

[JonathanPlasse]

Hi vvanglro, this is a very cool addition, thank you! tada

[...] makes me wish that I never exposed connect and disconnect as part of the public interface. sweat_smile It's quite the maintenance burden!

Our documentation didn't document connect and disconnect for a long time, in fact, it states: "asyncio-mqtt doesn’t support manual calls to connect and disconnect."

It'd be a breaking change nonetheless, but I think it's valid to remove it as well. I'm always for simplifying things smile

Should we deprecate first before removing?

[frederikaalund]

It'd be a breaking change nonetheless, but I think it's valid to remove it as well. I'm always for simplifying things 😄

Good to know that I'm not the only one. :)

Should we deprecate first before removing?

Yes!

I'd also like to point out that our users may still manually call "connect/disconnect": They simply call "__aenter__/__aexit__". :)) Hopefully, these oddly-named functions discourage that pattern. In any case, we can document that as a "compatibility workaround for existing code that relies on manual 'connect/disconnect' calls".

[frederikaalund]

Thank you again for your great work on this PR, @vvanglro. 👍 I plan to do the code review this weekend. I'm really tied up until then. :)

[vvanglro]

I'm thinking about some problems,🤯 if we remove connect/disconnect, based main code and the current PR, there will be the following three situations.

The first situation(Base main code, This will establish 10 connections simultaneously):

from asyncio_mqtt import Client


async def run1(message):
    async with Client("192.168.5.24", username="admin", password="123456") as client:
        await client.publish("humidity/outside", payload=message)
        # Maybe it can only be concurrently in context.
        # tasks = [asyncio.create_task(client.publish("humidity/outside", payload=i)) for i in range(10)]
        # await asyncio.gather(*tasks)


async def gather_task1():
    tasks = [asyncio.create_task(run1(i)) for i in range(10)]
    await asyncio.gather(*tasks)
    
    
if __name__ == '__main__':
    import asyncio
    asyncio.get_event_loop().run_until_complete(gather_task1())

The two situation(Base current PR, Not reentrant will result in an error):

from asyncio_mqtt import Client


client = Client("192.168.5.24", username="admin", password="123456")


async def run2(message):
    async with client:
        await client.publish("humidity/outside", payload=message)
        # Maybe it can only be concurrently in context.
        # tasks = [asyncio.create_task(client.publish("humidity/outside", payload=i)) for i in range(10)]
        # await asyncio.gather(*tasks)


async def gather_task2():
    tasks = [asyncio.create_task(run2(i)) for i in range(10)]
    await asyncio.gather(*tasks)


if __name__ == '__main__':
    import asyncio
    asyncio.get_event_loop().run_until_complete(gather_task2())

The three situation(There is only 1 connection, and no error will be generated):

from asyncio_mqtt import Client


client = Client("192.168.5.24", username="admin", password="123456")

async def run3(message):
    await client.publish("humidity/outside", payload=message)


async def gather_task3():
    await client.connect()
    tasks = [asyncio.create_task(run3(i)) for i in range(10)]
    await asyncio.gather(*tasks)
    await client.disconnect()

if __name__ == '__main__':
    import asyncio
    asyncio.get_event_loop().run_until_complete(gather_task3())

So is it possible to consider from other angles and then keep connect/disconnect.
We can not recommend manually calling connect/disconnect unless you understand the logic behind it.

[frederikaalund]

First of all, I apologize about the long delay.

In any case, this PR now looks good to me. 👍 Though I can't merge it in because GitHub says "This branch cannot be rebased due to conflicts". I guess that the main branch updated (probably a bot-based commit) in the meantime.

@vvanglro Can I get you to rebase this PR on top of the master branch? Thanks! Then I'll merge it in right away. :)

[vvanglro]

I can't reproduce the error in CI locally. This seems to be a problem with mypy?

I use python3.8 locally, and python3.10 in CI throws an error.

[frederikaalund]

It was just a simple merge conflict. I added a merge commit on it to fix it. :)

Thank you for a great contribution to asyncio-mqtt. 👍

[thalesmaoa]

This is on the main branch now? Do I still need to call disconnect?

[frederikaalund]

This is on the main branch now?

Yes, and part of the v1.0.0 release. :)

Do I still need to call disconnect?

We did not remove disconnect (too big of a change). You may still call connect/disconnect instead of using the context manager approach.