Receive protobuf message from socket Args: conn (socket.socket): Socket connection. data (Message): Placehold for protobuf message. pack_format (str, optional): Length of protobuf message pack format. Defaults to "Q", which means unsigned long long. Returns:
(conn: socket.socket, data: Message, pack_format: str = "Q")
| 90 | |
| 91 | |
| 92 | def receive_message(conn: socket.socket, data: Message, pack_format: str = "Q") -> Message: |
| 93 | """Receive protobuf message from socket |
| 94 | |
| 95 | Args: |
| 96 | conn (socket.socket): Socket connection. |
| 97 | data (Message): Placehold for protobuf message. |
| 98 | pack_format (str, optional): Length of protobuf message pack format. Defaults to "Q", which means unsigned long long. |
| 99 | |
| 100 | Returns: |
| 101 | Message: The protobuf message. |
| 102 | """ |
| 103 | data_len = receive_int(conn, pack_format) |
| 104 | data.ParseFromString(receive_all(conn, data_len)) |
| 105 | return data |
| 106 | |
| 107 | |
| 108 | def serialize_tensor(t: tensor.Tensor) -> bytes: |
nothing calls this directly
no test coverage detected