Receive a given length of bytes from socket. Args: conn (socket.socket): Socket connection. size (int): Length of bytes to receive. Raises: RuntimeError: If connection closed before chunk was read, it will raise an error. Returns: bytes: Received bytes.
(conn: socket.socket, size: int)
| 27 | |
| 28 | |
| 29 | def receive_all(conn: socket.socket, size: int) -> bytes: |
| 30 | """Receive a given length of bytes from socket. |
| 31 | |
| 32 | Args: |
| 33 | conn (socket.socket): Socket connection. |
| 34 | size (int): Length of bytes to receive. |
| 35 | |
| 36 | Raises: |
| 37 | RuntimeError: If connection closed before chunk was read, it will raise an error. |
| 38 | |
| 39 | Returns: |
| 40 | bytes: Received bytes. |
| 41 | """ |
| 42 | buffer = b"" |
| 43 | while size > 0: |
| 44 | chunk = conn.recv(size) |
| 45 | if not chunk: |
| 46 | raise RuntimeError("connection closed before chunk was read") |
| 47 | buffer += chunk |
| 48 | size -= len(chunk) |
| 49 | return buffer |
| 50 | |
| 51 | |
| 52 | def send_int(conn: socket.socket, i: int, pack_format: str = "Q") -> None: |
no test coverage detected