diff --git a/CHANGELOG.md b/CHANGELOG.md index f6debb3b3..f80a6cf3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed issues with Ctrl-C program termination. +- Fixed an issue that was causing `StopTaskFrame` to actually not exit the + `PipelineTask`. + ## [0.0.16] - 2024-05-16 ### Fixed diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 221d588f5..400adcbfd 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -61,8 +61,6 @@ class PipelineTask: self._process_up_task = asyncio.create_task(self._process_up_queue()) self._process_down_task = asyncio.create_task(self._process_down_queue()) await asyncio.gather(self._process_up_task, self._process_down_task) - await self._source.cleanup() - await self._pipeline.cleanup() async def queue_frame(self, frame: Frame): await self._down_queue.put(frame) @@ -81,14 +79,20 @@ class PipelineTask: await self._source.process_frame( StartFrame(allow_interruptions=self._allow_interruptions), FrameDirection.DOWNSTREAM) running = True + should_cleanup = True while running: try: frame = await self._down_queue.get() await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) - self._down_queue.task_done() running = not (isinstance(frame, StopTaskFrame) or isinstance(frame, EndFrame)) + should_cleanup = not isinstance(frame, StopTaskFrame) + self._down_queue.task_done() except asyncio.CancelledError: break + # Cleanup only if we need to. + if should_cleanup: + await self._source.cleanup() + await self._pipeline.cleanup() # We just enqueue None to terminate the task gracefully. self._process_up_task.cancel()