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

task: start heartbeats task when push task starts

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-01-17 10:03:13 -08:00
parent 2503f76107
commit f22a00570d
1 changed files with 8 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -201,15 +201,16 @@ class PipelineTask:
tasks = [self._process_up_task, self._process_down_task, self._process_push_task] 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: if self._params.enable_heartbeats:
self._heartbeat_push_task = asyncio.create_task(self._heartbeat_push_handler()) self._heartbeat_push_task = asyncio.create_task(self._heartbeat_push_handler())
self._heartbeat_monitor_task = asyncio.create_task(self._heartbeat_monitor_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): async def _cancel_tasks(self, cancel_push: bool):
await self._maybe_cancel_heartbeat_tasks()
if cancel_push: if cancel_push:
self._process_push_task.cancel() self._process_push_task.cancel()
await self._process_push_task await self._process_push_task
@@ -220,6 +221,7 @@ class PipelineTask:
self._process_down_task.cancel() self._process_down_task.cancel()
await self._process_down_task await self._process_down_task
async def _maybe_cancel_heartbeat_tasks(self):
if self._params.enable_heartbeats: if self._params.enable_heartbeats:
self._heartbeat_push_task.cancel() self._heartbeat_push_task.cancel()
await self._heartbeat_push_task await self._heartbeat_push_task
@@ -246,6 +248,8 @@ class PipelineTask:
""" """
self._clock.start() self._clock.start()
self._maybe_start_heartbeat_tasks()
start_frame = StartFrame( start_frame = StartFrame(
allow_interruptions=self._params.allow_interruptions, allow_interruptions=self._params.allow_interruptions,
enable_metrics=self._params.enable_metrics, enable_metrics=self._params.enable_metrics,