Merge pull request #809 from pipecat-ai/aleix/parallel-pipeline-fix-system-frames

fix system frames parallel pipeline

CHANGELOG.md

@@ -75,6 +75,8 @@ async def on_audio_data(processor, audio, sample_rate, num_channels):
 
 ### Fixed
 
+- Fixed a `ParallelPipeline` issue that would cause system frames to be queued.
+
 - Fixed `FastAPIWebsocketTransport` so it can work with binary data (e.g. using
   the protobuf serializer).
 

src/pipecat/pipeline/parallel_pipeline.py

@@ -12,7 +12,7 @@ from typing import List
 from pipecat.pipeline.base_pipeline import BasePipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame
+from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame, SystemFrame
 
 from loguru import logger
 
@@ -27,7 +27,12 @@ class Source(FrameProcessor):
 
         match direction:
             case FrameDirection.UPSTREAM:
-                await self._up_queue.put(frame)
+                # We don't want to queue system frames as they would be
+                # processed by a separate task.
+                if isinstance(frame, SystemFrame):
+                    await self.push_frame(frame, direction)
+                else:
+                    await self._up_queue.put(frame)
             case FrameDirection.DOWNSTREAM:
                 await self.push_frame(frame, direction)
 
@@ -44,7 +49,12 @@ class Sink(FrameProcessor):
             case FrameDirection.UPSTREAM:
                 await self.push_frame(frame, direction)
             case FrameDirection.DOWNSTREAM:
-                await self._down_queue.put(frame)
+                # We don't want to queue system frames as they would be
+                # processed by a separate task.
+                if isinstance(frame, SystemFrame):
+                    await self.push_frame(frame, direction)
+                else:
+                    await self._down_queue.put(frame)
 
 
 class ParallelPipeline(BasePipeline):

src/pipecat/transports/services/daily.py

@@ -200,7 +200,6 @@ class DailyTransportClient(EventHandler):
         self._other_participant_has_joined = False
 
         self._joined = False
-        self._joining = False
         self._leave_counter = 0
 
         self._executor = ThreadPoolExecutor(max_workers=5)
@@ -315,12 +314,11 @@ class DailyTransportClient(EventHandler):
 
     async def join(self):
         # Transport already joined, ignore.
-        if self._joined or self._joining:
+        if self._joined:
             # Increment leave counter if we already joined.
             self._leave_counter += 1
             return
 
-        self._joining = True
         logger.info(f"Joining {self._room_url}")
 
         # For performance reasons, never subscribe to video streams (unless a
@@ -353,7 +351,6 @@ class DailyTransportClient(EventHandler):
             error_msg = f"Time out joining {self._room_url}"
             logger.error(error_msg)
             await self._callbacks.on_error(error_msg)
-        self._joining = False
 
     async def _start_transcription(self):
         if not self._token: