cleanup subscriber_task in client during unsubscribing

rabbitmq-community · Aug 27, 2024 · 3d7a4f8 · 3d7a4f8
1 parent da75e53
commit 3d7a4f8
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 0 deletions.
diff --git a/rstream/client.py b/rstream/client.py
@@ -246,6 +246,10 @@ async def run_queue_listener_task(self, subscriber_name: str, handler: HT[FT]) -
                 self._run_delivery_handlers(subscriber_name, handler),
             )
 
+    async def destroy_queue_listener_task(self, subscriber_name: str) -> None:
+        if subscriber_name in self._frames:
+            self.stop_task(f"run_delivery_handlers_{subscriber_name}")
+
     async def _run_delivery_handlers(self, subscriber_name: str, handler: HT[FT]):
         while self.is_connection_alive():
             frame_entry = await self._frames[subscriber_name].get()

diff --git a/rstream/consumer.py b/rstream/consumer.py
@@ -325,6 +325,8 @@ async def unsubscribe(self, subscriber_name: str) -> None:
             await self._clients[stream].remove_stream(stream)
             await self._clients[stream].free_available_id(subscriber.subscription_id)
 
+        subscriber.client.destroy_queue_listener_task(subscriber_name)
+
         del self._subscribers[subscriber_name]
 
     async def query_offset(self, stream: str, subscriber_name: str) -> int: