Class ServerWebSocketRoute

playwright/_impl/_network.py:580–633

class ServerWebSocketRoute:
    def __init__(self, ws: "WebSocketRoute"):
        self._ws = ws

    def on_message(self, handler: Callable[[Union[str, bytes]], Any]) -> None:
        self._ws._on_server_message = handler

    def on_close(self, handler: Callable[[Optional[int], Optional[str]], Any]) -> None:
        self._ws._on_server_close = handler

    def connect_to_server(self) -> None:
        raise NotImplementedError(
            "connectToServer must be called on the page-side WebSocketRoute"
        )

    @property
    def url(self) -> str:
        return self._ws._initializer["url"]

    @property
    def protocols(self) -> List[str]:
        return list(self._ws._initializer.get("protocols", []))

    def close(self, code: int = None, reason: str = None) -> None:
        _create_task_and_ignore_exception(
            self._ws._loop,
            self._ws._channel.send(
                "closeServer",
                None,
                {
                    "code": code,
                    "reason": reason,
                    "wasClean": True,
                },
            ),
        )

    def send(self, message: Union[str, bytes]) -> None:
        if isinstance(message, str):
            _create_task_and_ignore_exception(
                self._ws._loop,
                self._ws._channel.send(
                    "sendToServer", None, {"message": message, "isBase64": False}
                ),
            )
        else:
            _create_task_and_ignore_exception(
                self._ws._loop,
                self._ws._channel.send(
                    "sendToServer",
                    None,
                    {"message": base64.b64encode(message).decode(), "isBase64": True},
                ),
            )


class WebSocketRoute(ChannelOwner):