Overview
The Server class manages multiple client connections, creating a Channel instance for each one. It can run in the main thread or in a background daemon thread for host/client P2P setups.
Class Definition
A generic async TCP server parameterized by the Channel type.
Class Attributes
The Channel subclass instantiated for each new connection. Must be set in your subclass.
Hostname or IP address the server is bound to.
Port number the server is listening on.
List of currently connected client channels.
Constructor
def __init__(
self,
host: str = "127.0.0.1",
port: int = 5071,
) -> None
Hostname or IP to bind to. Use "0.0.0.0" for all interfaces.
Port number to listen on.
Properties
address
@property
def address(self) -> tuple[str, int]
(host, port) tuple.
A tuple containing the host and port the server is bound to.
Methods
start
async def start(self) -> None
host:port and begins listening for incoming TCP connections.
This is an async method that must be awaited.
run
async def run(self) -> None
launch
Start the server and block forever. Convenience wrapper that hides asyncio boilerplate. Equivalent to calling start() then run() inside asyncio.run(). Handles KeyboardInterrupt gracefully.
Example:
GameServer(host="0.0.0.0", port=5071).launch()
stop
async def stop(self) -> None
start_background
def start_background(self) -> threading.Thread
The thread running the server.
on_connect
def on_connect(self, channel: C, addr: tuple[str, int]) -> None
The newly created channel for this client.
(host, port) of the remote client.
on_disconnect
def on_disconnect(self, channel: C) -> None
The channel that disconnected.
send_to_all
def send_to_all(self, data: dict[str, Any]) -> None
Message dictionary to send to all channels.
Complete Example
from repod import Server, Channel
class GameChannel(Channel):
def Network_chat(self, data: dict) -> None:
self.server.send_to_all(
{"action": "chat", "text": data["text"]}
)
class GameServer(Server):
channel_class = GameChannel
GameServer(host="0.0.0.0", port=5071).launch()