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

processors: cancel input task and empty queue with interruptions

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-10-20 22:35:15 -07:00
parent 84f26ac1ca
commit 22ca4c5a02
1 changed files with 23 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -63,11 +63,11 @@ class FrameProcessor:
self._metrics = metrics or FrameProcessorMetrics() self._metrics = metrics or FrameProcessorMetrics()
self._metrics.set_processor_name(self.name) self._metrics.set_processor_name(self.name)
# Processors have an input queue. The input queue can be processed # Processors have an input queue. The input queue will be processed
# immediately (default) or it can be processed when the processor asks # immediately (default) or it will block if one of the `block_on_frames`
# to process a new frame (via process_next_frame()). # is found. To resume processing frames we need to call
# `resume_processing_frames()`.
self._block_on_frames = block_on_frames self._block_on_frames = block_on_frames
self._should_queue_input_frames = False
self.__create_input_task() self.__create_input_task()
# Every processor in Pipecat should only output frames from a single # Every processor in Pipecat should only output frames from a single
@@ -134,7 +134,8 @@ class FrameProcessor:
await self.stop_processing_metrics() await self.stop_processing_metrics()
async def cleanup(self): async def cleanup(self):
pass await self.__cancel_input_task()
await self.__cancel_push_task()
def link(self, processor: "FrameProcessor"): def link(self, processor: "FrameProcessor"):
self._next = processor self._next = processor
@@ -210,13 +211,18 @@ class FrameProcessor:
# #
async def _start_interruption(self): async def _start_interruption(self):
# Cancel the task. This will stop pushing frames downstream. # Cancel the input task. This will stop processing queued frames.
self.__push_frame_task.cancel() await self.__cancel_input_task()
await self.__push_frame_task
# Create a new queue and task. # Cancel the push frame task. This will stop pushing frames downstream.
await self.__cancel_push_task()
# Create a new output queue and task.
self.__create_push_task() self.__create_push_task()
# Create a new input queue and task.
self.__create_input_task()
async def _stop_interruption(self): async def _stop_interruption(self):
# Nothing to do right now. # Nothing to do right now.
pass pass
@@ -239,6 +245,10 @@ class FrameProcessor:
) )
self.__input_event = asyncio.Event() self.__input_event = asyncio.Event()
async def __cancel_input_task(self):
self.__input_frame_task.cancel()
await self.__input_frame_task
async def __input_frame_task_handler(self): async def __input_frame_task_handler(self):
running = True running = True
should_block_frames = False should_block_frames = False
@@ -269,6 +279,10 @@ class FrameProcessor:
self.__push_queue = asyncio.Queue() self.__push_queue = asyncio.Queue()
self.__push_frame_task = self.get_event_loop().create_task(self.__push_frame_task_handler()) self.__push_frame_task = self.get_event_loop().create_task(self.__push_frame_task_handler())
async def __cancel_push_task(self):
self.__push_frame_task.cancel()
await self.__push_frame_task
async def __push_frame_task_handler(self): async def __push_frame_task_handler(self):
running = True running = True
while running: while running: