#!/usr/bin/python2 import collections import threading class Pipe(collections.Iterable, collections.Sized): def __init__(self): self.deque = collections.deque() self.closed = False self.condition = threading.Condition() def put(self, element): with self.condition: if self.closed: raise EOFError("Pipe closed") self.deque.append(element) self.condition.notify() def pop(self): with self.condition: while not (self.deque or self.closed): self.condition.wait() try: return self.deque.popleft() except IndexError: raise EOFError("Pipe closed") def close(self): with self.condition: self.closed = True self.condition.notify_all() def __len__(self): with self.condition: return len(self.deque) def __iter__(self): return PipeIter(self) class PipeIter(collections.Iterator): def __init__(self, pipe): self.pipe = pipe def __iter__(self): return self def next(self): try: return self.pipe.pop() except EOFError: raise StopIteration() def main(): pipe = Pipe() elements = ["Foo", "Bar"] for element in elements: pipe.put(element) pipe.close() popped = list(pipe) assert popped == elements if __name__ == '__main__': main()