add bounded buffer to audio_stream, and use 10 frames as the default

.changeset/add_bounded_buffer_to_audio_stream_and_use_10_frames_as_the_default.md

@@ -0,0 +1,19 @@
+---
+livekit-protocol: patch
+livekit-wakeword: patch
+livekit-ffi: patch
+livekit-api: patch
+soxr-sys: patch
+webrtc-sys: patch
+webrtc-sys-build: patch
+imgproc: patch
+yuv-sys: patch
+libwebrtc: patch
+livekit: patch
+---
+
+# add bounded buffer to audio_stream, and use 10 frames as the default
+
+#945 by @xianshijing-lk
+
+

libwebrtc/Cargo.toml

@@ -35,6 +35,7 @@ lazy_static = { workspace = true }
 parking_lot = { workspace = true }
 tokio = { workspace = true, default-features = false, features = ["sync", "macros"] }
 cxx = "1.0"
+rtrb = "0.3.3"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 wasm-bindgen = "0.2"

libwebrtc/src/audio_stream.rs

@@ -27,6 +27,22 @@ pub mod native {
     use super::stream_imp;
     use crate::{audio_frame::AudioFrame, audio_track::RtcAudioTrack};
 
+    const DEFAULT_QUEUE_SIZE_FRAMES: usize = 10;
+
+    #[derive(Clone, Debug, Default)]
+    pub struct NativeAudioStreamOptions {
+        /// Maximum number of queued WebRTC sink frames after the audio callback.
+        ///
+        /// Each queued frame corresponds to roughly 10 ms of decoded PCM audio
+        /// on the WebRTC sink path.
+        ///
+        /// `None` uses the default bounded queue size of 10 frames. `Some(0)`
+        /// opts into unbounded buffering. Positive values bound the queue, and
+        /// the stream drops the oldest queued frames on overflow so latency
+        /// stays bounded.
+        pub queue_size_frames: Option<usize>,
+    }
+
     pub struct NativeAudioStream {
         pub(crate) handle: stream_imp::NativeAudioStream,
     }
@@ -40,7 +56,28 @@ pub mod native {
     impl NativeAudioStream {
         pub fn new(audio_track: RtcAudioTrack, sample_rate: i32, num_channels: i32) -> Self {
             Self {
-                handle: stream_imp::NativeAudioStream::new(audio_track, sample_rate, num_channels),
+                handle: stream_imp::NativeAudioStream::new(
+                    audio_track,
+                    sample_rate,
+                    num_channels,
+                    Some(DEFAULT_QUEUE_SIZE_FRAMES),
+                ),
+            }
+        }
+
+        pub fn with_options(
+            audio_track: RtcAudioTrack,
+            sample_rate: i32,
+            num_channels: i32,
+            options: NativeAudioStreamOptions,
+        ) -> Self {
+            Self {
+                handle: stream_imp::NativeAudioStream::new(
+                    audio_track,
+                    sample_rate,
+                    num_channels,
+                    normalize_queue_size_frames(options.queue_size_frames),
+                ),
             }
         }
 
@@ -60,4 +97,12 @@ pub mod native {
             Pin::new(&mut self.get_mut().handle).poll_next(cx)
         }
     }
+
+    fn normalize_queue_size_frames(queue_size_frames: Option<usize>) -> Option<usize> {
+        match queue_size_frames {
+            None => Some(DEFAULT_QUEUE_SIZE_FRAMES),
+            Some(0) => None,
+            Some(value) => Some(value),
+        }
+    }
 }