Socket Programming

Master the fundamentals of socket programming! Learn how to create networked games using TCP and UDP protocols, implement real-time communication, and handle multiple clients! 🌐🎮🔌

🔌 What Are Sockets?

A socket is an endpoint for network communication. Think of it as a telephone - you need one on each end to have a conversation!

• IP Address: The "phone number" - identifies the computer
• Port: The "extension" - identifies the specific application
• Protocol: The "language" - how data is formatted (TCP or UDP)
• Socket: The combination of IP + Port + Protocol

📡 Choosing the Right Protocol

🔧 TCP Server and Client

TCP Server

import socket
import threading
import json
from typing import Dict, List, Optional, Any, Tuple

class GameServer:
    def __init__(self, host: str = 'localhost', port: int = 5555) -> None:
        self.host: str = host
        self.port: int = port
        self.server: socket.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.clients: Dict[socket.socket, Dict[str, Any]] = {}  # {client_socket: player_data}
        self.running: bool = False
        
    def start(self) -> None:
        """Start the server"""
        # Allow reuse of address
        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        
        # Bind to address and port
        self.server.bind((self.host, self.port))
        
        # Listen for connections (max 5 queued)
        self.server.listen(5)
        self.running = True
        
        print(f"Server listening on {self.host}:{self.port}")
        
        # Accept connections in a loop
        while self.running:
            try:
                client_socket, address = self.server.accept()
                print(f"New connection from {address}")
                
                # Handle each client in a separate thread
                client_thread = threading.Thread(
                    target=self.handle_client,
                    args=(client_socket, address)
                )
                client_thread.start()
                
            except Exception as e:
                print(f"Error accepting connection: {e}")
    
    def handle_client(self, client_socket: socket.socket, address: Tuple[str, int]) -> None:
        """Handle individual client connection"""
        # Initialize player data
        player_id = f"player_{len(self.clients)}"
        self.clients[client_socket] = {
            'id': player_id,
            'address': address,
            'x': 100,
            'y': 100,
            'score': 0
        }
        
        # Send welcome message
        welcome_msg = {
            'type': 'welcome',
            'player_id': player_id,
            'message': 'Connected to game server!'
        }
        self.send_to_client(client_socket, welcome_msg)
        
        # Broadcast new player to others
        self.broadcast({
            'type': 'player_joined',
            'player_id': player_id
        }, exclude=client_socket)
        
        # Main client loop
        while self.running:
            try:
                # Receive data (max 4096 bytes)
                data = client_socket.recv(4096).decode('utf-8')
                
                if not data:
                    break
                
                # Parse JSON message
                message = json.loads(data)
                self.process_message(client_socket, message)
                
            except ConnectionResetError:
                print(f"Client {address} disconnected abruptly")
                break
            except json.JSONDecodeError:
                print(f"Invalid JSON from {address}")
            except Exception as e:
                print(f"Error handling client {address}: {e}")
                break
        
        # Clean up on disconnect
        self.remove_client(client_socket)
    
    def process_message(self, client_socket: socket.socket, message: Dict[str, Any]) -> None:
        """Process incoming message from client"""
        msg_type = message.get('type')
        player = self.clients[client_socket]
        
        if msg_type == 'move':
            # Update player position
            player['x'] = message.get('x', player['x'])
            player['y'] = message.get('y', player['y'])
            
            # Broadcast position to all clients
            self.broadcast({
                'type': 'player_moved',
                'player_id': player['id'],
                'x': player['x'],
                'y': player['y']
            })
            
        elif msg_type == 'chat':
            # Broadcast chat message
            self.broadcast({
                'type': 'chat',
                'player_id': player['id'],
                'message': message.get('message', '')
            })
            
        elif msg_type == 'get_players':
            # Send list of all players
            players_list = [
                {
                    'id': p['id'],
                    'x': p['x'],
                    'y': p['y'],
                    'score': p['score']
                }
                for p in self.clients.values()
            ]
            self.send_to_client(client_socket, {
                'type': 'players_list',
                'players': players_list
            })
    
    def send_to_client(self, client_socket: socket.socket, message: Dict[str, Any]) -> None:
        """Send message to specific client"""
        try:
            json_msg = json.dumps(message)
            client_socket.send(json_msg.encode('utf-8'))
        except Exception as e:
            print(f"Error sending to client: {e}")
    
    def broadcast(self, message: Dict[str, Any], exclude: Optional[socket.socket] = None) -> None:
        """Broadcast message to all clients"""
        json_msg = json.dumps(message)
        
        for client_socket in list(self.clients.keys()):
            if client_socket != exclude:
                try:
                    client_socket.send(json_msg.encode('utf-8'))
                except:
                    # Remove dead connections
                    self.remove_client(client_socket)
    
    def remove_client(self, client_socket: socket.socket) -> None:
        """Remove client from server"""
        if client_socket in self.clients:
            player_id = self.clients[client_socket]['id']
            del self.clients[client_socket]
            
            # Notify others
            self.broadcast({
                'type': 'player_left',
                'player_id': player_id
            })
            
            client_socket.close()
            print(f"Client {player_id} disconnected")
    
    def stop(self) -> None:
        """Stop the server"""
        self.running = False
        for client in list(self.clients.keys()):
            client.close()
        self.server.close()

