Advance the current buffer position by ``size`` bytes.
(self, size: int)
| 176 | return memoryview(b)[pos : pos + size] |
| 177 | |
| 178 | def advance(self, size: int) -> None: |
| 179 | """ |
| 180 | Advance the current buffer position by ``size`` bytes. |
| 181 | """ |
| 182 | assert 0 < size <= self._size |
| 183 | self._size -= size |
| 184 | pos = self._first_pos |
| 185 | |
| 186 | buffers = self._buffers |
| 187 | while buffers and size > 0: |
| 188 | is_large, b = buffers[0] |
| 189 | b_remain = len(b) - size - pos |
| 190 | if b_remain <= 0: |
| 191 | buffers.popleft() |
| 192 | size -= len(b) - pos |
| 193 | pos = 0 |
| 194 | elif is_large: |
| 195 | pos += size |
| 196 | size = 0 |
| 197 | else: |
| 198 | # Amortized O(1) shrink for Python 2 |
| 199 | pos += size |
| 200 | if len(b) <= 2 * pos: |
| 201 | del typing.cast(bytearray, b)[:pos] |
| 202 | pos = 0 |
| 203 | size = 0 |
| 204 | |
| 205 | assert size == 0 |
| 206 | self._first_pos = pos |
| 207 | |
| 208 | |
| 209 | class BaseIOStream(object): |
no outgoing calls