Added smallqueue class.

src/structures.py

@@ -39,9 +39,10 @@ from fix import *
 
 import types
 
-__all__ = ['RingBuffer', 'queue', 'MaxLengthQueue']
+__all__ = ['RingBuffer', 'queue', 'smallqueue', 'MaxLengthQueue']
 
 class RingBuffer(object):
+    """Class to represent a fixed-size ring buffer."""
     __slots__ = ('L', 'i', 'full', 'maxSize')
     def __init__(self, maxSize, seq=()):
         if maxSize <= 0:
@@ -159,6 +160,9 @@ class RingBuffer(object):
 
 
 class queue(object):
+    """Queue class for handling large queues.  Queues smaller than 1,000 or so
+    elements are probably better served by the smallqueue class.
+    """
     __slots__ = ('front', 'back')
     def __init__(self, seq=()):
         self.back = []
@@ -279,6 +283,17 @@ class queue(object):
         self.front = L
         self.back = []
 
+class smallqueue(list):
+    __slots__ = ()
+    def enqueue(self, elt):
+        self.append(elt)
+
+    def dequeue(self):
+        return self.pop(0)
+
+    def peek(self):
+        return self[0]
+
 
 class MaxLengthQueue(queue):
     __slots__ = ('length',)

test/test_structures.py

@@ -354,6 +354,164 @@ class QueueTest(unittest.TestCase):
         q = queue(range(10))
         self.assertEqual(q, pickle.loads(pickle.dumps(q)))
 
