refactor dispatchers

Simplify them greatly by making them normal functions.
This avoids a lot of indirection and complexity caused
by inheritance.

In addition this fixes some impropper usages of TaskGroups
which was causing problems in the SharedClientStateServer
implementations.

This refactor also allowed us to get rid of the clunky
HasAsyncResources util class. In the end it just added
confusion.

Author: rmorshea

docs/source/core-concepts.rst

@@ -63,7 +63,7 @@ ever be removed from the model. Then you'll just need to call and await a
 
 .. testcode::
 
-    async with idom.Layout(ClickCount()) as layout:
+    with idom.Layout(ClickCount()) as layout:
         patch = await layout.render()
 
 The layout also handles the triggering of event handlers. Normally these are
@@ -88,7 +88,7 @@ which we can re-render and see what changed:
 
         return idom.html.button({"onClick": handler}, [f"Click count: {count}"])
 
-    async with idom.Layout(ClickCount()) as layout:
+    with idom.Layout(ClickCount()) as layout:
         patch_1 = await layout.render()
 
         fake_event = LayoutEvent(target=static_handler.target, data=[{}])
@@ -111,20 +111,20 @@ which we can re-render and see what changed:
 Layout Dispatcher
 -----------------
 
-An :class:`~idom.core.dispatcher.AbstractDispatcher` implementation is a relatively thin layer
-of logic around a :class:`~idom.core.layout.Layout` which drives the triggering of
-events and layout updates by scheduling an asynchronous loop that will run forever -
-effectively animating the model. To execute the loop, the dispatcher's
-:meth:`~idom.core.dispatcher.AbstractDispatcher.run` method accepts two callbacks. One is a
-"send" callback to which the dispatcher passes updates, while the other is "receive"
-callback that's called by the dispatcher to events it should execute.
+A "dispatcher" implementation is a relatively thin layer of logic around a
+:class:`~idom.core.layout.Layout` which drives the triggering of events and updates by
+scheduling an asynchronous loop that will run forever - effectively animating the model.
+The simplest dispatcher is :func:`~idom.core.dispatcher.dispatch_single_view` which
+accepts three arguments. The first is a :class:`~idom.core.layout.Layout`, the second is
+a "send" callback to which the dispatcher passes updates, and the third is a "receive"
+callback that's called by the dispatcher to collect events it should execute.
 
 .. testcode::
 
     import asyncio
 
-    from idom.core import SingleViewDispatcher, EventHandler
     from idom.core.layout import LayoutEvent
+    from idom.core.dispatch import dispatch_single_view
 
 
     sent_patches = []
@@ -148,20 +148,14 @@ callback that's called by the dispatcher to events it should execute.
         return event
 
 
-    async with SingleViewDispatcher(idom.Layout(ClickCount())) as dispatcher:
-        context = None  # see note below
-        await dispatcher.run(send, recv, context)
-
+    await dispatch_single_view(idom.Layout(ClickCount()), send, recv)
     assert len(sent_patches) == 5
 
 
 .. note::
 
-    ``context`` is information that's specific to the
-    :class:`~idom.core.dispatcher.AbstractDispatcher` implementation. In the case of
-    the :class:`~idom.core.dispatcher.SingleViewDispatcher` it doesn't require any
-    context. On the other hand the :class:`~idom.core.dispatcher.SharedViewDispatcher`
-    requires a client ID as its piece of contextual information.
+    The :func:`~idom.core.dispatcher.create_shared_view_dispatcher`, while more complex
+    in its usage, allows multiple clients to share one synchronized view.
 
 
 Layout Server

requirements/pkg-deps.txt

@@ -1,7 +1,6 @@
 typing-extensions >=3.7.4
 mypy-extensions >=0.4.3
 anyio >=2.0
-async_generator >=1.10; python_version<"3.7"
 async_exit_stack >=1.0.1; python_version<"3.7"
 jsonpatch >=1.26
 typer >=0.3.2

