A FIFO queue. If the queue has a max size, the oldest item on the queue is dropped if that size id exceeded.
| 123 | |
| 124 | |
| 125 | class Queue(object): |
| 126 | """A FIFO queue. If the queue has a max size, the oldest item on the |
| 127 | queue is dropped if that size id exceeded. |
| 128 | """ |
| 129 | |
| 130 | def __init__(self, size=0, dispose_oldest=True): |
| 131 | self._queue = [] |
| 132 | self._size = size |
| 133 | self._dispose_oldest = dispose_oldest |
| 134 | |
| 135 | def put(self, item): |
| 136 | """Put item on the queue. If the queue size is limited ... |
| 137 | """ |
| 138 | if self._size > 0 and len(self._queue) >= self._size: |
| 139 | if self._dispose_oldest: |
| 140 | self.get() |
| 141 | else: |
| 142 | raise QueueFull |
| 143 | |
| 144 | self._queue.insert(0, item) |
| 145 | |
| 146 | def get(self): |
| 147 | """Get the oldest item off the queue. |
| 148 | QueueEmpty is raised if no items are left on the queue. |
| 149 | """ |
| 150 | try: |
| 151 | return self._queue.pop() |
| 152 | except IndexError: |
| 153 | raise QueueEmpty |
| 154 | |
| 155 | |
| 156 | if __name__ == '__main__': |
no outgoing calls
no test coverage detected