Extra functionality for Pipe added

threads/pipe.py

@@ -10,6 +10,10 @@ class Pipe(Queue.Queue):
     kinda persistent. 
     
     Call .close() to terminate that cyclic reference.
+    
+    Throws Queue.Empty where it makes sense. You can
+    use Pipe.Empty - contains a handy shortcut to the same
+    exception
     """
     
     Empty = Queue.Empty
@@ -21,13 +25,27 @@ class Pipe(Queue.Queue):
     def put(self, item, block=True, timeout=None):
         Queue.Queue.put(self.other_queue, item, block, timeout)
 
-def create_pipes():
-    """Returns a pair (tuple) of Pipes connecting to each
-    other"""    
+    def close(self):
+        self.other_queue = None
+        
+def create_pipes(queue_for_a=None, queue_for_b=None):
+    """Returns a pair (tuple - a, b) of Pipes connecting to each
+    other
+    
+    @param queue_for_a: if defined, should be a Queue that a's .put()
+    will send messages to
+    @param queue_for_b: same as queue_for_a, but for b
+    """    
     a = Pipe()
     b = Pipe()
-    a.other_queue = b
-    b.other_queue = a
+    if queue_for_a == None:
+        a.other_queue = b
+    else:
+        a.other_queue = queue_for_a
+    if queue_for_b == None:
+        b.other_queue = a
+    else:
+        b.other_queue = queue_for_a
     return a, b
     
     
\ No newline at end of file

threads/unittests/pipe.py

-from satella.threads import create_pipes
+from satella.threads import create_pipes, Pipe
 from time import sleep
+import Queue
 
 import unittest
 
@@ -12,4 +13,22 @@ class PipeTest(unittest.TestCase):
         
         a.put('Lol')
         
-        self.assertEqual(b.get(), 'Lol')
\ No newline at end of file
+        self.assertEqual(b.get(), 'Lol')
+        a.close()
+        b.close()
+        
+    def test_pipe_others(self):
+        """Tests queue_for_a"""
+
+        someq = Queue.Queue()
+        
+        a, b = create_pipes(queue_for_a=someq)
+        c, d = create_pipes(queue_for_a=someq)
+        
+        a.put('Lol')
+        c.put('Hey')
+        
+        self.assertEqual(someq.get(), 'Lol')        
+        self.assertEqual(someq.get(), 'Hey') 
+        self.assertRaises(Pipe.Empty, b.get, False)
+        self.assertRaises(Queue.Empty, b.get, False)        
\ No newline at end of file