# TCP Client
class GameClient:
    def __init__(self) -> None:
        self.socket: Optional[socket.socket] = None
        self.connected: bool = False
        self.player_id: Optional[str] = None
        self.receive_thread: Optional[threading.Thread] = None
        
    def connect(self, host: str = 'localhost', port: int = 5555) -> bool:
        """Connect to game server"""
        try:
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socket.connect((host, port))
            self.connected = True
            
            # Start receiving thread
            self.receive_thread = threading.Thread(target=self.receive_loop)
            self.receive_thread.daemon = True
            self.receive_thread.start()
            
            print(f"Connected to {host}:{port}")
            return True
            
        except Exception as e:
            print(f"Failed to connect: {e}")
            return False
    
    def receive_loop(self) -> None:
        """Continuously receive messages from server"""
        while self.connected:
            try:
                data = self.socket.recv(4096).decode('utf-8')
                if not data:
                    break
                    
                message = json.loads(data)
                self.handle_message(message)
                
            except Exception as e:
                print(f"Receive error: {e}")
                break
        
        self.disconnect()
    
    def handle_message(self, message: Dict[str, Any]) -> None:
        """Handle incoming message from server"""
        msg_type = message.get('type')
        
        if msg_type == 'welcome':
            self.player_id = message.get('player_id')
            print(f"Assigned ID: {self.player_id}")
            print(message.get('message'))
            
        elif msg_type == 'player_joined':
            print(f"Player {message.get('player_id')} joined the game")
            
        elif msg_type == 'player_left':
            print(f"Player {message.get('player_id')} left the game")
            
        elif msg_type == 'player_moved':
            player_id = message.get('player_id')
            x = message.get('x')
            y = message.get('y')
            print(f"Player {player_id} moved to ({x}, {y})")
            
        elif msg_type == 'chat':
            player_id = message.get('player_id')
            chat_msg = message.get('message')
            print(f"[{player_id}]: {chat_msg}")
    
    def send_message(self, message: Dict[str, Any]) -> bool:
        """Send message to server"""
        if self.connected and self.socket:
            try:
                json_msg = json.dumps(message)
                self.socket.send(json_msg.encode('utf-8'))
                return True
            except Exception as e:
                print(f"Send error: {e}")
                return False
        return False
    
    def move(self, x: int, y: int) -> bool:
        """Send movement update"""
        return self.send_message({
            'type': 'move',
            'x': x,
            'y': y
        })
    
    def chat(self, message: str) -> bool:
        """Send chat message"""
        return self.send_message({
            'type': 'chat',
            'message': message
        })
    
    def disconnect(self) -> None:
        """Disconnect from server"""
        self.connected = False
        if self.socket:
            self.socket.close()
            print("Disconnected from server")

# Example usage
if __name__ == "__main__":
    import sys
    
    if len(sys.argv) > 1 and sys.argv[1] == 'server':
        # Run as server
        server = GameServer()
        try:
            server.start()
        except KeyboardInterrupt:
            print("\nShutting down server...")
            server.stop()
    else:
        # Run as client
        client = GameClient()
        if client.connect():
            # Simple command loop
            while client.connected:
                try:
                    cmd = input("Enter command (move x y, chat msg, quit): ")
                    parts = cmd.split(' ', 2)
                    
                    if parts[0] == 'quit':
                        break
                    elif parts[0] == 'move' and len(parts) >= 3:
                        x = int(parts[1])
                        y = int(parts[2])
                        client.move(x, y)
                    elif parts[0] == 'chat' and len(parts) >= 2:
                        client.chat(parts[1])
                    
                except KeyboardInterrupt:
                    break
                except Exception as e:
                    print(f"Error: {e}")
            
            client.disconnect()
        

⚡ UDP for Real-Time Games

import socket
import json
import time
import struct
import threading
from typing import Dict, Optional, Any, Tuple

class UDPGameServer:
    def __init__(self, host: str = 'localhost', port: int = 5556) -> None:
        self.host: str = host
        self.port: int = port
        self.socket: socket.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.clients: Dict[Tuple[str, int], Dict[str, Any]] = {}  # {address: {last_seen, player_data}}
        self.running: bool = False
        self.sequence_number: int = 0
        
    def start(self) -> None:
        """Start UDP server"""
        self.socket.bind((self.host, self.port))
        self.running = True
        
        print(f"UDP Server listening on {self.host}:{self.port}")
        
        # Start cleanup thread
        cleanup_thread = threading.Thread(target=self.cleanup_inactive_clients)
        cleanup_thread.daemon = True
        cleanup_thread.start()
        
        # Main receive loop
        while self.running:
            try:
                # Receive datagram
                data, address = self.socket.recvfrom(1024)
                
                # Process in separate thread for non-blocking
                threading.Thread(
                    target=self.handle_packet,
                    args=(data, address)
                ).start()
                
            except Exception as e:
                print(f"Error receiving packet: {e}")
    
    def handle_packet(self, data: bytes, address: Tuple[str, int]) -> None:
        """Handle incoming UDP packet"""
        try:
            # Parse packet
            message = json.loads(data.decode('utf-8'))
            
            # Update client last seen time
            if address not in self.clients:
                # New client
                player_id = f"player_{len(self.clients)}"
                self.clients[address] = {
                    'last_seen': time.time(),
                    'player_id': player_id,
                    'x': 100,
                    'y': 100,
                    'vx': 0,
                    'vy': 0,
                    'sequence': 0
                }
                
                # Send acknowledgment
                self.send_to_client(address, {
                    'type': 'connected',
                    'player_id': player_id
                })
                
                print(f"New UDP client: {address}")
            else:
                self.clients[address]['last_seen'] = time.time()
            
            # Process message
            self.process_packet(address, message)
            
        except json.JSONDecodeError:
            print(f"Invalid JSON from {address}")
        except Exception as e:
            print(f"Error handling packet: {e}")
    
    def process_packet(self, address: Tuple[str, int], message: Dict[str, Any]) -> None:
        """Process UDP message"""
        client = self.clients[address]
        msg_type = message.get('type')
        
        # Check sequence number for ordering
        seq = message.get('sequence', 0)
        if seq <= client['sequence'] and msg_type != 'ping':
            return  # Ignore old packets
        client['sequence'] = seq
        
        if msg_type == 'update':
            # Position update
            client['x'] = message.get('x', client['x'])
            client['y'] = message.get('y', client['y'])
            client['vx'] = message.get('vx', 0)
            client['vy'] = message.get('vy', 0)
            
            # Broadcast state to all clients
            self.broadcast_state()
            
        elif msg_type == 'ping':
            # Respond to ping
            self.send_to_client(address, {
                'type': 'pong',
                'timestamp': message.get('timestamp', time.time())
            })
    
    def broadcast_state(self) -> None:
        """Broadcast game state to all clients"""
        # Prepare state snapshot
        state = {
            'type': 'state',
            'timestamp': time.time(),
            'sequence': self.sequence_number,
            'players': []
        }
        
        for addr, client in self.clients.items():
            state['players'].append({
                'id': client['player_id'],
                'x': client['x'],
                'y': client['y'],
                'vx': client['vx'],
                'vy': client['vy']
            })
        
        # Send to all clients
        for address in self.clients.keys():
            self.send_to_client(address, state)
        
        self.sequence_number += 1
    
    def send_to_client(self, address: Tuple[str, int], message: Dict[str, Any]) -> None:
        """Send message to specific client"""
        try:
            data = json.dumps(message).encode('utf-8')
            self.socket.sendto(data, address)
        except Exception as e:
            print(f"Error sending to {address}: {e}")
    
    def cleanup_inactive_clients(self) -> None:
        """Remove inactive clients"""
        while self.running:
            time.sleep(5)  # Check every 5 seconds
            current_time = time.time()
            
            # Remove clients inactive for 10+ seconds
            inactive = []
            for address, client in self.clients.items():
                if current_time - client['last_seen'] > 10:
                    inactive.append(address)
            
            for address in inactive:
                print(f"Removing inactive client: {address}")
                del self.clients[address]
    
    def stop(self) -> None:
        """Stop server"""
        self.running = False
        self.socket.close()

class UDPGameClient:
    def __init__(self) -> None:
        self.socket: socket.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.server_address: Optional[Tuple[str, int]] = None
        self.player_id: Optional[str] = None
        self.sequence_number: int = 0
        self.last_ping: float = 0
        self.latency: float = 0
        self.running: bool = False
        
    def connect(self, host: str = 'localhost', port: int = 5556) -> bool:
        """Connect to UDP server"""
        self.server_address = (host, port)
        self.running = True
        
        # Start receive thread
        receive_thread = threading.Thread(target=self.receive_loop)
        receive_thread.daemon = True
        receive_thread.start()
        
        # Send initial connection packet
        self.send_message({'type': 'connect'})
        
        # Start ping thread
        ping_thread = threading.Thread(target=self.ping_loop)
        ping_thread.daemon = True
        ping_thread.start()
        
        print(f"Connecting to UDP server at {host}:{port}")
        return True
    
    def receive_loop(self) -> None:
        """Receive packets from server"""
        while self.running:
            try:
                data, address = self.socket.recvfrom(1024)
                message = json.loads(data.decode('utf-8'))
                self.handle_message(message)
                
            except Exception as e:
                print(f"Receive error: {e}")
    
    def handle_message(self, message: Dict[str, Any]) -> None:
        """Handle server message"""
        msg_type = message.get('type')
        
        if msg_type == 'connected':
            self.player_id = message.get('player_id')
            print(f"Connected as {self.player_id}")
            
        elif msg_type == 'state':
            # Game state update
            players = message.get('players', [])
            for player in players:
                if player['id'] != self.player_id:
                    # Update other players
                    print(f"Player {player['id']}: ({player['x']}, {player['y']})")
            
        elif msg_type == 'pong':
            # Calculate latency
            self.latency = (time.time() - message.get('timestamp', 0)) * 1000
            print(f"Latency: {self.latency:.1f}ms")
    
    def send_message(self, message: Dict[str, Any]) -> bool:
        """Send message to server"""
        if self.server_address:
            try:
                message['sequence'] = self.sequence_number
                self.sequence_number += 1
                
                data = json.dumps(message).encode('utf-8')
                self.socket.sendto(data, self.server_address)
                return True
                
            except Exception as e:
                print(f"Send error: {e}")
                return False
        return False
    
    def send_position(self, x: float, y: float, vx: float = 0, vy: float = 0) -> bool:
        """Send position update"""
        return self.send_message({
            'type': 'update',
            'x': x,
            'y': y,
            'vx': vx,
            'vy': vy
        })
    
    def ping_loop(self) -> None:
        """Send periodic pings to measure latency"""
        while self.running:
            time.sleep(1)  # Ping every second
            self.send_message({
                'type': 'ping',
                'timestamp': time.time()
            })
    
    def disconnect(self) -> None:
        """Disconnect from server"""
        self.running = False
        self.socket.close()
        print("Disconnected")
        

🎮 Pygame Network Game

import pygame
import threading
from typing import Dict, Optional, Any
from game_client import GameClient  # Our TCP client from above

