Fixed: PubSub is not shared between pages in the same FastAPI app

sdk/python/packages/flet-core/src/flet_core/page.py

@@ -390,13 +390,15 @@ async def _clean_async(self, control: Control):
                 await c.will_unmount_async()
 
     def _close(self):
+        self.__pubsub.unsubscribe_all()
         removed_controls = self._remove_control_recursively(self.index, self)
         for c in removed_controls:
             c.will_unmount()
             c._dispose()
         self.__close_internal()
 
     async def _close_async(self):
+        await self.__pubsub.unsubscribe_all_async()
         removed_controls = self._remove_control_recursively(self.index, self)
         for c in removed_controls:
             await c.will_unmount_async()
@@ -413,13 +415,19 @@ def __close_internal(self):
         self.__query = None
 
     def __update(self, *controls) -> Tuple[List[Control], List[Control]]:
+        if self.__conn is None:
+            logger.warning(f"Page is disconnected: {self._session_id}")
+            return [], []
         commands, added_controls, removed_controls = self.__prepare_update(*controls)
         self.__validate_controls_page(added_controls)
         results = self.__conn.send_commands(self._session_id, commands).results
         self.__update_control_ids(added_controls, results)
         return added_controls, removed_controls
 
     async def __update_async(self, *controls) -> Tuple[List[Control], List[Control]]:
+        if self.__conn is None:
+            logger.warning(f"Page is disconnected: {self._session_id}")
+            return [], []
         commands, added_controls, removed_controls = self.__prepare_update(*controls)
         self.__validate_controls_page(added_controls)
         results = (

sdk/python/packages/flet-fastapi/src/flet_fastapi/flet_app.py

@@ -7,9 +7,8 @@
 from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional
 
-from fastapi import WebSocket
-
 import flet_fastapi
+from fastapi import WebSocket
 from flet_core.event import Event
 from flet_core.local_connection import LocalConnection
 from flet_core.page import Page
@@ -25,13 +24,17 @@
 from flet_core.utils import is_coroutine, random_string
 from flet_fastapi.flet_app_manager import app_manager
 from flet_fastapi.oauth_state import OAuthState
+from flet_runtime.pubsub import PubSubHub
 from flet_runtime.uploads import build_upload_url
 
 logger = logging.getLogger(flet_fastapi.__name__)
 
 DEFAULT_FLET_SESSION_TIMEOUT = 3600
 DEFAULT_FLET_OAUTH_STATE_TIMEOUT = 600
 
+_pubsubhubs_lock = asyncio.Lock()
+_pubsubhubs = {}
+
 
 class FletApp(LocalConnection):
     def __init__(
@@ -81,6 +84,14 @@ async def handle(self, websocket: WebSocket):
         * `websocket` (WebSocket) - Websocket instance.
         """
         self.__websocket = websocket
+
+        async with _pubsubhubs_lock:
+            psh = _pubsubhubs.get(self.__session_handler, None)
+            if psh is None:
+                psh = PubSubHub()
+                _pubsubhubs[self.__session_handler] = psh
+            self.pubsubhub = psh
+
         self.page_url = str(websocket.url).rsplit("/", 1)[0]
         self.page_name = websocket.url.path.rsplit("/", 1)[0].lstrip("/")
 

sdk/python/packages/flet-runtime/src/flet_runtime/pubsub.py

@@ -12,6 +12,7 @@
 
 class PubSubHub:
     def __init__(self):
+        logger.debug("Creating new PubSubHub instance")
         self.__lock = threading.Lock() if not is_asyncio() else NopeLock()
         self.__async_lock = asyncio.Lock() if is_asyncio() else AsyncNopeLock()
         self.__subscribers: Dict[str, Callable] = {}  # key: session_id, value: handler