High-Level WebSocket Application

class WebSocketApp:
    def __init__(
        self,
        url: str,
        header: Union[list, dict, Callable, None] = None,
        on_open: Optional[Callable[[WebSocket], None]] = None,
        on_reconnect: Optional[Callable[[WebSocket], None]] = None,
        on_message: Optional[Callable[[WebSocket, Any], None]] = None,
        on_error: Optional[Callable[[WebSocket, Any], None]] = None,
        on_close: Optional[Callable[[WebSocket, Any, Any], None]] = None,
        on_ping: Optional[Callable] = None,
        on_pong: Optional[Callable] = None,
        on_cont_message: Optional[Callable] = None,
        keep_running: bool = True,
        get_mask_key: Optional[Callable] = None,
        cookie: Optional[str] = None,
        subprotocols: Optional[list] = None,
        on_data: Optional[Callable] = None,
        socket: Optional[socket.socket] = None,
    ) -> None:
        """
        WebSocketApp initialization.

        Parameters:
        - url: WebSocket URL (ws:// or wss://)
        - header: Custom headers as list, dict, or callable returning headers
        - on_open: Callback called when connection opens (ws)
        - on_reconnect: Callback called when reconnecting (ws)
        - on_message: Callback called when message received (ws, message)
        - on_error: Callback called on error (ws, error)
        - on_close: Callback called when connection closes (ws, status_code, reason)
        - on_ping: Callback called when ping received
        - on_pong: Callback called when pong received
        - on_cont_message: Callback called for continuation frames (ws, data, is_final)
        - keep_running: Obsolete parameter, ignored
        - get_mask_key: Function to generate mask keys
        - cookie: Cookie string for authentication
        - subprotocols: List of supported subprotocols
        - on_data: Callback called before on_message (ws, data, opcode, is_final)
        - socket: Pre-initialized socket object
        """

def run_forever(
    self,
    sockopt: tuple = None,
    sslopt: dict = None,
    ping_interval: Union[float, int] = 0,
    ping_timeout: Union[float, int, None] = None,
    ping_payload: str = "",
    http_proxy_host: str = None,
    http_proxy_port: Union[int, str] = None,
    http_no_proxy: list = None,
    http_proxy_auth: tuple = None,
    http_proxy_timeout: Optional[float] = None,
    skip_utf8_validation: bool = False,
    host: str = None,
    origin: str = None,
    dispatcher=None,
    suppress_origin: bool = False,
    proxy_type: str = None,
    reconnect: int = None,
) -> bool:
    """
    Run WebSocket event loop until connection closes.

    Parameters:
    - sockopt: Socket options tuple for socket.setsockopt
    - sslopt: SSL options dictionary
    - ping_interval: Auto-ping interval in seconds (0 = disabled)
    - ping_timeout: Pong timeout in seconds
    - ping_payload: Payload for ping messages
    - http_proxy_host: HTTP proxy hostname
    - http_proxy_port: HTTP proxy port
    - http_no_proxy: Hosts that bypass proxy
    - http_proxy_auth: Proxy authentication (username, password)
    - http_proxy_timeout: Proxy connection timeout
    - skip_utf8_validation: Skip UTF-8 validation for performance
    - host: Custom Host header
    - origin: Custom Origin header
    - dispatcher: External event dispatcher (e.g., rel)
    - suppress_origin: Don't send Origin header
    - proxy_type: Proxy type (http, socks4, socks4a, socks5, socks5h)
    - reconnect: Reconnection delay in seconds

    Returns:
    bool: True if error occurred, False if clean shutdown
    """

def send(self, data: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> None:
    """
    Send message with specified opcode.

    Parameters:
    - data: Message data (string for text, bytes for binary)
    - opcode: WebSocket opcode (OPCODE_TEXT, OPCODE_BINARY, etc.)

    Raises:
    WebSocketConnectionClosedException: If connection is closed
    """

def send_text(self, text_data: str) -> None:
    """
    Send UTF-8 text message.

    Parameters:
    - text_data: Text string to send

    Raises:
    WebSocketConnectionClosedException: If connection is closed
    """

def send_bytes(self, data: Union[bytes, bytearray]) -> None:
    """
    Send binary message.

    Parameters:
    - data: Binary data to send

    Raises:
    WebSocketConnectionClosedException: If connection is closed
    """

def close(self, **kwargs) -> None:
    """
    Close WebSocket connection.

    Parameters:
    - **kwargs: Additional arguments passed to underlying close method
    """

def setReconnect(reconnectInterval: int) -> None:
    """
    Set global reconnection interval.

    Parameters:
    - reconnectInterval: Delay in seconds between reconnection attempts
    """

import websocket

def on_message(ws, message):
    print(f"Received: {message}")
    ws.send("Echo: " + message)

def on_error(ws, error):
    print(f"Error: {error}")

def on_close(ws, close_status_code, close_msg):
    print("Connection closed")

def on_open(ws):
    print("Connection opened")
    ws.send("Hello Server!")

ws = websocket.WebSocketApp("ws://echo.websocket.events",
                          on_open=on_open,
                          on_message=on_message,
                          on_error=on_error,
                          on_close=on_close)
ws.run_forever()

import websocket

def on_open(ws):
    print("Authenticated connection established")

headers = {
    "Authorization": "Bearer your-token-here",
    "User-Agent": "MyApp/1.0"
}

ws = websocket.WebSocketApp("wss://api.example.com/websocket",
                          header=headers,
                          on_open=on_open)
ws.run_forever()

import websocket
import rel

def on_message(ws, message):
    print(f"Market data: {message}")

def on_error(ws, error):
    print(f"Connection error: {error}")

def on_close(ws, close_status_code, close_msg):
    print("Connection closed, will reconnect...")

ws = websocket.WebSocketApp("wss://api.gemini.com/v1/marketdata/BTCUSD",
                          on_message=on_message,
                          on_error=on_error,
                          on_close=on_close)

# Run with automatic reconnection every 5 seconds
ws.run_forever(dispatcher=rel, reconnect=5)
rel.signal(2, rel.abort)  # Handle Ctrl+C
rel.dispatch()

import websocket

def on_open(ws):
    print("Connected through proxy")

ws = websocket.WebSocketApp("ws://echo.websocket.events")
ws.run_forever(
    http_proxy_host="proxy.company.com",
    http_proxy_port=8080,
    http_proxy_auth=("username", "password")
)

import websocket
import ssl

def on_open(ws):
    print("Secure connection established")

sslopt = {
    "cert_reqs": ssl.CERT_REQUIRED,
    "ca_certs": "/path/to/ca-certificates.crt",
    "certfile": "/path/to/client.crt",
    "keyfile": "/path/to/client.key"
}

ws = websocket.WebSocketApp("wss://secure.example.com/websocket",
                          on_open=on_open)
ws.run_forever(sslopt=sslopt)

# Callback function signatures
OnOpenCallback = Callable[[WebSocket], None]
OnReconnectCallback = Callable[[WebSocket], None]  
OnMessageCallback = Callable[[WebSocket, Any], None]
OnErrorCallback = Callable[[WebSocket, Any], None]
OnCloseCallback = Callable[[WebSocket, Any, Any], None]
OnDataCallback = Callable[[WebSocket, Any, int, bool], None]
OnPingCallback = Callable[[WebSocket, str], None]
OnPongCallback = Callable[[WebSocket, str], None]
OnContMessageCallback = Callable[[WebSocket, Any, bool], None]