From 25165d6e2b221d0ca9b181ccac50c48650ef98ee Mon Sep 17 00:00:00 2001 From: filipi87 Date: Mon, 9 Mar 2026 09:47:45 -0300 Subject: [PATCH] Queuing the messages received before the data channel is ready to send them. --- src/pipecat/transports/smallwebrtc/connection.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/pipecat/transports/smallwebrtc/connection.py b/src/pipecat/transports/smallwebrtc/connection.py index b9b6538f1..bc990d736 100644 --- a/src/pipecat/transports/smallwebrtc/connection.py +++ b/src/pipecat/transports/smallwebrtc/connection.py @@ -283,6 +283,7 @@ class SmallWebRTCConnection(BaseObject): self._data_channel = None self._renegotiation_in_progress = False self._last_received_time = None + self._message_queue = [] self._pending_app_messages = [] self._connecting_timeout_task = None @@ -297,6 +298,7 @@ class SmallWebRTCConnection(BaseObject): @channel.on("open") async def on_open(): logger.debug("Data channel is open!") + self._flush_message_queue() @channel.on("message") async def on_message(message): @@ -499,6 +501,7 @@ class SmallWebRTCConnection(BaseObject): self._track_map.clear() if self._pc: await self._pc.close() + self._message_queue.clear() self._pending_app_messages.clear() self._track_map = {} self._cancel_monitoring_connecting_state() @@ -664,8 +667,14 @@ class SmallWebRTCConnection(BaseObject): if self._data_channel and self._data_channel.readyState == "open": self._data_channel.send(json_message) else: - # The client might choose never to create a data channel. - logger.trace("Data channel not ready, discarding message!") + logger.debug("Data channel not ready, queuing message") + self._message_queue.append(json_message) + + def _flush_message_queue(self): + logger.debug("Data channel is open, flushing queued messages") + while self._message_queue: + message = self._message_queue.pop(0) + self._data_channel.send(message) def ask_to_renegotiate(self): """Request renegotiation of the WebRTC connection."""