from argparse import ArgumentParser
+from itertools import chain
+from secrets import token_urlsafe
+import socket
+from subprocess import run
+import sys
+from traceback import print_exception
+from typing import List, Optional
+from urllib.parse import urlunsplit
from falcon.asgi import App
from falcon.constants import MEDIA_HTML
+from uvicorn import Config, Server
-from .cryptresource import CryptResource
+from .staticresource import StaticResource
+from .hub import Hub
+
+
+class HubApp(App):
+ def __init__(self, secret, browser, hubapp, *args, **kwargs):
+ kwargs.setdefault("media_type", MEDIA_HTML)
+ super().__init__(*args, **kwargs)
+ self.secret = secret or token_urlsafe(64)
+ self.hub = Hub(self.secret)
+ self.sr = StaticResource(self.secret, hubapp)
+ self.browser = browser
+ self.hub.add_routes(self)
+ self.sr.add_routes(self)
+ self.hub.update_context_vars(self.sr.context_vars)
+ self.add_error_handler(Exception, self.print_exception)
+
+ async def print_exception(self, req, resp, ex, params):
+ print_exception(*sys.exc_info())
def get_app():
- app = App(media_type=MEDIA_HTML)
ap = ArgumentParser()
ap.add_argument("--secret")
+ ap.add_argument("--browser", default="firefox")
+ ap.add_argument("--hubapp", default="first")
args = ap.parse_args()
- cr = CryptResource(args.secret)
- cr.register_on(app)
- return app
+ return HubApp(args.secret, args.browser, args.hubapp)
+
+
+class HubServer(Server):
+ async def startup(self, sockets: Optional[List[socket.socket]] = None) -> None:
+ await super().startup(sockets)
+ config = self.config
+ protocol_name = "https" if config.ssl else "http"
+ host = "0.0.0.0" if config.host is None else config.host
+ if ":" in host:
+ # It's an IPv6 address.
+ host = f"[{host.rstrip(']').lstrip('[')}]"
+
+ port = config.port
+ if port == 0:
+ try:
+ port = next(
+ chain.from_iterable((server.sockets for server in getattr(self, "servers", ())))
+ ).getsockname()[1]
+ except StopIteration:
+ pass
+ if {"http": 80, "https": 443}[protocol_name] != port:
+ host = f"{host}:{port}"
+ app = config.loaded_app.app
+ print("master_uri", app.sr.master_uri)
+ master_url = urlunsplit((protocol_name, host, app.sr.master_uri, "", ""))
+ print("secret:", app.secret)
+ if not app.browser:
+ print("master url", master_url)
+ else:
+ run([app.browser, master_url])
+
+
+def main():
+ HubServer(
+ Config("hub.app:get_app", factory=True, port=5000, log_level="info")
+ ).run()