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

Use inherited create_task/cancel_task in PipelineTask

Browse Source 
PipelineTask owns its TaskManager but is itself a BaseObject, so it
inherits create_task/cancel_task. Replace the explicit
self._task_manager.create_task(coro, f"{self}::name") call sites with
self.create_task(coro, "name") for consistency with other BaseObject
subclasses.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-08 13:44:59 -07:00
parent df1b071a13
commit dc035df0aa
1 changed files with 12 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -654,32 +654,32 @@ class PipelineTask(BasePipelineTask):
async def _create_tasks(self): async def _create_tasks(self):
"""Create and start all pipeline processing tasks.""" """Create and start all pipeline processing tasks."""
self._process_push_task = self._task_manager.create_task( self._process_push_task = self.create_task(
self._process_push_queue(), f"{self}::_process_push_queue" self._process_push_queue(), "_process_push_queue"
) )
return self._process_push_task return self._process_push_task
def _maybe_start_heartbeat_tasks(self): def _maybe_start_heartbeat_tasks(self):
"""Start heartbeat tasks if heartbeats are enabled and not already running.""" """Start heartbeat tasks if heartbeats are enabled and not already running."""
if self._params.enable_heartbeats and self._heartbeat_push_task is None: 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.create_task(
self._heartbeat_push_handler(), f"{self}::_heartbeat_push_handler" self._heartbeat_push_handler(), "_heartbeat_push_handler"
) )
self._heartbeat_monitor_task = self._task_manager.create_task( self._heartbeat_monitor_task = self.create_task(
self._heartbeat_monitor_handler(), f"{self}::_heartbeat_monitor_handler" self._heartbeat_monitor_handler(), "_heartbeat_monitor_handler"
) )
def _maybe_start_idle_task(self): def _maybe_start_idle_task(self):
"""Start idle monitoring task if idle timeout is configured.""" """Start idle monitoring task if idle timeout is configured."""
if self._idle_timeout_secs: if self._idle_timeout_secs:
self._idle_monitor_task = self._task_manager.create_task( self._idle_monitor_task = self.create_task(
self._idle_monitor_handler(), f"{self}::_idle_monitor_handler" self._idle_monitor_handler(), "_idle_monitor_handler"
) )
async def _cancel_tasks(self): async def _cancel_tasks(self):
"""Cancel all running pipeline tasks.""" """Cancel all running pipeline tasks."""
if self._process_push_task: if self._process_push_task:
await self._task_manager.cancel_task(self._process_push_task) await self.cancel_task(self._process_push_task)
self._process_push_task = None self._process_push_task = None
await self._maybe_cancel_heartbeat_tasks() await self._maybe_cancel_heartbeat_tasks()
@@ -691,17 +691,17 @@ class PipelineTask(BasePipelineTask):
return return
if self._heartbeat_push_task: if self._heartbeat_push_task:
await self._task_manager.cancel_task(self._heartbeat_push_task) await self.cancel_task(self._heartbeat_push_task)
self._heartbeat_push_task = None self._heartbeat_push_task = None
if self._heartbeat_monitor_task: if self._heartbeat_monitor_task:
await self._task_manager.cancel_task(self._heartbeat_monitor_task) await self.cancel_task(self._heartbeat_monitor_task)
self._heartbeat_monitor_task = None self._heartbeat_monitor_task = None
async def _maybe_cancel_idle_task(self): async def _maybe_cancel_idle_task(self):
"""Cancel idle monitoring task if it is running.""" """Cancel idle monitoring task if it is running."""
if self._idle_monitor_task: if self._idle_monitor_task:
await self._task_manager.cancel_task(self._idle_monitor_task) await self.cancel_task(self._idle_monitor_task)
self._idle_monitor_task = None self._idle_monitor_task = None
def _initial_metrics_frame(self) -> MetricsFrame: def _initial_metrics_frame(self) -> MetricsFrame: