Exception Handling

class SocketIOError(Exception):
    """
    Base exception class for all Socket.IO errors.
    
    Inherits from: Exception
    
    This is the base class for all Socket.IO-specific exceptions.
    Catch this exception to handle any Socket.IO-related error.
    """

import socketio

sio = socketio.Client()

try:
    sio.connect('http://localhost:5000')
    sio.emit('message', {'text': 'Hello'})
except socketio.SocketIOError as e:
    print(f'Socket.IO error occurred: {e}')
    # Handle any Socket.IO-related error

class ConnectionError(SocketIOError):
    """
    Base class for connection-related errors.
    
    Inherits from: SocketIOError
    
    This exception is raised when there are problems with Socket.IO connections,
    such as network issues, server unavailability, or connection failures.
    """

import socketio

sio = socketio.Client()

try:
    sio.connect('http://unreachable-server:5000')
except socketio.ConnectionError as e:
    print(f'Connection error: {e}')
    # Handle connection-related issues
    # Try alternative server, show offline mode, etc.

class ConnectionRefusedError(ConnectionError):
    """
    Connection refused exception.
    
    Inherits from: ConnectionError
    
    Attributes:
        error_args (dict): Dictionary containing error details with 'message' key
                          and optional 'data' key for additional information
    """
    
    def __init__(self, *args):
        """
        Initialize the connection refused error.
        
        Args:
            *args: Error arguments, typically a message string or error dict
        """

import socketio

sio = socketio.Client()

@sio.event
def connect_error(data):
    print(f'Connection failed: {data}')

try:
    sio.connect('http://localhost:5000', auth={'token': 'invalid_token'})
except socketio.ConnectionRefusedError as e:
    print(f'Connection refused: {e}')
    if hasattr(e, 'error_args'):
        error_info = e.error_args
        print(f'Error message: {error_info.get("message")}')
        if 'data' in error_info:
            print(f'Additional data: {error_info["data"]}')
    
    # Handle authentication failure
    # Prompt for new credentials, redirect to login, etc.

# Server-side example that raises ConnectionRefusedError
sio_server = socketio.Server()

@sio_server.event
def connect(sid, environ, auth):
    if not auth or auth.get('token') != 'valid_token':
        raise socketio.ConnectionRefusedError('Authentication failed')
    print(f'Client {sid} authenticated successfully')

class TimeoutError(SocketIOError):
    """
    Timeout-related exception.
    
    Inherits from: SocketIOError
    
    This exception is raised when Socket.IO operations take longer than
    the specified timeout period. Common scenarios include call() operations
    waiting for responses and connection attempts timing out.
    """

import socketio

sio = socketio.Client()
sio.connect('http://localhost:5000')

try:
    # This will raise TimeoutError if server doesn't respond within 5 seconds
    response = sio.call('slow_operation', {'data': 'test'}, timeout=5)
    print(f'Response: {response}')
except socketio.TimeoutError as e:
    print(f'Operation timed out: {e}')
    # Handle timeout scenario
    # Show loading indicator, retry, or provide fallback

# Async client example
async def async_example():
    sio = socketio.AsyncClient()
    await sio.connect('http://localhost:5000')
    
    try:
        response = await sio.call('async_operation', timeout=3)
        print(f'Async response: {response}')
    except socketio.TimeoutError as e:
        print(f'Async operation timed out: {e}')
    finally:
        await sio.disconnect()

# Server-side handler that might cause timeout
@sio_server.event
def slow_operation(sid, data):
    import time
    time.sleep(10)  # This will cause TimeoutError on client
    return {'result': 'completed'}

class BadNamespaceError(SocketIOError):
    """
    Invalid namespace exception.
    
    Inherits from: SocketIOError
    
    This exception is raised when:
    - Attempting to connect to a namespace that doesn't exist
    - Using a namespace that the client is not authorized to access
    - Malformed namespace names
    """

import socketio

sio = socketio.Client()

try:
    sio.connect('http://localhost:5000', namespaces=['/chat', '/invalid-namespace'])
except socketio.BadNamespaceError as e:
    print(f'Invalid namespace: {e}')
    # Handle namespace error
    # Connect to default namespace or show error message

# Using specific namespace that might not exist
try:
    sio.emit('message', {'text': 'Hello'}, namespace='/nonexistent')
except socketio.BadNamespaceError as e:
    print(f'Namespace error: {e}')
    # Fall back to default namespace
    sio.emit('message', {'text': 'Hello'})

# Server-side namespace configuration
sio_server = socketio.Server()

# Only allow specific namespaces
@sio_server.event
def connect(sid, environ, auth):
    # Validate namespace access based on authentication
    namespace = environ.get('PATH_INFO', '/').split('/')[1]
    if namespace not in ['chat', 'game'] and auth.get('role') != 'admin':
        raise socketio.BadNamespaceError(f'Access denied to namespace: /{namespace}')

class DisconnectedError(SocketIOError):
    """
    Client disconnection exception.
    
    Inherits from: SocketIOError
    
    This exception is raised when:
    - Attempting to emit events on a disconnected client
    - Trying to access session data for disconnected clients
    - Performing operations that require an active connection
    """

import socketio

sio = socketio.Client()

@sio.event
def disconnect():
    print('Disconnected from server')

try:
    sio.connect('http://localhost:5000')
    
    # Simulate network interruption or server shutdown
    # ... connection is lost ...
    
    # This will raise DisconnectedError
    sio.emit('message', {'text': 'Hello'})
    
