diff --git a/src/pipecat/bus/bus.py b/src/pipecat/bus/bus.py index 009419cdb..e2e7169dc 100644 --- a/src/pipecat/bus/bus.py +++ b/src/pipecat/bus/bus.py @@ -15,10 +15,9 @@ from abc import abstractmethod from dataclasses import dataclass, field from typing import cast -from pipecat.bus.messages import BusLocalMessage, BusMessage +from pipecat.bus.messages import BusLocalMessage, BusMessage, BusSystemMessage from pipecat.bus.queue import BusMessageQueue from pipecat.bus.subscriber import BusSubscriber -from pipecat.frames.frames import SystemFrame from pipecat.utils.base_object import BaseObject @@ -167,7 +166,7 @@ class WorkerBus(BaseObject): try: while True: message = await sub.queue.get() - if isinstance(message, SystemFrame): + if isinstance(message, BusSystemMessage): await sub.subscriber.on_bus_message(cast(BusMessage, message)) else: sub.data_queue.put_nowait(message) diff --git a/src/pipecat/bus/messages.py b/src/pipecat/bus/messages.py index 3ed32eac6..ad031e8ce 100644 --- a/src/pipecat/bus/messages.py +++ b/src/pipecat/bus/messages.py @@ -15,7 +15,7 @@ from __future__ import annotations from dataclasses import dataclass from typing import TYPE_CHECKING -from pipecat.frames.frames import DataFrame, Frame, SystemFrame +from pipecat.frames.frames import Frame from pipecat.processors.frame_processor import FrameDirection from pipecat.registry.types import WorkerRegistryEntry @@ -28,11 +28,19 @@ if TYPE_CHECKING: # --------------------------------------------------------------------------- +@dataclass(kw_only=True) class BusMessage: - """Mixin carrying source/target metadata for bus messages. + """Base class for messages carried by the `WorkerBus`. - Not a frame itself. Combined with `DataFrame` or `SystemFrame` - to create concrete message types with appropriate priority. + Bus messages are independent of pipeline `Frame`s — if a worker needs + to ship a frame between pipelines it wraps it in a `BusFrameMessage`. + Subclasses choose delivery priority by extending :class:`BusDataMessage` + (normal priority, FIFO) or :class:`BusSystemMessage` (high priority, + delivered ahead of queued data messages). + + Parameters: + source: Name of the worker or component that sent this message. + target: Name of the intended recipient worker, or None for broadcast. """ source: str @@ -49,29 +57,24 @@ class BusLocalMessage: @dataclass(kw_only=True) -class BusDataMessage(BusMessage, DataFrame): +class BusDataMessage(BusMessage): """Normal-priority bus message. - Parameters: - source: Name of the worker or component that sent this message. - target: Name of the intended recipient worker, or None for broadcast. + Delivered in FIFO order on the subscriber's data queue. """ - source: str - target: str | None = None + pass @dataclass(kw_only=True) -class BusSystemMessage(BusMessage, SystemFrame): - """High-priority bus message that preempts normal messages in subscriber queues. +class BusSystemMessage(BusMessage): + """High-priority bus message. - Parameters: - source: Name of the worker or component that sent this message. - target: Name of the intended recipient worker, or None for broadcast. + Delivered ahead of any queued :class:`BusDataMessage` on the + subscriber's priority queue. """ - source: str - target: str | None = None + pass # --------------------------------------------------------------------------- diff --git a/src/pipecat/bus/queue.py b/src/pipecat/bus/queue.py index 343edd1f4..c673b394f 100644 --- a/src/pipecat/bus/queue.py +++ b/src/pipecat/bus/queue.py @@ -9,7 +9,7 @@ import asyncio from typing import Any -from pipecat.frames.frames import SystemFrame +from pipecat.bus.messages import BusSystemMessage HIGH_PRIORITY = 1 LOW_PRIORITY = 2 @@ -18,9 +18,9 @@ LOW_PRIORITY = 2 class BusMessageQueue(asyncio.PriorityQueue): """Priority queue that delivers system messages before normal messages. - Messages that extend ``SystemFrame`` (e.g. cancel messages) get high - priority. All other messages are delivered in FIFO order at normal - priority. + Messages that extend :class:`BusSystemMessage` (e.g. cancel messages) + get high priority. All other messages are delivered in FIFO order at + normal priority. """ def __init__(self): @@ -35,7 +35,7 @@ class BusMessageQueue(asyncio.PriorityQueue): Args: item: The bus message to enqueue. """ - if isinstance(item, SystemFrame): + if isinstance(item, BusSystemMessage): self._high_counter += 1 super().put_nowait((HIGH_PRIORITY, self._high_counter, item)) else: @@ -48,7 +48,7 @@ class BusMessageQueue(asyncio.PriorityQueue): Args: item: The bus message to enqueue. """ - if isinstance(item, SystemFrame): + if isinstance(item, BusSystemMessage): self._high_counter += 1 await super().put((HIGH_PRIORITY, self._high_counter, item)) else: diff --git a/tests/test_serializers.py b/tests/test_serializers.py index 775f83d32..ef5f80f41 100644 --- a/tests/test_serializers.py +++ b/tests/test_serializers.py @@ -5,6 +5,7 @@ # import unittest +from dataclasses import dataclass, field from pydantic import BaseModel @@ -35,6 +36,11 @@ class _UserInfo(BaseModel): address: _Address | None = None +@dataclass(kw_only=True) +class _MessageWithNonInit(BusDataMessage): + tag: str = field(init=False, default="default") + + class TestJSONMessageSerializer(unittest.TestCase): def setUp(self): self.serializer = JSONMessageSerializer() @@ -232,14 +238,13 @@ class TestJSONMessageSerializer(unittest.TestCase): def test_non_init_fields_preserved(self): """Non-init dataclass fields survive round-trip via setattr.""" - msg = BusDataMessage(source="task_a", target="task_b") - # name is a non-init field on DataFrame - msg.name = "custom_name" + msg = _MessageWithNonInit(source="worker_a", target="worker_b") + msg.tag = "custom_tag" data = self.serializer.serialize(msg) restored = self.serializer.deserialize(data) - self.assertEqual(restored.name, "custom_name") + self.assertEqual(restored.tag, "custom_tag") if __name__ == "__main__":