src/judas_server/backend/client.py

@@ -0,0 +1,58 @@
+# -*- coding: utf-8 -*-
+"""Client representation."""
+
+from __future__ import annotations
+
+import logging as lg
+import socket
+
+from enum import Enum
+
+
+class ClientStatus(str, Enum):
+    CONNECTED = "connected"
+    DISCONNECTED = "disconnected"
+
+
+class Client:
+    """Represents a client."""
+
+    def __init__(
+        self, id_: str, addr: tuple[str, int], socket: socket.socket
+    ) -> None:
+        """Initialize the client.
+
+        Args:
+            id_ (str): The unique identifier for the client.
+            addr (tuple[str, int]): The (IP, port) address of the client.
+            socket (socket.socket): The socket object for communication.
+        """
+        self.logger: lg.Logger = lg.getLogger(
+            f"{__name__}.{self.__class__.__name__}"
+        )
+        self.logger.debug(f"Initializing Client {addr}...")
+
+        self.id: str = id_
+        self.last_seen: float = 0.0  # unix timestanp of last inbound message
+        self.status: ClientStatus = ClientStatus.CONNECTED
+
+        self.socket: socket.socket = socket
+        self.addr: tuple[str, int] = addr
+        self.inbound: bytes = b""
+        self.outbound: bytes = b""
+
+    def __str__(self) -> str:
+        return f"Client({self.id} ({self.addr[0]}:{self.addr[1]}))"
+
+    def __repr__(self) -> str:
+        return f"Client({self.id}, {self.addr})"
+
+    def disconnect(self) -> None:
+        """Disconnect the client and close the socket."""
+        self.logger.debug(f"Disconnecting Client {self}...")
+        try:
+            self.socket.close()
+        except Exception as e:
+            self.logger.error(f"Error closing socket for Client {self}: {e}")
+        self.status = ClientStatus.DISCONNECTED
+        self.logger.info(f"Client {self} disconnected.")