class NetworkedGame:
    def __init__(self) -> None:
        pygame.init()
        self.screen: pygame.Surface = pygame.display.set_mode((800, 600))
        pygame.display.set_caption("Networked Game")
        self.clock: pygame.time.Clock = pygame.time.Clock()
        
        # Network client
        self.client: GameClient = GameClient()
        self.connected: bool = False
        
        # Game state
        self.players: Dict[str, Dict[str, Any]] = {}  # {player_id: {'x': x, 'y': y, 'color': color}}
        self.my_player: Optional[str] = None
        
        # Connect to server
        self.connect_to_server()
    
    def connect_to_server(self) -> None:
        """Connect to game server"""
        if self.client.connect('localhost', 5555):
            self.connected = True
            
            # Override client's message handler
            self.client.handle_message = self.handle_network_message
            
            print("Connected to server!")
        else:
            print("Failed to connect to server")
    
    def handle_network_message(self, message: Dict[str, Any]) -> None:
        """Handle messages from server"""
        msg_type = message.get('type')
        
        if msg_type == 'welcome':
            # Create our player
            player_id = message.get('player_id')
            self.my_player = player_id
            self.players[player_id] = {
                'x': 400,
                'y': 300,
                'color': (0, 255, 0)  # Green for our player
            }
            
        elif msg_type == 'player_joined':
            # Add new player
            player_id = message.get('player_id')
            if player_id not in self.players:
                self.players[player_id] = {
                    'x': 100,
                    'y': 100,
                    'color': (255, 0, 0)  # Red for others
                }
            
        elif msg_type == 'player_left':
            # Remove player
            player_id = message.get('player_id')
            if player_id in self.players:
                del self.players[player_id]
            
        elif msg_type == 'player_moved':
            # Update player position
            player_id = message.get('player_id')
            if player_id in self.players:
                self.players[player_id]['x'] = message.get('x')
                self.players[player_id]['y'] = message.get('y')
            
        elif msg_type == 'players_list':
            # Sync all players
            for player_data in message.get('players', []):
                player_id = player_data['id']
                if player_id not in self.players:
                    color = (0, 255, 0) if player_id == self.my_player else (255, 0, 0)
                    self.players[player_id] = {
                        'x': player_data['x'],
                        'y': player_data['y'],
                        'color': color
                    }
    
    def run(self) -> None:
        """Main game loop"""
        running = True
        
        while running:
            dt = self.clock.tick(60) / 1000.0  # 60 FPS
            
            # Handle events
            for event in pygame.event.get():
                if event.type == pygame.QUIT:
                    running = False
                elif event.type == pygame.KEYDOWN:
                    if event.key == pygame.K_SPACE:
                        # Request player list
                        self.client.send_message({'type': 'get_players'})
            
            # Handle input
            if self.connected and self.my_player:
                keys = pygame.key.get_pressed()
                player = self.players.get(self.my_player)
                
                if player:
                    moved = False
                    speed = 200  # pixels per second
                    
                    if keys[pygame.K_LEFT]:
                        player['x'] -= speed * dt
                        moved = True
                    if keys[pygame.K_RIGHT]:
                        player['x'] += speed * dt
                        moved = True
                    if keys[pygame.K_UP]:
                        player['y'] -= speed * dt
                        moved = True
                    if keys[pygame.K_DOWN]:
                        player['y'] += speed * dt
                        moved = True
                    
                    # Keep player on screen
                    player['x'] = max(20, min(780, player['x']))
                    player['y'] = max(20, min(580, player['y']))
                    
                    # Send position to server
                    if moved:
                        self.client.move(int(player['x']), int(player['y']))
            
            # Draw everything
            self.screen.fill((30, 30, 30))
            
            # Draw players
            for player_id, player_data in self.players.items():
                x = int(player_data['x'])
                y = int(player_data['y'])
                color = player_data['color']
                
                # Draw player circle
                pygame.draw.circle(self.screen, color, (x, y), 20)
                
                # Draw player ID
                font = pygame.font.Font(None, 24)
                text = font.render(player_id, True, (255, 255, 255))
                text_rect = text.get_rect(center=(x, y - 30))
                self.screen.blit(text, text_rect)
            
            # Draw connection status
            status_color = (0, 255, 0) if self.connected else (255, 0, 0)
            status_text = "Connected" if self.connected else "Disconnected"
            font = pygame.font.Font(None, 36)
            text = font.render(status_text, True, status_color)
            self.screen.blit(text, (10, 10))
            
            # Draw instructions
            font = pygame.font.Font(None, 24)
            instructions = [
                "Arrow Keys: Move",
                "Space: Sync Players",
                "ESC: Quit"
            ]
            for i, instruction in enumerate(instructions):
                text = font.render(instruction, True, (200, 200, 200))
                self.screen.blit(text, (10, 550 + i * 20))
            
            pygame.display.flip()
        
        # Clean up
        if self.connected:
            self.client.disconnect()
        pygame.quit()

