From 1cf93f1dcb5bc2451c8ed2e2e0653976393bc094 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 3 Dec 2024 16:26:29 -0800 Subject: [PATCH] FrameProcessor: ignore other frames during CancelFrame --- CHANGELOG.md | 3 +++ src/pipecat/processors/frame_processor.py | 14 ++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff84cc17d..19dd9aab5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,9 @@ async def on_audio_data(processor, audio, sample_rate, num_channels): ### Fixed +- Fixed an issue where other frames were being processed while a `CancelFrame` + was being pushed down the pipeline. + - `AudioBufferProcessor` now handles interruptions properly. - Fixed a `WebsocketServerTransport` issue that would prevent interruptions with diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 93829c8fc..52066b4f4 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -13,6 +13,7 @@ from loguru import logger from pipecat.clocks.base_clock import BaseClock from pipecat.frames.frames import ( + CancelFrame, EndFrame, ErrorFrame, Frame, @@ -58,6 +59,13 @@ class FrameProcessor: self._enable_usage_metrics = False self._report_only_initial_ttfb = False + # Cancellation is done through CancelFrame (a system frame). This could + # cause other events being triggered (e.g. closing a transport) which + # could also cause other frames to be pushed from other tasks + # (e.g. EndFrame). So, when we are cancelling we don't want anything + # else to be pushed. + self._cancelling = False + # Metrics self._metrics = metrics or FrameProcessorMetrics() self._metrics.set_processor_name(self.name) @@ -161,6 +169,10 @@ class FrameProcessor: Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]] ] = None, ): + # If we are cancelling we don't want to process any other frame. + if self._cancelling: + return + if isinstance(frame, SystemFrame): # We don't want to queue system frames. await self.process_frame(frame, direction) @@ -187,6 +199,8 @@ class FrameProcessor: await self.stop_all_metrics() elif isinstance(frame, StopInterruptionFrame): self._should_report_ttfb = True + elif isinstance(frame, CancelFrame): + self._cancelling = True async def push_error(self, error: ErrorFrame): await self.push_frame(error, FrameDirection.UPSTREAM)