(self, size: int)
| 257 | return data |
| 258 | |
| 259 | def read_exact(self, size: int) -> bytes: |
| 260 | while len(self._buffer) < size: |
| 261 | chunk = self._sock.recv(4096) |
| 262 | if not chunk: |
| 263 | raise ConnectionError("socket closed before expected data") |
| 264 | self._buffer.extend(chunk) |
| 265 | |
| 266 | data = bytes(self._buffer[:size]) |
| 267 | del self._buffer[:size] |
| 268 | return data |
| 269 | |
| 270 | |
| 271 | def _encode_client_ws_frame(opcode: int, payload: bytes = b"") -> bytes: |
no test coverage detected