messenger pattern

src/py/reactpy/reactpy/backend/hooks.py

@@ -3,12 +3,15 @@
 from collections.abc import MutableMapping
 from typing import Any
 
+from reactpy.backend.messenger import Messenger
 from reactpy.backend.types import Connection, Location
 from reactpy.core.hooks import Context, create_context, use_context
 
 # backend implementations should establish this context at the root of an app
 ConnectionContext: Context[Connection[Any] | None] = create_context(None)
 
+MessengerContext: Context[Messenger | None] = create_context(None)
+
 
 def use_connection() -> Connection[Any]:
     """Get the current :class:`~reactpy.backend.types.Connection`."""
@@ -27,3 +30,12 @@ def use_scope() -> MutableMapping[str, Any]:
 def use_location() -> Location:
     """Get the current :class:`~reactpy.backend.types.Connection`'s location."""
     return use_connection().location
+
+
+def use_messenger() -> Messenger:
+    """Get the current :class:`~reactpy.core.serve.Messenger`."""
+    messenger = use_context(MessengerContext)
+    if messenger is None:  # nocov
+        msg = "No backend established a messenger."
+        raise RuntimeError(msg)
+    return messenger

src/py/reactpy/reactpy/backend/messenger.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Awaitable
+from typing import Callable
+
+from anyio import Event, create_memory_object_stream, create_task_group
+from anyio.abc import ObjectReceiveStream, ObjectSendStream
+
+from reactpy.core.types import Message
+
+_MessageStream = tuple[ObjectSendStream[Message], ObjectReceiveStream[Message]]
+
+
+class Messenger:
+    """A messenger for sending and receiving messages"""
+
+    def __init__(self) -> None:
+        self._task_group = create_task_group()
+        self._streams: dict[str, list[_MessageStream]] = {}
+        self._recv_began: dict[str, Event] = {}
+
+    def start_producer(self, producer: Callable[[], AsyncIterator[Message]]) -> None:
+        """Add a message producer"""
+
+        async def _producer() -> None:
+            async for message in producer():
+                await self.send(message)
+
+        self._task_group.start_soon(_producer)
+
+    def start_consumer(
+        self, message_type: str, consumer: Callable[[Message], Awaitable[None]]
+    ) -> None:
+        """Add a message consumer"""
+
+        async def _consumer() -> None:
+            async for message in self.receive(message_type):
+                self._task_group.start_soon(consumer, message)
+
+        self._task_group.start_soon(_consumer)
+
+    async def send(self, message: Message) -> None:
+        """Send a message to all consumers of the message type"""
+        for send, _ in self._streams.get(message["type"], []):
+            await send.send(message)
+
+    async def receive(self, message_type: str) -> AsyncIterator[Message]:
+        """Receive messages of a given type"""
+        send, recv = create_memory_object_stream()
+        self._streams.setdefault(message_type, []).append((send, recv))
+        async with recv:
+            async with send:
+                async for message in recv:
+                    yield message
+
+    async def __aenter__(self) -> Messenger:
+        await self._task_group.__aenter__()
+        return self
+
+    async def __aexit__(self, *args) -> None:
+        await self._task_group.__aexit__(*args)

src/py/reactpy/reactpy/backend/starlette.py

@@ -3,10 +3,12 @@
 import asyncio
 import json
 import logging
-from collections.abc import Awaitable
+from collections.abc import AsyncIterator, Awaitable
 from dataclasses import dataclass
 from typing import Any, Callable
 
+from py.reactpy.reactpy.backend.hooks import MessengerContext
+from py.reactpy.reactpy.core.serve import Messenger
 from starlette.applications import Starlette
 from starlette.middleware.cors import CORSMiddleware
 from starlette.requests import Request
@@ -141,6 +143,25 @@ async def model_stream(socket: WebSocket) -> None:
         pathname = pathname[len(options.url_prefix) :] or "/"
         search = socket.scope["query_string"].decode()
 
