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)