From 28a12652a3418dc8abd80d75d100271a5e88c7ce Mon Sep 17 00:00:00 2001
From: ManiMozaffar <mani.mozaffar@gmail.com>
Date: Thu, 21 Sep 2023 14:13:04 +0300
Subject: [PATCH] Fix connection leak issue

---
 undetected_chromedriver/reactor.py | 50 +++++++++++++++++-------------
 1 file changed, 28 insertions(+), 22 deletions(-)

diff --git a/undetected_chromedriver/reactor.py b/undetected_chromedriver/reactor.py
index 58b0264..3cc73bf 100644
--- a/undetected_chromedriver/reactor.py
+++ b/undetected_chromedriver/reactor.py
@@ -1,27 +1,26 @@
-#!/usr/bin/env python3
-# this module is part of undetected_chromedriver
-
 import asyncio
 import json
 import logging
 import threading
+import time
+
+from selenium import webdriver
 
 logger = logging.getLogger(__name__)
 
 
 class Reactor(threading.Thread):
-    def __init__(self, driver: "Chrome"):
+    def __init__(self, driver: webdriver.Chrome):
         super().__init__()
-
         self.driver = driver
         self.loop = asyncio.new_event_loop()
-
+        self.paused = False
         self.lock = threading.Lock()
         self.event = threading.Event()
         self.daemon = True
         self.handlers = {}
 
-    def add_event_handler(self, method_name, callback: callable):
+    def add_event_handler(self, method_name, callback):
         """
 
         Parameters
@@ -39,9 +38,16 @@ class Reactor(threading.Thread):
         with self.lock:
             self.handlers[method_name.lower()] = callback
 
-    @property
-    def running(self):
-        return not self.event.is_set()
+    def terminate(self, timeout=10):
+        self.event.set()
+        start_time = time.time()
+        while not self.paused:
+            elapsed_time = time.time() - start_time
+            if elapsed_time >= timeout:
+                break
+            time.sleep(0.1)
+        self.loop.close()
+        return None
 
     def run(self):
         try:
@@ -58,12 +64,12 @@ class Reactor(threading.Thread):
                     and getattr(self.driver.service, "process", None)
                     and self.driver.service.process.poll()
                 ):
-                    await asyncio.sleep(self.driver._delay or 0.25)
+                    await asyncio.sleep(self.driver._delay or 0.25)  # type: ignore
                 else:
                     break
 
     async def listen(self):
-        while self.running:
+        while not self.event.is_set():
             await self._wait_service_started()
             await asyncio.sleep(1)
 
@@ -74,9 +80,11 @@ class Reactor(threading.Thread):
                 for entry in log_entries:
                     try:
                         obj_serialized: str = entry.get("message")
-                        obj = json.loads(obj_serialized)
+                        obj: dict[str, dict] = json.loads(obj_serialized)
                         message = obj.get("message")
-                        method = message.get("method")
+                        assert message is not None
+                        method: str | None = message.get("method")
+                        assert isinstance(method, str)
 
                         if "*" in self.handlers:
                             await self.loop.run_in_executor(
@@ -86,13 +94,11 @@ class Reactor(threading.Thread):
                             await self.loop.run_in_executor(
                                 None, self.handlers[method.lower()], message
                             )
+                    except Exception as error:
+                        logging.debug("exception ignored :", error)
+                        raise error from None
 
-                        # print(type(message), message)
-                    except Exception as e:
-                        raise e from None
+            except Exception as error:
+                logging.debug("exception ignored :", error)
 
-            except Exception as e:
-                if "invalid session id" in str(e):
-                    pass
-                else:
-                    logging.debug("exception ignored :", e)
+        self.paused = True