src/idom/core/__init__.py

@@ -1,23 +0,0 @@
-from .component import AbstractComponent, Component, ComponentConstructor, component
-from .dispatcher import AbstractDispatcher, SharedViewDispatcher, SingleViewDispatcher
-from .events import EventHandler, Events, event
-from .layout import Layout
-from .vdom import vdom
-
-
-__all__ = [
-    "AbstractComponent",
-    "Layout",
-    "AbstractDispatcher",
-    "component",
-    "Component",
-    "EventHandler",
-    "ComponentConstructor",
-    "event",
-    "Events",
-    "hooks",
-    "Layout",
-    "vdom",
-    "SharedViewDispatcher",
-    "SingleViewDispatcher",
-]

src/idom/core/dispatcher.py

@@ -1,13 +1,23 @@
-import abc
-import asyncio
+from __future__ import annotations
+
+import sys
+from asyncio import Future, Queue
+from asyncio.tasks import FIRST_COMPLETED, ensure_future, gather, wait
 from logging import getLogger
-from typing import Any, AsyncIterator, Awaitable, Callable, Dict
+from typing import Any, AsyncIterator, Awaitable, Callable, List, Sequence, Tuple
+from weakref import WeakSet
 
 from anyio import create_task_group
-from anyio.abc import TaskGroup
+
+from idom.utils import Ref
 
 from .layout import Layout, LayoutEvent, LayoutUpdate
-from .utils import HasAsyncResources, async_resource
+
+
+if sys.version_info >= (3, 7):  # pragma: no cover
+    from contextlib import asynccontextmanager  # noqa
+else:  # pragma: no cover
+    from async_generator import asynccontextmanager
 
 
 logger = getLogger(__name__)
@@ -16,136 +26,139 @@
 RecvCoroutine = Callable[[], Awaitable[LayoutEvent]]
 
 
-class AbstractDispatcher(HasAsyncResources, abc.ABC):
-    """A base class for implementing :class:`~idom.core.layout.Layout` dispatchers."""
+async def dispatch_single_view(
+    layout: Layout,
+    send: SendCoroutine,
+    recv: RecvCoroutine,
+) -> None:
+    with layout:
+        async with create_task_group() as task_group:
+            task_group.start_soon(_single_outgoing_loop, layout, send)
+            task_group.start_soon(_single_incoming_loop, layout, recv)
 
-    __slots__ = "_layout"
 
-    def __init__(self, layout: Layout) -> None:
-        super().__init__()
-        self._layout = layout
+_SharedDispatchFuture = Callable[[SendCoroutine, RecvCoroutine], Future]
 
-    async def start(self) -> None:
-        await self.__aenter__()
 
-    async def stop(self) -> None:
-        await self.task_group.cancel_scope.cancel()
-        await self.__aexit__(None, None, None)
+@asynccontextmanager
+async def create_shared_view_dispatcher(
+    layout: Layout, run_forever: bool = False
+) -> AsyncIterator[_SharedDispatchFuture]:
+    with layout:
+        (
+            dispatch_shared_view,
+            model_state,
+            all_update_queues,
+        ) = await _make_shared_view_dispatcher(layout)
 
-    @async_resource
-    async def layout(self) -> AsyncIterator[Layout]:
-        async with self._layout as layout:
-            yield layout
+        dispatch_tasks: List[Future] = []
 
-    @async_resource
-    async def task_group(self) -> AsyncIterator[TaskGroup]:
-        async with create_task_group() as group:
-            yield group
+        def dispatch_shared_view_soon(
+            send: SendCoroutine, recv: RecvCoroutine
+        ) -> Future:
+            future = ensure_future(dispatch_shared_view(send, recv))
+            dispatch_tasks.append(future)
+            return future
 
