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

Merge pull request #2822 from pipecat-ai/pk/make-pause-processing-frames-more-robust

Browse Source 
Make `pause_processing_frames()` and `pause_processing_system_frames(…
This commit is contained in:
kompfner
2025-10-10 10:16:27 -04:00
committed by GitHub
parent 0c102ce70b 10069719e4
commit d293cee372
1 changed files with 4 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -877,6 +877,8 @@ class FrameProcessor(BaseObject):
""" """
while True: while True:
(frame, direction, callback) = await self.__input_queue.get()
if self.__should_block_system_frames and self.__input_event: if self.__should_block_system_frames and self.__input_event:
logger.trace(f"{self}: system frame processing paused") logger.trace(f"{self}: system frame processing paused")
await self.__input_event.wait() await self.__input_event.wait()
@@ -884,8 +886,6 @@ class FrameProcessor(BaseObject):
self.__should_block_system_frames = False self.__should_block_system_frames = False
logger.trace(f"{self}: system frame processing resumed") logger.trace(f"{self}: system frame processing resumed")
(frame, direction, callback) = await self.__input_queue.get()
if isinstance(frame, SystemFrame): if isinstance(frame, SystemFrame):
await self.__process_frame(frame, direction, callback) await self.__process_frame(frame, direction, callback)
elif self.__process_queue: elif self.__process_queue:
@@ -900,6 +900,8 @@ class FrameProcessor(BaseObject):
async def __process_frame_task_handler(self): async def __process_frame_task_handler(self):
"""Handle non-system frames from the process queue.""" """Handle non-system frames from the process queue."""
while True: while True:
(frame, direction, callback) = await self.__process_queue.get()
if self.__should_block_frames and self.__process_event: if self.__should_block_frames and self.__process_event:
logger.trace(f"{self}: frame processing paused") logger.trace(f"{self}: frame processing paused")
await self.__process_event.wait() await self.__process_event.wait()
@@ -907,8 +909,6 @@ class FrameProcessor(BaseObject):
self.__should_block_frames = False self.__should_block_frames = False
logger.trace(f"{self}: frame processing resumed") logger.trace(f"{self}: frame processing resumed")
(frame, direction, callback) = await self.__process_queue.get()
await self.__process_frame(frame, direction, callback) await self.__process_frame(frame, direction, callback)
self.__process_queue.task_done() self.__process_queue.task_done()