Migrate to asyncio

.gitignore

@@ -4,3 +4,5 @@ polysh.egg-info
 build/
 dist/
 .devcontainer/
+flake.lock
+result

flake.nix

@@ -0,0 +1,60 @@
+# Flake for polysh, only used for easier development and testing on nix-based systems.
+# Might be used to build and test the package on multiple platforms, but is not intended/supported for production use.
+{
+  description = "Remote shell multiplexer for executing commands on multiple hosts";
+
+  inputs = {
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+  };
+
+  outputs = { self, nixpkgs }:
+    let
+      supportedSystems = [ "x86_64-linux" "aarch64-linux" "x86_64-darwin" "aarch64-darwin" ];
+      forAllSystems = nixpkgs.lib.genAttrs supportedSystems;
+    in
+    {
+      packages = forAllSystems (system:
+        let
+          pkgs = nixpkgs.legacyPackages.${system};
+        in
+        {
+          polysh = pkgs.python3Packages.buildPythonApplication {
+            pname = "polysh";
+            version = "0.15";
+
+            pyproject = true;
+
+            build-system = with pkgs.python3Packages; [
+              hatchling
+            ];
+
+            src = ./.;
+
+            meta = with pkgs.lib; {
+              description = "Remote shell multiplexer for executing commands on multiple hosts";
+              homepage = "https://github.com/innogames/polysh";
+              license = licenses.gpl2Plus;
+              maintainers = with maintainers; [ seqizz ];
+              platforms = platforms.unix;
+            };
+          };
+
+          default = self.packages.${system}.polysh;
+        }
+      );
+
+      devShells = forAllSystems (system:
+        let
+          pkgs = nixpkgs.legacyPackages.${system};
+        in
+        {
+          default = pkgs.mkShell {
+            packages = with pkgs; [
+              python3
+              python3Packages.hatchling
+            ];
+          };
+        }
+      );
+    };
+}

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "polysh"
 authors = [{ email = "it@innogames.com" }]
-version = "0.15.0"
+version = "0.16.0"
 description = "Control thousands of SSH sessions from a single prompt"
 readme = "README.rst"
 requires-python = ">=3.5,<=3.12"

src/polysh/buffered_dispatcher.py

@@ -16,31 +16,56 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import asyncore
 import errno
+import fcntl
+import os
 
+from polysh import dispatcher_registry
 from polysh.console import console_output
+from polysh.exceptions import ExitNow
 
 
-class BufferedDispatcher(asyncore.file_dispatcher):
+class BufferedDispatcher:
     """A dispatcher with a write buffer to allow asynchronous writers, and a
     read buffer to permit line oriented manipulations"""
 
     # 1 MiB should be enough for everybody
     MAX_BUFFER_SIZE = 1 * 1024 * 1024
 
     def __init__(self, fd: int) -> None:
-        asyncore.file_dispatcher.__init__(self, fd)
         self.fd = fd
-        self.read_buffer = b""
-        self.write_buffer = b""
+        self.read_buffer = b''
+        self.write_buffer = b''
+
+        # Set non-blocking mode
+        flags = fcntl.fcntl(fd, fcntl.F_GETFL)
+        fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
+
+        # Register with the dispatcher registry
+        dispatcher_registry.register(fd, self)
+
+    def recv(self, buffer_size: int) -> bytes:
+        """Read from the file descriptor."""
+        return os.read(self.fd, buffer_size)
+
+    def send(self, data: bytes) -> int:
+        """Write to the file descriptor."""
+        return os.write(self.fd, data)
+
+    def close(self) -> None:
+        """Unregister and close the file descriptor."""
+        dispatcher_registry.unregister(self.fd)
+        try:
+            os.close(self.fd)
+        except OSError:
+            pass
 
     def handle_read(self) -> None:
         self._handle_read_chunk()
 
     def _handle_read_chunk(self) -> bytes:
         """Some data can be read"""
-        new_data = b""
+        new_data = b''
         buffer_length = len(self.read_buffer)
         try:
             while buffer_length < self.MAX_BUFFER_SIZE:
@@ -50,12 +75,11 @@ def _handle_read_chunk(self) -> bytes:
                     if e.errno == errno.EAGAIN:
                         # End of the available data
                         break
