Add event queue (#1338)

Author: Archmonger

.github/workflows/check.yml

@@ -15,7 +15,8 @@ jobs:
         uses: ./.github/workflows/.hatch-run.yml
         with:
             job-name: "python-{0}"
-            run-cmd: "hatch test --parallel --cover"
+            # Retries needed because GitHub workers sometimes lag enough to crash parallel workers
+            run-cmd: "hatch test --parallel --cover --retries 10"
     lint-python:
         uses: ./.github/workflows/.hatch-run.yml
         with:
@@ -25,7 +26,7 @@ jobs:
         uses: ./.github/workflows/.hatch-run.yml
         with:
             job-name: "python-{0} {1}"
-            run-cmd: "hatch test --parallel"
+            run-cmd: "hatch test --parallel --retries 10"
             runs-on: '["ubuntu-latest", "macos-latest", "windows-latest"]'
             python-version: '["3.11", "3.12", "3.13", "3.14"]'
     test-documentation:

CHANGELOG.md

@@ -52,7 +52,7 @@ Don't forget to remove deprecated code on each major release!
 - Rewrite the `event-to-object` package to be more robust at handling properties on events.
 - Custom JS components will now automatically assume you are using ReactJS in the absence of a `bind` function.
 - Refactor layout rendering logic to improve readability and maintainability.
-- `@reactpy/client` now exports `React` and `ReactDOM`.
+- The JavaScript package `@reactpy/client` now exports `React` and `ReactDOM`, which allows third-party components to re-use the same React instance as ReactPy.
 - `reactpy.html` will now automatically flatten lists recursively (ex. `reactpy.html(["child1", ["child2"]])`)
 - `reactpy.utils.reactpy_to_string` will now retain the user's original casing for `data-*` and `aria-*` attributes.
 - `reactpy.utils.string_to_reactpy` has been upgraded to handle more complex scenarios without causing ReactJS rendering errors.

src/js/packages/@reactpy/client/src/bind.tsx

@@ -8,9 +8,8 @@ export async function infer_bind_from_environment() {
     const ReactDOM = await import("react-dom/client");
     return (node: HTMLElement) => reactjs_bind(node, React, ReactDOM);
   } catch {
-    console.error(
-      "Unknown error occurred: 'react' is missing within this ReactPy environment! \
-      Your JavaScript components may not work as expected!",
+    console.debug(
+      "ReactPy will render JavaScript components using internal bindings for 'react'.",
     );
     return (node: HTMLElement) => local_preact_bind(node);
   }

src/js/packages/@reactpy/client/src/client.ts

@@ -59,6 +59,7 @@ export class ReactPyClient
   urls: ReactPyUrls;
   socket: { current?: WebSocket };
   mountElement: HTMLElement;
+  private readonly messageQueue: any[] = [];
 
   constructor(props: GenericReactPyClientProps) {
     super();
@@ -69,12 +70,24 @@ export class ReactPyClient
       url: this.urls.componentUrl,
       readyPromise: this.ready,
       ...props.reconnectOptions,
+      onOpen: () => {
+        while (this.messageQueue.length > 0) {
+          this.sendMessage(this.messageQueue.shift());
+        }
+      },
       onMessage: async ({ data }) => this.handleIncoming(JSON.parse(data)),
     });
   }
 
   sendMessage(message: any): void {
-    this.socket.current?.send(JSON.stringify(message));
+    if (
+      this.socket.current &&
+      this.socket.current.readyState === WebSocket.OPEN
+    ) {
+      this.socket.current.send(JSON.stringify(message));
+    } else {
+      this.messageQueue.push(message);
+    }
   }
 
   loadModule(moduleName: string): Promise<ReactPyModule> {

src/reactpy/core/hooks.py

@@ -155,8 +155,6 @@ def decorator(func: _SyncEffectFunc) -> None:
             )
 
         async def effect(stop: asyncio.Event) -> None:
-            # Since the effect is asynchronous, we need to make sure we
-            # always clean up the previous effect's resources
             run_effect_cleanup(cleanup_func)
 
             # Execute the effect and store the clean-up function
