(
self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
)
| 949 | ) |
| 950 | |
| 951 | def __init__( |
| 952 | self, parent: ChannelOwner, type: str, guid: str, initializer: Dict |
| 953 | ) -> None: |
| 954 | super().__init__(parent, type, guid, initializer) |
| 955 | self._is_closed = False |
| 956 | self._page = cast("Page", parent) |
| 957 | self._channel.on( |
| 958 | "frameSent", |
| 959 | lambda params: self._on_frame_sent(params["opcode"], params["data"]), |
| 960 | ) |
| 961 | self._channel.on( |
| 962 | "frameReceived", |
| 963 | lambda params: self._on_frame_received(params["opcode"], params["data"]), |
| 964 | ) |
| 965 | self._channel.on( |
| 966 | "socketError", |
| 967 | lambda params: self.emit(WebSocket.Events.Error, params["error"]), |
| 968 | ) |
| 969 | self._channel.on("close", lambda params: self._on_close()) |
| 970 | |
| 971 | def __repr__(self) -> str: |
| 972 | return f"<WebSocket url={self.url!r}>" |
nothing calls this directly
no test coverage detected