wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

frame_processor: allow pushing more frames after EndFrame

Browse Source 
This can be useful for testing purposes. In real practice, there shouldn't be
any frames after an EndFrame is pushed.
This commit is contained in:
Aleix Conchillo Flaqué
2025-01-21 09:42:15 -08:00
parent 2f23693bf3
commit 2e0fb198bf
1 changed files with 2 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/pipecat/processors/frame_processor.py
View File

@@ -293,8 +293,7 @@ class FrameProcessor:
await self.__input_frame_task await self.__input_frame_task
async def __input_frame_task_handler(self): async def __input_frame_task_handler(self):
running = True while True:
while running:
try: try:
if self.__should_block_frames: if self.__should_block_frames:
logger.trace(f"{self}: frame processing paused") logger.trace(f"{self}: frame processing paused")
@@ -311,8 +310,6 @@ class FrameProcessor:
if callback: if callback:
await callback(self, frame, direction) await callback(self, frame, direction)
running = not isinstance(frame, EndFrame)
self.__input_queue.task_done() self.__input_queue.task_done()
except asyncio.CancelledError: except asyncio.CancelledError:
logger.trace(f"{self}: cancelled input task") logger.trace(f"{self}: cancelled input task")
@@ -330,12 +327,10 @@ class FrameProcessor:
await self.__push_frame_task await self.__push_frame_task
async def __push_frame_task_handler(self): async def __push_frame_task_handler(self):
running = True while True:
while running:
try: try:
(frame, direction) = await self.__push_queue.get() (frame, direction) = await self.__push_queue.get()
await self.__internal_push_frame(frame, direction) await self.__internal_push_frame(frame, direction)
running = not isinstance(frame, EndFrame)
self.__push_queue.task_done() self.__push_queue.task_done()
except asyncio.CancelledError: except asyncio.CancelledError:
logger.trace(f"{self}: cancelled push task") logger.trace(f"{self}: cancelled push task")