From 37e23887586edebea358c2641806ddd1503042ad Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 29 Aug 2024 22:43:01 -0700
Subject: [PATCH] StartFrame should be the first frame every processor receives

Fixes #427
---
 src/pipecat/processors/frameworks/rtvi.py           | 4 +++-
 src/pipecat/processors/gstreamer/pipeline_source.py | 4 +++-
 src/pipecat/transports/base_input.py                | 4 +++-
 src/pipecat/transports/base_output.py               | 2 +-
 4 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py
index 653a883d0..dd28e7252 100644
--- a/src/pipecat/processors/frameworks/rtvi.py
+++ b/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.
diff --git a/src/pipecat/processors/gstreamer/pipeline_source.py b/src/pipecat/processors/gstreamer/pipeline_source.py
index 50e0baabf..eef0d56cc 100644
--- a/src/pipecat/processors/gstreamer/pipeline_source.py
+++ b/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.
diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py
index 49a685d0e..3d1d0c4d7 100644
--- a/src/pipecat/transports/base_input.py
+++ b/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.
diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py
index 584d3bb24..d2c6add2b 100644
--- a/src/pipecat/transports/base_output.py
+++ b/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)