capture blank frames on AudioSource & VideoSource (#258)

libwebrtc/src/audio_source.rs

@@ -34,6 +34,8 @@ impl RtcAudioSource {
         [Native];
         fn set_audio_options(self: &Self, options: AudioSourceOptions) -> ();
         fn audio_options(self: &Self) -> AudioSourceOptions;
+        fn sample_rate(self: &Self) -> u32;
+        fn num_channels(self: &Self) -> u32;
     );
 }
 

libwebrtc/src/native/audio_source.rs

@@ -33,6 +33,8 @@ pub struct NativeAudioSource {
 struct AudioSourceInner {
     buf: Box<[i16]>,
 
+    captured_frames: usize,
+
     // Amount of data from the previous frame that hasn't been sent to the libwebrtc source
     // (because it requires 10ms of data)
     len: usize,
@@ -51,18 +53,47 @@ impl NativeAudioSource {
     ) -> NativeAudioSource {
         let samples_10ms = (sample_rate / 100 * num_channels) as usize;
 
-        Self {
+        let source = Self {
             sys_handle: sys_at::ffi::new_audio_track_source(options.into()),
             inner: Arc::new(AsyncMutex::new(AudioSourceInner {
                 buf: vec![0; samples_10ms].into_boxed_slice(),
+                captured_frames: 0,
                 len: 0,
                 read_offset: 0,
                 interval: None, // interval must be created from a tokio runtime context
             })),
             sample_rate,
             num_channels,
             samples_10ms,
-        }
+        };
+
+        tokio::spawn({
+            let source = source.clone();
+            async move {
+                let mut interval = interval(Duration::from_millis(10));
+
+                loop {
+                    // We directly use the sys_handle instead of the capture_frame function
+                    // (We don't want to increase the captured_frames count and no need to buffer)
+                    interval.tick().await;
+
+                    let inner = source.inner.lock().await;
+                    if inner.captured_frames > 0 {
+                        break; // User captured something, stop injecting silence
+                    }
+
+                    let data = vec![0; samples_10ms];
+                    source.sys_handle.on_captured_frame(
+                        &data,
+                        sample_rate,
+                        num_channels,
+                        sample_rate as usize / 100,
+                    );
+                }
+            }
+        });
+
+        source
     }
 
     pub fn sys_handle(&self) -> SharedPtr<sys_at::ffi::AudioTrackSource> {
@@ -131,6 +162,8 @@ impl NativeAudioSource {
         }
 
         let mut inner = self.inner.lock().await;
+        inner.captured_frames += 1;
+
         let mut interval = inner.interval.take().unwrap_or_else(|| {
             let mut interval = interval(Duration::from_millis(10));
             interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
@@ -145,11 +178,12 @@ impl NativeAudioSource {
 
             interval.tick().await;
 
+            // samples per channel = number of frames
             let samples_per_channel = data.len() / self.num_channels as usize;
             self.sys_handle.on_captured_frame(
                 data,
-                self.sample_rate as i32,
-                self.num_channels as usize,
+                self.sample_rate,
+                self.num_channels,
                 samples_per_channel,
             );
         }

libwebrtc/src/native/video_source.rs

@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::video_frame::{VideoBuffer, VideoFrame};
+use crate::video_frame::{I420Buffer, VideoBuffer, VideoFrame};
 use crate::video_source::VideoResolution;
 use cxx::SharedPtr;
-use std::time::{SystemTime, UNIX_EPOCH};
+use parking_lot::Mutex;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use webrtc_sys::video_frame as vf_sys;
+use webrtc_sys::video_frame::ffi::VideoRotation;
 use webrtc_sys::video_track as vt_sys;
 
 impl From<vt_sys::ffi::VideoResolution> for VideoResolution {
@@ -40,22 +43,65 @@ impl From<VideoResolution> for vt_sys::ffi::VideoResolution {
 #[derive(Clone)]
 pub struct NativeVideoSource {
     sys_handle: SharedPtr<vt_sys::ffi::VideoTrackSource>,
+    inner: Arc<Mutex<VideoSourceInner>>,
+}
+
+struct VideoSourceInner {
+    captured_frames: usize,
 }
 
 impl NativeVideoSource {
     pub fn new(resolution: VideoResolution) -> NativeVideoSource {
-        Self {
+        let source = Self {
             sys_handle: vt_sys::ffi::new_video_track_source(&vt_sys::ffi::VideoResolution::from(
-                resolution,
+                resolution.clone(),
             )),
-        }
+            inner: Arc::new(Mutex::new(VideoSourceInner { captured_frames: 0 })),
+        };
+
+        tokio::spawn({
+            let source = source.clone();
+            let i420 = I420Buffer::new(resolution.width, resolution.height);
+            async move {
+                let mut interval = tokio::time::interval(Duration::from_millis(100)); // 10 fps
+
+                loop {
+                    interval.tick().await;
+
+                    let inner = source.inner.lock();
+                    if inner.captured_frames > 0 {
+                        break;
+                    }
+
+                    let mut builder = vf_sys::ffi::new_video_frame_builder();
+                    builder
+                        .pin_mut()
+                        .set_rotation(VideoRotation::VideoRotation0);
+                    builder
+                        .pin_mut()
+                        .set_video_frame_buffer(i420.as_ref().sys_handle());
+
+                    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
+                    builder.pin_mut().set_timestamp_us(now.as_micros() as i64);
+
+                    source
+                        .sys_handle
+                        .on_captured_frame(&builder.pin_mut().build());
+                }
+            }
+        });
+
+        source
     }
 
     pub fn sys_handle(&self) -> SharedPtr<vt_sys::ffi::VideoTrackSource> {
         self.sys_handle.clone()
     }
 
     pub fn capture_frame<T: AsRef<dyn VideoBuffer>>(&self, frame: &VideoFrame<T>) {
+        let mut inner = self.inner.lock();
+        inner.captured_frames += 1;
+
         let mut builder = vf_sys::ffi::new_video_frame_builder();
         builder.pin_mut().set_rotation(frame.rotation.into());
         builder

libwebrtc/src/video_source.rs

@@ -1,4 +1,4 @@
-// Copyright 2023 LiveKit, Inc.
+// Copyright 2024 LiveKit, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

livekit/src/room/participant/local_participant.rs

@@ -29,6 +29,8 @@ use parking_lot::Mutex;
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::sleep;
 
 type LocalTrackPublishedHandler = Box<dyn Fn(LocalParticipant, LocalTrackPublication) + Send>;
 type LocalTrackUnpublishedHandler = Box<dyn Fn(LocalParticipant, LocalTrackPublication) + Send>;
@@ -227,7 +229,6 @@ impl LocalParticipant {
         {
             local_track_published(self.clone(), publication.clone());
         }
-
         track.enable();
 
         Ok(publication)

webrtc-sys/include/livekit/audio_track.h

@@ -94,8 +94,8 @@ class AudioTrackSource {
     // AudioFrame should always contain 10 ms worth of data (see index.md of
     // acm)
     void on_captured_frame(rust::Slice<const int16_t> audio_data,
-                           int sample_rate,
-                           size_t number_of_channels,
+                           uint32_t sample_rate,
+                           uint32_t number_of_channels,
                            size_t number_of_frames);
 
    private:
@@ -112,8 +112,8 @@ class AudioTrackSource {
   void set_audio_options(const AudioSourceOptions& options) const;
 
   void on_captured_frame(rust::Slice<const int16_t> audio_data,
-                         int sample_rate,
-                         size_t number_of_channels,
+                         uint32_t sample_rate,
+                         uint32_t number_of_channels,
                          size_t number_of_frames) const;
 
   rtc::scoped_refptr<InternalSource> get() const;

webrtc-sys/src/audio_track.cpp

@@ -131,8 +131,8 @@ void AudioTrackSource::InternalSource::RemoveSink(
 
 void AudioTrackSource::InternalSource::on_captured_frame(
     rust::Slice<const int16_t> data,
-    int sample_rate,
-    size_t number_of_channels,
+    uint32_t sample_rate,
+    uint32_t number_of_channels,
     size_t number_of_frames) {
   webrtc::MutexLock lock(&mutex_);
   for (auto sink : sinks_) {
@@ -156,8 +156,8 @@ void AudioTrackSource::set_audio_options(
 }
 
 void AudioTrackSource::on_captured_frame(rust::Slice<const int16_t> audio_data,
-                                         int sample_rate,
-                                         size_t number_of_channels,
+                                         uint32_t sample_rate,
+                                         uint32_t number_of_channels,
                                          size_t number_of_frames) const {
   source_->on_captured_frame(audio_data, sample_rate, number_of_channels,
                              number_of_frames);

webrtc-sys/src/audio_track.rs

@@ -44,8 +44,8 @@ pub mod ffi {
         fn on_captured_frame(
             self: &AudioTrackSource,
             data: &[i16],
-            sample_rate: i32,
-            nb_channels: usize,
+            sample_rate: u32,
+            nb_channels: u32,
             nb_frames: usize,
         );
         fn audio_options(self: &AudioTrackSource) -> AudioSourceOptions;