-    async def run(self, send: SendCoroutine, recv: RecvCoroutine, context: Any) -> None:
-        """Start an unending loop which will drive the layout.
+        yield dispatch_shared_view_soon
 
-        This will call :meth:`AbstractLayout.render` and :meth:`Layout.dispatch`
-        to render new models and execute events respectively.
-        """
-        await self.task_group.spawn(self._outgoing_loop, send, context)
-        await self.task_group.spawn(self._incoming_loop, recv, context)
-        return None
+        gathered_dispatch_tasks = gather(*dispatch_tasks, return_exceptions=True)
 
-    async def _outgoing_loop(self, send: SendCoroutine, context: Any) -> None:
-        try:
-            while True:
-                await send(await self._outgoing(self.layout, context))
-        except Exception:
-            logger.info("Failed to send outgoing update", exc_info=True)
-            raise
+        while True:
+            (
+                update_future,
+                dispatchers_completed_future,
+            ) = await _wait_until_first_complete(
+                layout.render(),
+                gathered_dispatch_tasks,
+            )
+
+            if dispatchers_completed_future.done():
+                update_future.cancel()
+                break
+            else:
+                update: LayoutUpdate = update_future.result()
+
+            model_state.current = update.apply_to(model_state.current)
+            # push updates to all dispatcher callbacks
+            for queue in all_update_queues:
+                queue.put_nowait(update)
+
+
+def ensure_shared_view_dispatcher_future(
+    layout: Layout,
+) -> Tuple[Future, _SharedDispatchFuture]:
+    dispatcher_future = Future()
+
+    async def dispatch_shared_view_forever():
+        with layout:
+            (
+                dispatch_shared_view,
+                model_state,
+                all_update_queues,
+            ) = await _make_shared_view_dispatcher(layout)
+
+            dispatcher_future.set_result(dispatch_shared_view)
 
-    async def _incoming_loop(self, recv: RecvCoroutine, context: Any) -> None:
-        try:
             while True:
-                await self._incoming(self.layout, context, await recv())
-        except Exception:
-            logger.info("Failed to receive incoming event", exc_info=True)
-            raise
-
-    @abc.abstractmethod
-    async def _outgoing(self, layout: Layout, context: Any) -> Any:
-        ...
+                update = await layout.render()
+                model_state.current = update.apply_to(model_state.current)
+                # push updates to all dispatcher callbacks
+                for queue in all_update_queues:
+                    queue.put_nowait(update)
 
-    @abc.abstractmethod
-    async def _incoming(self, layout: Layout, context: Any, message: Any) -> None:
-        ...
+    async def dispatch(send: SendCoroutine, recv: RecvCoroutine) -> None:
+        await (await dispatcher_future)(send, recv)
 
+    return ensure_future(dispatch_shared_view_forever()), dispatch
 
-class SingleViewDispatcher(AbstractDispatcher):
-    """Each client of the dispatcher will get its own model.
 
-    ..note::
-        The ``context`` parameter of :meth:`SingleViewDispatcher.run` should just
-        be ``None`` since it's not used.
-    """
+_SharedDispatchCoroutine = Callable[[SendCoroutine, RecvCoroutine], Awaitable[None]]
 
-    __slots__ = "_current_model_as_json"
 
-    def __init__(self, layout: Layout) -> None:
-        super().__init__(layout)
-        self._current_model_as_json = ""
+async def _make_shared_view_dispatcher(
+    layout: Layout,
+) -> Tuple[_SharedDispatchCoroutine, Ref[Any], WeakSet[Queue[LayoutUpdate]]]:
+    initial_update = await layout.render()
+    model_state = Ref(initial_update.apply_to({}))
 
-    async def _outgoing(self, layout: Layout, context: Any) -> LayoutUpdate:
-        return await layout.render()
+    # We push updates to queues instead of pushing directly to send() callbacks in
+    # order to isolate the render loop from any errors dispatch callbacks might
+    # raise.
+    all_update_queues: WeakSet[Queue[LayoutUpdate]] = WeakSet()
 
