refactor: allow consuming without blocking the listening loop (#147)

* refactor: kotlin-like flows to allow consuming without blocking the listening loop

* update changelogs

* Revert "refactor: kotlin-like flows to allow consuming without blocking the listening loop"

This reverts commit b1520ba.

* Refactor to a asynchronous handlers to avoid blocking the listening loop

* typo

Author: nathanfallet

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Refactor to asynchronous handlers to avoid blocking the listening loop @nathanfallet
+
 ### Added
 
 ### Changed

zendriver/core/browser.py

@@ -59,6 +59,7 @@ class Browser:
     _process_pid: int | None
     _http: HTTPApi | None = None
     _cookies: CookieJar | None = None
+    _update_target_info_mutex: asyncio.Lock = asyncio.Lock()
 
     config: Config
     connection: Connection | None
@@ -182,7 +183,7 @@ async def wait(self, time: Union[float, int] = 1) -> Browser:
     sleep = wait
     """alias for wait"""
 
-    def _handle_target_update(
+    async def _handle_target_update(
         self,
         event: Union[
             cdp.target.TargetInfoChanged,
@@ -193,56 +194,58 @@ def _handle_target_update(
     ):
         """this is an internal handler which updates the targets when chrome emits the corresponding event"""
 
-        if isinstance(event, cdp.target.TargetInfoChanged):
-            target_info = event.target_info
+        async with self._update_target_info_mutex:
+            if isinstance(event, cdp.target.TargetInfoChanged):
+                target_info = event.target_info
 
-            current_tab = next(
-                filter(
-                    lambda item: item.target_id == target_info.target_id, self.targets
-                )
-            )
-            current_target = current_tab.target
-
-            if logger.getEffectiveLevel() <= 10:
-                changes = util.compare_target_info(current_target, target_info)
-                changes_string = ""
-                for change in changes:
-                    key, old, new = change
-                    changes_string += f"\n{key}: {old} => {new}\n"
-                logger.debug(
-                    "target #%d has changed: %s"
-                    % (self.targets.index(current_tab), changes_string)
+                current_tab = next(
+                    filter(
+                        lambda item: item.target_id == target_info.target_id,
+                        self.targets,
+                    )
                 )
+                current_target = current_tab.target
+
+                if logger.getEffectiveLevel() <= 10:
+                    changes = util.compare_target_info(current_target, target_info)
+                    changes_string = ""
+                    for change in changes:
+                        key, old, new = change
+                        changes_string += f"\n{key}: {old} => {new}\n"
+                    logger.debug(
+                        "target #%d has changed: %s"
+                        % (self.targets.index(current_tab), changes_string)
+                    )
 
-                current_tab.target = target_info
+                    current_tab.target = target_info
 
-        elif isinstance(event, cdp.target.TargetCreated):
-            target_info = event.target_info
-            from .tab import Tab
+            elif isinstance(event, cdp.target.TargetCreated):
+                target_info = event.target_info
+                from .tab import Tab
 
-            new_target = Tab(
-                (
-                    f"ws://{self.config.host}:{self.config.port}"
-                    f"/devtools/{target_info.type_ or 'page'}"  # all types are 'page' internally in chrome apparently
-                    f"/{target_info.target_id}"
-                ),
-                target=target_info,
-                browser=self,
-            )
+                new_target = Tab(
+                    (
+                        f"ws://{self.config.host}:{self.config.port}"
+                        f"/devtools/{target_info.type_ or 'page'}"  # all types are 'page' internally in chrome apparently
+                        f"/{target_info.target_id}"
+                    ),
+                    target=target_info,
+                    browser=self,
+                )
 
-            self.targets.append(new_target)
+                self.targets.append(new_target)
 
-            logger.debug("target #%d created => %s", len(self.targets), new_target)
+                logger.debug("target #%d created => %s", len(self.targets), new_target)
 
-        elif isinstance(event, cdp.target.TargetDestroyed):
-            current_tab = next(
-                filter(lambda item: item.target_id == event.target_id, self.targets)
-            )
-            logger.debug(
-                "target removed. id # %d => %s"
-                % (self.targets.index(current_tab), current_tab)
-            )
-            self.targets.remove(current_tab)
+            elif isinstance(event, cdp.target.TargetDestroyed):
+                current_tab = next(
+                    filter(lambda item: item.target_id == event.target_id, self.targets)
+                )
+                logger.debug(
+                    "target removed. id # %d => %s"
+                    % (self.targets.index(current_tab), current_tab)
+                )
+                self.targets.remove(current_tab)
 
     async def get(
         self, url="about:blank", new_tab: bool = False, new_window: bool = False

zendriver/core/connection.py

@@ -723,15 +723,21 @@ async def listener_loop(self):
                         try:
                             if iscoroutinefunction(callback):
                                 try:
-                                    await callback(event, self.connection)
+                                    asyncio.create_task(
+                                        callback(event, self.connection)
+                                    )
                                 except TypeError:
-                                    await callback(event)
+                                    asyncio.create_task(callback(event))
                             else:
                                 callback = typing.cast(Callable, callback)
-                                try:
-                                    callback(event, self.connection)
-                                except TypeError:
-                                    callback(event)
+
+                                def run_callback():
+                                    try:
+                                        callback(event, self.connection)
+                                    except TypeError:
+                                        callback(event)
+
+                                asyncio.create_task(asyncio.to_thread(run_callback))
                         except Exception as e:
                             logger.warning(
                                 "exception in callback %s for event %s => %s",