From 6554479d39ba36d56dc4e3c91813db01dee42adf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 10 Jun 2024 20:56:01 -0700 Subject: [PATCH] transports: don't queue system frames --- src/pipecat/transports/base_input.py | 2 +- src/pipecat/transports/base_output.py | 11 ++++++++--- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 1162708cc..9496cd3b4 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -83,9 +83,9 @@ class BaseInputTransport(FrameProcessor): await super().process_frame(frame, direction) if isinstance(frame, CancelFrame): + await self.stop() # We don't queue a CancelFrame since we want to stop ASAP. await self.push_frame(frame, direction) - await self.stop() elif isinstance(frame, StartFrame): await self.start(frame) await self._internal_push_frame(frame, direction) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 4c19a0b55..b484c8fce 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -28,6 +28,7 @@ from pipecat.frames.frames import ( ImageRawFrame, StartInterruptionFrame, StopInterruptionFrame, + SystemFrame, TransportMessageFrame) from pipecat.transports.base_transport import TransportParams @@ -53,6 +54,7 @@ class BaseOutputTransport(FrameProcessor): if self._params.camera_out_enabled: self._camera_out_queue = queue.Queue() self._sink_queue = queue.Queue() + self._sink_thread = None self._stopped_event = asyncio.Event() self._is_interrupted = threading.Event() @@ -106,7 +108,8 @@ class BaseOutputTransport(FrameProcessor): if self._params.camera_out_enabled: await self._camera_out_thread - await self._sink_thread + if self._sink_thread: + await self._sink_thread async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -121,11 +124,13 @@ class BaseOutputTransport(FrameProcessor): self._sink_queue.put_nowait(frame) # EndFrame is managed in the queue handler. elif isinstance(frame, CancelFrame): - await self.push_frame(frame, direction) await self.stop() - elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame): await self.push_frame(frame, direction) + elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame): await self._handle_interruptions(frame) + await self.push_frame(frame, direction) + elif isinstance(frame, SystemFrame): + await self.push_frame(frame, direction) else: self._sink_queue.put_nowait(frame)