except socketio.DisconnectedError as e:
    print(f'Client disconnected: {e}')
    # Handle disconnection
    # Attempt reconnection, switch to offline mode, etc.
    
    try:
        print('Attempting to reconnect...')
        sio.connect('http://localhost:5000')
        sio.emit('message', {'text': 'Reconnected'})
    except socketio.ConnectionError as reconnect_error:
        print(f'Reconnection failed: {reconnect_error}')

# Server-side example
sio_server = socketio.Server()

@sio_server.event
def handle_message(sid, data):
    try:
        # Try to access session data
        session = sio_server.get_session(sid)
        # Process message with session data
        sio_server.emit('response', {'processed': data}, room=sid)
    except socketio.DisconnectedError as e:
        print(f'Client {sid} disconnected during processing: {e}')
        # Clean up any resources associated with this client

# Async server example
@async_sio_server.event
async def async_handle_message(sid, data):
    try:
        async with async_sio_server.session(sid) as session:
            session['last_message'] = data
        await async_sio_server.emit('response', {'received': data}, room=sid)
    except socketio.DisconnectedError as e:
        print(f'Client {sid} disconnected: {e}')

import socketio
import logging

logger = logging.getLogger(__name__)

sio = socketio.Client(logger=True)

def safe_emit(event, data, **kwargs):
    """Safely emit events with comprehensive error handling."""
    try:
        return sio.emit(event, data, **kwargs)
    except socketio.DisconnectedError:
        logger.error('Cannot emit: client disconnected')
        # Attempt reconnection
        try_reconnect()
    except socketio.BadNamespaceError as e:
        logger.error(f'Invalid namespace: {e}')
        # Fall back to default namespace
        return sio.emit(event, data)
    except socketio.SocketIOError as e:
        logger.error(f'Socket.IO error: {e}')
        # Handle generic Socket.IO errors
    except Exception as e:
        logger.error(f'Unexpected error: {e}')

def safe_call(event, data, timeout=30):
    """Safely call with response handling."""
    try:
        return sio.call(event, data, timeout=timeout)
    except socketio.TimeoutError:
        logger.warning(f'Call to {event} timed out after {timeout}s')
        return None
    except socketio.DisconnectedError:
        logger.error('Cannot call: client disconnected')
        if try_reconnect():
            # Retry after reconnection
            return sio.call(event, data, timeout=timeout)
        return None
    except socketio.SocketIOError as e:
        logger.error(f'Call failed: {e}')
        return None

def try_reconnect(max_attempts=3):
    """Attempt to reconnect with exponential backoff."""
    import time
    
    for attempt in range(max_attempts):
        try:
            delay = 2 ** attempt  # Exponential backoff
            logger.info(f'Reconnection attempt {attempt + 1} in {delay}s')
            time.sleep(delay)
            
            sio.connect('http://localhost:5000')
            logger.info('Reconnected successfully')
            return True
            
        except socketio.ConnectionRefusedError as e:
            logger.error(f'Connection refused: {e}')
            # Don't retry on authentication failures
            break
        except socketio.ConnectionError as e:
            logger.warning(f'Reconnection attempt {attempt + 1} failed: {e}')
            
    logger.error('All reconnection attempts failed')
    return False

import socketio

sio = socketio.Server()

@sio.event
def connect(sid, environ, auth):
    try:
        # Validate authentication
        if not validate_auth(auth):
            raise socketio.ConnectionRefusedError('Invalid credentials')
        
        # Initialize client session
        sio.save_session(sid, {
            'user_id': auth['user_id'],
            'connected_at': time.time()
        })
        
        sio.enter_room(sid, 'authenticated_users')
        sio.emit('welcome', {'message': 'Connected successfully'}, room=sid)
        
    except socketio.ConnectionRefusedError:
        # Re-raise authentication errors
        raise
    except Exception as e:
        logger.error(f'Connection error for {sid}: {e}')
        raise socketio.ConnectionRefusedError('Internal server error')

@sio.event
def handle_message(sid, data):
    try:
        # Validate client connection
        session = sio.get_session(sid)
        
        # Process message
        result = process_message(data, session)
        sio.emit('message_result', result, room=sid)
        
    except socketio.DisconnectedError:
        logger.warning(f'Client {sid} disconnected during message processing')
    except Exception as e:
        logger.error(f'Message processing error for {sid}: {e}')
        sio.emit('error', {'message': 'Message processing failed'}, room=sid)

def validate_auth(auth):
    """Validate authentication credentials."""
    if not auth or 'token' not in auth:
        return False
    # Implement your authentication logic
    return auth['token'] == 'valid_token'

def process_message(data, session):
    """Process incoming message with error handling."""
    if not data or 'content' not in data:
        raise ValueError('Invalid message format')
    
    # Process the message
    return {
        'status': 'processed',
        'user_id': session['user_id'],
        'processed_at': time.time()
    }

import socketio
from contextlib import contextmanager

@contextmanager
def socket_connection(url, **kwargs):
    """Context manager for safe Socket.IO connections."""
    sio = socketio.Client()
    try:
        sio.connect(url, **kwargs)
        yield sio
    except socketio.ConnectionError as e:
        print(f'Failed to connect: {e}')
        raise
    finally:
        if sio.connected:
            sio.disconnect()

# Usage
try:
    with socket_connection('http://localhost:5000') as sio:
        sio.emit('message', {'text': 'Hello'})
        response = sio.call('get_data', timeout=5)
        print(f'Response: {response}')
except socketio.TimeoutError:
    print('Operation timed out')
except socketio.SocketIOError as e:
    print(f'Socket.IO error: {e}')