Merge pull request #2272 from zauberzeug/outbox-delay

Keep updates and messages in outbox until client is connected

nicegui/air.py

@@ -206,16 +206,13 @@ class Air:
         return target_id in core.sio.manager.rooms
 
 
-instance: Optional[Air] = None
-
-
 def connect() -> None:
     """Connect to the NiceGUI On Air server if there is an air instance."""
-    if instance:
-        background_tasks.create(instance.connect())
+    if core.air:
+        background_tasks.create(core.air.connect())
 
 
 def disconnect() -> None:
     """Disconnect from the NiceGUI On Air server if there is an air instance."""
-    if instance:
-        background_tasks.create(instance.disconnect())
+    if core.air:
+        background_tasks.create(core.air.disconnect())

nicegui/client.py

@@ -13,12 +13,13 @@ from fastapi.responses import Response
 from fastapi.templating import Jinja2Templates
 from typing_extensions import Self
 
-from . import background_tasks, binding, core, helpers, json, outbox
+from . import background_tasks, binding, core, helpers, json
 from .awaitable_response import AwaitableResponse
 from .dependencies import generate_resources
 from .element import Element
 from .favicon import get_favicon_url
 from .logging import log
+from .outbox import Outbox
 from .version import __version__
 
 if TYPE_CHECKING:
@@ -57,6 +58,8 @@ class Client:
         self.on_air = False
         self._disconnect_task: Optional[asyncio.Task] = None
 
+        self.outbox = Outbox(self)
+
         with Element('q-layout', _client=self).props('view="hhh lpr fff"').classes('nicegui-layout') as self.layout:
             with Element('q-page-container') as self.page_container:
                 with Element('q-page'):
@@ -198,10 +201,10 @@ class Client:
         target_id = self._temporary_socket_id or self.id
 
         def send_and_forget():
-            outbox.enqueue_message('run_javascript', {'code': code}, target_id)
+            self.outbox.enqueue_message('run_javascript', {'code': code}, target_id)
 
         async def send_and_wait():
-            outbox.enqueue_message('run_javascript', {'code': code, 'request_id': request_id}, target_id)
+            self.outbox.enqueue_message('run_javascript', {'code': code, 'request_id': request_id}, target_id)
             deadline = time.time() + timeout
             while request_id not in self.waiting_javascript_commands:
                 if time.time() > deadline:
@@ -214,11 +217,11 @@ class Client:
     def open(self, target: Union[Callable[..., Any], str], new_tab: bool = False) -> None:
         """Open a new page in the client."""
         path = target if isinstance(target, str) else self.page_routes[target]
-        outbox.enqueue_message('open', {'path': path, 'new_tab': new_tab}, self.id)
+        self.outbox.enqueue_message('open', {'path': path, 'new_tab': new_tab}, self.id)
 
     def download(self, src: Union[str, bytes], filename: Optional[str] = None) -> None:
         """Download a file from a given URL or raw bytes."""
-        outbox.enqueue_message('download', {'src': src, 'filename': filename}, self.id)
+        self.outbox.enqueue_message('download', {'src': src, 'filename': filename}, self.id)
 
     def on_connect(self, handler: Union[Callable[..., Any], Awaitable]) -> None:
         """Register a callback to be called when the client connects."""
@@ -296,7 +299,7 @@ class Client:
         for element in elements:
             element._handle_delete()  # pylint: disable=protected-access
             element._deleted = True  # pylint: disable=protected-access
-            outbox.enqueue_delete(element)
+            self.outbox.enqueue_delete(element)
 
     def remove_all_elements(self) -> None:
         """Remove all elements from the client."""
@@ -309,6 +312,7 @@ class Client:
         Normally this should never happen, but has been observed (see #1826).
         """
         self.remove_all_elements()
+        self.outbox.stop()
         del Client.instances[self.id]
 
     @contextmanager

nicegui/core.py

@@ -6,8 +6,10 @@ from typing import TYPE_CHECKING, Optional
 from socketio import AsyncServer
 
 if TYPE_CHECKING:
+    from .air import Air
     from .app import App
 
 app: App
 sio: AsyncServer
 loop: Optional[asyncio.AbstractEventLoop] = None
+air: Optional[Air] = None

nicegui/element.py

@@ -9,7 +9,7 @@ from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Optional,
 
 from typing_extensions import Self
 
-from . import context, core, events, helpers, json, outbox, storage
+from . import context, core, events, helpers, json, storage
 from .awaitable_response import AwaitableResponse, NullResponse
 from .dependencies import Component, Library, register_library, register_resource, register_vue_component
 from .elements.mixins.visibility import Visibility
@@ -99,9 +99,9 @@ class Element(Visibility):
 
         self.tailwind = Tailwind(self)
 
-        outbox.enqueue_update(self)
+        self.client.outbox.enqueue_update(self)
         if self.parent_slot:
-            outbox.enqueue_update(self.parent_slot.parent)
+            self.client.outbox.enqueue_update(self.parent_slot.parent)
 
     def __init_subclass__(cls, *,
                           component: Union[str, Path, None] = None,
@@ -419,7 +419,7 @@ class Element(Visibility):
         """Update the element on the client side."""
         if self.is_deleted:
             return
-        outbox.enqueue_update(self)
+        self.client.outbox.enqueue_update(self)
 
     def run_method(self, name: str, *args: Any, timeout: float = 1, check_interval: float = 0.01) -> AwaitableResponse:
         """Run a method on the client side.

nicegui/functions/javascript.py

@@ -32,9 +32,4 @@ def run_javascript(code: str, *,
                          'Now the function always returns an AwaitableResponse that can be awaited. '
                          'Please remove the "respond=False" argument and call the function without awaiting.')
 
-    client = context.get_client()
-    if not client.has_socket_connection:
-        raise RuntimeError('Cannot run JavaScript before client is connected; '
-                           'try "await client.connected()" or "client.on_connect(...)".')
-
-    return client.run_javascript(code, timeout=timeout, check_interval=check_interval)
+    return context.get_client().run_javascript(code, timeout=timeout, check_interval=check_interval)

nicegui/functions/notify.py

@@ -1,7 +1,6 @@
 from typing import Any, Literal, Optional, Union
 
-from .. import context, outbox
-from ..logging import log
+from .. import context
 
 ARG_MAP = {
     'close_button': 'closeBtn',
@@ -50,7 +49,5 @@ def notify(message: Any, *,
     options = {ARG_MAP.get(key, key): value for key, value in locals().items() if key != 'kwargs' and value is not None}
     options['message'] = str(message)
     options.update(kwargs)
-    if context.get_client().has_socket_connection:
-        outbox.enqueue_message('notify', options, context.get_client().id)
-    else:
-        log.warning(f'Ignoring notification "{message}" because the client is not connected.')
+    client = context.get_client()
+    client.outbox.enqueue_message('notify', options, client.id)

nicegui/nicegui.py

@@ -11,7 +11,7 @@ from fastapi.responses import FileResponse, Response
 from fastapi.staticfiles import StaticFiles
 from fastapi_socketio import SocketManager
 
-from . import air, background_tasks, binding, core, favicon, helpers, json, outbox, run, welcome
+from . import air, background_tasks, binding, core, favicon, helpers, json, run, welcome
 from .app import App
 from .client import Client
 from .dependencies import js_components, libraries, resources
@@ -111,7 +111,6 @@ async def _startup() -> None:
     core.loop = asyncio.get_running_loop()
     app.start()
     background_tasks.create(binding.refresh_loop(), name='refresh bindings')
-    background_tasks.create(outbox.loop(air.instance), name='send outbox')
     background_tasks.create(Client.prune_instances(), name='prune clients')
     background_tasks.create(Slot.prune_stacks(), name='prune slot stacks')
     air.connect()

nicegui/outbox.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
 import asyncio
-from collections import defaultdict, deque
-from typing import TYPE_CHECKING, Any, DefaultDict, Deque, Dict, Optional, Tuple
+from collections import deque
+from typing import TYPE_CHECKING, Any, Deque, Dict, Optional, Tuple
 
-from . import core
+from . import background_tasks, core
 
 if TYPE_CHECKING:
-    from .air import Air
+    from .client import Client
     from .element import Element
 
 ClientId = str
@@ -15,56 +15,70 @@ ElementId = int
 MessageType = str
 Message = Tuple[ClientId, MessageType, Any]
 
-update_queue: DefaultDict[ClientId, Dict[ElementId, Optional[Element]]] = defaultdict(dict)
-message_queue: Deque[Message] = deque()
 
+class Outbox:
 
-def enqueue_update(element: Element) -> None:
-    """Enqueue an update for the given element."""
-    update_queue[element.client.id][element.id] = element
+    def __init__(self, client: Client) -> None:
+        self.client = client
+        self.updates: Dict[ElementId, Optional[Element]] = {}
+        self.messages: Deque[Message] = deque()
+        self._should_stop = False
+        if core.app.is_started:
+            background_tasks.create(self.loop(), name=f'outbox loop {client.id}')
+        else:
+            core.app.on_startup(self.loop)
 
+    def enqueue_update(self, element: Element) -> None:
+        """Enqueue an update for the given element."""
+        self.updates[element.id] = element
 
-def enqueue_delete(element: Element) -> None:
-    """Enqueue a deletion for the given element."""
-    update_queue[element.client.id][element.id] = None
+    def enqueue_delete(self, element: Element) -> None:
+        """Enqueue a deletion for the given element."""
+        self.updates[element.id] = None
 
+    def enqueue_message(self, message_type: MessageType, data: Any, target_id: ClientId) -> None:
+        """Enqueue a message for the given client."""
+        self.messages.append((target_id, message_type, data))
 
-def enqueue_message(message_type: MessageType, data: Any, target_id: ClientId) -> None:
-    """Enqueue a message for the given client."""
-    message_queue.append((target_id, message_type, data))
+    async def loop(self) -> None:
+        """Send updates and messages to all clients in an endless loop."""
+        while not self._should_stop:
+            try:
+                await asyncio.sleep(0.01)
 
+                if not self.updates and not self.messages:
+                    continue
 
-async def loop(air: Optional[Air]) -> None:
-    """Emit queued updates and messages in an endless loop."""
-    async def emit(message_type: MessageType, data: Any, target_id: ClientId) -> None:
-        await core.sio.emit(message_type, data, room=target_id)
-        if air is not None and air.is_air_target(target_id):
-            await air.emit(message_type, data, room=target_id)
-
-    while True:
-        if not update_queue and not message_queue:
-            await asyncio.sleep(0.01)
-            continue
+                if not self.client.has_socket_connection:
+                    continue
 
-        coros = []
-        try:
-            for client_id, elements in update_queue.items():
+                coros = []
                 data = {
                     element_id: None if element is None else element._to_dict()  # pylint: disable=protected-access
-                    for element_id, element in elements.items()
+                    for element_id, element in self.updates.items()
                 }
-                coros.append(emit('update', data, client_id))
-            update_queue.clear()
-
-            for target_id, message_type, data in message_queue:
-                coros.append(emit(message_type, data, target_id))
-            message_queue.clear()
-
-            for coro in coros:
-                try:
-                    await coro
-                except Exception as e:
-                    core.app.handle_exception(e)
-        except Exception as e:
-            core.app.handle_exception(e)
-            await asyncio.sleep(0.1)
+                coros.append(self._emit('update', data, self.client.id))
+                self.updates.clear()
+
+                for target_id, message_type, data in self.messages:
+                    coros.append(self._emit(message_type, data, target_id))
+                self.messages.clear()
+
+                for coro in coros:
+                    try:
+                        await coro
+                    except Exception as e:
+                        core.app.handle_exception(e)
+
+            except Exception as e:
+                core.app.handle_exception(e)
+                await asyncio.sleep(0.1)
+
+    async def _emit(self, message_type: MessageType, data: Any, target_id: ClientId) -> None:
+        await core.sio.emit(message_type, data, room=target_id)
+        if core.air is not None and core.air.is_air_target(target_id):
+            await core.air.emit(message_type, data, room=target_id)
+
+    def stop(self) -> None:
+        """Stop the outbox loop."""
+        self._should_stop = True

nicegui/testing/conftest.py

@@ -69,8 +69,8 @@ def reset_globals() -> Generator[None, None, None]:
     importlib.reload(core)
     Client.instances.clear()
     Client.page_routes.clear()
-    Client.auto_index_client = Client(page('/'), shared=True).__enter__()  # pylint: disable=unnecessary-dunder-call
     app.reset()
+    Client.auto_index_client = Client(page('/'), shared=True).__enter__()  # pylint: disable=unnecessary-dunder-call
     # NOTE we need to re-add the auto index route because we removed all routes above
     app.get('/')(Client.auto_index_client.build_response)
     binding.reset()

nicegui/ui_run.py

@@ -9,8 +9,9 @@ from starlette.routing import Route
 from uvicorn.main import STARTUP_FAILURE
 from uvicorn.supervisors import ChangeReload, Multiprocess
 
-from . import air, core, helpers
+from . import core, helpers
 from . import native as native_module
+from .air import Air
 from .client import Client
 from .language import Language
 from .logging import log
@@ -103,7 +104,7 @@ def run(*,
             route.include_in_schema = endpoint_documentation in {'page', 'all'}
 
     if on_air:
-        air.instance = air.Air('' if on_air is True else on_air)
+        core.air = Air('' if on_air is True else on_air)
 
     if multiprocessing.current_process().name != 'MainProcess':
         return

tests/test_element.py

@@ -1,7 +1,7 @@
 import pytest
 from selenium.webdriver.common.by import By
 
-from nicegui import ui
+from nicegui import background_tasks, ui
 from nicegui.testing import Screen
 
 
@@ -281,3 +281,16 @@ def test_bad_characters(screen: Screen):
 
     screen.open('/')
     screen.should_contain(r'& <test> ` ${foo}')
+
+
+def test_update_before_client_connection(screen: Screen):
+    @ui.page('/')
+    def page():
+        label = ui.label('Hello world!')
+
+        async def update():
+            label.text = 'Hello again!'
+        background_tasks.create(update())
+
+    screen.open('/')
+    screen.should_contain('Hello again!')

tests/test_javascript.py

@@ -1,5 +1,3 @@
-import pytest
-
 from nicegui import Client, ui
 from nicegui.testing import Screen
 
@@ -36,14 +34,13 @@ def test_run_javascript_before_client_connected(screen: Screen):
     @ui.page('/')
     def page():
         ui.label('before js')
-        with pytest.raises(RuntimeError):
-            ui.run_javascript('document.title = "A New Title"')
+        ui.run_javascript('document.title = "New Title"')
         ui.label('after js')
 
     screen.open('/')
-    assert screen.selenium.title == 'NiceGUI'
     screen.should_contain('before js')
     screen.should_contain('after js')
+    screen.should_contain('New Title')
 
 
 def test_response_from_javascript(screen: Screen):

tests/test_outbox.py

@@ -0,0 +1,25 @@
+import asyncio
+
+from nicegui import ui
+from nicegui.testing import Screen
+
+
+def test_removing_outbox_loops(screen: Screen):
+    @ui.page('/page', reconnect_timeout=0.1)
+    def page():
+        ui.button('Click me', on_click=lambda: ui.notify('Hello world!'))
+
+    def count_outbox_loop_tasks() -> int:
+        return len([t for t in asyncio.all_tasks() if t.get_name().startswith('outbox loop')])
+
+    count = ui.label()
+    ui.timer(0.1, lambda: count.set_text(f'{count_outbox_loop_tasks()} tasks'))
+
+    screen.open('/page')
+    screen.click('Click me')
+    screen.should_contain('Hello world!')
+    assert count.text == '1 tasks'
+
+    screen.open('/')
+    screen.wait(0.5)
+    assert count.text == '0 tasks'

tests/test_refreshable.py

@@ -162,24 +162,25 @@ def test_refresh_with_function_reference(screen: Screen):
 
         def __init__(self, name):
             self.name = name
+            self.count = 0
             self.ui()
 
         @ui.refreshable
         def ui(self):
-            ui.notify(f'Refreshing {self.name}')
+            ui.notify(f'Refreshing {self.name} ({self.count})')
+            self.count += 1
             ui.button(self.name, on_click=self.ui.refresh)
 
     Test('A')
-    screen.assert_py_logger('WARNING', 'Ignoring notification "Refreshing A" because the client is not connected.')
-
     Test('B')
-    screen.assert_py_logger('WARNING', 'Ignoring notification "Refreshing B" because the client is not connected.')
 
     screen.open('/')
+    screen.should_contain('Refreshing A (0)')
+    screen.should_contain('Refreshing B (0)')
     screen.click('A')
-    screen.should_contain('Refreshing A')
+    screen.should_contain('Refreshing A (1)')
     screen.click('B')
-    screen.should_contain('Refreshing B')
+    screen.should_contain('Refreshing B (1)')
 
 
 def test_refreshable_with_state(screen: Screen):