Queuing the messages received before the data channel is ready to send them.

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."""