Receive a raw BSON message or raise socket.error.
(
conn: Connection, request_id: Optional[int], max_message_size: int = MAX_MESSAGE_SIZE
)
| 744 | |
| 745 | |
| 746 | def receive_message( |
| 747 | conn: Connection, request_id: Optional[int], max_message_size: int = MAX_MESSAGE_SIZE |
| 748 | ) -> Union[_OpReply, _OpMsg]: |
| 749 | """Receive a raw BSON message or raise socket.error.""" |
| 750 | if _csot.get_timeout(): |
| 751 | deadline = _csot.get_deadline() |
| 752 | else: |
| 753 | timeout = conn.conn.gettimeout() |
| 754 | if timeout: |
| 755 | deadline = time.monotonic() + timeout |
| 756 | else: |
| 757 | deadline = None |
| 758 | # Ignore the response's request id. |
| 759 | length, _, response_to, op_code = _UNPACK_HEADER(receive_data(conn, 16, deadline)) |
| 760 | # No request_id for exhaust cursor "getMore". |
| 761 | if request_id is not None: |
| 762 | if request_id != response_to: |
| 763 | raise ProtocolError(f"Got response id {response_to!r} but expected {request_id!r}") |
| 764 | if length <= 16: |
| 765 | raise ProtocolError( |
| 766 | f"Message length ({length!r}) not longer than standard message header size (16)" |
| 767 | ) |
| 768 | if length > max_message_size: |
| 769 | raise ProtocolError( |
| 770 | f"Message length ({length!r}) is larger than server max " |
| 771 | f"message size ({max_message_size!r})" |
| 772 | ) |
| 773 | data: memoryview | bytes |
| 774 | if op_code == 2012: |
| 775 | op_code, _, compressor_id = _UNPACK_COMPRESSION_HEADER(receive_data(conn, 9, deadline)) |
| 776 | data = decompress(receive_data(conn, length - 25, deadline), compressor_id) |
| 777 | else: |
| 778 | data = receive_data(conn, length - 16, deadline) |
| 779 | |
| 780 | try: |
| 781 | unpack_reply = _UNPACK_REPLY[op_code] |
| 782 | except KeyError: |
| 783 | raise ProtocolError( |
| 784 | f"Got opcode {op_code!r} but expected {_UNPACK_REPLY.keys()!r}" |
| 785 | ) from None |
| 786 | return unpack_reply(data) |
no test coverage detected