Merge pull request 'chore(release): 0.3.0' (#5) from release/0.3.0 into main

Reviewed-on: #5

src/judas_client/__main__.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 import logging as lg
 
-from judas_client.connector import Connector
+from judas_client.client import Client
 
 if __name__ == "__main__":
     lg.basicConfig(
@@ -12,5 +12,5 @@ if __name__ == "__main__":
     )
     logger = lg.getLogger(__name__)
 
-    connector = Connector("127.0.0.1", 3692)
+    client = Client(server_host="localhost", server_port=3692)
-    connector.run()
+    client.run()

src/judas_client/client.py

@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+"""Client module for handling messages."""
+
+from __future__ import annotations
+
+import logging as lg
+import uuid
+
+from judas_protocol import Message
+
+from judas_client.connector import Connector
+
+
+class Client:
+    """Client class for handling messages."""
+
+    def __init__(self, server_host: str, server_port: int) -> None:
+        """Initialize the Client.
+
+        Args:
+            server_host (str): The server host address.
+            server_port (int): The server port number.
+        """
+        self.logger: lg.Logger = lg.getLogger(
+            f"{__name__}.{self.__class__.__name__}"
+        )
+        self.logger.debug("Initializing Client...")
+
+        self.server_host: str = server_host
+        self.server_port: int = server_port
+
+        self.mac_address: str = self._get_mac_address()
+        self.logger.debug(f"MAC address: {self.mac_address}")
+
+        self.connector: Connector = Connector(
+            mac_address=self.mac_address,
+            host=self.server_host,
+            port=self.server_port,
+            on_message=self.handle_message,
+        )
+
+    def _get_mac_address(self) -> str:
+        """Get the MAC address of the client.
+
+        Returns:
+            str: The MAC address in standard lowercase format (e.g., "0a:1b:2c:3d:4e:5f").
+        """
+        mac_address: str = ":".join(
+            [
+                "{:02x}".format((uuid.getnode() >> ele) & 0xFF)
+                for ele in range(0, 8 * 6, 8)
+            ][::-1]
+        )
+        return mac_address
+
+    def handle_message(self, message: Message) -> None:
+        """Handle incoming messages.
+
+        Args:
+            message (Message): The incoming message to handle.
+        """
+        self.logger.info(f"[v] Received message: {message}")
+
+    def run(self) -> None:
+        """Run the client."""
+        self.logger.info(f"Starting Client ({self.mac_address})...")
+        self.connector.run()

src/judas_client/connector.py

@@ -2,21 +2,33 @@
 from __future__ import annotations
 
 import logging as lg
+import selectors
 import socket
 import time
-import uuid
+from typing import Callable
 
-from judas_protocol import Message
+from judas_protocol import Category, ControlAction, Message
 
 
 class Connector:
+    """Connector class for managing TCP connection and message exchange."""
+
     def __init__(
         self,
+        mac_address: str,
         host: str,
         port: int,
-        connect_timeout: float = 5.0,
+        *,
-        ack_timeout: float = None,
+        on_message: Callable[[Message], None],
     ) -> None:
+        """Initialize the Connector.
+
+        Args:
+            mac_address (str): The MAC address of the client.
+            host (str): The server host address.
+            port (int): The server port number.
+            on_message (Callable[[Message], None]): Callback for handling incoming messages.
+        """
         self.logger: lg.Logger = lg.getLogger(
             f"{__name__}.{self.__class__.__name__}"
         )
@@ -24,167 +36,170 @@ class Connector:
 
         self.host: str = host
         self.port: int = port
-        self.socket_timeout: None = None
-        self.connect_timeout: float = connect_timeout
-        self.ack_timeout: float = ack_timeout
 
+        self.selector = selectors.DefaultSelector()
         self.socket: socket.socket = socket.socket(
             socket.AF_INET, socket.SOCK_STREAM
         )
+        self.socket.setblocking(False)
 
-        self.mac_address: str = self._get_mac_address()
+        self.selector.register(
-
+            self.socket,
-    def _get_mac_address(self) -> str:
+            selectors.EVENT_READ | selectors.EVENT_WRITE,
-        mac_address: str = ":".join(
+            data=None,
-            [
-                "{:02x}".format((uuid.getnode() >> ele) & 0xFF)
-                for ele in range(0, 8 * 6, 8)
-            ][::-1]
         )
-        self.logger.debug(f"MAC address: {mac_address}")
-        return mac_address
 
-    def _send_ack(self) -> None:
+        self.mac_address: str = mac_address
-        self.logger.debug("[>] Sending ACK...")
-        try:
-            self.socket.sendall(Message.ack().to_bytes())
-            self.logger.debug("[<] ACK sent")
-        except socket.error as e:
-            self.logger.error(f"[!] Failed to send ACK: {e}")
 
-    def _check_ack(self) -> bool:
+        self.inbound_buffer: bytes = b""
-        self.logger.debug("[.] Waiting for ACK...")
+        self.outbound_buffer: bytes = b""
-        try:
-            self.socket.settimeout(self.ack_timeout)
-            ack: bytes = self.socket.recv(1024)
-            self.socket.settimeout(self.socket_timeout)
 
-            if ack == Message.ack().to_bytes():
+        self.pending_acks: dict[str, tuple[Message, float]] = {}
-                self.logger.debug("[<] ACK received")
-                return True
-            else:
-                self.logger.error(f"[!] Invalid ACK received: {ack}")
 
-        except TimeoutError as e:
+        self.running: bool = True
-            self.logger.error(f"[!] ACK timeout: {e}")
 
-        except socket.error as e:
+        self.on_message: Callable[[Message], None] = on_message
-            self.logger.error(f"[!] Failed to receive ACK: {e}")
 
-        return False
+    def _send_outbound(self) -> None:
-
+        """Send data from the outbound buffer."""
-    def connect(self, retry_interval: int = 1) -> None:
+        while self.outbound_buffer:
-        self.logger.debug(
-            f"Connecting to {self.host}:{self.port} with timeout {self.connect_timeout}s..."
-        )
-        try:
-            self.socket.settimeout(self.connect_timeout)
-            self.socket.connect((self.host, self.port))
-            self.socket.settimeout(self.socket_timeout)
-            self.logger.info(f"[+] Connected to {self.host}:{self.port}")
-            self.send_hello()
-        except (
-            socket.timeout,
-            ConnectionRefusedError,
-            ConnectionAbortedError,
-        ) as e:
-            self.logger.error(
-                f"[!] Connection to {self.host}:{self.port} failed: {e}"
-            )
-            self.logger.info(
-                f"[.] Retrying connection in {retry_interval} s..."
-            )
-            time.sleep(retry_interval)
-            self.connect(retry_interval=min(30, retry_interval * 2))
-
-    def send(self, data: bytes, no_check_ack: bool = False) -> None:
-        self.logger.debug(f"[>] Sending data: {data}")
-        while True:
             try:
-                self.socket.sendall(data)
+                sent = self.socket.send(self.outbound_buffer)
+                self.logger.debug(
+                    f"[>] Sent {sent} bytes: {self.outbound_buffer[:sent]!r}"
+                )
 
-                if no_check_ack:
+                self.outbound_buffer = self.outbound_buffer[sent:]
-                    self.logger.debug("[>] Data sent without ACK check")
+            except BlockingIOError:
-                    break
+                # OS buffer full, wait for next EVENT_WRITE
-                else:
+                break
-                    self.logger.info("[>] Data sent")
+            except socket.error as e:
+                self.logger.error(f"[!] Socket error: {e}")
+                self.reconnect()
+                break
 
-                acknowledged: bool = self._check_ack()
+    def _receive_inbound(self) -> None:
-                if acknowledged:
+        """Receive data into the inbound buffer."""
-                    self.logger.debug("[.] Data acknowledged")
-                    break
-                else:
-                    self.logger.warning(
-                        "[!] Data not acknowledged, retrying..."
-                    )
-
-            except BrokenPipeError as e:
-                self.logger.error(f"[!] Broken pipe: {e}")
-                self.logger.info("[.] Reconnecting...")
-                self.connect()
-            except (socket.error, ValueError) as e:
-                self.logger.error(f"[!] Failed to send data: {e}")
-            time.sleep(1)
-
-    def receive(self) -> bytes:
-        self.logger.debug("[.] Waiting to receive data...")
         try:
             data: bytes = self.socket.recv(4096)
-            if not data:
+            if data:
-                self.logger.warning("[!] Received empty message")
+                self.logger.debug(f"[<] Received {len(data)} bytes: {data!r}")
-                return b""
+                self.inbound_buffer += data
-            self.logger.debug(f"[<] Received data: {data}")
+            else:
-            return data
+                self.logger.debug("[!] Connection closed by the server.")
+                self.reconnect()
         except socket.error as e:
-            self.logger.error(f"[!] Failed to receive data: {e}")
+            self.logger.error(f"[!] Socket error: {e}")
-            return b""
+            self.reconnect()
 
-    def close(self) -> None:
+    def send(self, message: Message) -> None:
-        self.logger.debug("Closing connection...")
+        """Send a message to the server.
-        self.socket.close()
-        self.logger.info("Connection closed.")
 
-    def reconnect(self) -> None:
+        Args:
-        self.logger.debug("Reconnecting...")
+            message (Message): The message to send.
-        self.close()
+        """
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.logger.debug(f"[>] Queueing message to send: {message}")
-        self.connect()
+        if message.ack_required:
+            self.pending_acks[message.id] = (message, time.time())
+        self.outbound_buffer += message.to_bytes()
 
     def send_hello(self) -> None:
-        self.logger.debug("[.] Sending hello message...")
+        """Send a HELLO message to the server."""
+        self.logger.debug("[*] Sending HELLO message...")
         hello_message: Message = Message.hello(self.mac_address)
-        acknowledged: bool = False
+        self.send(hello_message)
-        while not acknowledged:
-            self.send(hello_message.to_bytes(), no_check_ack=True)
-            self.logger.debug("[.] Hello message sent, waiting for ACK...")
-            acknowledged = self._check_ack()
-            if not acknowledged:
-                self.logger.warning(
-                    "[!] Hello message not acknowledged, retrying..."
-                )
-                time.sleep(1)
 
-    def _loop(self) -> None:
+    def close(self) -> None:
-        self.logger.debug("Starting main loop...")
+        """Close the connection and clean up resources."""
-        while True:
+        self.logger.debug("[*] Closing connection...")
-            time.sleep(1)
+        try:
-            data: bytes = self.receive()
+            self.selector.unregister(self.socket)
-            if not data:
+        except Exception as e:
-                self.reconnect()
+            self.logger.error(f"[!] Error unregistering socket: {e}")
-                continue
+        self.socket.close()
-            message = Message.from_bytes(data.strip())
+        self.logger.debug("[.] Connection closed.")
-            self.logger.info(f"[<] Message received: {message}")
+
-            # if self._check_ack():
+    def reconnect(self) -> None:
-            #     self.logger.debug("[.] ACK verified")
+        """Reconnect to the server."""
-            # else:
+        self.logger.debug("[*] Reconnecting...")
-            #     self.logger.error("[!] ACK verification failed")
+        self.close()
+
+        # reinit socket
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.setblocking(False)
+        self.selector.register(
+            self.socket,
+            selectors.EVENT_READ | selectors.EVENT_WRITE,
+            data=None,
+        )
+        self.connect()
+
+    def connect(self) -> None:
+        """Establish a connection to the server."""
+        self.logger.debug(f"Connecting to {self.host}:{self.port}...")
+        connected: bool = False
+        delay: float = 1.0
+        while not connected:
+            try:
+                self.socket.connect((self.host, self.port))
+                connected = True
+            except BlockingIOError:
+                # Connection in progress
+                time.sleep(0.1)
+            except socket.error as e:
+                self.logger.error(f"[!] Connection error: {e}")
+                self.logger.debug(f"[.] Retrying in {delay} seconds...")
+                time.sleep(delay)
+                delay = min(delay * 2, 30)  # exponential backoff
+
+        self.logger.debug("[*] Connected, sending HELLO...")
+        self.send_hello()
 
     def run(self) -> None:
-        self.logger.debug("Running Connector...")
+        """Run the main event loop."""
+        self.connect()
         try:
-            self.connect()
+            while self.running:
-            self._loop()
+                events = self.selector.select(timeout=1)
+                for key, mask in events:
+                    if mask & selectors.EVENT_READ:
+                        self._receive_inbound()
+                    if mask & selectors.EVENT_WRITE:
+                        self._send_outbound()
+
+                # Process inbound buffer for complete messages
+                while b"\n" in self.inbound_buffer:
+                    message_bytes, self.inbound_buffer = (
+                        self.inbound_buffer.split(b"\n", 1)
+                    )
+                    try:
+                        message: Message = Message.from_bytes(message_bytes)
+                        # handle incoming ACKs
+                        if (
+                            message.category == Category.CONTROL
+                            and message.action == ControlAction.ACK
+                        ):
+                            if (
+                                message.payload.get("target_id")
+                                in self.pending_acks
+                            ):
+                                target_id = message.payload["target_id"]
+                                self.logger.debug(
+                                    f"[.] Received ACK for message ID {target_id}"
+                                )
+                                del self.pending_acks[target_id]
+                        else:
+                            self.on_message(message)
+
+                        if message.ack_required:
+                            ack_message: Message = Message.ack(message.id)
+                            self.send(ack_message)
+                            self._send_outbound()
+                    except Exception as e:
+                        self.logger.error(f"[!] Failed to parse message: {e}")
+
+                time.sleep(0.1)
         except KeyboardInterrupt:
-            self.logger.info("Interrupted by user.")
+            self.logger.debug("[*] Interrupted by user.")
         finally:
             self.close()

uv.lock

@@ -292,8 +292,8 @@ test = [
 
 [[package]]
 name = "judas-protocol"
-version = "0.1.0"
+version = "0.5.0"
-source = { git = "https://gitea.pufereq.pl/judas/judas_protocol.git#fd070b176347a0f7b81f937b189d8f50736f3514" }
+source = { git = "https://gitea.pufereq.pl/judas/judas_protocol.git#c48b69ecee16f5824ffd8bce8921341d5fa326b7" }
 
 [[package]]
 name = "markdown-it-py"