refactor distributed modules and reorganize tests for quality improvements

utils/cloudflare/process_reader.py

@@ -3,6 +3,7 @@
 import asyncio
 import re
 import threading
+from typing import Any
 
 from ..constants import CLOUDFLARE_LOG_BUFFER_SIZE
 from ..logging import debug_log
@@ -15,6 +16,7 @@ class ProcessReader:
     def __init__(self, log_file=None):
         self._process = None
         self._thread = None
+        self._task = None
         self._loop = None
         self._url_event = None
         self._public_url = None
@@ -37,6 +39,23 @@ class ProcessReader:
         if len(self._recent_logs) > CLOUDFLARE_LOG_BUFFER_SIZE:
             self._recent_logs = self._recent_logs[-CLOUDFLARE_LOG_BUFFER_SIZE:]
 
+    @staticmethod
+    def _normalize_line(raw_line):
+        if isinstance(raw_line, bytes):
+            return raw_line.decode("utf-8", errors="replace").strip()
+        return str(raw_line).strip()
+
+    def _process_log_line(self, line):
+        self._append_log(line)
+        match = PUBLIC_URL_PATTERN.search(line)
+        if match and not self._public_url:
+            self._public_url = match.group(1).rstrip("/")
+            return True
+
+        if "error" in line.lower() and not self._last_error:
+            self._last_error = line
+        return False
+
     def _reader(self):
         process = self._process
         if process is None:
@@ -44,34 +63,63 @@ class ProcessReader:
 
         loop = self._loop
         for raw_line in iter(process.stdout.readline, ""):
-            line = raw_line.strip()
+            line = self._normalize_line(raw_line)
             if not line:
                 continue
 
-            self._append_log(line)
-            match = PUBLIC_URL_PATTERN.search(line)
-            if match and not self._public_url:
-                self._public_url = match.group(1).rstrip("/")
-                if self._url_event and loop:
-                    loop.call_soon_threadsafe(self._url_event.set)
-
-            if "error" in line.lower() and not self._last_error:
-                self._last_error = line
+            found_url = self._process_log_line(line)
+            if found_url and self._url_event and loop:
+                loop.call_soon_threadsafe(self._url_event.set)
 
         if self._url_event and loop:
             if not self._last_error and not self._public_url:
                 self._last_error = "Cloudflare tunnel exited before becoming ready"
             loop.call_soon_threadsafe(self._url_event.set)
 
-    def start(self, process, loop):
+    async def _async_reader(self):
+        process = self._process
+        if process is None or process.stdout is None:
+            if self._url_event:
+                self._url_event.set()
+            return
+
+        while True:
+            raw_line = await process.stdout.readline()
+            if not raw_line:
+                break
+
+            line = self._normalize_line(raw_line)
+            if not line:
+                continue
+
+            found_url = self._process_log_line(line)
+            if found_url and self._url_event:
+                self._url_event.set()
+
+        if self._url_event:
+            if not self._last_error and not self._public_url:
+                self._last_error = "Cloudflare tunnel exited before becoming ready"
+            self._url_event.set()
+
+    def start(self, process: Any, loop: asyncio.AbstractEventLoop) -> None:
         self._process = process
         self._loop = loop
         self._url_event = asyncio.Event()
         self._public_url = None
         self._last_error = None
         self._recent_logs = []
-        self._thread = threading.Thread(target=self._reader, daemon=True)
-        self._thread.start()
+        self._thread = None
+        self._task = None
+
+        stdout = getattr(process, "stdout", None)
+        if stdout is None:
+            return
+
+        if asyncio.iscoroutinefunction(getattr(stdout, "readline", None)):
+            self._task = loop.create_task(self._async_reader())
+        else:
+            self._thread = threading.Thread(target=self._reader, daemon=True)
+            self._thread.start()
 
     async def wait_for_url(self, timeout):
         if not self._url_event:
@@ -79,7 +127,11 @@ class ProcessReader:
         await asyncio.wait_for(self._url_event.wait(), timeout=timeout)
         return self._public_url
 
-    def stop(self):
+    def stop(self) -> None:
+        if self._task and not self._task.done():
+            self._task.cancel()
+        self._task = None
+
         if self._thread and self._thread.is_alive():
             self._thread.join(timeout=1)
         self._thread = None