Switch node to use new module interface

Rename some confusing things

Author: matburt

receptor/config.py

@@ -154,7 +154,7 @@ def __init__(self, args=None):
         )
         self.add_config_option(
             section='node',
-            key='ping_interval',
+            key='keepalive_interval',
             default_value=-1,
             value_type='int',
             hint='If specified, the node will ping all other known nodes in the mesh every N seconds. The default is -1, meaning no pings are sent.',

receptor/controller.py

@@ -2,6 +2,7 @@
 import datetime
 import logging
 import uuid
+from urllib.parse import urlparse
 
 from .ws import WSServer
 from .protocol import BasicProtocol, create_peer
@@ -11,13 +12,6 @@
 logger = logging.getLogger(__name__)
 
 
-def connect_to_socket(socket_path):
-    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-    sock.connect(socket_path)
-    return sock
-
-
-# TODO: track stats
 class Controller:
 
     def __init__(self, config, loop=asyncio.get_event_loop(), queue=None):
@@ -46,16 +40,17 @@ def enable_websocket_server(self, listen_address, listen_port):
         self.loop.create_task(listener)
 
     async def add_peer(self, peer):
-        # NOTE: Not signing or serializing
+        if "://" not in peer:
+            peer = f"receptor://{peer}"
+        peer = urlparse(peer)
         logger.info("Connecting to peer {}".format(peer))
-        await self.loop.create_task(create_peer(self.receptor, self.loop,
-                                                *peer.strip().split(":", 1)))
+        await self.loop.create_task(create_peer(self.receptor, self.loop, peer.hostname, peer.port))
 
     async def recv(self):
         inner = await self.receptor.response_queue.get()
         return inner.raw_payload
 
-    async def send(self, message):
+    async def send(self, message, expect_response=True):
         inner_env = envelope.Inner(
             receptor=self.receptor,
             message_id=str(uuid.uuid4()),
@@ -66,16 +61,16 @@ async def send(self, message):
             timestamp=datetime.datetime.utcnow().isoformat(),
             raw_payload=message.fd.read(),
         )
-        await self.receptor.router.send(inner_env, expected_response=True)
+        await self.receptor.router.send(inner_env, expected_response=expect_response)
 
-    async def ping(self, destination):
-        await self.receptor.router.ping_node(destination)
+    async def ping(self, destination, expected_response=True):
+        await self.receptor.router.ping_node(destination, expected_response)
 
-    def run(self, coro=None):
+    def run(self, app=None):
         try:
-            if coro is None:
-                coro = self.receptor.shutdown_handler
-            self.loop.run_until_complete(coro())
+            if app is None:
+                app = self.receptor.shutdown_handler
+            self.loop.run_until_complete(app())
         except KeyboardInterrupt:
             pass
         finally:

receptor/entrypoints.py

@@ -4,20 +4,34 @@
 
 from prometheus_client import start_http_server
 
-from .receptor import Receptor
 from .controller import Controller
-from . import node
 
 logger = logging.getLogger(__name__)
 
 
 def run_as_node(config):
-    receptor = Receptor(config)
-    logger.info(f'Running as Receptor node with ID: {receptor.node_id}')
+    async def node_keepalive():
+        for node_id in controller.receptor.router.get_nodes():
+            await controller.ping(node_id, expected_response=False)
+        absolute_call_time = (((int(controller.loop.time()) + 1) // config.node_keepalive_interval) + 1) * config.node_keepalive_interval
+        controller.loop.call_at(absolute_call_time,
+                                controller.loop.create_task,
+                                node_keepalive())
+
+    controller = Controller(config)
+    logger.info(f'Running as Receptor node with ID: {controller.receptor.node_id}')
     if config.node_stats_enable:
         logger.info(f'Starting stats on port {config.node_stats_port}')
         start_http_server(config.node_stats_port)
-    node.mainloop(receptor, config.node_ping_interval)
+    if not config.node_server_disable:
+        controller.enable_server(config.node_listen_address,
+                                 config.node_listen_port)
+    for peer in config.node_peers:
+        controller.loop.create_task(controller.add_peer(peer))
+    if config.node_keepalive_interval > 1:
+        controller.loop.create_task(node_keepalive())
+    controller.loop.create_task(controller.receptor.watch_expire())
+    controller.run()
 
 
 def run_as_controller(config):
@@ -27,6 +41,7 @@ def run_as_controller(config):
         start_http_server(config.controller_stats_port)
     controller.enable_server(config.controller_listen_address,
                              config.controller_listen_port)
+    controller.loop.create_task(controller.receptor.watch_expire())
     controller.run()
 
 
@@ -39,28 +54,28 @@ def ping_iter():
             while True:
                 yield 0
 
-    async def handle_ping():
+    async def ping_entrypoint():
         read_task = controller.loop.create_task(read_responses())
         await controller.add_peer(config.ping_peer)
         start_wait = time.time()
         while not controller.receptor.router.node_is_known(config.ping_recipient) and (time.time() - start_wait < 5):
             await asyncio.sleep(0.1)
-        await do_ping()
+        await send_pings()
         await read_task
 
     async def read_responses():
         for _ in ping_iter():
             payload = await controller.recv()
             print("{}".format(payload))
 
-    async def do_ping():
+    async def send_pings():
         for _ in ping_iter():
             await controller.ping(config.ping_recipient)
             await asyncio.sleep(config.ping_delay)
 
     logger.info(f'Sending ping to {config.ping_recipient} via {config.ping_peer}.')
     controller = Controller(config)
-    controller.run(handle_ping)
+    controller.run(ping_entrypoint)
 
 
 def run_as_send(config):

receptor/node.py

@@ -5,25 +5,13 @@
 import uuid
 from collections import defaultdict
 
-from dateutil import parser
-
 from .exceptions import ReceptorBufferError, UnrouteableError
 from .messages import envelope
 from .stats import route_counter
 
 logger = logging.getLogger(__name__)
 
 
-async def log_ping(response):
-    pong_received = datetime.datetime.utcnow()
-    ping_sent, ping_received = response.raw_payload.split('|')
-    ping_time = parser.parse(ping_received) - parser.parse(ping_sent)
-    pong_time = pong_received - parser.parse(ping_received)
-    logger.info(f'Ping report for {response.sender}: '
-                f'ping={ping_time.total_seconds()}s; '
-                f'pong={pong_time.total_seconds()}s')
-
-
 class MeshRouter:
     _nodes = set()
     _edges = set()
@@ -94,7 +82,7 @@ def get_edges(self):
     def get_nodes(self):
         return self._nodes
 
-    async def ping_node(self, node_id, callback=log_ping):
+    async def ping_node(self, node_id, expected_response=True):
         logger.info(f'Sending ping to node {node_id}')
         now = datetime.datetime.utcnow().isoformat()
         ping_envelope = envelope.Inner(
@@ -108,7 +96,7 @@ async def ping_node(self, node_id, callback=log_ping):
             directive='receptor:ping',
             ttl=15
         )
-        await self.send(ping_envelope, callback)
+        await self.send(ping_envelope, expected_response)
 
     def find_shortest_path(self, to_node_id):
         """Implementation of Dijkstra algorithm"""