+queue = smallqueue
+
+class SmallQueueTest(unittest.TestCase):
+    def testGetitem(self):
+        q = queue()
+        n = 10
+        self.assertRaises(IndexError, q.__getitem__, 0)
+        for i in xrange(n):
+            q.enqueue(i)
+        for i in xrange(n):
+            self.assertEqual(q[i], i)
+        for i in xrange(n, 0, -1):
+            self.assertEqual(q[-i], n-i)
+        for i in xrange(len(q)):
+            self.assertEqual(list(q), list(q[:i]) + list(q[i:]))
+        self.assertRaises(IndexError, q.__getitem__, -(n+1))
+        self.assertRaises(IndexError, q.__getitem__, n)
+        self.assertEqual(q[3:7], queue([3, 4, 5, 6]))
+
+    def testSetitem(self):
+        q1 = queue()
+        self.assertRaises(IndexError, q1.__setitem__, 0, 0)
+        for i in xrange(10):
+            q1.enqueue(i)
+        q2 = eval(repr(q1))
+        for (i, elt) in enumerate(q2):
+            q2[i] = elt*2
+        self.assertEqual([x*2 for x in q1], list(q2))
+
+    def testNonzero(self):
+        q = queue()
+        self.failIf(q, 'queue not zero after initialization')
+        q.enqueue(1)
+        self.failUnless(q, 'queue zero after adding element')
+        q.dequeue()
+        self.failIf(q, 'queue not zero after dequeue of only element')
+
+    def testLen(self):
+        q = queue()
+        self.assertEqual(0, len(q), 'queue len not 0 after initialization')
+        q.enqueue(1)
+        self.assertEqual(1, len(q), 'queue len not 1 after enqueue')
+        q.enqueue(2)
+        self.assertEqual(2, len(q), 'queue len not 2 after enqueue')
+        q.dequeue()
+        self.assertEqual(1, len(q), 'queue len not 1 after dequeue')
+        q.dequeue()
+        self.assertEqual(0, len(q), 'queue len not 0 after dequeue')
+        for i in range(10):
+            L = range(i)
+            q = queue(L)
+            self.assertEqual(len(q), i)
+
+    def testEq(self):
+        q1 = queue()
+        q2 = queue()
+        self.failUnless(q1 == q1, 'queue not equal to itself')
+        self.failUnless(q2 == q2, 'queue not equal to itself')
+        self.failUnless(q1 == q2, 'initialized queues not equal')
+        q1.enqueue(1)
+        self.failUnless(q1 == q1, 'queue not equal to itself')
+        self.failUnless(q2 == q2, 'queue not equal to itself')
+        q2.enqueue(1)
+        self.failUnless(q1 == q1, 'queue not equal to itself')
+        self.failUnless(q2 == q2, 'queue not equal to itself')
+        self.failUnless(q1 == q2, 'queues not equal after identical enqueue')
+        q1.dequeue()
+        self.failUnless(q1 == q1, 'queue not equal to itself')
+        self.failUnless(q2 == q2, 'queue not equal to itself')
+        self.failIf(q1 == q2, 'queues equal after one dequeue')
+        q2.dequeue()
+        self.failUnless(q1 == q2, 'queues not equal after both are dequeued')
+        self.failUnless(q1 == q1, 'queue not equal to itself')
+        self.failUnless(q2 == q2, 'queue not equal to itself')
+
+    def testInit(self):
+        self.assertEqual(len(queue()), 0, 'queue len not 0 after init')
+        q = queue()
+        q.enqueue(1)
+        q.enqueue(2)
+        q.enqueue(3)
+        self.assertEqual(queue((1, 2, 3)),q, 'init not equivalent to enqueues')
+        q = queue((1, 2, 3))
+        self.assertEqual(q.dequeue(), 1, 'values not returned in proper order')
+        self.assertEqual(q.dequeue(), 2, 'values not returned in proper order')
+        self.assertEqual(q.dequeue(), 3, 'values not returned in proper order')
+
+    def testRepr(self):
+        q = queue()
+        q.enqueue(1)
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+        q.enqueue('foo')
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+        q.enqueue(None)
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+        q.enqueue(1.0)
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+        q.enqueue([])
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+        q.enqueue(())
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+        q.enqueue([1])
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+        q.enqueue((1,))
+        self.assertEqual(q, eval(repr(q)), 'repr doesn\'t eval to same queue')
+
+    def testEnqueueDequeue(self):
+        q = queue()
+        self.assertRaises(IndexError, q.dequeue)
+        q.enqueue(1)
+        self.assertEqual(q.dequeue(), 1,
+                         'first dequeue didn\'t return same as first enqueue')
+        q.enqueue(1)
+        q.enqueue(2)
+        q.enqueue(3)
+        self.assertEqual(q.dequeue(), 1)
+        self.assertEqual(q.dequeue(), 2)
+        self.assertEqual(q.dequeue(), 3)
+
+    def testPeek(self):
+        q = queue()
+        self.assertRaises(IndexError, q.peek)
+        q.enqueue(1)
+        self.assertEqual(q.peek(), 1, 'peek didn\'t return first enqueue')
+        q.enqueue(2)
+        self.assertEqual(q.peek(), 1, 'peek didn\'t return first enqueue')
+        q.dequeue()
+        self.assertEqual(q.peek(), 2, 'peek didn\'t return second enqueue')
+        q.dequeue()
+        self.assertRaises(IndexError, q.peek)
+
+    def testContains(self):
+        q = queue()
+        self.failIf(1 in q, 'empty queue cannot have elements')
+        q.enqueue(1)
+        self.failUnless(1 in q, 'recent enqueued element not in q')
+        q.enqueue(2)
+        self.failUnless(1 in q, 'original enqueued element not in q')
+        self.failUnless(2 in q, 'second enqueued element not in q')
+        q.dequeue()
+        self.failIf(1 in q, 'dequeued element in q')
+        self.failUnless(2 in q, 'not dequeued element not in q')
+        q.dequeue()
+        self.failIf(2 in q, 'dequeued element in q')
+
+    def testIter(self):
+        q1 = queue((1, 2, 3))
+        q2 = queue()
+        for i in q1:
+            q2.enqueue(i)
+        self.assertEqual(q1, q2, 'iterate didn\'t return all elements')
+        for _ in queue():
+            self.fail('no elements should be in empty queue')
+
+    def testPickleCopy(self):
+        q = queue(range(10))
+        self.assertEqual(q, pickle.loads(pickle.dumps(q)))
+
 
 class MaxLengthQueueTestCase(unittest.TestCase):
     def testInit(self):