Import incomplete econtext.tcp implementation.

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)