Set maxMessageSize on transports and...

transports/daily/CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to **Pipecat Daily WebRTC Transport** will be documented in
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+- Added error handling around `daily.sendAppMessage`
+- Set `_maxMessageSize` to the underlyling daily instance's max size (defaults to 10MB)
+
 ## [1.6.0]
 
 - Bump client-js version to work with latest 1.6.0 and support latest features

transports/daily/src/transport.ts

@@ -16,10 +16,12 @@ import Daily, {
   DailyEventObjectTrack,
   DailyFactoryOptions,
   DailyParticipant,
+  DailyRoomInfo,
 } from "@daily-co/daily-js";
 import {
   DeviceArray,
   DeviceError,
+  MessageTooLargeError,
   Participant,
   PipecatClientOptions,
   RTVIError,
@@ -482,6 +484,10 @@ export class DailyTransport extends Transport {
 
     if (this._abortController?.signal.aborted) return;
 
+    const r = await this._daily.room();
+    this._maxMessageSize =
+      (r as DailyRoomInfo)?.domainConfig?.max_app_message_size ||
+      10 * 1024 * 1024;
     this.state = "connected";
 
     this._callbacks.onConnected?.();
@@ -597,7 +603,17 @@ export class DailyTransport extends Transport {
   }
 
   public sendMessage(message: RTVIMessage) {
-    this._daily.sendAppMessage(message, "*");
+    try {
+      this._daily.sendAppMessage(message, "*");
+    } catch (error: unknown) {
+      if (
+        error instanceof Error &&
+        error.message.includes("Message data too large")
+      ) {
+        throw new MessageTooLargeError(error.message);
+      }
+      throw error;
+    }
   }
 
   private handleAppMessage(ev: DailyEventObjectAppMessage) {

transports/openai-realtime-webrtc-transport/CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to **Pipecat OpenAIRealTimeWebRTCTransport** will be documen
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+- Added default case to `sendMessage` to properly throw an `UnsupportedFeatureError`
+
 ## [1.5.0]
 
 - Bump client-js version to work with latest 1.6.0 and support latest features

transports/openai-realtime-webrtc-transport/src/OpenAIRealTimeWebRTCTransport.ts

@@ -11,6 +11,7 @@ import {
   Transport,
   TransportStartError,
   TransportState,
+  UnsupportedFeatureError,
   logger,
 } from "@pipecat-ai/client-js";
 
@@ -388,6 +389,11 @@ export class OpenAIRealTimeWebRTCTransport extends Transport {
         );
         break;
       }
+      default:
+        throw new UnsupportedFeatureError(
+          message.type,
+          "OpenAIRealTimeWebRTCTransport",
+        );
     }
   }
 

transports/small-webrtc-transport/CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to **Pipecat Small WebRTC Transport** will be documented in
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+- Added message size checking to `sendMessage()` to ensure no large
+  messages are sent, causing the data channel to fail and not recover.
+- Set `_maxMessageSize` to the peer connections's max size (defaults to 64KB)
+
 ## [1.9.0]
 
 ### Added

transports/small-webrtc-transport/src/smallWebRTCTransport.ts

@@ -1,7 +1,11 @@
 import cloneDeep from "lodash/cloneDeep";
 import {
+  APIRequest,
+  isAPIRequest,
   logger,
   makeRequest,
+  messageSizeWithinLimit,
+  MessageTooLargeError,
   RTVIError,
   RTVIMessage,
   PipecatClientOptions,
@@ -10,8 +14,6 @@ import {
   TransportStartError,
   TransportState,
   UnsupportedFeatureError,
-  APIRequest,
-  isAPIRequest,
 } from "@pipecat-ai/client-js";
 import { MediaManager } from "../../../lib/media-mgmt/mediaManager";
 import { DailyMediaManager } from "../../../lib/media-mgmt/dailyMediaManager";
@@ -428,6 +430,13 @@ export class SmallWebRTCTransport extends Transport {
       logger.warn(`Datachannel is not ready. Message not sent: ${message}`);
       return;
     }
+
+    // Note: This shouldn't happen since client-js should have already checked this
+    if (!messageSizeWithinLimit(message, this._maxMessageSize)) {
+      throw new MessageTooLargeError(
+        "Message data too large. Max size is " + this._maxMessageSize,
+      );
+    }
     this.dc?.send(JSON.stringify(message));
   }
 
@@ -903,6 +912,7 @@ export class SmallWebRTCTransport extends Transport {
 
     dc.addEventListener("open", () => {
       logger.debug("datachannel opened");
+      this._maxMessageSize = this.pc?.sctp?.maxMessageSize ?? 64 * 1024;
       if (this._connectResolved) {
         this.syncTrackStatus();
         this._connectResolved();

transports/websocket-transport/CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to **Pipecat Websocket Transport** will be documented in thi
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+- Set `_maxMessageSize` to the server's supported max size (1 MB)
+
 ## [1.6.0]
 
 - Bump client-js version to work with latest 1.6.0 and support latest features

transports/websocket-transport/src/webSocketTransport.ts

@@ -61,6 +61,7 @@ export class WebSocketTransport extends Transport {
     );
     this._ws = null;
     this._serializer = opts.serializer || new ProtobufFrameSerializer();
+    this._maxMessageSize = 1024 * 1024; // python websockets default to 1MB
   }
 
   initialize(