transports(daily): send transport messages in a task

We queue transport messages and send them in a task to avoid potential hangs by
sending urgent transport messages from a transport event handler.
This commit is contained in:
Aleix Conchillo Flaqué
2024-10-14 16:19:53 -07:00
parent 96bfcc3dca
commit 7b9712daad
2 changed files with 30 additions and 2 deletions

View File

@@ -13,6 +13,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
`filter_code` to filter code from text and `filter_tables` to filter tables
from text.
### Fixed
- Fixed an issue in Daily transport that would cause tasks to be hanging if
urgent transport messages were being sent from a transport event handler.
## [0.0.43] - 2024-10-10
### Added

View File

@@ -720,15 +720,27 @@ class DailyOutputTransport(BaseOutputTransport):
self._client = client
# Task to process outgoing messages.
self._messages_task = None
self._messages_queue = asyncio.Queue()
async def start(self, frame: StartFrame):
# Parent start.
await super().start(frame)
# Join the room.
await self._client.join()
# Start messages task
self._messages_task = self.get_event_loop().create_task(self._messages_task_handler())
async def stop(self, frame: EndFrame):
# Parent stop.
await super().stop(frame)
# Cancel messages task
if self._messages_task:
self._messages_task.cancel()
await self._messages_task
self._messages_task = None
self._messages_task = None
# Leave the room.
await self._client.leave()
@@ -743,7 +755,7 @@ class DailyOutputTransport(BaseOutputTransport):
await self._client.cleanup()
async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
await self._client.send_message(frame)
await self._messages_queue.put(frame)
async def send_metrics(self, frame: MetricsFrame):
metrics = {}
@@ -768,7 +780,7 @@ class DailyOutputTransport(BaseOutputTransport):
message = DailyTransportMessageFrame(
message={"type": "pipecat-metrics", "metrics": metrics}
)
await self._client.send_message(message)
await self._messages_queue.put(message)
async def write_raw_audio_frames(self, frames: bytes):
await self._client.write_raw_audio_frames(frames)
@@ -776,6 +788,17 @@ class DailyOutputTransport(BaseOutputTransport):
async def write_frame_to_camera(self, frame: OutputImageRawFrame):
await self._client.write_frame_to_camera(frame)
async def _messages_task_handler(self):
while True:
try:
message = await self._messages_queue.get()
await self._client.send_message(message)
self._messages_queue.task_done()
except asyncio.CancelledError:
break
except Exception as e:
logger.exception(f"{self} error processing message queue: {e}")
class DailyTransport(BaseTransport):
def __init__(