From 571cc00dc3595f2e627884d70fa6e3181a13d47b Mon Sep 17 00:00:00 2001 From: David Wilson Date: Fri, 26 Aug 2016 20:12:04 +0100 Subject: [PATCH] Import incomplete econtext.tcp implementation. --- econtext/tcp.py | 42 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) create mode 100644 econtext/tcp.py diff --git a/econtext/tcp.py b/econtext/tcp.py new file mode 100644 index 00000000..b45688a8 --- /dev/null +++ b/econtext/tcp.py @@ -0,0 +1,42 @@ +""" +Functionality to allow a slave context to reconnect back to its master using a +plain TCP connection. +""" + +import socket + +import econtext.core + + +class Listener(econtext.core.BasicStream): + def __init__(self, broker, address=None, backlog=30): + self._broker = broker + self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._sock.bind(address or ('0.0.0.0', 0)) + self._sock.listen(backlog) + econtext.core.set_cloexec(self._sock.fileno()) + self.address = self._sock.getsockname() + self.receive_side = econtext.core.Side(self, self._sock.fileno()) + broker.start_receive(self) + + def on_receive(self, broker): + sock, addr = self._sock.accept() + context = Context(self._broker, name=addr) + stream = econtext.core.Stream(context) + stream.accept(sock.fileno(), sock.fileno()) + + +def listen(broker, address=None, backlog=30): + """Listen on `address` for connections from newly spawned contexts.""" + return Listener(broker, address, backlog) + + +def connect(context): + """Connect to a Broker at the address specified in our associated + Context.""" + LOG.debug('%s.connect()', __name__) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.receive_side = econtext.core.Side(self, sock.fileno()) + self.transmit_side = econtext.core.Side(self, sock.fileno()) + sock.connect(self._context.parent_addr) + self.enqueue(0, self._context.name)