Trim the buffer by the passed in bytes_to_remove by creating a new buffer that is at a minimum the default max size.
(self, bytes_to_remove)
| 411 | return True |
| 412 | |
| 413 | def trim_buffer_and_resize(self, bytes_to_remove): |
| 414 | """ |
| 415 | Trim the buffer by the passed in bytes_to_remove by creating a new buffer |
| 416 | that is at a minimum the default max size. |
| 417 | """ |
| 418 | current_buffer_size = len(self.buffer) |
| 419 | # Create a new buffer with either minumum size or leftover size. |
| 420 | new_buffer = bytearray(max(current_buffer_size - |
| 421 | bytes_to_remove, self.DEFAULT_BUFFER_SIZE)) |
| 422 | |
| 423 | # if we have content we did not read, copy that portion to the new |
| 424 | # buffer. |
| 425 | if bytes_to_remove <= current_buffer_size: |
| 426 | new_buffer[:self.buffer_end_offset - |
| 427 | bytes_to_remove] = self.buffer[bytes_to_remove:self.buffer_end_offset] |
| 428 | |
| 429 | # Point to the new buffer. |
| 430 | self.buffer = new_buffer |
| 431 | |
| 432 | # reset pointers after the shift. |
| 433 | self.read_offset = 0 |
| 434 | self.buffer_end_offset -= bytes_to_remove |
| 435 | |
| 436 | def close(self): |
| 437 | """ |
no outgoing calls
no test coverage detected