Python Client/Server

client.py

import socket
import threading
import queue
from typing import Callable


class GameClient:
    def __init__(self, host: str, port: int):
        self.host = socket.gethostbyname(host) if host else None
        self.port = port
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connection = None
        self.inbox = queue.Queue()
        self.outbox = queue.Queue()
        self.listen_thread = threading.Thread(
            target=self.listen_for_messages, daemon=True
        )
        self.send_thread = threading.Thread(target=self.process_outbox, daemon=True)

    def start(self):
        """Starts the client and connects it to the server."""
        self.socket.connect((self.host, self.port))
        self.connection = self.socket

        self.listen_thread.start()
        self.send_thread.start()
        print("Connected to the server.")

    def send_message(self, message: bytes):
        """Queues a message to be sent to the server."""
        if self.connection:
            self.outbox.put(message)

    def listen_for_messages(self):
        """Listens for incoming messages from the server and puts them in the inbox."""
        while self.connection:
            try:
                message = self.connection.recv(1024).decode()
                if message:
                    self.inbox.put(message)
                else:
                    print("Disconnected from the server.")
                    self.connection = None
                    break
            except socket.error as e:
                print(f"Error receiving data: {e}")
                return None
        else:
            """If the connection is closed, stop the client."""
            self.close()

    def process_outbox(self):
        """Processes the outbox, sending messages to the server."""
        while self.connection:
            if not self.outbox.empty():
                try:
                    message = self.outbox.get()
                    self.connection.sendall(message)
                except socket.error as e:
                    print(f"Error sending data: {e}")
                    return None

    def close(self):
        """Closes the connection to the server."""
        if self.connection:
            self.connection.close()
            self.connection = None
        self.socket.close()


if __name__ == "__main__":
    client = GameClient("localhost", 12345)
    client.start()

    while True:
        message = input("Enter a message to send (type 'quit' to exit): ")
        client.send_message(message.encode())
        if message == "quit":
            break

    client.close()

server.py

import threading
import queue
import socket
import uuid


class GameServer:
    def __init__(self, host: str, port: int) -> None:
        self.host: str = host
        self.port: int = port
        self.clients: dict[
            uuid.UUID, socket.socket
        ] = {}  # Stores the client sockets by UUID
        self.inbox: queue.Queue = queue.Queue()
        self.outbox: queue.Queue = queue.Queue()
        self.server_socket: socket.socket = socket.socket(
            socket.AF_INET, socket.SOCK_STREAM
        )

    def start_server(self) -> None:
        self.server_socket.bind((self.host, self.port))
        self.server_socket.listen()
        print("Server started, waiting for connections...")

        threading.Thread(target=self.process_messages, daemon=True).start()

        try:
            while True:
                client_socket, (addr, port) = self.server_socket.accept()
                client_uuid = uuid.uuid4()
                self.clients[client_uuid] = client_socket
                print(f"Connected to client with UUID: {client_uuid} ({addr}:{port})")
                threading.Thread(
                    target=self.handle_client,
                    args=(client_socket, client_uuid),
                    daemon=True,
                ).start()
        except KeyboardInterrupt:
            print("Server is shutting down.")
        finally:
            self.server_socket.close()

    def handle_client(
        self, client_socket: socket.socket, client_uuid: uuid.UUID
    ) -> None:
        while True:
            try:
                data: str = client_socket.recv(1024).decode()
                if not data:
                    break
                self.inbox.put((client_uuid, data))
            except socket.error:
                break

        client_socket.close()
        del self.clients[client_uuid]
        print(f"Connection with UUID {client_uuid} closed.")

    def process_messages(self) -> None:
        while True:
            if not self.inbox.empty():
                client_uuid, data = self.inbox.get()
                self.handle_incoming_message(client_uuid, data)

            if not self.outbox.empty():
                client_uuid, message = self.outbox.get()
                self.send_message(client_uuid, message)

    def handle_incoming_message(self, client_uuid: uuid.UUID, message: str) -> None:
        print(f"Message from {client_uuid}: {message}")
        # Process incoming messages here
        self.outbox.put((client_uuid, f"Processed: {message}"))

    def send_message(self, client_uuid: uuid.UUID, message: str) -> None:
        if client_uuid in self.clients:
            try:
                self.clients[client_uuid].sendall(message.encode())
            except socket.error:
                pass

    def broadcast(
        self, message: str, specific_user_uuid: uuid.UUID | None = None
    ) -> None:
        if specific_user_uuid:
            self.outbox.put((specific_user_uuid, message))
        else:
            for client_uuid in self.clients:
                self.outbox.put((client_uuid, message))


if __name__ == "__main__":
    server = GameServer("localhost", 12345)
    server.start_server()