-    async def _incoming(self, layout: Layout, context: Any, event: LayoutEvent) -> None:
-        await layout.dispatch(event)
+    async def dispatch_shared_view(send: SendCoroutine, recv: RecvCoroutine) -> None:
+        update_queue: Queue[LayoutUpdate] = Queue()
+        async with create_task_group() as inner_task_group:
+            all_update_queues.add(update_queue)
+            await send(LayoutUpdate.create_from({}, model_state.current))
+            inner_task_group.start_soon(_single_incoming_loop, layout, recv)
+            inner_task_group.start_soon(_shared_outgoing_loop, send, update_queue)
         return None
 
+    return dispatch_shared_view, model_state, all_update_queues
 
-class SharedViewDispatcher(SingleViewDispatcher):
-    """Each client of the dispatcher shares the same model.
 
-    The client's ID is indicated by the ``context`` argument of
-    :meth:`SharedViewDispatcher.run`
-    """
+async def _single_outgoing_loop(layout: Layout, send: SendCoroutine) -> None:
+    while True:
+        await send(await layout.render())
 
-    __slots__ = "_update_queues", "_model_state"
 
-    def __init__(self, layout: Layout) -> None:
-        super().__init__(layout)
-        self._model_state: Any = {}
-        self._update_queues: Dict[str, asyncio.Queue[LayoutUpdate]] = {}
+async def _single_incoming_loop(layout: Layout, recv: RecvCoroutine) -> None:
+    while True:
+        await layout.dispatch(await recv())
 
-    @async_resource
-    async def task_group(self) -> AsyncIterator[TaskGroup]:
-        async with create_task_group() as group:
-            await group.spawn(self._render_loop)
-            yield group
 
-    async def run(
-        self, send: SendCoroutine, recv: RecvCoroutine, context: str, join: bool = False
-    ) -> None:
-        await super().run(send, recv, context)
-        if join:
-            await self._join_event.wait()
+async def _shared_outgoing_loop(
+    send: SendCoroutine, queue: Queue[LayoutUpdate]
+) -> None:
+    while True:
+        await send(await queue.get())
 
-    async def _render_loop(self) -> None:
-        while True:
-            update = await super()._outgoing(self.layout, None)
-            self._model_state = update.apply_to(self._model_state)
-            # append updates to all other contexts
-            for queue in self._update_queues.values():
-                await queue.put(update)
-
-    async def _outgoing_loop(self, send: SendCoroutine, context: Any) -> None:
-        self._update_queues[context] = asyncio.Queue()
-        await send(LayoutUpdate.create_from({}, self._model_state))
-        await super()._outgoing_loop(send, context)
-
-    async def _outgoing(self, layout: Layout, context: str) -> LayoutUpdate:
-        return await self._update_queues[context].get()
-
-    @async_resource
-    async def _join_event(self) -> AsyncIterator[asyncio.Event]:
-        event = asyncio.Event()
-        try:
-            yield event
-        finally:
-            event.set()
+
+async def _wait_until_first_complete(
+    *tasks: Awaitable[Any],
+) -> Sequence[Future]:
+    futures = [ensure_future(t) for t in tasks]
+    await wait(futures, return_when=FIRST_COMPLETED)
+    return futures

src/idom/core/events.py

@@ -192,7 +192,7 @@ async def __call__(self, data: List[Any]) -> Any:
         if self._coro_handlers:
             async with create_task_group() as group:
                 for handler in self._coro_handlers:
-                    await group.spawn(handler, *data)
+                    group.start_soon(handler, *data)
         for handler in self._func_handlers:
             handler(*data)
 

src/idom/core/hooks.py

@@ -387,8 +387,8 @@ class LifeCycleHook:
 
     def __init__(
         self,
-        component: AbstractComponent,
         layout: idom.core.layout.Layout,
+        component: AbstractComponent,
     ) -> None:
         self.component = component
         self._layout = weakref.ref(layout)