Refactor buffer manager to take receptor object

In order to get the local node-id
Fix an issue importing exception in the buffer object itself

Author: matburt

receptor/buffers/base.py

@@ -3,16 +3,16 @@
 
 
 class BaseBufferManager:
-    def get_buffer_for_node(self, node_id, config):
+    def get_buffer_for_node(self, node_id, receptor):
         raise NotImplementedError()
 
 
 class BaseBuffer:
     node_id = None
 
-    def __init__(self, node_id, config):
+    def __init__(self, node_id, receptor):
         self.node_id = node_id
-        self.config = config
+        self.receptor = receptor
 
     def push(self, message):
         raise NotImplementedError()

receptor/buffers/file.py

@@ -5,26 +5,26 @@
 import os
 
 from .base import BaseBufferManager, BaseBuffer
-from .exceptions import ReceptorBufferError
+from ..exceptions import ReceptorBufferError
 
 logger = logging.getLogger(__name__)
 
 
 class FileBufferManager(BaseBufferManager):
     _buffers = {}
 
-    def get_buffer_for_node(self, node_id, config):
-        return self._buffers.setdefault(node_id, FileBuffer(node_id, config))
+    def get_buffer_for_node(self, node_id, receptor):
+        return self._buffers.setdefault(node_id, FileBuffer(node_id, receptor))
 
 
 class FileBuffer(BaseBuffer):
 
-    def __init__(self, node_id, config):
-        super().__init__(node_id, config)
+    def __init__(self, node_id, receptor):
+        super().__init__(node_id, receptor)
         self.node_id = node_id
-        self.base_path = os.path.join(os.path.expanduser(self.config.server.data_dir))
-        self.message_path = os.path.join(self.base_path, "messages")
-        self.manifest_path = os.path.join(self.base_path, "manifest-{}".format(node_id))
+        self.base_path = os.path.join(os.path.expanduser(self.receptor.config.server.data_dir))
+        self.message_path = os.path.join(self.base_path, self.receptor.node_id, "messages")
+        self.manifest_path = os.path.join(self.base_path, self.receptor.node_id, "manifest-{}".format(node_id))
         if not os.path.exists(self.message_path):
             os.makedirs(self.message_path, mode=0o700)
 

receptor/buffers/memory.py

@@ -9,14 +9,14 @@
 class InMemoryBufferManager(BaseBufferManager):
     _buffers = {}
 
-    def get_buffer_for_node(self, node_id, config):
+    def get_buffer_for_node(self, node_id, receptor):
         return self._buffers.setdefault(node_id, InMemoryBuffer(node_id))
 
 
 # NOTE: This is not thread safe
 class InMemoryBuffer(BaseBuffer):
-    def __init__(self, node_id, config):
-        super().__init__(node_id, config)
+    def __init__(self, node_id, receptor):
+        super().__init__(node_id, receptor)
         self._buffer = collections.deque()
 
     def push(self, message):

receptor/connection.py

@@ -47,7 +47,7 @@ def send_route_advertisement(self, edges=None, seen=[]):
 
         # TODO: This should be a broadcast call to the connection manager
         for target in destinations:
-            buf = self.buffer_mgr.get_buffer_for_node(target, self.receptor.config)
+            buf = self.buffer_mgr.get_buffer_for_node(target, self.receptor)
             try:
                 buf.push(json.dumps({
                     "cmd": "ROUTE",

receptor/protocol.py

@@ -36,13 +36,13 @@ def __init__(self, receptor, loop):
 
     async def watch_queue(self, node, transport):
         buffer_mgr = self.receptor.config.components.buffer_manager
-        buffer_obj = buffer_mgr.get_buffer_for_node(node, self.receptor.config)
+        buffer_obj = buffer_mgr.get_buffer_for_node(node, self.receptor)
         while not transport.is_closing():
             try:
                 msg = buffer_obj.pop()
                 transport.write(msg + DELIM)
             except IndexError:
-                await asyncio.sleep(1)
+                await asyncio.sleep(0.1)
             except ReceptorBufferError as e:
                 logger.exception("Receptor Buffer Read Error: {}".format(e))
                 # TODO: We need to try to send this message along somewhere else
@@ -82,7 +82,7 @@ async def wait_greeting(self):
                     # TODO: Trigger disconnection
             await asyncio.sleep(.1)
         logger.debug("handshake complete, starting normal handle loop")
-        self.loop.create_task(self.connection.message_handler(self.incoming_buffer))
+        self.loop.create_task(self.connection.message_handler(self.incoming_buffer)) # Duplicated?
 
     def handle_handshake(self, data):
         self.greeted = True

receptor/router.py

@@ -133,7 +133,7 @@ async def forward(self, outer_envelope, next_hop):
         Forward a message on to the next hop closer to its destination
         """
         buffer_mgr = self.receptor.config.components.buffer_manager
-        buffer_obj = buffer_mgr.get_buffer_for_node(next_hop, self.receptor.config)
+        buffer_obj = buffer_mgr.get_buffer_for_node(next_hop, self.receptor)
         outer_envelope.route_list.append(self.node_id)
         logger.debug(f'Forwarding frame {outer_envelope.frame_id} to {next_hop}')
         try: