From da75e537cd84e2ab2017e2841fb168df6bad4b5a Mon Sep 17 00:00:00 2001 From: Daniele Palaia Date: Tue, 27 Aug 2024 11:11:30 +0200 Subject: [PATCH] add cleanup in unsubscribe --- rstream/consumer.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/rstream/consumer.py b/rstream/consumer.py index 3786601..32ade3f 100644 --- a/rstream/consumer.py +++ b/rstream/consumer.py @@ -319,6 +319,12 @@ async def unsubscribe(self, subscriber_name: str) -> None: except BaseException as exc: logger.warning("exception in unsubscribe of Consumer:" + str(exc)) + stream = subscriber.stream + + if stream in self._clients: + await self._clients[stream].remove_stream(stream) + await self._clients[stream].free_available_id(subscriber.subscription_id) + del self._subscribers[subscriber_name] async def query_offset(self, stream: str, subscriber_name: str) -> int: