Client doesn’t unblock when calling `Subscription.unsubscribe` on last subscription if `Client.__aexit__` was reached

[lorenzobenvenuti]

Hi,

sorry to bother you again. I don't know if this is a relevant use case, but I've noticed that subscription.unsubscribe only works within the with ... as client context manager: this means we can't stop the consumer from a callback (say after N messages have been consumed or a certain message is received, or even when a timeout occurs). AFAICT this happens because Client.__aexit__ waits for a Future that is never set (see [here]).

I'm wondering if we can change this behavior and set the Future result (or just use an asyncio.Event instead of a Future) when the last subscription has been removed. This could be implemented, for example, by converting ActiveSubscriptions into an "observable" dictionary that calls Event.cancel when a subscription is added and Event.set it when a subscription is deleted and there aren't any subscriptions left e.g.

class ActiveSubscriptions(MutableMappings):
    
    # ...

    def __delitem__(self, key):
        value = self._subcriptions.__delitem__(key)
        if not self._subcriptions:
            self._event.set()
        return value
    
    def __setitem__(self, key, value):
        self._event.clear()
        return self._subcriptions.__setitem__(key, value)

    async def wait(self):
        return await self._event.wait()

Thoughts?

If you think this is worth doing I'd happily put a PR together.

Thanks,

lorenzo

[vrslev]

Hi! I see that it can be useful.

I agree that we should make shutdown react on changes to active subscriptions, but I generally do not support using Python magic unless it is absolutely necessary.

Can you suggest an implementation that wouldn’t require such magic?

[lorenzobenvenuti]

Hi, sure!

[lorenzobenvenuti]

I had a look at the code, I see these options:

• Transform ActiveSubscriptions into an actual class, adding some methods and logic. It won't be much different than the approach described above but the class won't use dunder methods (e.g. instead of id in active_subscriptions the subscription will call active_subscriptions.is_active(id), instead of del active_subscriptions[id] it will call active_subscriptions.remove(id) and so on)
• Add an event listener to the subscription objects. Events can be triggered every time a subscription is created/removed, and Client could listen to those events and update an asyncio.Event accordingly
• Create a task that runs a loop that sleeps for a while and exits when active_subscriptions is empty, and await this task in the context manager
  The third one is pretty hacky, TBH. Considering that Client and BaseSubscription are already "coupled" by ActiveSubscriptions, I think the first approach would be simpler.

Thanks,

lorenzo

[vrslev]

The first option is the best, I think. Don’t want to clutter Client even more 😄

[lorenzobenvenuti]

Created a PR here: #121

Thanks,

lorenzo

[vrslev]

I’ve merged the PR, the changes are available in 2.2.0.

Thanks!

[lorenzobenvenuti]

Thank you!