@@ -219,17 +217,23 @@ def use_async_effect(
     dependencies = _try_to_infer_closure_values(function, dependencies)
     memoize = use_memo(dependencies=dependencies)
     cleanup_func: Ref[_EffectCleanFunc | None] = use_ref(None)
+    pending_task: Ref[asyncio.Task[_EffectCleanFunc | None] | None] = use_ref(None)
 
     def decorator(func: _AsyncEffectFunc) -> None:
         async def effect(stop: asyncio.Event) -> None:
-            # Since the effect is asynchronous, we need to make sure we
-            # always clean up the previous effect's resources
+            # Make sure we always clean up the previous effect's resources
+            if pending_task.current:
+                pending_task.current.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await pending_task.current
+
             run_effect_cleanup(cleanup_func)
 
             # Execute the effect and store the clean-up function.
             # We run this in a task so it can be cancelled if the stop signal
             # is set before the effect completes.
             task = asyncio.create_task(func())
+            pending_task.current = task
 
             # Wait for either the effect to complete or the stop signal
             stop_task = asyncio.create_task(stop.wait())
@@ -240,15 +244,17 @@ async def effect(stop: asyncio.Event) -> None:
 
             # If the effect completed first, store the cleanup function
             if task in done:
-                cleanup_func.current = task.result()
+                pending_task.current = None
+                with contextlib.suppress(asyncio.CancelledError):
+                    cleanup_func.current = task.result()
                 # Cancel the stop task since we don't need it anymore
                 stop_task.cancel()
                 with contextlib.suppress(asyncio.CancelledError):
                     await stop_task
                 # Now wait for the stop signal to run cleanup
                 await stop.wait()
+            # Stop signal came first - cancel the effect task
             else:
-                # Stop signal came first - cancel the effect task
                 task.cancel()
                 with contextlib.suppress(asyncio.CancelledError):
                     await task

src/reactpy/core/layout.py

@@ -8,6 +8,7 @@
     create_task,
     current_task,
     get_running_loop,
+    sleep,
     wait,
 )
 from collections import Counter
@@ -65,6 +66,8 @@ def __init__(self, root: Component | Context[Any] | ContextProvider[Any]) -> Non
     async def __aenter__(self) -> Layout:
         # create attributes here to avoid access before entering context manager
         self._event_handlers: EventHandlerDict = {}
