The ConnectionListener class provides a high-level interface for building game clients. It runs networking in a background thread, keeping your main game loop synchronous and simple.
Creating a Client Subclass
Clients are built by subclassing ConnectionListener and defining message handlers:
Subclass ConnectionListener
Create a class that inherits from ConnectionListener:from repod import ConnectionListener
class GameClient(ConnectionListener):
"""Client that connects to a game server."""
def __init__(self):
self.running = True
self.player_name = "Guest"
Define Network_* handlers
Create methods that handle specific message types from the server:def Network_connected(self, data: dict) -> None:
print("Connected to server!")
# Send initial data to server
self.send({
"action": "join",
"name": self.player_name
})
def Network_game_state(self, data: dict) -> None:
# Update local game state from server
self.update_game(data["state"])
Connect and pump
Call connect() to establish the connection, then call pump() in your game loop:client = GameClient()
client.connect("localhost", 5071)
while client.running:
client.pump() # Process network messages
# ... rest of your game loop
time.sleep(0.01)
Connecting to a Server
The connect() method creates a background network thread and attempts to connect:
class GameClient(ConnectionListener):
def __init__(self, host: str = "127.0.0.1", port: int = 5071):
self.running = True
self.connect(host, port) # Non-blocking
print("Connecting...")
The connect() method returns immediately. The actual connection happens asynchronously in the background thread. Use Network_connected to know when the connection succeeds.
Connection Events
The client automatically sends these action types:
connected - Connection established successfullysocketConnect - Low-level socket connected (internal)error - Connection failed (includes error message)disconnected - Connection closed
class GameClient(ConnectionListener):
def Network_connected(self, data: dict) -> None:
print("Connected to server!")
def Network_error(self, data: dict) -> None:
print(f"Connection error: {data.get('error', 'Unknown')}")
self.running = False
def Network_disconnected(self, data: dict) -> None:
print("Disconnected from server")
self.running = False
The Pump Loop
The pump() method processes all pending network messages. Call it once per frame in your game loop:
import time
client = GameClient()
client.connect("localhost", 5071)
while client.running:
client.pump() # Dispatch queued messages
# Your game logic here
update_game()
render_frame()
time.sleep(0.01) # ~100 FPS
How Pump Works
When you call pump(), it:
- Checks the receive queue for new messages
- For each message, looks for a
Network_{action} method
- Calls the matching handler with the message data
- Falls back to
network_received() for unhandled actions
class GameClient(ConnectionListener):
def network_received(self, data: dict) -> None:
"""Fallback for unhandled message types."""
print(f"Unhandled message: {data.get('action')}")
Call pump() frequently (every frame) to minimize input lag. The method returns immediately if there are no queued messages.
Sending Messages
Use the send() method to send messages to the server:
class GameClient(ConnectionListener):
def send_chat_message(self, text: str) -> None:
self.send({
"action": "message",
"text": text
})
def send_player_input(self, keys: dict) -> None:
self.send({
"action": "input",
"keys": keys,
"timestamp": time.time()
})
send() method returns the number of bytes queued, or 0 if disconnected:
bytes_sent = self.send({"action": "ping"})
if bytes_sent == 0:
print("Not connected!")
Complete Example
Here’s a complete chat client implementation:
import sys
import threading
import time
from repod import ConnectionListener
class ChatClient(ConnectionListener):
"""Chat client that connects to a chat server."""
def __init__(self, host: str = "127.0.0.1", port: int = 5071):
self.running = True
self.connect(host, port)
print("Chat client started")
def set_nickname(self, nickname: str) -> None:
"""Send the chosen nickname to the server."""
self.send({"action": "nickname", "nickname": nickname})
def send_message(self, text: str) -> None:
"""Send a chat message to the server."""
self.send({"action": "message", "text": text})
def run(self) -> None:
"""Synchronous main loop (ideal for game loops like pygame)."""
while self.running:
self.pump()
time.sleep(0.01)
def Network_connected(self, data: dict) -> None:
print("Connected to server!")
def Network_system(self, data: dict) -> None:
print(f"SYSTEM: {data['text']}")
def Network_message(self, data: dict) -> None:
print(f"{data['nickname']}: {data['text']}")
def Network_players(self, data: dict) -> None:
print(f"Players: {', '.join(data['list'])}")
def Network_error(self, data: dict) -> None:
print(f"Error: {data.get('error', 'Unknown error')}")
self.running = False
def Network_disconnected(self, data: dict) -> None:
print("Disconnected from server")
self.running = False
def input_thread(client: ChatClient) -> None:
"""Read user input in a separate thread to avoid blocking the pump loop."""
nickname = input("Enter your nickname: ")
client.set_nickname(nickname)
print("Type your messages (Ctrl+C to quit):")
while client.running:
try:
msg = input()
if msg and client.running:
client.send_message(msg)
except (EOFError, KeyboardInterrupt):
client.running = False
break
def main() -> None:
host = "127.0.0.1"
port = 5071
if len(sys.argv) > 1:
host = sys.argv[1]
if len(sys.argv) > 2:
port = int(sys.argv[2])
client = ChatClient(host, port)
t = threading.Thread(target=input_thread, args=(client,), daemon=True)
t.start()
try:
client.run()
except KeyboardInterrupt:
print("\nDisconnecting...")
client.running = False
if __name__ == "__main__":
main()
Low-Level Client API
For advanced use cases, you can use the Client class directly instead of ConnectionListener:
from repod.client import Client
client = Client(host="localhost", port=5071)
client.start_background()
# Send messages
client.send({"action": "ping"})
# Poll for messages manually
while not client._receive_queue.empty():
data = client._receive_queue.get_nowait()
print(f"Received: {data}")
# Close when done
client.close()
The low-level Client API requires you to manage the receive queue manually. Most applications should use ConnectionListener for its automatic message dispatching and cleaner API.
Properties and Methods
Properties
connection - The underlying Client instance (or None if not connected)
Methods
connect(host, port) - Connect to a serverpump() - Process all pending messagessend(data) - Send a message to the servernetwork_received(data) - Fallback handler for unrecognized actions
Check client.connection to determine if the client has attempted to connect. This doesn’t guarantee the connection is active—use connection state handlers for that.