An efficient FIFO queue that splits received elements into chunks.
| 53 | |
| 54 | |
| 55 | class EfficientCollectionQueue: |
| 56 | """ |
| 57 | An efficient FIFO queue that splits received elements into chunks. |
| 58 | """ |
| 59 | |
| 60 | class SizeUnderflow(Error): |
| 61 | """Could not pop the first {} elements; collection only has {} elements.""" |
| 62 | |
| 63 | def __init__(self, split_size, member_type): |
| 64 | """ |
| 65 | Initialize an empty queue. |
| 66 | split_size defines the maximum chunk size. |
| 67 | member_type is the type that defines what the base collection looks like. |
| 68 | """ |
| 69 | self.buffers = [] |
| 70 | self.size = 0 |
| 71 | self.split_size = split_size |
| 72 | self.member_type = member_type |
| 73 | |
| 74 | def peek_front(self): |
| 75 | """ |
| 76 | Return the first chunk from the queue without removing it. |
| 77 | The returned collection will have between 1 and split_size elements. |
| 78 | Returns an empty collection when nothing is queued. |
| 79 | """ |
| 80 | if not self.buffers: |
| 81 | return self.member_type() |
| 82 | buffer = self.buffers[0] |
| 83 | return buffer |
| 84 | |
| 85 | def pop_front(self, size): |
| 86 | """ |
| 87 | Remove the first `size` elements from the queue. |
| 88 | Raises an error if the requested removal size is larger than the entire queue. |
| 89 | """ |
| 90 | if size > self.size: |
| 91 | raise EfficientCollectionQueue.SizeUnderflow(size, self.size) |
| 92 | while size > 0: |
| 93 | buffer = self.buffers[0] |
| 94 | to_remove = min(size, len(buffer)) |
| 95 | buffer = buffer[to_remove:] |
| 96 | if buffer: |
| 97 | self.buffers[0] = buffer |
| 98 | else: |
| 99 | del self.buffers[0] |
| 100 | size -= to_remove |
| 101 | self.size -= to_remove |
| 102 | |
| 103 | def push_back(self, data): |
| 104 | """ |
| 105 | Add data at the end of the queue. |
| 106 | Chunks data into elements of size up to split_size. |
| 107 | """ |
| 108 | if not self.buffers: |
| 109 | self.buffers = [self.member_type()] |
| 110 | while data: |
| 111 | buffer = self.buffers[-1] |
| 112 | if len(buffer) >= self.split_size: |
no outgoing calls