diff --git a/libwebrtc/src/audio_source.rs b/libwebrtc/src/audio_source.rs index ac35a46a..f72bb1fa 100644 --- a/libwebrtc/src/audio_source.rs +++ b/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; ); } diff --git a/libwebrtc/src/native/audio_source.rs b/libwebrtc/src/native/audio_source.rs index 64cd58e6..0695d52d 100644 --- a/libwebrtc/src/native/audio_source.rs +++ b/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,10 +53,11 @@ 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 @@ -62,7 +65,35 @@ impl NativeAudioSource { 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 { @@ -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, ); } diff --git a/libwebrtc/src/native/video_source.rs b/libwebrtc/src/native/video_source.rs index 1f66ae5b..4e55048a 100644 --- a/libwebrtc/src/native/video_source.rs +++ b/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 for VideoResolution { @@ -40,15 +43,55 @@ impl From for vt_sys::ffi::VideoResolution { #[derive(Clone)] pub struct NativeVideoSource { sys_handle: SharedPtr, + inner: Arc>, +} + +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 { @@ -56,6 +99,9 @@ impl NativeVideoSource { } pub fn capture_frame>(&self, frame: &VideoFrame) { + 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 diff --git a/libwebrtc/src/video_source.rs b/libwebrtc/src/video_source.rs index 11480408..6791e79a 100644 --- a/libwebrtc/src/video_source.rs +++ b/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. diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs index 9d76367a..2c8bff4d 100644 --- a/livekit/src/room/participant/local_participant.rs +++ b/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; type LocalTrackUnpublishedHandler = Box; @@ -227,7 +229,6 @@ impl LocalParticipant { { local_track_published(self.clone(), publication.clone()); } - track.enable(); Ok(publication) diff --git a/webrtc-sys/include/livekit/audio_track.h b/webrtc-sys/include/livekit/audio_track.h index 0d08de65..a67ca811 100644 --- a/webrtc-sys/include/livekit/audio_track.h +++ b/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 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 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 get() const; diff --git a/webrtc-sys/src/audio_track.cpp b/webrtc-sys/src/audio_track.cpp index eff92a48..951208d2 100644 --- a/webrtc-sys/src/audio_track.cpp +++ b/webrtc-sys/src/audio_track.cpp @@ -131,8 +131,8 @@ void AudioTrackSource::InternalSource::RemoveSink( void AudioTrackSource::InternalSource::on_captured_frame( rust::Slice 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 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); diff --git a/webrtc-sys/src/audio_track.rs b/webrtc-sys/src/audio_track.rs index dd92dd4b..c3c35cd4 100644 --- a/webrtc-sys/src/audio_track.rs +++ b/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;