Module: anynet.websocket

Provides a client and server for the websocket protocol.

class WSError(Exception)
Raised when a websocket error occurs.

class WebSocketClient
A websocket client.

async with connect(url: str, context: TLSContext = None, *, protocols: list[str] = None, disconnect_timeout: float = None) -> WebSocketClient
Creates a websocket client and connects it to the given address. Blocks until the connection is ready and the handshake has been performed.

url must contain at least the hostname or IP address of the server, and the path. Scheme and port are optional. Example: wss://example.com:8080/test.

If no scheme is provided, the connection is secured with TLS precisely if a TLS context is provided. If the scheme is wss but no TLS context is provided the connection is secured with the default TLS context.

async with serve(handler: Callable, host: str = "", port: int = 0, context: TLSContext = None, *, path: str = "/", protocol: str = None, disconnect_timeout: float = None) -> None
Creates a websocket server and binds it to the given address. If host is empty, the local address of the default gateway is used. If port is 0, it is chosen by the operating system. handler must be an async function that accepts a WebSocketClient. When handler returns, the closing handshake is started with the given timeout. If context is provided, the server is secured with TLS.

async with route(handler: Callable, router: HTTPRouter, path: str, *, protocol: str = None, disconnect_timeout: float = None) -> None
Creates a websocket server and binds it to the given path. handler must be an async function that accepts a WebSocketClient. When handler returns, the closing handshake is started with the given timeout.

WebSocketClient

async def send(data: bytes) -> None
Sends a binary packet to the server. Blocks if the send buffer is full.

async def recv() -> bytes
Receives a single binary packet from the server. Blocks if no binary data is available.

async def send_text(data: str) -> None
Sends a text packet to the server. Blocks if the send buffer is full.

async def recv_text() -> str
Receives a single text packet from the server. Blocks if no text data is available.

async def disconnect() -> None
Closes the connection gracefully by sending a Close frame and waiting for the response. Blocks until the closing handshake is complete. Then closes the underlying TCP connection.

async def close() -> None
Closes the underlying TCP connection.

def local_address() -> tuple[str, int]
Returns the local address of the client.

def remote_address() -> tuple[str, int]
Returns the address that the client is connected to.

def remote_ceritifcate() -> TLSCertificate
Returns the certificate that was provided by the other side of the connection. Returns None if the connection is not secured with TLS, or if the other side of the connection did not provide a client certificate.