raw sockets are functional now, logging now has node id

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

Author: jhjaggars

receptor/__main__.py

@@ -21,7 +21,7 @@ def main(args=None):
             'disable_existing_loggers': False,
             'formatters': {
                 'verbose': {
-                    'format': '{levelname} {asctime} {module} {message}',
+                    'format': '{levelname} {asctime} {node_id} {module} {message}',
                     'style': '{',
                 }
             },
@@ -40,6 +40,13 @@ def main(args=None):
         }
     )
 
+    def _f(record):
+        record.node_id = config.default_node_id
+        return True
+
+    for h in logging.getLogger('receptor').handlers:
+        h.addFilter(_f)
+
     try:
         config.go()
     except Exception:

receptor/connection/__init__.py

@@ -1,28 +1,27 @@
 import logging
 
 import asyncio
+from collections.abc import AsyncIterator
+from abc import abstractmethod, abstractproperty
 
 from ..messages.envelope import FramedBuffer
 
 logger = logging.getLogger(__name__)
 
 
-class Transport:
-    def __aiter__(self):
-        return self
-
-    async def __anext__(self):
-        raise NotImplementedError("subclasses should implement this")
+class Transport(AsyncIterator):
 
+    @abstractmethod
     async def close(self):
-        raise NotImplementedError("subclasses should implement this")
+        pass
 
-    @property
+    @abstractproperty
     def closed(self):
-        raise NotImplementedError("subclasses should implement this")
+        pass
 
+    @abstractmethod
     async def send(self, bytes_):
-        raise NotImplementedError("subclasses should implement this")
+        pass
 
 
 async def watch_queue(conn, buf):
@@ -41,7 +40,6 @@ async def watch_queue(conn, buf):
             logger.exception("watch_queue: error received trying to write")
             await buf.put(msg)
             return await conn.close()
-    logger.debug("watch_queue: ws is now closed")
 
 
 class Worker:
@@ -56,7 +54,6 @@ def __init__(self, receptor, loop):
         self.write_task = None
 
     def start_receiving(self):
-        logger.debug("starting recv")
         self.read_task = self.loop.create_task(self.receive())
 
     async def receive(self):
@@ -98,7 +95,7 @@ async def start_processing(self):
         return await self.write_task
 
     async def _wait_handshake(self):
-        logger.debug("serve: waiting for HI")
+        logger.debug("waiting for HI")
         response = await self.buf.get()  # TODO: deal with timeout
         self.remote_id = response.header["id"]
         self.register()
@@ -110,7 +107,7 @@ async def client(self, transport):
             await self.hello()
             await self._wait_handshake()
             await self.start_processing()
-            logger.debug("connect: normal exit")
+            logger.debug("normal exit")
         finally:
             self.unregister()
 

receptor/connection/sock.py

@@ -6,13 +6,15 @@
 
 
 class RawSocket(Transport):
-    def __init__(self, reader, writer):
+    def __init__(self, reader, writer, chunk_size=2 ** 8):
         self.reader = reader
         self.writer = writer
         self._closed = False
+        self.chunk_size = chunk_size
 
     async def __anext__(self):
-        return await self.reader.read()
+        bytes_ = await self.reader.read(self.chunk_size)
+        return bytes_
 
     @property
     def closed(self):