Handle some expiration corner cases

Author: matburt

receptor/buffers/file.py

@@ -1,4 +1,3 @@
-import collections
 import logging
 import uuid
 import json

receptor/protocol.py

@@ -1,6 +1,7 @@
 import datetime
 import asyncio
 import logging
+import time
 import json
 import uuid
 from collections import deque
@@ -88,8 +89,7 @@ async def wait_greeting(self):
                     break
                 else:
                     logger.error("Handshake failed!")
-                    # TODO: Trigger disconnection
-                    # otherwise we could get stuck here
+                    self.transport.close()
             await asyncio.sleep(.1)
         logger.debug("handshake complete, starting normal handle loop")
         self.loop.create_task(self.connection.message_handler(self.incoming_buffer)) # Duplicated (see handle_handshake)?
@@ -104,6 +104,7 @@ def send_handshake(self):
         msg = json.dumps({
             "cmd": "HI",
             "id": self.receptor.node_id,
+            "expire_time": time.time() + 10,
         }).encode("utf-8")
         self.transport.write(msg + DELIM)
 
@@ -185,7 +186,6 @@ def data_received(self, data):
             raw_payload=payload,
             directive=directive
         )
-        # TODO: Response expiration task?
         # TODO: Persistent registry?
         self.loop.create_task(self.receptor.router.send(inner_env,
                                                         expected_response=True))

receptor/receptor.py

@@ -46,8 +46,8 @@ async def watch_expire(self):
                 for ident, message in buffer:
                     message_actual = json.loads(message)
                     if "expire_time" in message_actual and message_actual['expire_time'] < time.time():
+                        buffer.read_message(ident, remove=True)
                         logger.info("Expiring message {}:{}".format(ident, connection["id"]))
-                        expired_message = buffer.read_message(ident, remove=True)
                         # TODO: Do something with expired message
                 if connection["last"] + 86400 < time.time():
                     logger.info("Expiring connection {}".format(connection["id"]))
@@ -110,7 +110,6 @@ def remove_connection(self, protocol_obj):
                 self.router.debug_router()
                 self.update_connection_manifest(connection_node)
             notify_connections += self.connections[connection_node]
-        # TODO: Broadcast update, set timer for full expiration
         for active_connection in notify_connections:
             active_connection.send_route_advertisement(self.router.get_edges())