pybotters.WebSocketApp

class pybotters.WebSocketApp(session: ClientSession, url: str, *, send_str: str | list[str] | None = None, send_bytes: bytes | list[bytes] | None = None, send_json: dict | list[dict] | None = None, hdlr_str: Callable[[str, ClientWebSocketResponse], None] | list[Callable[[str, ClientWebSocketResponse], None]] | None = None, hdlr_bytes: Callable[[bytes, ClientWebSocketResponse], None] | list[Callable[[bytes, ClientWebSocketResponse], None]] | None = None, hdlr_json: Callable[[Any, ClientWebSocketResponse], None] | list[Callable[[Any, ClientWebSocketResponse], None]] | None = None, backoff: tuple[float, float, float, float] = (1.92, 60.0, 1.618, 5.0), **kwargs: Any)

WebSocket Application.

自動再接続、自動認証、自動 PING/PONG を備えた WebSocket アプリケーションです。

Usage example: WebSocketQueue

Methods

__init__(session, url, *[, send_str, ...])	WebSocket Application.
heartbeat([timeout])	Ensure WebSocket connection is open with Ping-Pong.
wait()	Wait WebSocketApp.

Attributes

current_ws	Current WebSocket connection.
url	WebSocket URL.

property current_ws: ClientWebSocketResponse | None

Current WebSocket connection.

現在の WebSocket コネクションです。 現在のコネクションが存在する場合は ClientWebSocketResponse を返します。 コネクションが存在しない場合は None を返します。

async heartbeat(timeout: float = 10.0) → None

Ensure WebSocket connection is open with Ping-Pong.

WebSocket の Ping-Pong で接続の疎通を確認します。 ユーザーコード内で WebSocket のデータを利用する前にこのコルーチンを待機することで、 WebSocket の接続性を保証するのに役に立ちます。

一定時間 Pong が返ってこない場合は再接続を試みます。 このメソッドは疎通が確認されるまで待機します。

パラメータ:

timeout -- WebSocket Ping-Pong ハートビート (デフォルト 10.0 秒)

property url: str

WebSocket URL.

WebSocket の接続 URL です。 接続中に値を変更した場合は、再接続時に設定 URL に接続されます。

async wait() → None

Wait WebSocketApp.

WebSocketApp の待機を待ちます。 ただし WebSocketApp は自動再接続の機構を備えるので、実質的に無限待機です。 プログラムの終了を防ぐのに役に立ちます。