Added TimeoutQueue.

This commit is contained in:
Jeremy Fincher 2004-09-01 21:58:21 +00:00
parent 012487b920
commit 13e77bac41
2 changed files with 48 additions and 1 deletions

View File

@ -37,8 +37,9 @@ __revision__ = "$Id$"
import supybot.fix as fix
import time
import types
from itertools import imap
from itertools import imap, ilen
class RingBuffer(object):
"""Class to represent a fixed-size ring buffer."""
@ -303,6 +304,39 @@ class smallqueue(list):
self[:] = []
class TimeoutQueue(object):
def __init__(self, timeout, queue=None):
if queue is None:
queue = smallqueue()
self.queue = queue
self.timeout = timeout
def _clearOldElements(self):
now = time.time()
while now - self.queue.peek()[0] > self.timeout:
self.queue.dequeue()
def enqueue(self, elt, at=None):
if at is None:
at = time.time()
self.queue.enqueue((at, elt))
def dequeue(self):
self._clearOldElements()
return self.queue.dequeue()[1]
def __iter__(self):
# We could _clearOldElements here, but what happens if someone stores
# the resulting generator and elements that should've timed out are
# yielded? Hmm? What happens then, smarty-pants?
for (t, elt) in self.queue:
if time.time() - t < self.timeout:
yield elt
def __len__(self):
return ilen(self)
class MaxLengthQueue(queue):
__slots__ = ('length',)
def __init__(self, length, seq=()):

View File

@ -576,5 +576,18 @@ class TwoWayDictionaryTestCase(SupyTestCase):
self.failIf('foo' in d)
class TestTimeoutQueue(SupyTestCase):
def test(self):
q = TimeoutQueue(1)
q.enqueue(1)
self.assertEqual(len(q), 1)
q.enqueue(2)
self.assertEqual(len(q), 2)
q.enqueue(3)
self.assertEqual(len(q), 3)
time.sleep(1)
self.assertEqual(len(q), 0)
# vim:set shiftwidth=4 tabstop=8 expandtab textwidth=78: