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

Rename BaseTask.send_message to send_bus_message

Browse Source 
Mirrors on_bus_message and makes it explicit that the call goes out on
the task bus, not on a transport (transports have their own
send_message for client/peer messaging).
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-20 15:57:29 -07:00
parent 373894fc65
commit d8947c68a9
4 changed files with 32 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/pipecat/pipeline/base_task.py
View File

@@ -360,11 +360,11 @@ class BaseTask(BaseObject, BusSubscriber):
Args: Args:
reason: Optional human-readable reason for ending. reason: Optional human-readable reason for ending.
""" """
await self.send_message(BusEndMessage(source=self.name, reason=reason)) await self.send_bus_message(BusEndMessage(source=self.name, reason=reason))
async def cancel(self) -> None: async def cancel(self) -> None:
"""Request an immediate cancellation of all tasks.""" """Request an immediate cancellation of all tasks."""
await self.send_message(BusCancelMessage(source=self.name)) await self.send_bus_message(BusCancelMessage(source=self.name))
async def wait(self) -> None: async def wait(self) -> None:
"""Wait for this task to finish.""" """Wait for this task to finish."""
@@ -520,7 +520,7 @@ class BaseTask(BaseObject, BusSubscriber):
""" """
pass pass
async def send_message(self, message: BusMessage) -> None: async def send_bus_message(self, message: BusMessage) -> None:
"""Send a message on the bus. """Send a message on the bus.
Args: Args:
@@ -539,9 +539,9 @@ class BaseTask(BaseObject, BusSubscriber):
error: Description of the error. error: Description of the error.
""" """
if self._parent: if self._parent:
await self.send_message(BusTaskLocalErrorMessage(source=self.name, error=error)) await self.send_bus_message(BusTaskLocalErrorMessage(source=self.name, error=error))
else: else:
await self.send_message(BusTaskErrorMessage(source=self.name, error=error)) await self.send_bus_message(BusTaskErrorMessage(source=self.name, error=error))
async def add_task(self, task: "BaseTask") -> None: async def add_task(self, task: "BaseTask") -> None:
"""Register a child task under this parent. """Register a child task under this parent.
@@ -558,7 +558,7 @@ class BaseTask(BaseObject, BusSubscriber):
return return
task._parent = self.name task._parent = self.name
self._children.append(task) self._children.append(task)
await self.send_message(BusAddTaskMessage(source=self.name, task=task)) await self.send_bus_message(BusAddTaskMessage(source=self.name, task=task))
async def activate_task( async def activate_task(
self, self,
@@ -581,7 +581,7 @@ class BaseTask(BaseObject, BusSubscriber):
""" """
if self._active and deactivate_self: if self._active and deactivate_self:
await self.deactivate_task(self.name) await self.deactivate_task(self.name)
await self.send_message( await self.send_bus_message(
BusActivateTaskMessage( BusActivateTaskMessage(
source=self.name, target=task_name, args=args.to_dict() if args else None source=self.name, target=task_name, args=args.to_dict() if args else None
) )
@@ -595,7 +595,7 @@ class BaseTask(BaseObject, BusSubscriber):
Args: Args:
task_name: The name of the task to deactivate. task_name: The name of the task to deactivate.
""" """
await self.send_message(BusDeactivateTaskMessage(source=self.name, target=task_name)) await self.send_bus_message(BusDeactivateTaskMessage(source=self.name, target=task_name))
async def watch_task(self, task_name: str) -> None: async def watch_task(self, task_name: str) -> None:
"""Request notification when a task registers. """Request notification when a task registers.
@@ -798,7 +798,7 @@ class BaseTask(BaseObject, BusSubscriber):
if group.timeout_task: if group.timeout_task:
await self.cancel_task(group.timeout_task) await self.cancel_task(group.timeout_task)
for task_name in group.task_names: for task_name in group.task_names:
await self.send_message( await self.send_bus_message(
BusJobCancelMessage( BusJobCancelMessage(
source=self.name, target=task_name, job_id=job_id, reason=reason source=self.name, target=task_name, job_id=job_id, reason=reason
) )
@@ -812,7 +812,7 @@ class BaseTask(BaseObject, BusSubscriber):
job_id: The job identifier. job_id: The job identifier.
task_name: The name of the worker to request an update from. task_name: The name of the worker to request an update from.
""" """
await self.send_message( await self.send_bus_message(
BusJobUpdateRequestMessage(source=self.name, target=task_name, job_id=job_id) BusJobUpdateRequestMessage(source=self.name, target=task_name, job_id=job_id)
) )
@@ -886,7 +886,7 @@ class BaseTask(BaseObject, BusSubscriber):
if request is None: if request is None:
raise RuntimeError(f"Task '{self}': no active job '{job_id}' to respond to") raise RuntimeError(f"Task '{self}': no active job '{job_id}' to respond to")
msg_class = BusJobResponseUrgentMessage if urgent else BusJobResponseMessage msg_class = BusJobResponseUrgentMessage if urgent else BusJobResponseMessage
await self.send_message( await self.send_bus_message(
msg_class( msg_class(
source=self.name, source=self.name,
target=request.source, target=request.source,
@@ -915,7 +915,7 @@ class BaseTask(BaseObject, BusSubscriber):
if request is None: if request is None:
raise RuntimeError(f"Task '{self}': no active job '{job_id}' to update") raise RuntimeError(f"Task '{self}': no active job '{job_id}' to update")
msg_class = BusJobUpdateUrgentMessage if urgent else BusJobUpdateMessage msg_class = BusJobUpdateUrgentMessage if urgent else BusJobUpdateMessage
await self.send_message( await self.send_bus_message(
msg_class( msg_class(
source=self.name, source=self.name,
target=request.source, target=request.source,
@@ -937,7 +937,7 @@ class BaseTask(BaseObject, BusSubscriber):
request = self._active_jobs.get(job_id) request = self._active_jobs.get(job_id)
if request is None: if request is None:
raise RuntimeError(f"Task '{self}': no active job '{job_id}' to stream") raise RuntimeError(f"Task '{self}': no active job '{job_id}' to stream")
await self.send_message( await self.send_bus_message(
BusJobStreamStartMessage( BusJobStreamStartMessage(
source=self.name, source=self.name,
target=request.source, target=request.source,
@@ -959,7 +959,7 @@ class BaseTask(BaseObject, BusSubscriber):
request = self._active_jobs.get(job_id) request = self._active_jobs.get(job_id)
if request is None: if request is None:
raise RuntimeError(f"Task '{self}': no active job '{job_id}' to stream") raise RuntimeError(f"Task '{self}': no active job '{job_id}' to stream")
await self.send_message( await self.send_bus_message(
BusJobStreamDataMessage( BusJobStreamDataMessage(
source=self.name, source=self.name,
target=request.source, target=request.source,
@@ -981,7 +981,7 @@ class BaseTask(BaseObject, BusSubscriber):
request = self._active_jobs.get(job_id) request = self._active_jobs.get(job_id)
if request is None: if request is None:
raise RuntimeError(f"Task '{self}': no active job '{job_id}' to stream") raise RuntimeError(f"Task '{self}': no active job '{job_id}' to stream")
await self.send_message( await self.send_bus_message(
BusJobStreamEndMessage( BusJobStreamEndMessage(
source=self.name, source=self.name,
target=request.source, target=request.source,
@@ -1001,7 +1001,7 @@ class BaseTask(BaseObject, BusSubscriber):
# fires watchers synchronously, which may send additional # fires watchers synchronously, which may send additional
# messages (e.g. ActivateTask). Sending the ready message # messages (e.g. ActivateTask). Sending the ready message
# first preserves correct chronological order for observers. # first preserves correct chronological order for observers.
await self.send_message( await self.send_bus_message(
BusTaskReadyMessage( BusTaskReadyMessage(
source=self.name, source=self.name,
runner=self._registry.runner_name, runner=self._registry.runner_name,
@@ -1111,7 +1111,7 @@ class BaseTask(BaseObject, BusSubscriber):
async def _propagate_end_to_children(self, message: BusEndTaskMessage) -> None: async def _propagate_end_to_children(self, message: BusEndTaskMessage) -> None:
"""Forward a ``BusEndTaskMessage`` to each child and wait for them.""" """Forward a ``BusEndTaskMessage`` to each child and wait for them."""
for child in self._children: for child in self._children:
await self.send_message( await self.send_bus_message(
BusEndTaskMessage(source=self.name, target=child.name, reason=message.reason) BusEndTaskMessage(source=self.name, target=child.name, reason=message.reason)
) )
await asyncio.gather(*(child.wait() for child in self._children)) await asyncio.gather(*(child.wait() for child in self._children))
@@ -1119,7 +1119,7 @@ class BaseTask(BaseObject, BusSubscriber):
async def _propagate_cancel_to_children(self, message: BusCancelTaskMessage) -> None: async def _propagate_cancel_to_children(self, message: BusCancelTaskMessage) -> None:
"""Forward a ``BusCancelTaskMessage`` to each child.""" """Forward a ``BusCancelTaskMessage`` to each child."""
for child in self._children: for child in self._children:
await self.send_message( await self.send_bus_message(
BusCancelTaskMessage(source=self.name, target=child.name, reason=message.reason) BusCancelTaskMessage(source=self.name, target=child.name, reason=message.reason)
) )
@@ -1298,7 +1298,7 @@ class BaseTask(BaseObject, BusSubscriber):
job_name: str | None = None, job_name: str | None = None,
payload: dict | None = None, payload: dict | None = None,
) -> None: ) -> None:
await self.send_message( await self.send_bus_message(
BusJobRequestMessage( BusJobRequestMessage(
source=self.name, source=self.name,
target=task_name, target=task_name,

6
src/pipecat/tasks/proxy/websocket/client.py
View File

@@ -188,13 +188,13 @@ class WebSocketProxyClientTask(BaseTask):
logger.trace( logger.trace(
f"Task '{self}': received registry from remote: {message.tasks}" f"Task '{self}': received registry from remote: {message.tasks}"
) )
await self.send_message(message) await self.send_bus_message(message)
continue continue
# Accept additional message types (e.g. BusFrameMessage). # Accept additional message types (e.g. BusFrameMessage).
if self._forward_messages and isinstance(message, self._forward_messages): if self._forward_messages and isinstance(message, self._forward_messages):
logger.trace(f"Task '{self}': received {message} from remote") logger.trace(f"Task '{self}': received {message} from remote")
await self.send_message(message) await self.send_bus_message(message)
continue continue
# Only accept other messages targeted at the local task. # Only accept other messages targeted at the local task.
@@ -206,7 +206,7 @@ class WebSocketProxyClientTask(BaseTask):
continue continue
logger.trace(f"Task '{self}': received {message} from remote") logger.trace(f"Task '{self}': received {message} from remote")
await self.send_message(message) await self.send_bus_message(message)
except Exception: except Exception:
logger.exception(f"Task '{self}': failed to deserialize remote message") logger.exception(f"Task '{self}': failed to deserialize remote message")
except websockets.exceptions.ConnectionClosed: except websockets.exceptions.ConnectionClosed:

4
src/pipecat/tasks/proxy/websocket/server.py
View File

@@ -197,7 +197,7 @@ class WebSocketProxyServerTask(BaseTask):
# Accept additional message types (e.g. BusFrameMessage). # Accept additional message types (e.g. BusFrameMessage).
if self._forward_messages and isinstance(message, self._forward_messages): if self._forward_messages and isinstance(message, self._forward_messages):
logger.trace(f"Task '{self}': received {message} from client") logger.trace(f"Task '{self}': received {message} from client")
await self.send_message(message) await self.send_bus_message(message)
continue continue
# Only accept other messages targeted at the local task. # Only accept other messages targeted at the local task.
@@ -209,7 +209,7 @@ class WebSocketProxyServerTask(BaseTask):
continue continue
logger.trace(f"Task '{self}': received {message} from client") logger.trace(f"Task '{self}': received {message} from client")
await self.send_message(message) await self.send_bus_message(message)
except Exception: except Exception:
logger.exception(f"Task '{self}': failed to deserialize client message") logger.exception(f"Task '{self}': failed to deserialize client message")
except WebSocketDisconnect: except WebSocketDisconnect:

16
tests/test_websocket_proxy.py
View File

@@ -154,13 +154,13 @@ class TestWebSocketProxyClientTask(unittest.IsolatedAsyncioTestCase):
task = await self._create_client(fake_ws) task = await self._create_client(fake_ws)
sent_to_bus = [] sent_to_bus = []
original_send = task.send_message original_send = task.send_bus_message
async def capture_send(message): async def capture_send(message):
sent_to_bus.append(message) sent_to_bus.append(message)
await original_send(message) await original_send(message)
task.send_message = capture_send task.send_bus_message = capture_send
msg = BusDataMessage(source="worker", target="voice") msg = BusDataMessage(source="worker", target="voice")
fake_ws.inject(self.serializer.serialize(msg)) fake_ws.inject(self.serializer.serialize(msg))
@@ -183,13 +183,13 @@ class TestWebSocketProxyClientTask(unittest.IsolatedAsyncioTestCase):
task = await self._create_client(fake_ws) task = await self._create_client(fake_ws)
sent_to_bus = [] sent_to_bus = []
original_send = task.send_message original_send = task.send_bus_message
async def capture_send(message): async def capture_send(message):
sent_to_bus.append(message) sent_to_bus.append(message)
await original_send(message) await original_send(message)
task.send_message = capture_send task.send_bus_message = capture_send
msg = BusDataMessage(source="worker", target="other_task") msg = BusDataMessage(source="worker", target="other_task")
fake_ws.inject(self.serializer.serialize(msg)) fake_ws.inject(self.serializer.serialize(msg))
@@ -264,13 +264,13 @@ class TestWebSocketProxyServerTask(unittest.IsolatedAsyncioTestCase):
task = await self._create_server(fake_ws) task = await self._create_server(fake_ws)
sent_to_bus = [] sent_to_bus = []
original_send = task.send_message original_send = task.send_bus_message
async def capture_send(message): async def capture_send(message):
sent_to_bus.append(message) sent_to_bus.append(message)
await original_send(message) await original_send(message)
task.send_message = capture_send task.send_bus_message = capture_send
msg = BusDataMessage(source="voice", target="worker") msg = BusDataMessage(source="voice", target="worker")
fake_ws.inject(self.serializer.serialize(msg)) fake_ws.inject(self.serializer.serialize(msg))
@@ -292,13 +292,13 @@ class TestWebSocketProxyServerTask(unittest.IsolatedAsyncioTestCase):
task = await self._create_server(fake_ws) task = await self._create_server(fake_ws)
sent_to_bus = [] sent_to_bus = []
original_send = task.send_message original_send = task.send_bus_message
async def capture_send(message): async def capture_send(message):
sent_to_bus.append(message) sent_to_bus.append(message)
await original_send(message) await original_send(message)
task.send_message = capture_send task.send_bus_message = capture_send
msg = BusDataMessage(source="voice", target="other_task") msg = BusDataMessage(source="voice", target="other_task")
fake_ws.inject(self.serializer.serialize(msg)) fake_ws.inject(self.serializer.serialize(msg))