diff --git a/CHANGELOG.md b/CHANGELOG.md index aa7f46d3e..f54ba1ab9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -81,6 +81,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `DailyTransport` issue that would result in an unhandled + `concurrent.futures.CancelledError` when a future is cancelled. + - Fixed a `RivaSTTService` issue that would result in an unhandled `concurrent.futures.CancelledError` when a future is cancelled when reading from the audio chunks from the incoming audio stream. diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 1ad443cb5..daf879be7 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -13,6 +13,7 @@ real-time communication features. import asyncio import time +from concurrent.futures import CancelledError as FuturesCancelledError from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from typing import Any, Awaitable, Callable, Dict, Mapping, Optional @@ -1320,10 +1321,13 @@ class DailyTransportClient(EventHandler): def _call_async_callback(self, queue: asyncio.Queue, callback, *args): """Queue a callback for async execution on the event loop.""" - future = asyncio.run_coroutine_threadsafe( - queue.put((callback, *args)), self._get_event_loop() - ) - future.result() + try: + future = asyncio.run_coroutine_threadsafe( + queue.put((callback, *args)), self._get_event_loop() + ) + future.result() + except FuturesCancelledError: + pass async def _callback_task_handler(self, queue: asyncio.Queue): """Handle queued callbacks from the specified queue."""