Merge pull request #3060 from pipecat-ai/aleix/consumer-queue-frames

ConsumerProcessor: queue frames internally instead of pushing them

CHANGELOG.md

@@ -24,6 +24,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `ConsumerProcessor` now queues frames from the producer internally instead of
+  pushing them directly. This allows us to subclass consumer processors and
+  manipulate frames before they are pushed.
+
 - `BaseTextFilter` only require subclasses to implement the `filter()` method.
 
 - Extracted the logic for retrying connections, and create a new `send_with_retry`

src/pipecat/processors/consumer_processor.py

@@ -83,4 +83,4 @@ class ConsumerProcessor(FrameProcessor):
         while True:
             frame = await self._queue.get()
             new_frame = await self._transformer(frame)
-            await self.push_frame(new_frame, self._direction)
+            await self.queue_frame(new_frame, self._direction)