jpayne@68: #
jpayne@68: # Analogue of `multiprocessing.connection` which uses queues instead of sockets
jpayne@68: #
jpayne@68: # multiprocessing/dummy/connection.py
jpayne@68: #
jpayne@68: # Copyright (c) 2006-2008, R Oudkerk
jpayne@68: # Licensed to PSF under a Contributor Agreement.
jpayne@68: #
jpayne@68: 
jpayne@68: __all__ = [ 'Client', 'Listener', 'Pipe' ]
jpayne@68: 
jpayne@68: from queue import Queue
jpayne@68: 
jpayne@68: 
jpayne@68: families = [None]
jpayne@68: 
jpayne@68: 
jpayne@68: class Listener(object):
jpayne@68: 
jpayne@68:     def __init__(self, address=None, family=None, backlog=1):
jpayne@68:         self._backlog_queue = Queue(backlog)
jpayne@68: 
jpayne@68:     def accept(self):
jpayne@68:         return Connection(*self._backlog_queue.get())
jpayne@68: 
jpayne@68:     def close(self):
jpayne@68:         self._backlog_queue = None
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def address(self):
jpayne@68:         return self._backlog_queue
jpayne@68: 
jpayne@68:     def __enter__(self):
jpayne@68:         return self
jpayne@68: 
jpayne@68:     def __exit__(self, exc_type, exc_value, exc_tb):
jpayne@68:         self.close()
jpayne@68: 
jpayne@68: 
jpayne@68: def Client(address):
jpayne@68:     _in, _out = Queue(), Queue()
jpayne@68:     address.put((_out, _in))
jpayne@68:     return Connection(_in, _out)
jpayne@68: 
jpayne@68: 
jpayne@68: def Pipe(duplex=True):
jpayne@68:     a, b = Queue(), Queue()
jpayne@68:     return Connection(a, b), Connection(b, a)
jpayne@68: 
jpayne@68: 
jpayne@68: class Connection(object):
jpayne@68: 
jpayne@68:     def __init__(self, _in, _out):
jpayne@68:         self._out = _out
jpayne@68:         self._in = _in
jpayne@68:         self.send = self.send_bytes = _out.put
jpayne@68:         self.recv = self.recv_bytes = _in.get
jpayne@68: 
jpayne@68:     def poll(self, timeout=0.0):
jpayne@68:         if self._in.qsize() > 0:
jpayne@68:             return True
jpayne@68:         if timeout <= 0.0:
jpayne@68:             return False
jpayne@68:         with self._in.not_empty:
jpayne@68:             self._in.not_empty.wait(timeout)
jpayne@68:         return self._in.qsize() > 0
jpayne@68: 
jpayne@68:     def close(self):
jpayne@68:         pass
jpayne@68: 
jpayne@68:     def __enter__(self):
jpayne@68:         return self
jpayne@68: 
jpayne@68:     def __exit__(self, exc_type, exc_value, exc_tb):
jpayne@68:         self.close()