batchqueue 0.1.0

A Queue type for handling traffic that arrives in bursts

batchqueue

This module provides an extension to the basic queue.Queue class in order to handle situations where incoming data for the queue tends to come in bursts to be processed as a group.

Example Usage:

def worker(queue=None):
    self.assertIsNotNone(queue)

    while True:
        items = queue.get_batch()
        for i in items:
            # Do stuff
            queue.task_done()

# Process a batch any time there's at least a 700ms lull in traffic
self.q = BatchQueue(lull_time=700)

# turn-on the worker thread
threading.Thread(target=worker, daemon=True, kwargs={'queue': self.q}).start()

# send ten task requests to the worker, pausing a random number of
# milliseconds (up to 1000)
for item in range(10):
    sleep(random.randrange(0, 1000) / 1000.0)
    self.q.put(item + 1)

# block until all tasks are done
self.q.join()