From 06fdb06d9ae3b5d26df371247ce16f54c1032587 Mon Sep 17 00:00:00 2001 From: Riccardo Balbo Date: Sat, 22 Feb 2025 13:05:53 +0000 Subject: [PATCH] garbage collect old tracked events --- nwcp.py | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/nwcp.py b/nwcp.py index ff8aff8..5aafbe8 100644 --- a/nwcp.py +++ b/nwcp.py @@ -47,6 +47,24 @@ class MainSubscription: if event_id not in self.responses: self.responses.append(event_id) + def gc(self, expire: int = 3 * 60 * 60): + """ + Garbage collection, remove all the events that have a response older than expire seconds. + """ + now = int(time.time()) + deleted_ids = [] + for [event_id, event] in self.events.items(): + if event_id in self.responses: + if now - event["created_at"] > expire: + del self.events[event_id] + deleted_ids.append(event_id) + self.responses = [event_id for event_id in self.responses if event_id not in deleted_ids] + + if len(deleted_ids) > 0: + logger.debug("Garbage collected " + str(len(deleted_ids)) + " events") + + + class Config: arbitrary_types_allowed = True @@ -92,6 +110,9 @@ class NWCServiceProvider: # Reconnect task (if the connection is lost) self.reconnect_task = None + # Garbage collection loop + self.gc_task = None + # Subscription self.sub = None self.rate_limit: Dict[str, RateLimit] = {} @@ -112,6 +133,13 @@ class NWCServiceProvider: + self.public_key_hex ) + async def _gc_loop(self): + while not self._is_shutting_down(): + if self.sub: + self.sub.gc() + await asyncio.sleep(60) + + def get_supported_methods(self): """ Returns the list of supported methods by this service provider. @@ -144,6 +172,7 @@ class NWCServiceProvider: Starts the NWC service provider. """ self.reconnect_task = asyncio.create_task(self._connect_to_relay()) + self.gc_task = asyncio.create_task(self._gc_loop()) def _json_dumps(self, data: Union[Dict, list]) -> str: """ @@ -602,6 +631,11 @@ class NWCServiceProvider: self.reconnect_task.cancel() except Exception as e: logger.warning("Error closing reconnection task: " + str(e)) + try: + if self.gc_task: + self.gc_task.cancel() + except Exception as e: + logger.warning("Error closing gc loop: " + str(e)) # close the websocket try: if self.ws: