add flask render server

also factor out assumption of asyncio from base render server

Author: rmorshea

idom/server/base.py

@@ -1,15 +1,8 @@
 import abc
-from asyncio import AbstractEventLoop, new_event_loop, set_event_loop, get_event_loop
-from typing import TypeVar, Dict, Any, Tuple, Type, Optional, Generic, TypeVar
+from typing import TypeVar, Dict, Any, Tuple, Optional, Generic, TypeVar
 from threading import Thread, Event
 
 from idom.core.element import ElementConstructor
-from idom.core.layout import Layout, Layout
-from idom.core.dispatcher import (
-    AbstractDispatcher,
-    SendCoroutine,
-    RecvCoroutine,
-)
 
 
 _App = TypeVar("_App", bound=Any)
@@ -30,26 +23,16 @@ class AbstractRenderServer(Generic[_App, _Config]):
         :meth:`AbstractServerExtension.register`
     """
 
-    _loop: AbstractEventLoop
-    _dispatcher_type: Type[AbstractDispatcher]
-    _layout_type: Type[Layout] = Layout
-    _daemon_server_did_start: Event
-
     def __init__(
         self,
         constructor: ElementConstructor,
         config: Optional[_Config] = None,
     ) -> None:
         self._app: Optional[_App] = None
-        self._make_root_element = constructor
+        self._root_element_constructor = constructor
         self._daemonized = False
-        self._config = self._init_config()
-        if config is not None:
-            self._config = self._update_config(self._config, config)
-
-    @property
-    def loop(self) -> AbstractEventLoop:
-        return self._loop
+        self._config = self._create_config(config)
+        self._server_did_start = Event()
 
     @property
     def application(self) -> _App:
@@ -59,45 +42,45 @@ def application(self) -> _App:
 
     def run(self, *args: Any, **kwargs: Any) -> None:
         """Run as a standalone application."""
-        self._loop = get_event_loop()
         if self._app is None:
             app = self._default_application(self._config)
             self.register(app)
         else:
             app = self._app
-        return self._run_application(app, self._config, args, kwargs)
+        if not self._daemonized:
+            return self._run_application(app, self._config, args, kwargs)
+        else:
+            return self._run_application_in_thread(app, self._config, args, kwargs)
 
     def daemon(self, *args: Any, **kwargs: Any) -> Thread:
         """Run the standalone application in a seperate thread."""
         self._daemonized = True
 
-        def run_in_thread() -> None:
-            set_event_loop(new_event_loop())
-            return self.run(*args, **kwargs)
-
-        thread = Thread(target=run_in_thread, daemon=True)
+        thread = Thread(target=lambda: self.run(*args, **kwargs), daemon=True)
         thread.start()
 
-        self._wait_until_daemon_server_start()
+        self.wait_until_server_start()
 
         return thread
 
     def register(self: _Self, app: Optional[_App]) -> _Self:
         """Register this as an extension."""
         self._setup_application(app, self._config)
+        self._setup_application_did_start_event(app, self._server_did_start)
         self._app = app
         return self
 
-    def stop(self) -> None:
-        """Stop the running application"""
-        self.loop.call_soon_threadsafe(self._stop)
+    def server_started(self) -> bool:
+        """Whether the underlying application has started"""
+        return self._server_did_start.set()
 
-    @abc.abstractmethod
-    def _stop(self) -> None:
-        raise NotImplementedError()
+    def wait_until_server_start(self, timeout: float = 3.0):
+        """Block until the underlying application has started"""
+        if not self._server_did_start.wait(timeout=timeout):
+            raise RuntimeError(f"Server did not start within {timeout} seconds")
 
     @abc.abstractmethod
-    def _init_config(self) -> _Config:
+    def _create_config(self, config: Optional[_Config]) -> _Config:
         """Return the default configuration options."""
 
     @abc.abstractmethod
@@ -107,49 +90,24 @@ def _default_application(self, config: _Config) -> _App:
 
     @abc.abstractmethod
     def _setup_application(self, app: _App, config: _Config) -> None:
-        ...
+        """General application setup - add routes, templates, static resource, etc."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def _setup_application_did_start_event(self, app: _App, event: Event) -> None:
+        """Register a callback to the app indicating whether the server has started"""
+        raise NotImplementedError()
 
     @abc.abstractmethod
     def _run_application(
         self, app: _App, config: _Config, args: Tuple[Any, ...], kwargs: Dict[str, Any]
     ) -> None:
+        """Run the application in the main thread"""
         raise NotImplementedError()
 
     @abc.abstractmethod
-    def _update_config(self, old: _Config, new: _Config) -> _Config:  # pragma: no cover
-        """Return the new configuration options
-
-        Parameters:
-            old: The existing configuration options
-            new: The new configuration options
-        """
-        raise NotImplementedError()
-
-    async def _run_dispatcher(
-        self,
-        send: SendCoroutine,
-        recv: RecvCoroutine,
-        params: Dict[str, Any],
+    def _run_application_in_thread(
+        self, app: _App, config: _Config, args: Tuple[Any, ...], kwargs: Dict[str, Any]
     ) -> None:
-        async with self._make_dispatcher(params) as dispatcher:
-            await dispatcher.run(send, recv, None)
-
-    def _make_dispatcher(
-        self,
-        params: Dict[str, Any],
-    ) -> AbstractDispatcher:
-        return self._dispatcher_type(self._make_layout(params))
-
-    def _make_layout(
-        self,
-        params: Dict[str, Any],
-    ) -> Layout:
-        return self._layout_type(self._make_root_element(**params))
-
-    def _wait_until_daemon_server_start(self):
-        try:
-            self._daemon_server_did_start.wait(timeout=5)
-        except AttributeError:  # pragma: no cover
-            raise NotImplementedError(
-                f"Server implementation {self} did not define a server started thread event"
-            )
+        """This function has been called inside a daemon thread to run the application"""
+        raise NotImplementedError()

idom/server/flask.py

@@ -0,0 +1,226 @@
+import json
+import asyncio
+import logging
+from urllib.parse import urljoin
+from asyncio import Queue as AsyncQueue
+from threading import Event as ThreadEvent, Thread
+from queue import Queue as ThreadQueue
+from typing import Union, Tuple, Dict, Any, Optional, Callable, NamedTuple
+
+from typing_extensions import TypedDict
+from flask import Flask, Blueprint, send_from_directory, redirect, url_for
+from flask_cors import CORS
+from flask_sockets import Sockets
+from geventwebsocket.websocket import WebSocket
+from gevent import pywsgi
+from geventwebsocket.handler import WebSocketHandler
+
+import idom
+from idom.client.manage import BUILD_DIR
+from idom.core.layout import LayoutEvent, Layout
+from idom.core.dispatcher import AbstractDispatcher, SingleViewDispatcher
+
+from .base import AbstractRenderServer
+
+
+class Config(TypedDict, total=False):
+    import_name: str
+    url_prefix: str
+    cors: Union[bool, Dict[str, Any]]
+    serve_static_files: bool
+    redirect_root_to_index: bool
+
+
+class FlaskRenderServer(AbstractRenderServer[Flask, Config]):
+    """Base class for render servers which use Flask"""
+
+    _dispatcher_type: AbstractDispatcher
+
+    def _create_config(self, config: Optional[Config]) -> Config:
+        return Config(
+            {
+                "import_name": __name__,
+                "url_prefix": "",
+                "cors": False,
+                "serve_static_files": True,
+                "redirect_root_to_index": True,
+                **(config or {}),
+            }
+        )
+
+    def _default_application(self, config: Config) -> Flask:
+        return Flask(config["import_name"])
+
+    def _setup_application(self, app: Flask, config: Config) -> None:
+        bp = Blueprint("idom", __name__, url_prefix=config["url_prefix"])
+
+        self._setup_blueprint_routes(bp, config)
+
+        cors_config = config["cors"]
+        if cors_config:
+            cors_params = cors_config if isinstance(cors_config, dict) else {}
+            CORS(bp, **cors_params)
+
+        app.register_blueprint(bp)
+
+        sockets = Sockets(app)
+
+        @sockets.route(urljoin(config["url_prefix"], "/stream"))
+        def model_stream(ws: WebSocket) -> None:
+            def send(value: Any) -> None:
+                ws.send(json.dumps(value))
+
+            def recv() -> LayoutEvent:
+                event = ws.receive()
+                if event is not None:
+                    return LayoutEvent(**json.loads(event))
+                else:
+                    return None
+
+            run_dispatcher_in_thread(
+                lambda: self._dispatcher_type(Layout(self._root_element_constructor())),
+                send,
+                recv,
+                None,
+            )
+
+    def _setup_blueprint_routes(self, blueprint: Blueprint, config: Config) -> None:
+        if config["serve_static_files"]:
+
+            @blueprint.route("/client/<path:path>")
+            def send_build_dir(path):
+                return send_from_directory(str(BUILD_DIR), path)
+
+            if config["redirect_root_to_index"]:
+
+                @blueprint.route("/")
+                def redirect_to_index():
+                    return redirect(url_for("idom.send_build_dir", path="index.html"))
+
+    def _setup_application_did_start_event(
+        self, app: Flask, event: ThreadEvent
+    ) -> None:
+        @app.before_first_request
+        def server_did_start():
+            event.set()
+
+    def _run_application(
+        self, app: Flask, config: Config, args: Tuple[Any, ...], kwargs: Dict[str, Any]
+    ) -> None:
+        self._generic_run_application(app, *args, **kwargs)
+
+    def _run_application_in_thread(
+        self, app: Flask, config: Config, args: Tuple[Any, ...], kwargs: Dict[str, Any]
+    ) -> None:
+        self._generic_run_application(app, *args, **kwargs)
+
+    def _generic_run_application(
+        self,
+        app: Flask,
+        host: str = "",
+        port: int = 5000,
+        debug: bool = False,
+        *args: Any,
+        **kwargs
+    ):
+        if debug:
+            logging.basicConfig(level=logging.DEBUG)
+        logging.debug("Starting server...")
+        _StartCallbackWSGIServer(
+            self._server_did_start.set,
+            (host, port),
+            app,
+            *args,
+            handler_class=WebSocketHandler,
+            **kwargs,
+        ).serve_forever()
+
+
+class PerClientStateServer(FlaskRenderServer):
+    _dispatcher_type = SingleViewDispatcher
+
+
+def run_dispatcher_in_thread(
+    make_dispatcher: Callable[[], AbstractDispatcher],
+    send: Callable[[Any], None],
+    recv: Callable[[], Optional[LayoutEvent]],
+    context: Optional[Any],
+) -> None:
+    dispatch_thread_info_created = ThreadEvent()
+    dispatch_thread_info_ref: idom.Ref[Optional[_DispatcherThreadInfo]] = idom.Ref(None)
+
+    def run_dispatcher():
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        thread_send_queue = ThreadQueue()
+        async_recv_queue = AsyncQueue()
+
+        async def send_coro(value: Any) -> None:
+            thread_send_queue.put(value)
+
+        async def recv_coro() -> Any:
+            return await async_recv_queue.get()
+
+        async def main():
+            async with make_dispatcher() as dispatcher:
+                await dispatcher.run(send_coro, recv_coro, context)
+
+        main_future = asyncio.ensure_future(main())
+
+        dispatch_thread_info_ref.current = _DispatcherThreadInfo(
+            dispatch_loop=loop,
+            dispatch_future=main_future,
+            thread_send_queue=thread_send_queue,
+            async_recv_queue=async_recv_queue,
+        )
+        dispatch_thread_info_created.set()
+
+        loop.run_until_complete(main_future)
+
+    Thread(target=run_dispatcher, daemon=True).start()
+    dispatch_thread_info_created.wait()
+
+    dispatch_thread_info = dispatch_thread_info_ref.current
+    assert dispatch_thread_info is not None
+
+    stop = ThreadEvent()
+
+    def run_send():
+        while not stop.is_set():
+            send(dispatch_thread_info.thread_send_queue.get())
+
+    Thread(target=run_send, daemon=True).start()
+
+    try:
+        while True:
+            value = recv()
+            if value is None:
+                stop.set()
+                break
+            dispatch_thread_info.dispatch_loop.call_soon_threadsafe(
+                dispatch_thread_info.async_recv_queue.put_nowait, value
+            )
+    finally:
+        dispatch_thread_info.dispatch_loop.call_soon_threadsafe(
+            dispatch_thread_info.dispatch_future.cancel
+        )
+
+    return None
+
+
+class _DispatcherThreadInfo(NamedTuple):
+    dispatch_loop: asyncio.AbstractEventLoop
+    dispatch_future: asyncio.Future
+    thread_send_queue: ThreadQueue
+    async_recv_queue: AsyncQueue
+
+
+class _StartCallbackWSGIServer(pywsgi.WSGIServer):
+    def __init__(self, before_first_request: Callable[[], None], *args, **kwargs):
+        self._before_first_request_callback = before_first_request
+        super().__init__(*args, **kwargs)
+
+    def init_socket(self):
+        self._before_first_request_callback()
+        return super().init_socket()