Called when the buffer was updated with the received data
(self, nbytes: int)
| 574 | return self._message[self._message_index :] # type: ignore[index] |
| 575 | |
| 576 | def buffer_updated(self, nbytes: int) -> None: |
| 577 | """Called when the buffer was updated with the received data""" |
| 578 | # Wrote 0 bytes into a non-empty buffer, signal connection closed |
| 579 | if nbytes == 0: |
| 580 | self.close(OSError("connection closed")) |
| 581 | return |
| 582 | if self._connection_lost: |
| 583 | return |
| 584 | if self._expecting_header: |
| 585 | self._header_index += nbytes |
| 586 | if self._header_index >= 16: |
| 587 | self._expecting_header = False |
| 588 | try: |
| 589 | ( |
| 590 | self._message_size, |
| 591 | self._op_code, |
| 592 | self._response_to, |
| 593 | self._expecting_compression, |
| 594 | ) = self.process_header() |
| 595 | except ProtocolError as exc: |
| 596 | self.close(exc) |
| 597 | return |
| 598 | self._message = memoryview(bytearray(self._message_size)) |
| 599 | return |
| 600 | if self._expecting_compression: |
| 601 | self._compression_index += nbytes |
| 602 | if self._compression_index >= 9: |
| 603 | self._expecting_compression = False |
| 604 | self._op_code, self._compressor_id = self.process_compression_header() |
| 605 | return |
| 606 | |
| 607 | self._message_index += nbytes |
| 608 | if self._message_index >= self._message_size: |
| 609 | self._expecting_header = True |
| 610 | # Pause reading to avoid storing an arbitrary number of messages in memory. |
| 611 | self.transport.pause_reading() |
| 612 | if self._pending_messages: |
| 613 | result = self._pending_messages.popleft() |
| 614 | else: |
| 615 | result = asyncio.get_running_loop().create_future() |
| 616 | # Future has been cancelled, close this connection |
| 617 | if result.done(): |
| 618 | self.close(None) |
| 619 | return |
| 620 | # Necessary values to reconstruct and verify message |
| 621 | result.set_result( |
| 622 | (self._op_code, self._compressor_id, self._response_to, self._message) |
| 623 | ) |
| 624 | self._done_messages.append(result) |
| 625 | # Reset internal state to expect a new message |
| 626 | self._header_index = 0 |
| 627 | self._compression_index = 0 |
| 628 | self._message_index = 0 |
| 629 | self._message_size = 0 |
| 630 | self._message = None |
| 631 | self._op_code = 0 |
| 632 | self._compressor_id = None |
| 633 | self._response_to = None |
nothing calls this directly
no test coverage detected