Skip to main content

​
Overview

A Channel represents one side of a TCP connection. On the server side, each connected client gets its own Channel instance. Subclass Channel and define Network_{action} methods to handle specific message types.

​
Class Definition

class Channel[S: Server]
A generic channel parameterized by the Server type.

​
Class Attributes

​
addr
tuple[str, int]
(host, port) tuple of the remote endpoint.
​
is_connected
bool
Whether the connection is currently active.

​
Constructor

def __init__(
    self,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    server: S | None = None,
) -> None
Initialize a channel with asyncio stream reader/writer.
​
reader
asyncio.StreamReader
Async stream reader for incoming data.
​
writer
asyncio.StreamWriter
Async stream writer for outgoing data.
​
server
S | None
default:"None"
Optional reference to the parent Server.

​
Properties

​
addr

@property
def addr(self) -> tuple[str, int]
Remote address as a (host, port) tuple.
​
returns
tuple[str, int]
The remote endpoint address.

​
is_connected

@property
def is_connected(self) -> bool
Whether this channel is still connected.
​
returns
bool
True if connected, False if closed.

​
server

@property
def server(self) -> S
The parent Server instance.
​
returns
S
The parent server instance.
​
raises
RuntimeError
If the channel is not connected to a server.

​
Methods

​
send

def send(self, data: dict[str, Any]) -> int
Queue a message to be sent to the remote endpoint. The dictionary is serialized with msgpack and framed with a 4-byte length prefix before being placed in the async send queue. Thread-safe — can be called from any thread, including the main game loop thread when using Server.start_background().
​
data
dict[str, Any]
Message dictionary. Should contain an action key to identify the message type on the receiver side.
​
returns
int
Number of bytes queued, or 0 if disconnected.

​
on_connect

def on_connect(self) -> None
Called when the connection is established. Override in your subclass to run setup logic when a client connects.

​
on_close

def on_close(self) -> None
Called when the connection is closed. Override in your subclass to run cleanup logic when a client disconnects.

​
on_error

def on_error(self, error: Exception) -> None
Called when a connection error occurs. Override to implement custom error handling.
​
error
Exception
The exception that was raised.

​
network_received

def network_received(self, data: dict[str, Any]) -> None
Fallback handler for messages with no specific handler. Called when a message’s action does not match any Network_{action} method. Override to handle unrecognized messages.
​
data
dict[str, Any]
The received message dictionary.

​
Message Handling

The Channel class uses a convention-based routing system for incoming messages. When a message is received with an action field, the channel looks for a method named Network_{action} and calls it with the message data.

​
Example

from repod import Channel

class GameChannel(Channel):
    def Network_chat(self, data: dict) -> None:
        print(f"Chat: {data['message']}")
    
    def Network_move(self, data: dict) -> None:
        print(f"Player moved to {data['x']}, {data['y']}")
    
    def on_connect(self) -> None:
        print(f"Client connected from {self.addr}")
    
    def on_close(self) -> None:
        print(f"Client {self.addr} disconnected")
    
    def on_error(self, error: Exception) -> None:
        print(f"Error on {self.addr}: {error}")
    
    def network_received(self, data: dict) -> None:
        print(f"Unhandled message: {data}")

​
Broadcasting Messages

To broadcast a message to all connected clients, use the server’s send_to_all() method: 
class GameChannel(Channel):
    def Network_chat(self, data: dict) -> None:
        # Broadcast chat message to all clients
        self.server.send_to_all(
            {"action": "chat", "text": data["text"]}
        )

Build docs developers (and LLMs) love

Get started for freeTalk to us