-                    elif e.errno == errno.EIO and new_data:
+                    if e.errno == errno.EIO and new_data:
                         # Hopefully we could read an error message before the
                         # actual termination
                         break
-                    else:
-                        raise
+                    raise
 
                 if not piece:
                     # A closed connection is indicated by signaling a read
@@ -66,7 +90,7 @@ def _handle_read_chunk(self) -> bytes:
                 buffer_length += len(piece)
 
         finally:
-            new_data = new_data.replace(b"\r", b"\n")
+            new_data = new_data.replace(b'\r', b'\n')
             self.read_buffer += new_data
         return new_data
 
@@ -76,16 +100,14 @@ def readable(self) -> bool:
 
     def writable(self) -> bool:
         """Do we have something to write?"""
-        return self.write_buffer != b""
+        return self.write_buffer != b''
 
     def dispatch_write(self, buf: bytes) -> bool:
         """Augment the buffer with stuff to write when possible"""
         self.write_buffer += buf
         if len(self.write_buffer) > self.MAX_BUFFER_SIZE:
             console_output(
-                "Buffer too big ({:d}) for {}\n".format(
-                    len(self.write_buffer), str(self)
-                ).encode()
+                f'Buffer too big ({len(self.write_buffer):d}) for {str(self)}\n'.encode()
             )
-            raise asyncore.ExitNow(1)
+            raise ExitNow(1)
         return True

src/polysh/control_commands.py

@@ -18,22 +18,20 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import asyncore
 import os
 import shlex
 from typing import List
 
+from polysh import dispatchers, remote_dispatcher, stdin
+from polysh.completion import add_to_history, complete_local_path
+from polysh.console import console_output
 from polysh.control_commands_helpers import (
     complete_shells,
     expand_local_path,
     selected_shells,
     toggle_shells,
 )
-from polysh.completion import complete_local_path, add_to_history
-from polysh.console import console_output
-from polysh import dispatchers
-from polysh import remote_dispatcher
-from polysh import stdin
+from polysh.exceptions import ExitNow
 
 
 def complete_list(line: str, text: str) -> List[str]:
@@ -43,11 +41,11 @@ def complete_list(line: str, text: str) -> List[str]:
 def do_list(command: str) -> None:
     instances = [i.get_info() for i in selected_shells(command)]
     flat_instances = dispatchers.format_info(instances)
-    console_output(b"".join(flat_instances))
+    console_output(b''.join(flat_instances))
 
 
 def do_quit(command: str) -> None:
-    raise asyncore.ExitNow(0)
+    raise ExitNow(0)
 
 
 def complete_chdir(line: str, text: str) -> List[str]:
@@ -58,29 +56,29 @@ def do_chdir(command: str) -> None:
     try:
         os.chdir(expand_local_path(command.strip()))
     except OSError as e:
-        console_output("{}\n".format(str(e)).encode())
+        console_output(f'{str(e)}\n'.encode())
 
 
 def complete_send_ctrl(line: str, text: str) -> List[str]:
     if len(line[:-1].split()) >= 2:
         # Control letter already given in command line
         return complete_shells(line, text, lambda i: i.enabled)
-    if text in ("c", "d", "z"):
-        return [text + " "]
-    return ["c ", "d ", "z "]
+    if text in ('c', 'd', 'z'):
+        return [text + ' ']
+    return ['c ', 'd ', 'z ']
 
 
 def do_send_ctrl(command: str) -> None:
     split = command.split()
     if not split:
-        console_output(b"Expected at least a letter\n")
+        console_output(b'Expected at least a letter\n')
         return
     letter = split[0]
     if len(letter) != 1:
-        console_output("Expected a single letter, got: {}\n".format(letter).encode())
+        console_output(f'Expected a single letter, got: {letter}\n'.encode())
         return
-    control_letter = chr(ord(letter.lower()) - ord("a") + 1)
-    for i in selected_shells(" ".join(split[1:])):
+    control_letter = chr(ord(letter.lower()) - ord('a') + 1)
+    for i in selected_shells(' '.join(split[1:])):
         if i.enabled:
             i.dispatch_write(control_letter.encode())
 
@@ -122,7 +120,9 @@ def complete_reconnect(line: str, text: str) -> List[str]:
 
 def do_reconnect(command: str) -> None:
     selec = selected_shells(command)