+        self._event_queues: dict[str, Queue[LayoutEventMessage | dict[str, Any]]] = {}
+        self._event_processing_tasks: dict[str, Task[None]] = {}
         self._render_tasks: set[Task[LayoutUpdateMessage]] = set()
         self._render_tasks_by_id: dict[
             _LifeCycleStateId, Task[LayoutUpdateMessage]
@@ -88,10 +91,18 @@ async def __aexit__(
             t.cancel()
             with suppress(CancelledError):
                 await t
+
+        for t in self._event_processing_tasks.values():
+            t.cancel()
+            with suppress(CancelledError):
+                await t
+
         await self._unmount_model_states([root_model_state])
 
         # delete attributes here to avoid access after exiting context manager
         del self._event_handlers
+        del self._event_queues
+        del self._event_processing_tasks
         del self._rendering_queue
         del self._render_tasks_by_id
         del self._root_life_cycle_state_id
@@ -103,20 +114,51 @@ async def deliver(self, event: LayoutEventMessage | dict[str, Any]) -> None:
         # associated with a backend model that has been deleted. We only handle
         # events if the element and the handler exist in the backend. Otherwise
         # we just ignore the event.
-        handler = self._event_handlers.get(event["target"])
-
-        if handler is not None:
-            try:
-                data = [Event(d) if isinstance(d, dict) else d for d in event["data"]]
-                await handler.function(data)
-            except Exception:
-                logger.exception(f"Failed to execute event handler {handler}")
-        else:
-            logger.info(
-                f"Ignored event - handler {event['target']!r} "
-                "does not exist or its component unmounted"
+        target = event["target"]
+        if target not in self._event_queues:
+            self._event_queues[target] = cast(
+                "Queue[LayoutEventMessage | dict[str, Any]]", Queue()
+            )
+            self._event_processing_tasks[target] = create_task(
+                self._process_event_queue(target, self._event_queues[target])
             )
 
+        await self._event_queues[target].put(event)
+
+        # In test environments, we yield to the event loop to let the processing tasks run.
+        if REACTPY_DEBUG.current:
+            await sleep(0)
+
+    async def _process_event_queue(
+        self, target: str, queue: Queue[LayoutEventMessage | dict[str, Any]]
+    ) -> None:
+        while True:
+            event = await queue.get()
+
+            # Retry a few times to handle potential re-render race conditions where
+            # the handler is temporarily removed and then re-added.
+            handler = self._event_handlers.get(target)
+            if handler is None:
+                for _ in range(3):
+                    await sleep(0.01)
+                    handler = self._event_handlers.get(target)
+                    if handler is not None:
+                        break
+
+            if handler is not None:
+                try:
+                    data = [
+                        Event(d) if isinstance(d, dict) else d for d in event["data"]
+                    ]
+                    await handler.function(data)
+                except Exception:
+                    logger.exception(f"Failed to execute event handler {handler}")
+            else:
+                logger.info(
+                    f"Ignored event - handler {event['target']!r} "
+                    "does not exist or its component unmounted"
+                )
+
     async def render(self) -> LayoutUpdateMessage:
         if REACTPY_ASYNC_RENDERING.current:
             return await self._parallel_render()
@@ -346,10 +388,11 @@ def _render_model_attributes(
 
         model_event_handlers = new_state.model.current["eventHandlers"] = {}
         for event, handler in handlers_by_event.items():
-            if event in old_state.targets_by_event:
-                target = old_state.targets_by_event[event]
+            if handler.target is not None:
+                target = handler.target
             else:
-                target = uuid4().hex if handler.target is None else handler.target
+                target = f"{new_state.key_path}:{event}"
+
             new_state.targets_by_event[event] = target
             self._event_handlers[target] = handler
             model_event_handlers[event] = {
@@ -370,7 +413,11 @@ def _render_model_event_handlers_without_old_state(
 
         model_event_handlers = new_state.model.current["eventHandlers"] = {}
         for event, handler in handlers_by_event.items():
-            target = uuid4().hex if handler.target is None else handler.target
+            if handler.target is not None:
+                target = handler.target
+            else:
+                target = f"{new_state.key_path}:{event}"
+
             new_state.targets_by_event[event] = target
             self._event_handlers[target] = handler
             model_event_handlers[event] = {
@@ -485,6 +532,7 @@ def _new_root_model_state(
         children_by_key={},
         targets_by_event={},
         life_cycle_state=_make_life_cycle_state(component, schedule_render),
+        key_path="",
     )
 
 
@@ -504,6 +552,7 @@ def _make_component_model_state(
         children_by_key={},
         targets_by_event={},
         life_cycle_state=_make_life_cycle_state(component, schedule_render),
+        key_path=f"{parent.key_path}/{key}" if parent else "",
     )
 
 
@@ -523,6 +572,7 @@ def _copy_component_model_state(old_model_state: _ModelState) -> _ModelState:
         children_by_key={},
         targets_by_event={},
         life_cycle_state=old_model_state.life_cycle_state,
+        key_path=old_model_state.key_path,
     )
 
 
@@ -546,6 +596,7 @@ def _update_component_model_state(
             if old_model_state.is_component_state
             else _make_life_cycle_state(new_component, schedule_render)
         ),
+        key_path=f"{new_parent.key_path}/{old_model_state.key}",
     )
 
 
@@ -562,6 +613,7 @@ def _make_element_model_state(
         patch_path=f"{parent.patch_path}/children/{index}",
         children_by_key={},
         targets_by_event={},
+        key_path=f"{parent.key_path}/{key}",
     )
 
 
@@ -575,9 +627,10 @@ def _update_element_model_state(
         index=new_index,
         key=old_model_state.key,
         model=Ref(),  # does not copy the model
-        patch_path=old_model_state.patch_path,
+        patch_path=f"{new_parent.patch_path}/children/{new_index}",
         children_by_key={},
         targets_by_event={},
+        key_path=f"{new_parent.key_path}/{old_model_state.key}",
     )
 
 
@@ -591,6 +644,7 @@ class _ModelState:
         "children_by_key",
         "index",
         "key",
+        "key_path",
         "life_cycle_state",
         "model",
         "patch_path",
@@ -607,6 +661,7 @@ def __init__(
         children_by_key: dict[Key, _ModelState],
         targets_by_event: dict[str, str],
         life_cycle_state: _LifeCycleState | None = None,
+        key_path: str = "",
     ):
         self.index = index
         """The index of the element amongst its siblings"""
@@ -626,6 +681,9 @@ def __init__(
         self.targets_by_event = targets_by_event
         """The element's event handler target strings indexed by their event name"""
 
+        self.key_path = key_path
+        """A slash-delimited path using element keys"""
+
         # === Conditionally Available Attributes ===
         # It's easier to conditionally assign than to force a null check on every usage
 

tests/__init__.py

@@ -1,5 +0,0 @@
-import pytest
-
-from reactpy.testing import GITHUB_ACTIONS
-
-pytestmark = [pytest.mark.flaky(reruns=10 if GITHUB_ACTIONS else 1)]

tests/conftest.py

@@ -15,8 +15,6 @@
 )
 from reactpy.testing.display import _playwright_visible
 
-from . import pytestmark  # noqa: F401
-
 REACTPY_ASYNC_RENDERING.set_current(True)
 REACTPY_DEBUG.set_current(True)
 

tests/test_asgi/test_middleware.py

@@ -15,8 +15,6 @@
 from reactpy.executors.asgi.middleware import ReactPyMiddleware
 from reactpy.testing import BackendFixture, DisplayFixture
 
-from .. import pytestmark  # noqa: F401
-
 
 @pytest.fixture(scope="module")
 async def display(browser):

tests/test_asgi/test_pyscript.py

@@ -12,8 +12,6 @@
 from reactpy.executors.asgi.pyscript import ReactPyCsr
 from reactpy.testing import BackendFixture, DisplayFixture
 
-from .. import pytestmark  # noqa: F401
-
 
 @pytest.fixture(scope="module")
 async def display(browser):