if __name__ == "__main__":
    game = NetworkedGame()
    game.run()
        

🔬 Advanced Techniques

1. Non-blocking Sockets

import socket
import select

# Make socket non-blocking
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(False)

# Use select for multiplexing
readable, writable, errors = select.select([sock], [], [], timeout=0.1)
if readable:
    data = sock.recv(1024)
        

2. Message Framing

def send_message_with_length(sock, message):
    """Send message with length prefix"""
    data = message.encode('utf-8')
    length = len(data)
    
    # Send 4-byte length prefix
    sock.send(struct.pack('!I', length))
    # Send actual data
    sock.send(data)

def receive_message_with_length(sock):
    """Receive message with length prefix"""
    # Receive length prefix
    length_data = sock.recv(4)
    if not length_data:
        return None
    
    length = struct.unpack('!I', length_data)[0]
    
    # Receive exact amount of data
    data = b''
    while len(data) < length:
        chunk = sock.recv(min(4096, length - len(data)))
        if not chunk:
            return None
        data += chunk
    
    return data.decode('utf-8')
        

3. Connection Pooling

class ConnectionPool:
    def __init__(self, max_connections: int = 10) -> None:
        self.pool: list = []
        self.max_connections: int = max_connections
        
    def get_connection(self, host: str, port: int):
        # Reuse existing connection if available
        for conn in self.pool:
            if not conn.in_use:
                conn.in_use = True
                return conn
        
        # Create new connection if under limit
        if len(self.pool) < self.max_connections:
            conn = self.create_connection(host, port)
            self.pool.append(conn)
            return conn
        
        return None  # Pool exhausted
        

🔒 Socket Security

• Input Validation: Always validate incoming data
• Buffer Limits: Set maximum message sizes
• Rate Limiting: Prevent spam/DoS attacks
• Encryption: Use SSL/TLS for sensitive data
• Authentication: Verify client identities
• Timeout Handling: Set socket timeouts

# Example: Rate limiting
class RateLimiter:
    def __init__(self, max_requests: int = 10, time_window: float = 1.0) -> None:
        self.max_requests: int = max_requests
        self.time_window: float = time_window
        self.clients: dict = {}  # {address: [timestamps]}
    
    def is_allowed(self, address) -> bool:
        current_time = time.time()
        
        if address not in self.clients:
            self.clients[address] = []
        
        # Remove old timestamps
        self.clients[address] = [
            t for t in self.clients[address] 
            if current_time - t < self.time_window
        ]
        
        # Check rate limit
        if len(self.clients[address]) < self.max_requests:
            self.clients[address].append(current_time)
            return True
        
        return False
        

🐛 Debugging Tools and Tips

1. Packet Logging

import logging

logging.basicConfig(level=logging.DEBUG,
                   format='%(asctime)s - %(levelname)s - %(message)s')

def log_packet(direction: str, address, data: str) -> None:
    logging.debug(f"{direction} {address}: {data[:100]}")  # First 100 chars
        

2. Network Monitoring Tools

• Wireshark: Packet analysis
• netstat: View active connections
• tcpdump: Command-line packet capture
• telnet: Test TCP connections
• nc (netcat): Swiss army knife for networking

3. Common Issues

• Connection Refused: Server not running or wrong port
• Address Already in Use: Port still bound from previous run
• Firewall Blocking: Check firewall rules
• Data Corruption: Encoding/decoding issues
• Deadlocks: Improper threading

✨ Socket Programming Best Practices

• Protocol Design: Define clear message formats
• Error Handling: Always handle network errors gracefully
• Resource Cleanup: Always close sockets properly
• Threading Safety: Use locks for shared resources
• Buffer Management: Don't assume message boundaries
• Timeouts: Set appropriate timeouts for operations
• Testing: Test with packet loss and high latency
• Documentation: Document your protocol
• Versioning: Plan for protocol changes
• Monitoring: Log important events

• 🔌 Sockets are the foundation of network communication
• 📡 Choose TCP for reliability, UDP for speed
• 🔄 Always handle network errors and disconnections
• 📦 Frame your messages properly for TCP
• ⏱️ Implement timeouts and cleanup inactive connections
• 🔒 Validate all incoming data for security
• 🧵 Use threading for handling multiple clients
• 📊 Monitor and log network activity for debugging