(self)
| 372 | self._sock.sendall(_encode_client_ws_frame(0x1, text.encode("utf-8"))) |
| 373 | |
| 374 | def close(self) -> None: |
| 375 | if self._sock.fileno() == -1: |
| 376 | return |
| 377 | |
| 378 | with contextlib.suppress(OSError): |
| 379 | self._sock.sendall( |
| 380 | _encode_client_ws_frame(0x8, struct.pack("!H", 1000)) |
| 381 | ) |
| 382 | |
| 383 | with contextlib.suppress(ConnectionError, OSError, socket.timeout): |
| 384 | _read_ws_frame(self._reader) |
| 385 | |
| 386 | with contextlib.suppress(OSError): |
| 387 | self._sock.close() |
| 388 | |
| 389 | |
| 390 | def _dechunk(data: bytes) -> bytes: |
no test coverage detected