| 470 | |
| 471 | |
| 472 | class PyMongoProtocol(BufferedProtocol): |
| 473 | def __init__(self, timeout: Optional[float] = None): |
| 474 | self.transport: Transport = None # type: ignore[assignment] |
| 475 | # Each message is reader in 2-3 parts: header, compression header, and message body |
| 476 | # The message buffer is allocated after the header is read. |
| 477 | self._header = memoryview(bytearray(16)) |
| 478 | self._header_index = 0 |
| 479 | self._compression_header = memoryview(bytearray(9)) |
| 480 | self._compression_index = 0 |
| 481 | self._message: Optional[memoryview] = None |
| 482 | self._message_index = 0 |
| 483 | # State. TODO: replace booleans with an enum? |
| 484 | self._expecting_header = True |
| 485 | self._expecting_compression = False |
| 486 | self._message_size = 0 |
| 487 | self._op_code = 0 |
| 488 | self._connection_lost = False |
| 489 | self._read_waiter: Optional[Future[Any]] = None |
| 490 | self._timeout = timeout |
| 491 | self._is_compressed = False |
| 492 | self._compressor_id: Optional[int] = None |
| 493 | self._max_message_size = MAX_MESSAGE_SIZE |
| 494 | self._response_to: Optional[int] = None |
| 495 | self._closed = asyncio.get_running_loop().create_future() |
| 496 | self._pending_messages: collections.deque[Future[Any]] = collections.deque() |
| 497 | self._done_messages: collections.deque[Future[Any]] = collections.deque() |
| 498 | |
| 499 | def settimeout(self, timeout: float | None) -> None: |
| 500 | self._timeout = timeout |
| 501 | |
| 502 | @property |
| 503 | def gettimeout(self) -> float | None: |
| 504 | """The configured timeout for the socket that underlies our protocol pair.""" |
| 505 | return self._timeout |
| 506 | |
| 507 | def connection_made(self, transport: BaseTransport) -> None: |
| 508 | """Called exactly once when a connection is made. |
| 509 | The transport argument is the transport representing the write side of the connection. |
| 510 | """ |
| 511 | self.transport = transport # type: ignore[assignment] |
| 512 | self.transport.set_write_buffer_limits(MAX_MESSAGE_SIZE, MAX_MESSAGE_SIZE) |
| 513 | |
| 514 | async def write(self, message: bytes) -> None: |
| 515 | """Write a message to this connection's transport.""" |
| 516 | if self.transport.is_closing(): |
| 517 | raise OSError("Connection is closed") |
| 518 | self.transport.write(message) |
| 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 |
no outgoing calls
no test coverage detected