diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f8930f44..aa324e93b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 520998988..6bf5dd688 100644 --- a/src/pipecat/pipeline/task.py +++ b/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): - await self._observer.add_observer(observer) + def add_observer(self, observer: BaseObserver): + self._observer.add_observer(observer) async def remove_observer(self, observer: BaseObserver): await self._observer.remove_observer(observer) diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index b7a54f58d..950ddc43b 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/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): - proxy = self._create_proxy(observer) - self._proxies[observer] = proxy + def add_observer(self, observer: BaseObserver): + # Add the observer to the list. 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( diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 9bc737b1a..84485098d 100644 --- a/tests/test_pipeline.py +++ b/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() - # Wait after the pipeline is started and add an observer. + observer2 = CustomAddObserver2() + # 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 - await task.remove_observer(observer) + # Remove both observers. + 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