Integrate websocket changes more cleanly

* Adjusted command line parameters
* Adding Controller methods for managing websocket services and peers

Author: matburt

receptor/config.py

@@ -4,6 +4,7 @@
 import logging
 import os
 import ssl
+import urllib.parse
 
 from .entrypoints import run_as_node, run_as_ping, run_as_send, run_as_controller
 from .exceptions import ReceptorRuntimeError, ReceptorConfigError
@@ -108,17 +109,18 @@ def __init__(self, args=None):
         # Receptor node options
         self.add_config_option(
             section='node',
-            key='listen_address',
-            default_value='0.0.0.0',
+            key='listen',
+            default_value='receptor://0.0.0.0:8888',
             value_type='str',
-            hint='Set/override IP address to listen on. If not set here or in a config file, the default is 0.0.0.0/0.',
+            hint='Set/override IP address and port to listen on. If not set here or in a config file, the default is receptor://0.0.0.0:8888.',
         )
         self.add_config_option(
             section='node',
-            key='listen_port',
-            default_value=8888,
-            value_type='int',
-            hint='Set/override TCP port to listen on. If not set here or in a config file, the default is 8888.',
+            key='websocket_listen',
+            default_value='',
+            value_type='str',
+            hint='Set IP address and port to listen on for websocket clients in the form ws[s]://addr:port. '
+                 'If not set here or in a config file then it is disabled',
         )
         self.add_config_option(
             section='node',
@@ -179,17 +181,18 @@ def __init__(self, args=None):
         )
         self.add_config_option(
             section='controller',
-            key='listen_address',
-            default_value='0.0.0.0',
+            key='listen',
+            default_value='receptor://0.0.0.0:8888',
             value_type='str',
-            hint='Set/override IP address to listen on. If not set here or in a config file, the default is 0.0.0.0/0.',
+            hint='Set IP address and port to listen on. If not set here or in a config file, the default is receptor://0.0.0.0/0:8888.',
         )
         self.add_config_option(
             section='controller',
-            key='listen_port',
-            default_value=8888,
-            value_type='int',
-            hint='Set/override TCP port to listen on. If not set here or in a config file, the default is 8888.',
+            key='websocket_listen',
+            default_value='',
+            value_type='str',
+            hint='Set IP address and port to listen on for websocket clients in the form ws://addr:port. '
+                 'If not set here or in a config file then it is disabled',
         )
         self.add_config_option(
             section='controller',
@@ -199,14 +202,6 @@ def __init__(self, args=None):
             value_type='str',
             hint='Set/override controller node identifier. If unspecified here or in a config file, one will be automatically generated.',
         )
-        # Ping options
-        # self.add_config_option(
-        #     section='ping',
-        #     key='socket_path',
-        #     default_value='/var/run/receptor_controller.sock',
-        #     value_type='path',
-        #     hint='Path to control socket for controller commands.',
-        # )
         self.add_config_option(
             section='ping',
             key='peer',
@@ -462,7 +457,6 @@ def go(self):
             raise ReceptorRuntimeError("you must specify a subcommand (%s)." % (", ".join(SUBCOMMAND_EXTRAS.keys()),))
         self._parsed_args.func(self)
 
-
     def get_client_ssl_context(self):
         if self.auth_ssl_cert:
             logger.debug("Loading SSL Client Context")

receptor/controller.py

@@ -4,7 +4,7 @@
 import uuid
 from urllib.parse import urlparse
 
-from .ws import WSServer
+from .ws import WSServer, WSClient
 from .protocol import BasicProtocol, create_peer
 from .receptor import Receptor
 from .messages import envelope
@@ -22,29 +22,37 @@ def __init__(self, config, loop=asyncio.get_event_loop(), queue=None):
             self.queue = asyncio.Queue(loop=loop)
         self.receptor.response_queue = self.queue
 
-    def enable_server(self, listen_address, listen_port):
+    def parse_peer(self, peer):
+        if "://" not in peer:
+            peer = f"receptor://{peer}"
+        return urlparse(peer)
+
+    def enable_server(self, listen_url):
+        service = self.parse_peer(listen_url)
         listener = self.loop.create_server(
             lambda: BasicProtocol(self.receptor, self.loop),
-            listen_address, listen_port,
+            service.hostname, service.port,
             ssl=self.receptor.config.get_server_ssl_context())
-        logger.info("Serving on {}:{}".format(listen_address, listen_port))
-        # TODO: Enable stats?
+        logger.info("Serving on {}:{}".format(service.hostname, service.port))
         self.loop.create_task(listener)
 
-    def enable_websocket_server(self, listen_address, listen_port):
+    def enable_websocket_server(self, listen_url):
+        service = urlparse(listen_url)
         listener = self.loop.create_server(
             WSServer(self.receptor, self.loop).app().make_handler(),
-            listen_address, listen_port,
+            service.hostname, service.port,
             ssl=self.receptor.config.get_server_ssl_context())
-        logger.info("Server ws on {}:{}".format(listen_address, listen_port))
+        logger.info("Serving websockets on {}:{}".format(service.hostname, service.port))
         self.loop.create_task(listener)
 
     async def add_peer(self, peer):
-        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.hostname, peer.port))
+        parsed = self.parse_peer(peer)
+        if parsed.scheme == 'receptor':
+            logger.info("Connecting to receptor peer {}".format(peer))
+            await self.loop.create_task(create_peer(self.receptor, self.loop, parsed.hostname, parsed.port))
+        elif parsed.scheme in ('ws', 'wss'):
+            logger.info("Connecting to websocket peer {}".format(peer))
+            await self.loop.create_task(WSClient(self.receptor, self.loop).connect(peer))
 
     async def recv(self):
         inner = await self.receptor.response_queue.get()

receptor/entrypoints.py

@@ -27,8 +27,9 @@ async def node_keepalive():
         logger.info(f'Starting stats on port {config.node_stats_port}')
         start_http_server(config.node_stats_port)
     if not config.node_server_disable:
-        controller.enable_server(config.node_listen_address,
-                                 config.node_listen_port)
+        controller.enable_server(config.node_listen)
+    if config.node_websocket_listen:
+        controller.enable_websocket_server(config.node_websocket_listen)
     for peer in config.node_peers:
         controller.loop.create_task(controller.add_peer(peer))
     if config.node_keepalive_interval > 1:
@@ -42,8 +43,9 @@ def run_as_controller(config):
     if config.controller_stats_enable:
         logger.info(f'Starting stats on port {config.node_stats_port}')
         start_http_server(config.controller_stats_port)
-    controller.enable_server(config.controller_listen_address,
-                             config.controller_listen_port)
+    controller.enable_server(config.controller_listen)
+    if config.controller_websocket_listen:
+        controller.enable_websocket_server(config.controller_websocket_listen)
     controller.loop.create_task(controller.receptor.watch_expire())
     controller.run()