|
1 |
| -import abc |
2 |
| -import asyncio |
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import sys |
| 4 | +from asyncio import Future |
| 5 | +from asyncio import Queue as AsyncQueue |
| 6 | +from asyncio.tasks import ensure_future |
3 | 7 | from logging import getLogger
|
4 |
| -from typing import Any, AsyncIterator, Awaitable, Callable, Dict |
| 8 | +from typing import Any, AsyncIterator, Awaitable, Callable, Tuple |
| 9 | +from weakref import WeakSet |
5 | 10 |
|
6 | 11 | from anyio import create_task_group
|
7 |
| -from anyio.abc import TaskGroup |
| 12 | + |
| 13 | +from idom.utils import Ref |
8 | 14 |
|
9 | 15 | from .layout import Layout, LayoutEvent, LayoutUpdate
|
10 |
| -from .utils import HasAsyncResources, async_resource |
| 16 | + |
| 17 | + |
| 18 | +if sys.version_info >= (3, 7): # pragma: no cover |
| 19 | + from contextlib import asynccontextmanager # noqa |
| 20 | +else: # pragma: no cover |
| 21 | + from async_generator import asynccontextmanager |
11 | 22 |
|
12 | 23 |
|
13 | 24 | logger = getLogger(__name__)
|
|
16 | 27 | RecvCoroutine = Callable[[], Awaitable[LayoutEvent]]
|
17 | 28 |
|
18 | 29 |
|
19 |
| -class AbstractDispatcher(HasAsyncResources, abc.ABC): |
20 |
| - """A base class for implementing :class:`~idom.core.layout.Layout` dispatchers.""" |
| 30 | +async def dispatch_single_view( |
| 31 | + layout: Layout, send: SendCoroutine, recv: RecvCoroutine |
| 32 | +) -> None: |
| 33 | + with layout: |
| 34 | + async with create_task_group() as task_group: |
| 35 | + task_group.start_soon(_single_outgoing_loop, layout, send) |
| 36 | + task_group.start_soon(_single_incoming_loop, layout, recv) |
21 | 37 |
|
22 |
| - __slots__ = "_layout" |
23 | 38 |
|
24 |
| - def __init__(self, layout: Layout) -> None: |
25 |
| - super().__init__() |
26 |
| - self._layout = layout |
| 39 | +_SharedDispatchCoro = Callable[[SendCoroutine, RecvCoroutine], Awaitable[None]] |
27 | 40 |
|
28 |
| - async def start(self) -> None: |
29 |
| - await self.__aenter__() |
30 | 41 |
|
31 |
| - async def stop(self) -> None: |
32 |
| - await self.task_group.cancel_scope.cancel() |
33 |
| - await self.__aexit__(None, None, None) |
| 42 | +@asynccontextmanager |
| 43 | +async def create_shared_view_dispatcher( |
| 44 | + layout: Layout, |
| 45 | +) -> AsyncIterator[_SharedDispatchCoro]: |
| 46 | + model_state: Ref[Any] = Ref({}) |
| 47 | + update_queues: WeakSet[AsyncQueue[LayoutUpdate]] = WeakSet() |
34 | 48 |
|
35 |
| - @async_resource |
36 |
| - async def layout(self) -> AsyncIterator[Layout]: |
37 |
| - async with self._layout as layout: |
38 |
| - yield layout |
39 |
| - |
40 |
| - @async_resource |
41 |
| - async def task_group(self) -> AsyncIterator[TaskGroup]: |
42 |
| - async with create_task_group() as group: |
43 |
| - yield group |
44 |
| - |
45 |
| - async def run(self, send: SendCoroutine, recv: RecvCoroutine, context: Any) -> None: |
46 |
| - """Start an unending loop which will drive the layout. |
47 |
| -
|
48 |
| - This will call :meth:`AbstractLayout.render` and :meth:`Layout.dispatch` |
49 |
| - to render new models and execute events respectively. |
50 |
| - """ |
51 |
| - await self.task_group.spawn(self._outgoing_loop, send, context) |
52 |
| - await self.task_group.spawn(self._incoming_loop, recv, context) |
| 49 | + async def dispatch_shared_view( |
| 50 | + send: SendCoroutine, |
| 51 | + recv: RecvCoroutine, |
| 52 | + ) -> None: |
| 53 | + queue = AsyncQueue() |
| 54 | + update_queues.add(queue) |
| 55 | + async with create_task_group() as inner_task_group: |
| 56 | + await send(LayoutUpdate.create_from({}, model_state.current)) |
| 57 | + inner_task_group.start_soon(_single_incoming_loop, layout, recv) |
| 58 | + inner_task_group.start_soon(_shared_outgoing_loop, send, queue) |
53 | 59 | return None
|
54 | 60 |
|
55 |
| - async def _outgoing_loop(self, send: SendCoroutine, context: Any) -> None: |
56 |
| - try: |
57 |
| - while True: |
58 |
| - await send(await self._outgoing(self.layout, context)) |
59 |
| - except Exception: |
60 |
| - logger.info("Failed to send outgoing update", exc_info=True) |
61 |
| - raise |
| 61 | + with layout: |
| 62 | + async with create_task_group() as task_group: |
| 63 | + task_group.start_soon( |
| 64 | + _shared_render_loop, |
| 65 | + layout, |
| 66 | + model_state, |
| 67 | + update_queues, |
| 68 | + ) |
| 69 | + yield dispatch_shared_view |
62 | 70 |
|
63 |
| - async def _incoming_loop(self, recv: RecvCoroutine, context: Any) -> None: |
64 |
| - try: |
65 |
| - while True: |
66 |
| - await self._incoming(self.layout, context, await recv()) |
67 |
| - except Exception: |
68 |
| - logger.info("Failed to receive incoming event", exc_info=True) |
69 |
| - raise |
70 | 71 |
|
71 |
| - @abc.abstractmethod |
72 |
| - async def _outgoing(self, layout: Layout, context: Any) -> Any: |
73 |
| - ... |
| 72 | +async def create_shared_view_dispatcher_future( |
| 73 | + layout: Layout, |
| 74 | +) -> Tuple[Future, _SharedDispatchCoro]: |
| 75 | + queue: AsyncQueue[_SharedDispatchCoro] = AsyncQueue() |
74 | 76 |
|
75 |
| - @abc.abstractmethod |
76 |
| - async def _incoming(self, layout: Layout, context: Any, message: Any) -> None: |
77 |
| - ... |
| 77 | + async def task(): |
| 78 | + async with create_shared_view_dispatcher(layout) as dispatch: |
| 79 | + queue.put_nowait(dispatch) |
78 | 80 |
|
| 81 | + return ensure_future(task()), (await queue.get()) |
79 | 82 |
|
80 |
| -class SingleViewDispatcher(AbstractDispatcher): |
81 |
| - """Each client of the dispatcher will get its own model. |
82 | 83 |
|
83 |
| - ..note:: |
84 |
| - The ``context`` parameter of :meth:`SingleViewDispatcher.run` should just |
85 |
| - be ``None`` since it's not used. |
86 |
| - """ |
| 84 | +async def _single_outgoing_loop(layout: Layout, send: SendCoroutine) -> None: |
| 85 | + while True: |
| 86 | + await send(await layout.render()) |
87 | 87 |
|
88 |
| - __slots__ = "_current_model_as_json" |
89 | 88 |
|
90 |
| - def __init__(self, layout: Layout) -> None: |
91 |
| - super().__init__(layout) |
92 |
| - self._current_model_as_json = "" |
| 89 | +async def _single_incoming_loop(layout: Layout, recv: RecvCoroutine) -> None: |
| 90 | + while True: |
| 91 | + await layout.dispatch(await recv()) |
93 | 92 |
|
94 |
| - async def _outgoing(self, layout: Layout, context: Any) -> LayoutUpdate: |
95 |
| - return await layout.render() |
96 | 93 |
|
97 |
| - async def _incoming(self, layout: Layout, context: Any, event: LayoutEvent) -> None: |
98 |
| - await layout.dispatch(event) |
99 |
| - return None |
100 |
| - |
101 |
| - |
102 |
| -class SharedViewDispatcher(SingleViewDispatcher): |
103 |
| - """Each client of the dispatcher shares the same model. |
104 |
| -
|
105 |
| - The client's ID is indicated by the ``context`` argument of |
106 |
| - :meth:`SharedViewDispatcher.run` |
107 |
| - """ |
| 94 | +async def _shared_render_loop( |
| 95 | + layout: Layout, |
| 96 | + model_state: Ref[Any], |
| 97 | + update_queues: WeakSet[AsyncQueue[LayoutUpdate]], |
| 98 | +) -> None: |
| 99 | + while True: |
| 100 | + update = await layout.render() |
| 101 | + model_state.current = update.apply_to(model_state.current) |
| 102 | + # append updates to all other contexts |
| 103 | + for queue in update_queues: |
| 104 | + await queue.put(update) |
108 | 105 |
|
109 |
| - __slots__ = "_update_queues", "_model_state" |
110 | 106 |
|
111 |
| - def __init__(self, layout: Layout) -> None: |
112 |
| - super().__init__(layout) |
113 |
| - self._model_state: Any = {} |
114 |
| - self._update_queues: Dict[str, asyncio.Queue[LayoutUpdate]] = {} |
115 |
| - |
116 |
| - @async_resource |
117 |
| - async def task_group(self) -> AsyncIterator[TaskGroup]: |
118 |
| - async with create_task_group() as group: |
119 |
| - await group.spawn(self._render_loop) |
120 |
| - yield group |
121 |
| - |
122 |
| - async def run( |
123 |
| - self, send: SendCoroutine, recv: RecvCoroutine, context: str, join: bool = False |
124 |
| - ) -> None: |
125 |
| - await super().run(send, recv, context) |
126 |
| - if join: |
127 |
| - await self._join_event.wait() |
128 |
| - |
129 |
| - async def _render_loop(self) -> None: |
130 |
| - while True: |
131 |
| - update = await super()._outgoing(self.layout, None) |
132 |
| - self._model_state = update.apply_to(self._model_state) |
133 |
| - # append updates to all other contexts |
134 |
| - for queue in self._update_queues.values(): |
135 |
| - await queue.put(update) |
136 |
| - |
137 |
| - async def _outgoing_loop(self, send: SendCoroutine, context: Any) -> None: |
138 |
| - self._update_queues[context] = asyncio.Queue() |
139 |
| - await send(LayoutUpdate.create_from({}, self._model_state)) |
140 |
| - await super()._outgoing_loop(send, context) |
141 |
| - |
142 |
| - async def _outgoing(self, layout: Layout, context: str) -> LayoutUpdate: |
143 |
| - return await self._update_queues[context].get() |
144 |
| - |
145 |
| - @async_resource |
146 |
| - async def _join_event(self) -> AsyncIterator[asyncio.Event]: |
147 |
| - event = asyncio.Event() |
148 |
| - try: |
149 |
| - yield event |
150 |
| - finally: |
151 |
| - event.set() |
| 107 | +async def _shared_outgoing_loop( |
| 108 | + send: SendCoroutine, queue: AsyncQueue[LayoutUpdate] |
| 109 | +) -> None: |
| 110 | + while True: |
| 111 | + await send(await queue.get()) |
0 commit comments