Merge pull request #433 from pipecat-ai/aleix/process-all-startframes

StartFrame should be the first frame every processor receives

src/pipecat/processors/frameworks/rtvi.py

@@ -356,8 +356,10 @@ class RTVIProcessor(FrameProcessor):
             await self.push_frame(frame, direction)
         # Control frames
         elif isinstance(frame, StartFrame):
-            await self._start(frame)
+            # Push StartFrame before start(), because we want StartFrame to be
+            # processed by every processor before any other frame is processed.
             await self.push_frame(frame, direction)
+            await self._start(frame)
         elif isinstance(frame, EndFrame):
             # Push EndFrame before stop(), because stop() waits on the task to
             # finish and the task finishes when EndFrame is processed.

src/pipecat/processors/gstreamer/pipeline_source.py

@@ -78,8 +78,10 @@ class GStreamerPipelineSource(FrameProcessor):
             await self.push_frame(frame, direction)
         # Control frames
         elif isinstance(frame, StartFrame):
-            await self._start(frame)
+            # Push StartFrame before start(), because we want StartFrame to be
+            # processed by every processor before any other frame is processed.
             await self._internal_push_frame(frame, direction)
+            await self._start(frame)
         elif isinstance(frame, EndFrame):
             # Push EndFrame before stop(), because stop() waits on the task to
             # finish and the task finishes when EndFrame is processed.

src/pipecat/transports/base_input.py

@@ -96,8 +96,10 @@ class BaseInputTransport(FrameProcessor):
             await self.push_frame(frame, direction)
         # Control frames
         elif isinstance(frame, StartFrame):
-            await self.start(frame)
+            # Push StartFrame before start(), because we want StartFrame to be
+            # processed by every processor before any other frame is processed.
             await self._internal_push_frame(frame, direction)
+            await self.start(frame)
         elif isinstance(frame, EndFrame):
             # Push EndFrame before stop(), because stop() waits on the task to
             # finish and the task finishes when EndFrame is processed.

src/pipecat/transports/base_output.py

@@ -134,8 +134,8 @@ class BaseOutputTransport(FrameProcessor):
         # queue.
         #
         if isinstance(frame, CancelFrame):
-            await self.push_frame(frame, direction)
             await self.cancel(frame)
+            await self.push_frame(frame, direction)
         elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
             await self.push_frame(frame, direction)
             await self._handle_interruptions(frame)