A specialized buffer that tries to avoid copies when large pieces of data are encountered.
| 115 | |
| 116 | |
| 117 | class _StreamBuffer(object): |
| 118 | """ |
| 119 | A specialized buffer that tries to avoid copies when large pieces |
| 120 | of data are encountered. |
| 121 | """ |
| 122 | |
| 123 | def __init__(self) -> None: |
| 124 | # A sequence of (False, bytearray) and (True, memoryview) objects |
| 125 | self._buffers = ( |
| 126 | collections.deque() |
| 127 | ) # type: Deque[Tuple[bool, Union[bytearray, memoryview]]] |
| 128 | # Position in the first buffer |
| 129 | self._first_pos = 0 |
| 130 | self._size = 0 |
| 131 | |
| 132 | def __len__(self) -> int: |
| 133 | return self._size |
| 134 | |
| 135 | # Data above this size will be appended separately instead |
| 136 | # of extending an existing bytearray |
| 137 | _large_buf_threshold = 2048 |
| 138 | |
| 139 | def append(self, data: Union[bytes, bytearray, memoryview]) -> None: |
| 140 | """ |
| 141 | Append the given piece of data (should be a buffer-compatible object). |
| 142 | """ |
| 143 | size = len(data) |
| 144 | if size > self._large_buf_threshold: |
| 145 | if not isinstance(data, memoryview): |
| 146 | data = memoryview(data) |
| 147 | self._buffers.append((True, data)) |
| 148 | elif size > 0: |
| 149 | if self._buffers: |
| 150 | is_memview, b = self._buffers[-1] |
| 151 | new_buf = is_memview or len(b) >= self._large_buf_threshold |
| 152 | else: |
| 153 | new_buf = True |
| 154 | if new_buf: |
| 155 | self._buffers.append((False, bytearray(data))) |
| 156 | else: |
| 157 | b += data # type: ignore |
| 158 | |
| 159 | self._size += size |
| 160 | |
| 161 | def peek(self, size: int) -> memoryview: |
| 162 | """ |
| 163 | Get a view over at most ``size`` bytes (possibly fewer) at the |
| 164 | current buffer position. |
| 165 | """ |
| 166 | assert size > 0 |
| 167 | try: |
| 168 | is_memview, b = self._buffers[0] |
| 169 | except IndexError: |
| 170 | return memoryview(b"") |
| 171 | |
| 172 | pos = self._first_pos |
| 173 | if is_memview: |
| 174 | return typing.cast(memoryview, b[pos : pos + size]) |
no outgoing calls