-    to_reconnect = [i for i in selec if i.state == remote_dispatcher.STATE_DEAD]
+    to_reconnect = [
+        i for i in selec if i.state == remote_dispatcher.STATE_DEAD
+    ]
     for i in to_reconnect:
         i.disconnect()
         i.close()
@@ -162,36 +162,36 @@ def do_hide_password(command: str) -> None:
             i.debug = False
             if not warned:
                 console_output(
-                    b"Debugging disabled to avoid displaying " b"passwords\n"
+                    b'Debugging disabled to avoid displaying passwords\n'
                 )
                 warned = True
     stdin.set_echo(False)
 
     if remote_dispatcher.options.log_file:
-        console_output(b"Logging disabled to avoid writing passwords\n")
+        console_output(b'Logging disabled to avoid writing passwords\n')
         remote_dispatcher.options.log_file = None
 
 
 def complete_set_debug(line: str, text: str) -> List[str]:
     if len(line[:-1].split()) >= 2:
         # Debug value already given in command line
         return complete_shells(line, text)
-    if text.lower() in ("y", "n"):
-        return [text + " "]
-    return ["y ", "n "]
+    if text.lower() in ('y', 'n'):
+        return [text + ' ']
+    return ['y ', 'n ']
 
 
 def do_set_debug(command: str) -> None:
     split = command.split()
     if not split:
-        console_output(b"Expected at least a letter\n")
+        console_output(b'Expected at least a letter\n')
         return
     letter = split[0].lower()
-    if letter not in ("y", "n"):
-        console_output("Expected 'y' or 'n', got: {}\n".format(split[0]).encode())
+    if letter not in ('y', 'n'):
+        console_output(f"Expected 'y' or 'n', got: {split[0]}\n".encode())
         return
-    debug = letter == "y"
-    for i in selected_shells(" ".join(split[1:])):
+    debug = letter == 'y'
+    for i in selected_shells(' '.join(split[1:])):
         i.debug = debug
 
 
@@ -200,25 +200,25 @@ def do_export_vars(command: str) -> None:
     for shell in dispatchers.all_instances():
         if shell.enabled:
             environment_variables = {
-                "POLYSH_RANK": str(rank),
-                "POLYSH_NAME": shell.hostname,
-                "POLYSH_DISPLAY_NAME": shell.display_name,
+                'POLYSH_RANK': str(rank),
+                'POLYSH_NAME': shell.hostname,
+                'POLYSH_DISPLAY_NAME': shell.display_name,
             }
             for name, value in environment_variables.items():
                 shell.dispatch_command(
-                    "export {}={}\n".format(name, shlex.quote(value)).encode()
+                    f'export {name}={shlex.quote(value)}\n'.encode()
                 )
             rank += 1
 
     for shell in dispatchers.all_instances():
         if shell.enabled:
             shell.dispatch_command(
-                "export POLYSH_NR_SHELLS={:d}\n".format(rank).encode()
+                f'export POLYSH_NR_SHELLS={rank:d}\n'.encode()
             )
 
 
-add_to_history("$POLYSH_RANK $POLYSH_NAME $POLYSH_DISPLAY_NAME")
-add_to_history("$POLYSH_NR_SHELLS")
+add_to_history('$POLYSH_RANK $POLYSH_NAME $POLYSH_DISPLAY_NAME')
+add_to_history('$POLYSH_NR_SHELLS')
 
 
 def complete_set_log(line: str, text: str) -> List[str]:
@@ -229,13 +229,13 @@ def do_set_log(command: str) -> None:
     command = command.strip()
     if command:
         try:
-            remote_dispatcher.options.log_file = open(command, "a")
-        except IOError as e:
-            console_output("{}\n".format(str(e)).encode())
+            remote_dispatcher.options.log_file = open(command, 'a')
+        except OSError as e:
+            console_output(f'{str(e)}\n'.encode())
             command = None
     if not command:
         remote_dispatcher.options.log_file = None
-        console_output(b"Logging disabled\n")
+        console_output(b'Logging disabled\n')
 
 
 def complete_show_read_buffer(line: str, text: str) -> List[str]:
@@ -248,4 +248,4 @@ def do_show_read_buffer(command: str) -> None:
     for i in selected_shells(command):
         if i.read_in_state_not_started:
             i.print_lines(i.read_in_state_not_started)
-            i.read_in_state_not_started = b""
+            i.read_in_state_not_started = b''