Merge pull request #1956 from pipecat-ai/pk/make-add-observer-sync

Make `PipelineTask.add_observer()` synchronous. This allows callers t…

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Make `PipelineTask.add_observer()` synchronous. This allows callers to call it before doing the
+  work of running the `PipelineTask` (i.e. without invoking `PipelineTask.set_event_loop()` first).
+
 - Pipecat 0.0.69 forced `uvloop` event loop on Linux on macOS. Unfortunately,
   this is causing issue in some systems. So, `uvloop` is not enabled by default
   anymore. If you want to use `uvloop` you can just set the `asyncio` event

src/pipecat/pipeline/task.py

@@ -310,8 +310,8 @@ class PipelineTask(BaseTask):
         """Return the turn trace observer if enabled."""
         return self._turn_trace_observer
 
-    async def add_observer(self, observer: BaseObserver):
+    def add_observer(self, observer: BaseObserver):
-        await self._observer.add_observer(observer)
+        self._observer.add_observer(observer)
 
     async def remove_observer(self, observer: BaseObserver):
         await self._observer.remove_observer(observer)

src/pipecat/pipeline/task_observer.py

@@ -49,21 +49,31 @@ class TaskObserver(BaseObserver):
         super().__init__(**kwargs)
         self._observers = observers or []
         self._task_manager = task_manager
-        self._proxies: Dict[BaseObserver, Proxy] = {}
+        self._proxies: Optional[Dict[BaseObserver, Proxy]] = (
+            None  # Becomes a dict after start() is called
+        )
 
-    async def add_observer(self, observer: BaseObserver):
+    def add_observer(self, observer: BaseObserver):
-        proxy = self._create_proxy(observer)
+        # Add the observer to the list.
-        self._proxies[observer] = proxy
         self._observers.append(observer)
 
+        # If we already started, create a new proxy for the observer.
+        # Otherwise, it will be created in start().
+        if self._started():
+            proxy = self._create_proxy(observer)
+            self._proxies[observer] = proxy
+
     async def remove_observer(self, observer: BaseObserver):
+        # If the observer has a proxy, remove it.
         if observer in self._proxies:
             proxy = self._proxies[observer]
             # Remove the proxy so it doesn't get called anymore.
             del self._proxies[observer]
             # Cancel the proxy task right away.
             await self._task_manager.cancel_task(proxy.task)
-            # Remove the observer.
+
+        # Remove the observer from the list.
+        if observer in self._observers:
             self._observers.remove(observer)
 
     async def start(self):
@@ -79,6 +89,9 @@ class TaskObserver(BaseObserver):
         for proxy in self._proxies.values():
             await proxy.queue.put(data)
 
+    def _started(self) -> bool:
+        return self._proxies is not None
+
     def _create_proxy(self, observer: BaseObserver) -> Proxy:
         queue = asyncio.Queue()
         task = self._task_manager.create_task(

tests/test_pipeline.py

@@ -117,7 +117,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
 
     async def test_task_add_observer(self):
         frame_received = False
-        frame_add_count = 0
+        frame_count_1 = 0
+        frame_count_2 = 0
 
         class CustomObserver(BaseObserver):
             async def on_push_frame(self, data: FramePushed):
@@ -126,28 +127,41 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
                 if isinstance(data.frame, TextFrame):
                     frame_received = True
 
-        class CustomAddObserver(BaseObserver):
+        class CustomAddObserver1(BaseObserver):
             async def on_push_frame(self, data: FramePushed):
-                nonlocal frame_add_count
+                nonlocal frame_count_1
 
                 if isinstance(data.source, IdentityFilter) and isinstance(data.frame, TextFrame):
-                    frame_add_count += 1
+                    frame_count_1 += 1
+
+        class CustomAddObserver2(BaseObserver):
+            async def on_push_frame(self, data: FramePushed):
+                nonlocal frame_count_2
+
+                if isinstance(data.source, IdentityFilter) and isinstance(data.frame, TextFrame):
+                    frame_count_2 += 1
 
         identity = IdentityFilter()
         pipeline = Pipeline([identity])
         task = PipelineTask(pipeline, observers=[CustomObserver()])
+
+        # Add a new observer right away, before doing anything else with the task.
+        observer1 = CustomAddObserver1()
+        task.add_observer(observer1)
+
         task.set_event_loop(asyncio.get_event_loop())
 
         async def delayed_add_observer():
-            observer = CustomAddObserver()
+            observer2 = CustomAddObserver2()
-            # Wait after the pipeline is started and add an observer.
+            # Wait after the pipeline is started and add another observer.
             await asyncio.sleep(0.1)
-            await task.add_observer(observer)
+            task.add_observer(observer2)
             # Push a TextFrame and wait for the observer to pick it up.
             await task.queue_frame(TextFrame(text="Hello Downstream!"))
             await asyncio.sleep(0.1)
-            # Remove the observer
+            # Remove both observers.
-            await task.remove_observer(observer)
+            await task.remove_observer(observer1)
+            await task.remove_observer(observer2)
             # Push another TextFrame. This time the counter should not
             # increments since we have removed the observer.
             await task.queue_frame(TextFrame(text="Hello Downstream!"))
@@ -158,7 +172,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         await asyncio.gather(task.run(), delayed_add_observer())
 
         assert frame_received
-        assert frame_add_count == 1
+        assert frame_count_1 == 1
+        assert frame_count_2 == 1
 
     async def test_task_started_ended_event_handler(self):
         start_received = False