From f22a00570d96480146093203d12a526e859a99db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 10:03:13 -0800 Subject: [PATCH] task: start heartbeats task when push task starts --- src/pipecat/pipeline/task.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e04e0d285..f0e2089e4 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -201,15 +201,16 @@ class PipelineTask: tasks = [self._process_up_task, self._process_down_task, self._process_push_task] + return tasks + + def _maybe_start_heartbeat_tasks(self): if self._params.enable_heartbeats: self._heartbeat_push_task = asyncio.create_task(self._heartbeat_push_handler()) self._heartbeat_monitor_task = asyncio.create_task(self._heartbeat_monitor_handler()) - tasks.append(self._heartbeat_push_task) - tasks.append(self._heartbeat_monitor_task) - - return tasks async def _cancel_tasks(self, cancel_push: bool): + await self._maybe_cancel_heartbeat_tasks() + if cancel_push: self._process_push_task.cancel() await self._process_push_task @@ -220,6 +221,7 @@ class PipelineTask: self._process_down_task.cancel() await self._process_down_task + async def _maybe_cancel_heartbeat_tasks(self): if self._params.enable_heartbeats: self._heartbeat_push_task.cancel() await self._heartbeat_push_task @@ -246,6 +248,8 @@ class PipelineTask: """ self._clock.start() + self._maybe_start_heartbeat_tasks() + start_frame = StartFrame( allow_interruptions=self._params.allow_interruptions, enable_metrics=self._params.enable_metrics,