+        async with Messenger() as msgr:
+            async with Layout(
+                MessengerContext(
+                    ConnectionContext(
+                        component(),
+                        value=Connection(
+                            scope=socket.scope,
+                            location=Location(pathname, f"?{search}" if search else ""),
+                            carrier=socket,
+                        ),
+                    ),
+                    value=msgr,
+                )
+            ) as layout:
+                msgr.start_consumer("layout-event", layout.deliver)
+                msgr.start_producer(layout.renders)
+                msgr.start_consumer("layout-update", send)
+                msgr.start_producer(recv)
+
         try:
             await serve_layout(
                 Layout(
@@ -166,7 +187,8 @@ def _make_send_recv_callbacks(
     async def sock_send(value: Any) -> None:
         await socket.send_text(json.dumps(value))
 
-    async def sock_recv() -> Any:
-        return json.loads(await socket.receive_text())
+    async def sock_recv() -> AsyncIterator[Any]:
+        while True:
+            yield json.loads(await socket.receive_text())
 
     return sock_send, sock_recv

src/py/reactpy/reactpy/core/events.py

@@ -2,7 +2,7 @@
 
 import asyncio
 from collections.abc import Sequence
-from typing import Any, Callable, Literal, overload
+from typing import Any, Callable, Literal, Optional, overload
 
 from anyio import create_task_group
 

src/py/reactpy/reactpy/core/layout.py

@@ -3,7 +3,7 @@
 import abc
 import asyncio
 from collections import Counter
-from collections.abc import Iterator
+from collections.abc import AsyncIterator, Iterator
 from contextlib import ExitStack
 from logging import getLogger
 from typing import (
@@ -99,6 +99,11 @@ async def deliver(self, event: LayoutEventMessage) -> None:
                 "does not exist or its component unmounted"
             )
 
+    async def renders(self) -> AsyncIterator[LayoutUpdateMessage]:
+        """Yield all available renders"""
+        while True:
+            yield await self.render()
+
     async def render(self) -> LayoutUpdateMessage:
         """Await the next available render. This will block until a component is updated"""
         while True:

src/py/reactpy/reactpy/core/serve.py

@@ -1,14 +1,20 @@
 from __future__ import annotations
 
-from collections.abc import Awaitable
+import warnings
+from collections.abc import AsyncIterator, Awaitable
 from logging import getLogger
 from typing import Callable
 
-from anyio import create_task_group
-from anyio.abc import TaskGroup
+from anyio import Event, create_memory_object_stream, create_task_group
+from anyio.abc import ObjectReceiveStream, ObjectSendStream, TaskGroup
 
 from reactpy.config import REACTPY_DEBUG_MODE
-from reactpy.core.types import LayoutEventMessage, LayoutType, LayoutUpdateMessage
+from reactpy.core.types import (
+    LayoutEventMessage,
+    LayoutType,
+    LayoutUpdateMessage,
+    Message,
+)
 
 logger = getLogger(__name__)
 
@@ -37,6 +43,11 @@ async def serve_layout(
     recv: RecvCoroutine,
 ) -> None:
     """Run a dispatch loop for a single view instance"""
+    warnings.warn(
+        "serve_layout is deprecated. Use a Messenger object instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
     async with layout:
         try:
             async with create_task_group() as task_group:

src/py/reactpy/reactpy/core/types.py

@@ -2,7 +2,7 @@
 
 import sys
 from collections import namedtuple
-from collections.abc import Mapping, Sequence
+from collections.abc import AsyncIterator, Mapping, Sequence
 from types import TracebackType
 from typing import (
     TYPE_CHECKING,
@@ -73,6 +73,9 @@ class LayoutType(Protocol[_Render, _Event]):
     async def render(self) -> _Render:
         """Render an update to a view"""
 
+    async def renders(self) -> AsyncIterator[_Render]:
+        """Render a series of updates to a view"""
+
     async def deliver(self, event: _Event) -> None:
         """Relay an event to its respective handler"""
 
@@ -213,7 +216,14 @@ def __call__(
         ...
 
 
-class LayoutUpdateMessage(TypedDict):
+class Message(TypedDict):
+    """Base class for all messages"""
+
+    type: str
+    """The type of message"""
+
+
+class LayoutUpdateMessage(Message):
     """A message describing an update to a layout"""
 
     type: Literal["layout-update"]
@@ -224,7 +234,7 @@ class LayoutUpdateMessage(TypedDict):
     """The model to assign at the given JSON Pointer path"""
 
 
-class LayoutEventMessage(TypedDict):
+class LayoutEventMessage(Message):
     """Message describing an event originating from an element in the layout"""
 
     type: Literal["layout-event"]