f

Author: marcin-usielski

moler/io/raw/pty_process_unicode.py

@@ -118,24 +118,32 @@ def read(self, size: int) -> str:
             return data_str
 
         except OSError as e:
-            if e.errno == errno.EAGAIN:
+            if e.errno == errno.EAGAIN:  # Resource temporarily unavailable
                 return ""  # No data available, return empty string
-            elif e.errno == errno.EIO:
+            elif e.errno == errno.EIO:  # Input/output error - process might be dead
                 self.close(force=True)
                 raise EOFError("End of file reached")
             else:
                 raise
 
-    def close(self, force: bool = False) -> None:
-        """
-        Close pty process.
-        :param force: if True, forcefully kill the process
-        :return: None
-        """
-        if self._closed:
-            return
-        self._closed = True
+    def _close_fds(self) -> None:
+        """Close file descriptors."""
+        if self.fd >= 0:
+            try:
+                os.close(self.fd)
+            except OSError:
+                pass
+            self.fd = -1
 
+        if self.slave_fd >= 0:
+            try:
+                os.close(self.slave_fd)
+            except OSError:
+                pass
+            self.slave_fd = -1
+
+    def _terminate_process(self, force: bool) -> None:
+        """Terminate the child process."""
         # Try to terminate the process gracefully first
         if self.process and self.isalive():
             try:
@@ -155,20 +163,21 @@ def close(self, force: bool = False) -> None:
             except Exception as e:
                 print(f"Error terminating process: {e}")
 
-        # Close file descriptors
-        if self.fd >= 0:
-            try:
-                os.close(self.fd)
-            except OSError:
-                pass
-            self.fd = -1
+    def close(self, force: bool = False) -> None:
+        """
+        Close pty process.
+        :param force: if True, forcefully kill the process
+        :return: None
+        """
+        if self._closed:
+            return
+        self._closed = True
 
-        if self.slave_fd >= 0:
-            try:
-                os.close(self.slave_fd)
-            except OSError:
-                pass
-            self.slave_fd = -1
+        # Terminate the child process
+        self._terminate_process(force=force)
+
+        # Close file descriptors
+        self._close_fds()
 
     def isalive(self) -> bool:
         """

moler/observer_thread_wrapper.py

@@ -62,6 +62,26 @@ def request_stop(self):
         if self._t:
             self._t = None
 
+    def _process_data_from_queue(self) -> None:
+        """Process data from queue."""
+        try:
+            data, timestamp = self._queue.get(True, self._timeout_for_get_from_queue)
+            try:
+                self.logger.log(level=TRACE, msg=f'notifying {self._observer}({repr(data)})')
+            except ReferenceError:
+                self._request_end.set()  # self._observer is no more valid.
+            try:
+                if self._observer_self:
+                    self._observer(self._observer_self, data, timestamp)
+                else:
+                    self._observer(data, timestamp)
+            except ReferenceError:
+                self._request_end.set()  # self._observer is no more valid.
+            except Exception as ex:
+                self._handle_unexpected_error_from_observer(exception=ex, data=data, timestamp=timestamp)
+        except queue.Empty:
+            pass  # No incoming data within self._timeout_for_get_from_queue
+
     @tracked_thread.log_exit_exception
     def _loop_for_observer(self):
         """
@@ -74,23 +94,7 @@ def _loop_for_observer(self):
         while not self._request_end.is_set():
             if next(heartbeat):
                 logging.getLogger("moler_threads").debug("ALIVE")
-            try:
-                data, timestamp = self._queue.get(True, self._timeout_for_get_from_queue)
-                try:
-                    self.logger.log(level=TRACE, msg=f'notifying {self._observer}({repr(data)})')
-                except ReferenceError:
-                    self._request_end.set()  # self._observer is no more valid.
-                try:
-                    if self._observer_self:
-                        self._observer(self._observer_self, data, timestamp)
-                    else:
-                        self._observer(data, timestamp)
-                except ReferenceError:
-                    self._request_end.set()  # self._observer is no more valid.
-                except Exception as ex:
-                    self._handle_unexpected_error_from_observer(exception=ex, data=data, timestamp=timestamp)
-            except queue.Empty:
-                pass  # No incoming data within self._timeout_for_get_from_queue
+            self._process_data_from_queue()
         self._observer = None
         self._observer_self = None
         logging.getLogger("moler_threads").debug("EXIT")