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