Receive exactly n bytes from a socket. Args: sock: Socket to read from n: Number of bytes to read Returns: bytes: Received data Raises: DirtyProtocolError: If read fails or connection closed
(sock: socket.socket, n: int)
| 490 | |
| 491 | @staticmethod |
| 492 | def _recv_exactly(sock: socket.socket, n: int) -> bytes: |
| 493 | """ |
| 494 | Receive exactly n bytes from a socket. |
| 495 | |
| 496 | Args: |
| 497 | sock: Socket to read from |
| 498 | n: Number of bytes to read |
| 499 | |
| 500 | Returns: |
| 501 | bytes: Received data |
| 502 | |
| 503 | Raises: |
| 504 | DirtyProtocolError: If read fails or connection closed |
| 505 | """ |
| 506 | data = b"" |
| 507 | while len(data) < n: |
| 508 | chunk = sock.recv(n - len(data)) |
| 509 | if not chunk: |
| 510 | if len(data) == 0: |
| 511 | raise DirtyProtocolError("Connection closed") |
| 512 | raise DirtyProtocolError( |
| 513 | f"Connection closed after {len(data)} bytes, expected {n}", |
| 514 | raw_data=data |
| 515 | ) |
| 516 | data += chunk |
| 517 | return data |
| 518 | |
| 519 | @staticmethod |
| 520 | def read_message(sock: socket.socket) -> dict: |
no test coverage detected