wiring up streams for regular sockets

Signed-off-by: Jesse Jaggars <jjaggars@redhat.com>

Author: jhjaggars

receptor/connection/sock.py

@@ -1,5 +1,9 @@
+import asyncio
+import logging
 from . import Transport
 
+logger = logging.getLogger(__name__)
+
 
 class RawSocket(Transport):
     def __init__(self, reader, writer):
@@ -21,3 +25,25 @@ async def close(self):
     async def send(self, bytes_):
         self.writer.write(bytes_)
         await self.writer.drain()
+
+
+async def connect(host, port, factory, loop=None):
+    if not loop:
+        loop = asyncio.get_event_loop()
+
+    worker = factory()
+    try:
+        r, w = await asyncio.open_connection(host, port, loop=loop)
+        t = RawSocket(r, w)
+        await worker.client(t)
+    except Exception:
+        logger.exception("sock.connect")
+    finally:
+        await asyncio.sleep(5)
+        logger.debug("sock.connect: reconnection")
+        loop.create_task(connect(host, port, factory, loop))
+
+
+async def serve(reader, writer, factory):
+    t = RawSocket(reader, writer)
+    await factory().server(t)

receptor/connection/ws.py

@@ -38,10 +38,10 @@ async def connect(uri, factory, loop=None):
             t = WebSocket(ws)
             await worker.client(t)
     except Exception:
-        logger.exception("connect")
+        logger.exception("ws.connect")
     finally:
         await asyncio.sleep(5)
-        logger.debug("reconnecting")
+        logger.debug("ws.connect: reconnecting")
         loop.create_task(connect(uri, factory=factory, loop=loop))
 
 

receptor/controller.py

@@ -1,5 +1,6 @@
 import asyncio
 import datetime
+import functools
 import logging
 import uuid
 from urllib.parse import urlparse
@@ -8,7 +9,7 @@
 from .receptor import Receptor
 from .messages import envelope
 from . import connection
-from .connection import ws, Worker
+from .connection import ws, sock, Worker
 from . import protocol
 
 logger = logging.getLogger(__name__)
@@ -31,9 +32,11 @@ def parse_peer(self, peer):
 
     def enable_server(self, listen_url):
         service = self.parse_peer(listen_url)
-        listener = self.loop.create_server(
-            lambda: BasicProtocol(self.receptor, self.loop),
-            service.hostname, service.port,
+        cb = functools.partial(sock.serve, factory=factory)
+        listener = asyncio.start_server(
+            client_connected_cb,
+            host=service.hostname,
+            port=service.port,
             ssl=self.receptor.config.get_server_ssl_context())
         logger.info("Serving on {}:{}".format(service.hostname, service.port))
         self.loop.create_task(listener)