fix(backend_server.py): fix server waiting for closed socket to dissapear by reusing it in __init__()

pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
     "flask>=3.1.1",
     "flask-login>=0.6.3",
     "flask-socketio>=5.5.1",
+    "judas-protocol",
 ]
 license = { text = "GPL-3.0+" }
 
@@ -85,3 +86,6 @@ allowed_tags = [
 minor_tags = ["feat"]
 patch_tags = ["fix", "perf"]
 default_bump_level = 0
+
+[tool.uv.sources]
+judas-protocol = { git = "https://gitea.pufereq.pl/judas/judas_protocol.git" }

src/judas_server/backend/backend_server.py

@@ -0,0 +1,182 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import logging as lg
+import threading
+import time
+import selectors
+import socket
+
+from judas_protocol import Message
+
+from judas_server.backend.client import Client
+
+
+class BackendServer:
+    def __init__(self, host: str = "0.0.0.0", port: int = 3692) -> None:
+        """Initialize the backend server.
+
+        Args:
+            host (str): The host IP address to bind the server to.
+            port (int): The port number to bind the server to.
+        """
+        self.logger: lg.Logger = lg.getLogger(
+            f"{__name__}.{self.__class__.__name__}"
+        )
+        self.logger.debug("Initializing Server...")
+
+        self.selector = selectors.DefaultSelector()
+        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.server_socket.setsockopt(
+            socket.SOL_SOCKET, socket.SO_REUSEADDR, 1
+        )
+        self._bind_socket(host, port)
+        self.server_socket.listen()
+        self.server_socket.setblocking(False)
+        self.selector.register(
+            self.server_socket, selectors.EVENT_READ, data=None
+        )
+
+        self.clients: dict[str, Client] = {}
+
+        self.running: bool = False
+
+    def _bind_socket(self, host: str, port: int) -> None:
+        """Bind the server socket to the specified host and port.
+
+        Args:
+            host (str): The host IP address to bind the server to.
+            port (int): The port number to bind the server to.
+        """
+        self.logger.debug(f"Binding socket to {host}:{port}")
+        while True:
+            try:
+                self.server_socket.bind((host, port))
+                self.logger.debug(f"Socket bound to {host}:{port}")
+                break
+            except OSError as e:
+                self.logger.error(
+                    f"Failed to bind socket to {host}:{port}, retrying...: {e}"
+                )
+                time.sleep(1)
+
+    def _accept_connection(self, sock: socket.socket) -> None:
+        """Accept a new client connection.
+
+        Args:
+            sock (socket.socket): The selected socket.
+        """
+        conn, addr = sock.accept()
+        self.logger.info(f"[+] Accepted connection from {addr}")
+        conn.setblocking(False)
+
+        # wait for hello message to get mac_id
+
+        conn.settimeout(5)
+        message = conn.recv(1024)
+        conn.settimeout(None)
+
+        message = Message.from_bytes(message)
+
+        mac_id = message.payload.get("mac", None)
+        if mac_id is None:
+            self.logger.error(
+                f"[-] No mac_id provided by {addr}, closing connection"
+            )
+            conn.close()
+            return
+
+        client = Client(id_=mac_id, addr=addr, socket=conn)
+        self.clients[mac_id] = client
+
+        events = selectors.EVENT_READ | selectors.EVENT_WRITE
+        self.selector.register(conn, events, data=client)
+
+        self.logger.info(f"[+] Registered client {client}")
+
+    def _disconnect(self, client: Client) -> None:
+        """Disconnect a client and clean up resources.
+
+        Args:
+            sock (socket.socket): The client socket to disconnect.
+        """
+        self.logger.info(f"[-] Disconnecting {client}")
+        self.selector.unregister(client.socket)
+        client.disconnect()
+
+    def _handle_connection(
+        self, key: selectors.SelectorKey, mask: int
+    ) -> None:
+        """Handle a client connection.
+
+        Args:
+            key (selectors.SelectorKey): The selector key for the client.
+            mask (int): The event mask.
+        """
+        sock: socket.socket = key.fileobj
+        client = key.data
+
+        try:
+            if mask & selectors.EVENT_READ:
+                recv_data = sock.recv(1024)
+                if recv_data:
+                    self.logger.debug(
+                        f"[<] Received data from {client}: {recv_data!r}"
+                    )
+                    client.inbound += recv_data
+                    while b"\n" in client.inbound:
+                        line, client.inbound = client.inbound.split(b"\n", 1)
+                        self.logger.info(
+                            f"[<] Complete message from {client}: {line!r}"
+                        )
+
+                    # send ACK
+                    ack = Message.ack().to_bytes()
+                    self.logger.debug(f"[>] Sending ACK to {client}")
+                    client.outbound += ack
+
+                else:
+                    self._disconnect(client)
+
+            if mask & selectors.EVENT_WRITE:
+                if client.outbound:
+                    self.logger.debug(
+                        f"[>] Sending data to {client}: {client.outbound!r}"
+                    )
+                    sent = sock.send(client.outbound)
+
+                    client.outbound = client.outbound[sent:]
+        except ConnectionResetError as e:
+            self.logger.error(f"Connection reset by {client}, disconnect: {e}")
+            self._disconnect(client)
+
+    def run(self) -> None:
+        """Start the backend server."""
+        self.running = True
+        threading.Thread(
+            name="BackendServer thread", target=self._loop, daemon=True
+        ).start()
+
+    def _loop(self) -> None:
+        """Main server loop to handle incoming connections and data."""
+        self.logger.info("Server is running...")
+        try:
+            while self.running:
+                events = self.selector.select(timeout=1)
+                for key, mask in events:
+                    if key.data is None:
+                        self._accept_connection(key.fileobj)
+                    else:
+                        self._handle_connection(key, mask)
+        except Exception as e:
+            self.logger.error(f"Server error: {e}")
+            raise e
+        finally:
+            self.selector.close()
+            self.server_socket.close()
+            self.logger.info("Server has stopped.")
+
+    # def get_client_data(
+    #     self, client_id: str
+    # ) -> dict[str, dict[str, Any]] | None:
+    #     return self.clients.get(client_id, None)

src/judas_server/backend/server.py

@@ -1,54 +0,0 @@
-# -*- coding: utf-8 -*-
-from __future__ import annotations
-
-from typing import Any
-
-import logging as lg
-import random as rn
-import threading
-import time
-
-
-class BackendServer:
-    def __init__(self) -> None:
-        self.logger: lg.Logger = lg.getLogger(
-            f"{__name__}.{self.__class__.__name__}"
-        )
-        self.logger.debug("Initializing Server...")
-
-        # TODO: add socket logic here
-
-        self.clients: dict[str, dict[str, dict[str, Any]]] = {
-            "C_01": {
-                "one_time": {
-                    "hostname": "mock-host",
-                    "platform": "windows 11",
-                    "cpu_info": "i7",
-                },
-                "polled": {"cpu_usage": 0, "ram_usage": 0},
-                "ondemand": {},
-            }
-        }
-
-        self.running: bool = False
-
-    def run(self) -> None:
-        self.running = True
-        threading.Thread(
-            name="BackendServer thread", target=self._loop, daemon=True
-        ).start()
-
-    def _loop(self) -> None:
-        self.logger.info("Starting server loop...")
-        while self.running:
-            for client in self.clients.values():
-                client["polled"]["cpu_usage"] = round(rn.uniform(0, 100), 1)
-                client["polled"]["ram_usage"] = round(rn.uniform(0, 100), 1)
-            time.sleep(1)
-
-        self.logger.info("Server loop stopped.")
-
-    def get_client_data(
-        self, client_id: str
-    ) -> dict[str, dict[str, Any]] | None:
-        return self.clients.get(client_id, None)

uv.lock

@@ -308,6 +308,11 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" },
 ]
 
+[[package]]
+name = "judas-protocol"
+version = "0.1.0"
+source = { git = "https://gitea.pufereq.pl/judas/judas_protocol.git#fd070b176347a0f7b81f937b189d8f50736f3514" }
+
 [[package]]
 name = "judas-server"
 version = "0.2.0"
@@ -316,6 +321,7 @@ dependencies = [
     { name = "flask" },
     { name = "flask-login" },
     { name = "flask-socketio" },
+    { name = "judas-protocol" },
 ]
 
 [package.dev-dependencies]
@@ -335,6 +341,7 @@ requires-dist = [
     { name = "flask", specifier = ">=3.1.1" },
     { name = "flask-login", specifier = ">=0.6.3" },
     { name = "flask-socketio", specifier = ">=5.5.1" },
+    { name = "judas-protocol", git = "https://gitea.pufereq.pl/judas/judas_protocol.git" },
 ]
 
 [package.metadata.requires-dev]