From 7c781ce816f58d394162e129d5dbdbb45486076e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 Aug 2025 10:51:16 -0700 Subject: [PATCH] WatchdogPriorityQueue: make WatchdogPriorityCancelSentinel public --- .../utils/asyncio/watchdog_priority_queue.py | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/src/pipecat/utils/asyncio/watchdog_priority_queue.py b/src/pipecat/utils/asyncio/watchdog_priority_queue.py index 900242d1d..f31b3b90a 100644 --- a/src/pipecat/utils/asyncio/watchdog_priority_queue.py +++ b/src/pipecat/utils/asyncio/watchdog_priority_queue.py @@ -20,7 +20,17 @@ from pipecat.utils.asyncio.task_manager import BaseTaskManager @dataclass -class _WatchdogPriorityCancelSentinel: +class WatchdogPriorityCancelSentinel: + """Sentinel object used in priority queues to force cancellation. + + An instance of this class is typically inserted into a + `WatchdogPriorityQueue` to act as a high-priority marker asyncio task + cancellation. The `__lt__` method always returns `True`, ensuring that the + sentinel is considered "less than" any other item in the queue, and + therefore processed before anything else. + + """ + def __lt__(self, other): return True @@ -62,7 +72,7 @@ class WatchdogPriorityQueue(asyncio.PriorityQueue): else: get_result = await super().get() - if isinstance(get_result, _WatchdogPriorityCancelSentinel): + if isinstance(get_result, WatchdogPriorityCancelSentinel): logger.trace( "Received WatchdogPriorityCancelSentinel, throwing CancelledError to force cancelling" ) @@ -91,7 +101,7 @@ class WatchdogPriorityQueue(asyncio.PriorityQueue): forces the task to raise CancelledError when consumed, ensuring proper task termination. """ - super().put_nowait(_WatchdogPriorityCancelSentinel()) + super().put_nowait(WatchdogPriorityCancelSentinel()) async def _watchdog_get(self): """Get item from queue while periodically resetting watchdog timer."""