Update WebRtcWorkerContext to unset video_transformer and video_receiver attrs

app.py

@@ -320,7 +320,12 @@ def transform(self, frame: av.VideoFrame) -> np.ndarray:
             # are not strictly synchronized.
             while True:
                 if webrtc_ctx.video_transformer:
-                    result = webrtc_ctx.video_transformer.result_queue.get(timeout=1.0)
+                    try:
+                        result = webrtc_ctx.video_transformer.result_queue.get(
+                            timeout=1.0
+                        )
+                    except queue.Empty:
+                        result = None
                     labels_placeholder.table(result)
                 else:
                     break

streamlit_webrtc/__init__.py

@@ -4,6 +4,7 @@
 import json
 import logging
 import os
+import weakref
 from typing import Dict, Hashable, NamedTuple, Optional, Union
 
 try:
@@ -78,14 +79,32 @@ class ClientSettings(TypedDict):
     media_stream_constraints: Optional[MediaStreamConstraints]
 
 
-class WebRtcWorkerState(NamedTuple):
+class WebRtcStreamerState(NamedTuple):
     playing: bool
 
 
-class WebRtcWorkerContext(NamedTuple):
-    state: WebRtcWorkerState
-    video_transformer: Optional[VideoTransformerBase]
-    video_receiver: Optional[VideoReceiver]
+class WebRtcStreamerContext:
+    state: WebRtcStreamerState
+    _worker_ref: "Optional[weakref.ReferenceType[WebRtcWorker]]"
+
+    def __init__(
+        self, worker: Optional[WebRtcWorker], state: WebRtcStreamerState
+    ) -> None:
+        self._worker_ref = weakref.ref(worker) if worker else None
+        self.state = state
+
+    def _get_worker(self) -> Optional[WebRtcWorker]:
+        return self._worker_ref() if self._worker_ref else None
+
+    @property
+    def video_transformer(self) -> Optional[VideoTransformerBase]:
+        worker = self._get_worker()
+        return worker.video_transformer if worker else None
+
+    @property
+    def video_receiver(self) -> Optional[VideoReceiver]:
+        worker = self._get_worker()
+        return worker.video_receiver if worker else None
 
 
 def webrtc_streamer(
@@ -97,7 +116,7 @@ def webrtc_streamer(
     out_recorder_factory: Optional[MediaRecorderFactory] = None,
     video_transformer_factory: Optional[VideoTransformerFactory] = None,
     async_transform: bool = True,
-) -> WebRtcWorkerContext:
+) -> WebRtcStreamerContext:
     webrtc_worker = _get_webrtc_worker(key)
 
     sdp_answer_json = None
@@ -140,10 +159,9 @@ def webrtc_streamer(
                 _set_webrtc_worker(key, webrtc_worker)
                 st.experimental_rerun()  # Rerun to send the SDP answer to frontend
 
-    ctx = WebRtcWorkerContext(
-        state=WebRtcWorkerState(playing=playing),
-        video_transformer=webrtc_worker.video_transformer if webrtc_worker else None,
-        video_receiver=webrtc_worker.video_receiver if webrtc_worker else None,
+    ctx = WebRtcStreamerContext(
+        state=WebRtcStreamerState(playing=playing),
+        worker=webrtc_worker,
     )
 
     return ctx

streamlit_webrtc/webrtc.py

@@ -309,6 +309,12 @@ def callback(localDescription):
         self._video_transformer = video_transformer
         self._video_receiver = video_receiver
 
+        @self.pc.on("iceconnectionstatechange")
+        async def on_iceconnectionstatechange():
+            iceConnectionState = self.pc.iceConnectionState
+            if iceConnectionState == "closed" or iceConnectionState == "failed":
+                self._unset_transformers()
+
         loop.create_task(
             _process_offer(
                 self.mode,
@@ -365,7 +371,12 @@ def process_offer(
 
         return result
 
+    def _unset_transformers(self):
+        self._video_transformer = None
+        self._video_receiver = None
+
     def stop(self, timeout: Union[float, None] = 1.0):
+        self._unset_transformers()
         if self._loop:
             self._loop.stop()
         if self._webrtc_thread: