diff --git a/CHANGELOG.md b/CHANGELOG.md index f7db1e97d..2830b1240 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -89,6 +89,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Check if `PipelineTask` has already been cancelled. + - Don't raise an exception if event handler is not registered. - Upgraded `deepgram-sdk` to 4.1.0. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 692f325d9..0fe330655 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -236,6 +236,7 @@ class PipelineTask(BaseTask): ) observers.append(self._turn_trace_observer) self._finished = False + self._cancelled = False # This queue receives frames coming from the pipeline upstream. self._up_queue = asyncio.Queue() @@ -346,7 +347,6 @@ class PipelineTask(BaseTask): async def cancel(self): """Stops the running pipeline immediately.""" - logger.debug(f"Canceling pipeline task {self}") await self._cancel() async def run(self): @@ -406,12 +406,15 @@ class PipelineTask(BaseTask): await self.queue_frame(frame) async def _cancel(self): - # Make sure everything is cleaned up downstream. This is sent - # out-of-band from the main streaming task which is what we want since - # we want to cancel right away. - await self._source.push_frame(CancelFrame()) - # Only cancel the push task. Everything else will be cancelled in run(). - await self._task_manager.cancel_task(self._process_push_task) + if not self._cancelled: + logger.debug(f"Canceling pipeline task {self}") + self._cancelled = True + # Make sure everything is cleaned up downstream. This is sent + # out-of-band from the main streaming task which is what we want since + # we want to cancel right away. + await self._source.push_frame(CancelFrame()) + # Only cancel the push task. Everything else will be cancelled in run(). + await self._task_manager.cancel_task(self._process_push_task) async def _create_tasks(self): self._process_up_task = self._task_manager.create_task(