diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 20a0be29f..37ae8a167 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -501,6 +501,8 @@ class DailyTransportClient(EventHandler): self._event_task = None self._audio_task = None self._video_task = None + self._message_queue: Optional[asyncio.Queue] = None + self._message_task = None # Input and ouput sample rates. They will be initialize on setup(). self._in_sample_rate = 0 @@ -567,7 +569,9 @@ class DailyTransportClient(EventHandler): error: An error description or None. """ if not self._joined: - return "Unable to send messages before joining." + if self._message_queue: + await self._message_queue.put((frame,)) + return None participant_id = None if isinstance( @@ -673,6 +677,12 @@ class DailyTransportClient(EventHandler): f"{self}::event_callback_task", ) + self._message_queue = asyncio.Queue() + self._message_task = self._task_manager.create_task( + self._message_task_handler(self._message_queue), + f"{self}::message_task", + ) + async def cleanup(self): """Cleanup client resources and cancel tasks.""" if self._event_task and self._task_manager: @@ -684,6 +694,9 @@ class DailyTransportClient(EventHandler): if self._video_task and self._task_manager: await self._task_manager.cancel_task(self._video_task) self._video_task = None + if self._message_task and self._task_manager: + await self._task_manager.cancel_task(self._message_task) + self._message_task = None # Make sure we don't block the event loop in case `client.release()` # takes extra time. await self._get_event_loop().run_in_executor(self._executor, self._cleanup) @@ -1541,6 +1554,14 @@ class DailyTransportClient(EventHandler): await callback(*args) queue.task_done() + async def _message_task_handler(self, queue: asyncio.Queue): + """Handle queued messages after transport is joined.""" + while True: + await self._joined_event.wait() + (frame,) = await queue.get() + await self.send_message(frame) + queue.task_done() + def _get_event_loop(self) -> asyncio.AbstractEventLoop: """Get the event loop from the task manager.""" if not self._task_manager: