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

Merge pull request #2073 from a6kme/patch-1

Browse Source 
Start HeartBeat when all processors have processed StartFrame
This commit is contained in:
Aleix Conchillo Flaqué
2025-06-26 11:34:46 -07:00
committed by GitHub
parent 539ca5856f 89c801f82c
commit 7947497f7e
1 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/pipecat/pipeline/task.py
View File

@@ -468,7 +468,7 @@ class PipelineTask(BasePipelineTask):
return self._process_push_task return self._process_push_task
def _maybe_start_heartbeat_tasks(self): def _maybe_start_heartbeat_tasks(self):
if self._params.enable_heartbeats: if self._params.enable_heartbeats and self._heartbeat_push_task is None:
self._heartbeat_push_task = self._task_manager.create_task( self._heartbeat_push_task = self._task_manager.create_task(
self._heartbeat_push_handler(), f"{self}::_heartbeat_push_handler" self._heartbeat_push_handler(), f"{self}::_heartbeat_push_handler"
) )
@@ -566,7 +566,6 @@ class PipelineTask(BasePipelineTask):
""" """
self._clock.start() self._clock.start()
self._maybe_start_heartbeat_tasks()
self._maybe_start_idle_task() self._maybe_start_idle_task()
start_frame = StartFrame( start_frame = StartFrame(
@@ -648,6 +647,10 @@ class PipelineTask(BasePipelineTask):
if isinstance(frame, StartFrame): if isinstance(frame, StartFrame):
await self._call_event_handler("on_pipeline_started", frame) await self._call_event_handler("on_pipeline_started", frame)
# Start heartbeat tasks now that StartFrame has been processed
# by all processors in the pipeline
self._maybe_start_heartbeat_tasks()
elif isinstance(frame, EndFrame): elif isinstance(frame, EndFrame):
await self._call_event_handler("on_pipeline_ended", frame) await self._call_event_handler("on_pipeline_ended", frame)
self._pipeline_end_event.set() self._pipeline_end_event.set()