Read a single MongoDB Wire Protocol message from this connection.
(self, request_id: Optional[int], max_message_size: int)
| 519 | self.transport.resume_reading() |
| 520 | |
| 521 | async def read(self, request_id: Optional[int], max_message_size: int) -> tuple[bytes, int]: |
| 522 | """Read a single MongoDB Wire Protocol message from this connection.""" |
| 523 | if self.transport: |
| 524 | try: |
| 525 | self.transport.resume_reading() |
| 526 | # Known bug in SSL Protocols, fixed in Python 3.11: https://github.com/python/cpython/issues/89322 |
| 527 | except AttributeError: |
| 528 | raise OSError("connection is already closed") from None |
| 529 | self._max_message_size = max_message_size |
| 530 | if self._done_messages: |
| 531 | message = await self._done_messages.popleft() |
| 532 | else: |
| 533 | if self.transport and self.transport.is_closing(): |
| 534 | raise OSError("connection is already closed") |
| 535 | read_waiter = asyncio.get_running_loop().create_future() |
| 536 | self._pending_messages.append(read_waiter) |
| 537 | try: |
| 538 | message = await read_waiter |
| 539 | finally: |
| 540 | if read_waiter in self._done_messages: |
| 541 | self._done_messages.remove(read_waiter) |
| 542 | if message: |
| 543 | op_code, compressor_id, response_to, data = message |
| 544 | # No request_id for exhaust cursor "getMore". |
| 545 | if request_id is not None: |
| 546 | if request_id != response_to: |
| 547 | raise ProtocolError( |
| 548 | f"Got response id {response_to!r} but expected {request_id!r}" |
| 549 | ) |
| 550 | if compressor_id is not None: |
| 551 | data = decompress(data, compressor_id) |
| 552 | return data, op_code |
| 553 | raise OSError("connection closed") |
| 554 | |
| 555 | def get_buffer(self, sizehint: int) -> memoryview: |
| 556 | """Called to allocate a new receive buffer. |
no test coverage detected