WebSocket Client

The WebSocket client module provides functionality

WebSocket Client Module

This module implements a WebSocket client for aiosonic, providing a WebSocketConnection class for managing WebSocket connections and a WebSocketClient class for performing the upgrade handshake and connecting to a WebSocket server.

Classes:

WebSocketConnection: Manages a WebSocket connection and provides methods

for sending and receiving text, binary, and JSON messages, as well as handling ping/pong and closing the connection.

WebSocketClient: Performs the HTTP upgrade request and establishes a

WebSocket connection with optional custom headers and subprotocol negotiation.

Exceptions:

ConnectionDisconnected: Raised when the connection is unexpectedly closed. ReadTimeout: Raised when a read operation times out.

class aiosonic.web_socket_client.ProtocolHandler

Bases: ABC

Base class for WebSocket subprotocol handlers.

To implement a custom protocol (for example, using MessagePack), subclass this base class and implement the encode and decode methods. For example:

import msgpack

class MsgpackHandler(ProtocolHandler):
    @property
    def name(self) -> str:
        return "msgpack"

    def encode(self, data) -> bytes:
        return msgpack.packb(data, use_bin_type=True)

    def decode(self, data: bytes):
        return msgpack.unpackb(data, raw=False)

abstract decode(data: bytes)

Decode bytes into data.

abstract encode(data) → bytes

Encode data into bytes.

abstract property name: str

Return the protocol name used for negotiation.

class aiosonic.web_socket_client.WebSocketClient(connector: TCPConnector | None = None)

Bases: object

WebSocket Client.

This class handles the HTTP upgrade process and establishes a WebSocket connection. It supports custom headers, subprotocol negotiation, and optional keep-alive functionality.

async connect(url: str, verify: bool = True, ssl: SSLContext | None = None, headers: Dict[str, str] | None = None, subprotocols: List[str] | None = None, start_keepalive: bool = True, keepalive_interval: float = 30.0, conn_opts: dict | None = None) → WebSocketConnection

Connect to a WebSocket server, optionally starting the keep-alive task.

Parameters:

• url – The WebSocket URL.

• verify – Whether to verify the SSL certificate.

• ssl – Custom SSL context if needed.

• headers – Optional custom headers.

• subprotocols – Optional list of subprotocols.

• start_keepalive – Whether to start the keep-alive ping task.

• keepalive_interval – Interval in seconds between pings.

• conn_opts – Options to override default options for WebSocketConnection.

Returns:

An instance of WebSocketConnection.

Raises:

ConnectionError – If the upgrade handshake fails.

class aiosonic.web_socket_client.WebSocketConnection(conn: Connection, text_queue_maxsize: int = 100, binary_queue_maxsize: int = 100, drop_frames: bool = False, protocol_handler: ProtocolHandler | None = None)

Bases: object

OPCODE_BINARY = 2

OPCODE_CLOSE = 8

OPCODE_PING = 9

OPCODE_PONG = 10

OPCODE_TEXT = 1

async close(code: int = 1000, reason: str = '')

async ping(data: bytes = b'')

async receive_bytes(timeout: float | None = None) → bytes

async receive_json(timeout: float | None = None)

async receive_pong(timeout: float | None = None) → bytes

async receive_protocol(timeout: float | None = None)

Receive data using the configured protocol handler.

async receive_text(timeout: float | None = None) → str

async send_bytes(data: bytes)

async send_json(data)

async send_protocol(data)

Send data using the configured protocol handler.

async send_text(message: str)

start_keep_alive(interval: float = 30.0)

stop_keep_alive()