diff --git a/examples/simple_room/src/app.rs b/examples/simple_room/src/app.rs index ab80c0f..0cdd06a 100644 --- a/examples/simple_room/src/app.rs +++ b/examples/simple_room/src/app.rs @@ -1,12 +1,15 @@ use crate::events::UiCmd; use crate::logo_track::LogoTrack; +use crate::sine_track::SineTrack; use crate::video_renderer::VideoRenderer; use crate::{events::AsyncCmd, video_grid::VideoGrid}; use egui::{Rounding, Stroke}; use egui_wgpu::WgpuConfiguration; +use futures::StreamExt; use image::ImageFormat; use livekit::options::{TrackPublishOptions, VideoCaptureOptions}; use livekit::prelude::*; +use livekit::webrtc::audio_stream::native::NativeAudioStream; use livekit::webrtc::native::yuv_helper; use livekit::webrtc::video_frame::native::I420BufferExt; use livekit::webrtc::video_frame::{I420Buffer, VideoFrame, VideoRotation}; @@ -36,6 +39,7 @@ use winit::{ struct Session { room: Room, logo_track: LogoTrack, + sine_track: SineTrack, close_tx: oneshot::Sender<()>, handle: tokio::task::JoinHandle<()>, } @@ -113,6 +117,7 @@ pub fn run(rt: tokio::runtime::Runtime) { if let Ok((room, room_events)) = res { let (close_tx, close_rx) = oneshot::channel(); let logo_track = LogoTrack::new(room.session()); + let sine_track = SineTrack::new(room.session()); let handle = tokio::spawn(room_task( state.clone(), room_events, @@ -123,6 +128,7 @@ pub fn run(rt: tokio::runtime::Runtime) { *state.session.lock() = Some(Session { room, logo_track, + sine_track, close_tx, handle, }); @@ -156,6 +162,12 @@ pub fn run(rt: tokio::runtime::Runtime) { } } } + AsyncCmd::ToggleSine => { + if let Some(session) = state.session.lock().as_mut() { + let sine_track = &mut session.sine_track; + sine_track.publish().await.unwrap(); + } + } } } }); @@ -213,8 +225,15 @@ impl App { self.video_renderers .insert((participant.sid(), track.sid()), video_renderer); } - RemoteTrack::Audio(_) => { - // The demo doesn't support Audio rendering at the moment. + RemoteTrack::Audio(audio_track) => { + tokio::spawn(async move { + let mut stream = + NativeAudioStream::new(audio_track.rtc_track()); + + while let Some(_frame) = stream.next().await { + // Received audio frames + } + }); } }; } @@ -323,9 +342,12 @@ impl App { }); ui.menu_button("Publish", |ui| { - if ui.button("CustomTrack - LK Logo").clicked() { + if ui.button("Logo").clicked() { let _ = self.cmd_tx.send(AsyncCmd::ToggleLogo); } + if ui.button("SineWave").clicked() { + let _ = self.cmd_tx.send(AsyncCmd::ToggleSine); + } }); }); }); diff --git a/examples/simple_room/src/events.rs b/examples/simple_room/src/events.rs index 46bd655..34cf9b3 100644 --- a/examples/simple_room/src/events.rs +++ b/examples/simple_room/src/events.rs @@ -7,6 +7,7 @@ pub enum AsyncCmd { RoomDisconnect, SimulateScenario { scenario: SimulateScenario }, ToggleLogo, // Unpublish/Publish a logo track + ToggleSine, } #[derive(Debug)] diff --git a/examples/simple_room/src/logo_track.rs b/examples/simple_room/src/logo_track.rs index 6dd802b..46c5c6a 100644 --- a/examples/simple_room/src/logo_track.rs +++ b/examples/simple_room/src/logo_track.rs @@ -56,7 +56,7 @@ impl LogoTrack { } pub async fn publish(&mut self) -> Result<(), RoomError> { - self.unpublish().await; + self.unpublish().await?; let (close_tx, close_rx) = oneshot::channel(); let track = LocalVideoTrack::create_video_track( diff --git a/examples/simple_room/src/main.rs b/examples/simple_room/src/main.rs index 1e87b11..7d2533c 100644 --- a/examples/simple_room/src/main.rs +++ b/examples/simple_room/src/main.rs @@ -1,6 +1,7 @@ mod app; mod events; mod logo_track; +mod sine_track; mod video_grid; mod video_renderer; diff --git a/examples/simple_room/src/sine_track.rs b/examples/simple_room/src/sine_track.rs new file mode 100644 index 0000000..43231a9 --- /dev/null +++ b/examples/simple_room/src/sine_track.rs @@ -0,0 +1,127 @@ +use livekit::options::{AudioCaptureOptions, TrackPublishOptions}; +use livekit::webrtc::audio_frame::AudioFrame; +use livekit::{prelude::*, webrtc::audio_source::native::NativeAudioSource}; +use parking_lot::Mutex; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; + +#[derive(Clone)] +struct FrameData { + pub sample_rate: u32, + pub freq: f64, + pub amplitude: f64, +} + +impl Default for FrameData { + fn default() -> Self { + Self { + sample_rate: 48000, + freq: 440.0, + amplitude: 1.0, + } + } +} + +struct TrackHandle { + frame_data: Arc>, + close_tx: oneshot::Sender<()>, + track: LocalAudioTrack, + task: JoinHandle<()>, +} + +pub struct SineTrack { + rtc_source: NativeAudioSource, + session: RoomSession, + handle: Option, +} + +impl SineTrack { + pub fn new(session: RoomSession) -> Self { + Self { + rtc_source: NativeAudioSource::default(), + session, + handle: None, + } + } + + pub async fn publish(&mut self) -> Result<(), RoomError> { + let (close_tx, close_rx) = oneshot::channel(); + let track = LocalAudioTrack::create_audio_track( + "sine_wave", + AudioCaptureOptions { + auto_gain_control: false, + echo_cancellation: false, + noise_suppression: false, + }, + self.rtc_source.clone(), + ); + + let data = Arc::new(Mutex::new(FrameData::default())); + let task = tokio::spawn(Self::track_task( + close_rx, + self.rtc_source.clone(), + data.clone(), + )); + + self.session + .local_participant() + .publish_track( + LocalTrack::Audio(track.clone()), + TrackPublishOptions { + source: TrackSource::Microphone, + ..Default::default() + }, + ) + .await?; + + let handle = TrackHandle { + frame_data: data, + close_tx, + track, + task, + }; + + self.handle = Some(handle); + Ok(()) + } + + async fn track_task( + mut close_rx: oneshot::Receiver<()>, + rtc_source: NativeAudioSource, + frame_options: Arc>, + ) { + let mut interval = tokio::time::interval(Duration::from_millis(10)); + let mut samples_10ms = Vec::::new(); + + loop { + interval.tick().await; + + let data = frame_options.lock(); + let samples_count_10ms = (data.sample_rate / 100) as usize; + + if samples_10ms.capacity() != samples_count_10ms { + samples_10ms.resize(samples_count_10ms, 0i16); + } + + for i in 0..samples_count_10ms { + let val = data.amplitude + * f64::sin( + std::f64::consts::PI * 2.0 * data.freq * i as f64 + / samples_count_10ms as f64, + ); + + // WebRTC uses 16-bit signed PCM + samples_10ms[i] = (val * 32768.0) as i16; + } + + rtc_source.capture_frame(AudioFrame { + data: samples_10ms.clone(), + sample_rate_hz: data.sample_rate, + num_channels: 1, + samples_per_channel: samples_count_10ms, + }); + } + } +} diff --git a/livekit-webrtc/src/audio_frame.rs b/livekit-webrtc/src/audio_frame.rs new file mode 100644 index 0000000..1a86f21 --- /dev/null +++ b/livekit-webrtc/src/audio_frame.rs @@ -0,0 +1,7 @@ +#[derive(Debug, Clone)] +pub struct AudioFrame { + pub data: Vec, + pub sample_rate_hz: u32, + pub num_channels: usize, + pub samples_per_channel: usize, +} diff --git a/livekit-webrtc/src/audio_source.rs b/livekit-webrtc/src/audio_source.rs new file mode 100644 index 0000000..a1ecec3 --- /dev/null +++ b/livekit-webrtc/src/audio_source.rs @@ -0,0 +1,25 @@ +use crate::imp::audio_source as imp_as; + +#[cfg(not(target_arch = "wasm32"))] +pub mod native { + use super::imp_as; + use crate::audio_frame::AudioFrame; + use std::fmt::{Debug, Formatter}; + + #[derive(Default, Clone)] + pub struct NativeAudioSource { + pub(crate) handle: imp_as::NativeAudioSource, + } + + impl Debug for NativeAudioSource { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + f.debug_struct("NativeAudioSource").finish() + } + } + + impl NativeAudioSource { + pub fn capture_frame(&self, frame: AudioFrame) { + self.handle.capture_frame(frame) + } + } +} diff --git a/livekit-webrtc/src/audio_stream.rs b/livekit-webrtc/src/audio_stream.rs new file mode 100644 index 0000000..8feeefa --- /dev/null +++ b/livekit-webrtc/src/audio_stream.rs @@ -0,0 +1,48 @@ +use crate::imp::audio_stream as stream_imp; + +#[cfg(not(target_arch = "wasm32"))] +pub mod native { + use super::stream_imp; + use crate::audio_frame::AudioFrame; + use crate::media_stream::RtcAudioTrack; + use futures::stream::Stream; + use std::fmt::{Debug, Formatter}; + use std::pin::Pin; + use std::task::{Context, Poll}; + + pub struct NativeAudioStream { + pub(crate) handle: stream_imp::NativeAudioStream, + } + + impl Debug for NativeAudioStream { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + f.debug_struct("NativeAudioStream") + .field("track", &self.track()) + .finish() + } + } + + impl NativeAudioStream { + pub fn new(audio_track: RtcAudioTrack) -> Self { + Self { + handle: stream_imp::NativeAudioStream::new(audio_track), + } + } + + pub fn track(&self) -> RtcAudioTrack { + self.handle.track() + } + + pub fn close(&mut self) { + self.handle.close() + } + } + + impl Stream for NativeAudioStream { + type Item = AudioFrame; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + Pin::new(&mut self.get_mut().handle).poll_next(cx) + } + } +} diff --git a/livekit-webrtc/src/lib.rs b/livekit-webrtc/src/lib.rs index 269ec6e..ee758f6 100644 --- a/livekit-webrtc/src/lib.rs +++ b/livekit-webrtc/src/lib.rs @@ -26,6 +26,9 @@ pub struct RtcError { pub message: String, } +pub mod audio_frame; +pub mod audio_source; +pub mod audio_stream; pub mod data_channel; pub mod ice_candidate; pub mod media_stream; diff --git a/livekit-webrtc/src/native/audio_source.rs b/livekit-webrtc/src/native/audio_source.rs new file mode 100644 index 0000000..37af08d --- /dev/null +++ b/livekit-webrtc/src/native/audio_source.rs @@ -0,0 +1,34 @@ +use crate::audio_frame::AudioFrame; +use cxx::SharedPtr; +use webrtc_sys::media_stream as sys_ms; + +#[derive(Clone)] +pub struct NativeAudioSource { + sys_handle: SharedPtr, +} + +impl Default for NativeAudioSource { + fn default() -> Self { + Self { + sys_handle: sys_ms::ffi::new_audio_track_source(), + } + } +} + +impl NativeAudioSource { + pub fn sys_handle(&self) -> SharedPtr { + self.sys_handle.clone() + } + + pub fn capture_frame(&self, frame: AudioFrame) { + // TODO(theomonnom): Should we check for 10ms worth of data here? + unsafe { + self.sys_handle.on_captured_frame( + frame.data.as_ptr(), + frame.sample_rate_hz as i32, + frame.num_channels, + frame.samples_per_channel, + ) + } + } +} diff --git a/livekit-webrtc/src/native/audio_stream.rs b/livekit-webrtc/src/native/audio_stream.rs new file mode 100644 index 0000000..03503f2 --- /dev/null +++ b/livekit-webrtc/src/native/audio_stream.rs @@ -0,0 +1,80 @@ +use crate::{audio_frame::AudioFrame, media_stream::RtcAudioTrack}; +use cxx::UniquePtr; +use futures::stream::Stream; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::sync::mpsc; +use webrtc_sys::media_stream as sys_ms; + +pub struct NativeAudioStream { + native_observer: UniquePtr, + _observer: Box, + audio_track: RtcAudioTrack, + frame_rx: mpsc::UnboundedReceiver, +} + +impl NativeAudioStream { + pub fn new(audio_track: RtcAudioTrack) -> Self { + let (frame_tx, frame_rx) = mpsc::unbounded_channel(); + let mut observer = Box::new(AudioTrackObserver { frame_tx }); + let mut native_observer = unsafe { + sys_ms::ffi::new_native_audio_sink(Box::new(sys_ms::AudioSinkWrapper::new( + &mut *observer, + ))) + }; + + unsafe { + sys_ms::ffi::media_to_audio(audio_track.sys_handle()) + .add_sink(native_observer.pin_mut()); + } + + Self { + native_observer, + _observer: observer, + audio_track, + frame_rx, + } + } + + pub fn track(&self) -> RtcAudioTrack { + self.audio_track.clone() + } + + pub fn close(&mut self) { + self.frame_rx.close(); + unsafe { + sys_ms::ffi::media_to_audio(self.audio_track.sys_handle()) + .remove_sink(self.native_observer.pin_mut()); + } + } +} + +impl Drop for NativeAudioStream { + fn drop(&mut self) { + self.close(); + } +} + +impl Stream for NativeAudioStream { + type Item = AudioFrame; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + self.frame_rx.poll_recv(cx) + } +} + +pub struct AudioTrackObserver { + frame_tx: mpsc::UnboundedSender, +} + +impl sys_ms::AudioSink for AudioTrackObserver { + fn on_data(&self, data: &[i16], sample_rate: i32, nb_channels: usize, nb_frames: usize) { + // TODO(theomonnom): Should we avoid copy here? + let _ = self.frame_tx.send(AudioFrame { + data: data.to_owned(), + sample_rate_hz: sample_rate as u32, + num_channels: nb_channels, + samples_per_channel: nb_frames, + }); + } +} diff --git a/livekit-webrtc/src/native/mod.rs b/livekit-webrtc/src/native/mod.rs index e890949..6c5f63a 100644 --- a/livekit-webrtc/src/native/mod.rs +++ b/livekit-webrtc/src/native/mod.rs @@ -1,3 +1,5 @@ +pub mod audio_source; +pub mod audio_stream; pub mod data_channel; pub mod ice_candidate; pub mod media_stream; diff --git a/livekit-webrtc/src/native/peer_connection_factory.rs b/livekit-webrtc/src/native/peer_connection_factory.rs index 13440a9..a4a2fac 100644 --- a/livekit-webrtc/src/native/peer_connection_factory.rs +++ b/livekit-webrtc/src/native/peer_connection_factory.rs @@ -1,6 +1,7 @@ +use crate::audio_source::native::NativeAudioSource; use crate::imp::media_stream as imp_ms; use crate::imp::peer_connection as imp_pc; -use crate::media_stream::RtcVideoTrack; +use crate::media_stream::{RtcAudioTrack, RtcVideoTrack}; use crate::peer_connection::PeerConnection; use crate::peer_connection_factory::{ ContinualGatheringPolicy, IceServer, IceTransportsType, RtcConfiguration, @@ -132,6 +133,16 @@ impl PeerConnectionFactory { } } + pub fn create_audio_track(&self, label: &str, source: NativeAudioSource) -> RtcAudioTrack { + RtcAudioTrack { + handle: imp_ms::RtcAudioTrack { + sys_handle: self + .sys_handle + .create_audio_track(label.to_string(), source.handle.sys_handle()), + }, + } + } + pub fn get_rtp_sender_capabilities(&self, media_type: MediaType) -> RtpCapabilities { self.sys_handle .get_rtp_sender_capabilities(media_type.into()) diff --git a/livekit-webrtc/src/native/rtp_transceiver.rs b/livekit-webrtc/src/native/rtp_transceiver.rs index a3119d5..c3d1a8c 100644 --- a/livekit-webrtc/src/native/rtp_transceiver.rs +++ b/livekit-webrtc/src/native/rtp_transceiver.rs @@ -19,6 +19,7 @@ impl From for RtpTransceiverDirection sys_webrtc::ffi::RtpTransceiverDirection::SendOnly => Self::SendOnly, sys_webrtc::ffi::RtpTransceiverDirection::RecvOnly => Self::RecvOnly, sys_webrtc::ffi::RtpTransceiverDirection::Inactive => Self::Inactive, + sys_webrtc::ffi::RtpTransceiverDirection::Stopped => Self::Stopped, _ => panic!("unknown RtpTransceiverDirection"), } } @@ -31,6 +32,7 @@ impl From for sys_webrtc::ffi::RtpTransceiverDirection RtpTransceiverDirection::SendOnly => Self::SendOnly, RtpTransceiverDirection::RecvOnly => Self::RecvOnly, RtpTransceiverDirection::Inactive => Self::Inactive, + RtpTransceiverDirection::Stopped => Self::Stopped, _ => panic!("unknown RtpTransceiverDirection"), } } diff --git a/livekit-webrtc/src/peer_connection_factory.rs b/livekit-webrtc/src/peer_connection_factory.rs index 7fedb86..6d1c6c5 100644 --- a/livekit-webrtc/src/peer_connection_factory.rs +++ b/livekit-webrtc/src/peer_connection_factory.rs @@ -63,16 +63,22 @@ impl PeerConnectionFactory { pub mod native { use super::PeerConnectionFactory; - use crate::media_stream::RtcVideoTrack; + use crate::audio_source::native::NativeAudioSource; + use crate::media_stream::{RtcAudioTrack, RtcVideoTrack}; use crate::video_source::native::NativeVideoSource; pub trait PeerConnectionFactoryExt { fn create_video_track(&self, label: &str, source: NativeVideoSource) -> RtcVideoTrack; + fn create_audio_track(&self, label: &str, source: NativeAudioSource) -> RtcAudioTrack; } impl PeerConnectionFactoryExt for PeerConnectionFactory { fn create_video_track(&self, label: &str, source: NativeVideoSource) -> RtcVideoTrack { self.handle.create_video_track(label, source) } + + fn create_audio_track(&self, label: &str, source: NativeAudioSource) -> RtcAudioTrack { + self.handle.create_audio_track(label, source) + } } } diff --git a/livekit-webrtc/src/prelude.rs b/livekit-webrtc/src/prelude.rs index d2d62d3..08fe57a 100644 --- a/livekit-webrtc/src/prelude.rs +++ b/livekit-webrtc/src/prelude.rs @@ -1,3 +1,4 @@ +pub use crate::audio_frame::AudioFrame; pub use crate::data_channel::{ DataBuffer, DataChannel, DataChannelError, DataChannelInit, DataState, }; diff --git a/livekit/src/room/options.rs b/livekit/src/room/options.rs index 118810a..8410488 100644 --- a/livekit/src/room/options.rs +++ b/livekit/src/room/options.rs @@ -40,38 +40,63 @@ pub struct VideoPreset { pub height: u32, } +#[derive(Debug, Clone)] +pub struct AudioEncoding { + pub max_bitrate: u64, +} + #[derive(Debug, Clone)] pub struct AudioPreset { - pub max_bitrate: u32, + pub encoding: AudioEncoding, } impl AudioPreset { - pub const fn new(max_bitrate: u32) -> Self { - Self { max_bitrate } + pub const fn new(max_bitrate: u64) -> Self { + Self { + encoding: AudioEncoding { max_bitrate }, + } + } +} + +#[derive(Debug, Clone)] +pub struct AudioCaptureOptions { + pub echo_cancellation: bool, + pub noise_suppression: bool, + pub auto_gain_control: bool, +} + +impl Default for AudioCaptureOptions { + fn default() -> Self { + Self { + echo_cancellation: true, + noise_suppression: true, + auto_gain_control: true, + } } } #[derive(Clone, Debug)] pub struct VideoCaptureOptions { - pub preset: VideoPreset, + pub resolution: VideoResolution, } impl Default for VideoCaptureOptions { fn default() -> Self { Self { - preset: video::H720, + resolution: video::H720.resolution(), } } } #[derive(Clone, Debug)] pub struct TrackPublishOptions { - pub dynacast: bool, + // If the encodings aren't set, LiveKit will compute the most appropriate ones + pub video_encoding: Option, + pub audio_encoding: Option, pub video_codec: VideoCodec, pub dtx: bool, pub red: bool, pub simulcast: bool, - pub screenshare: bool, pub name: String, pub source: TrackSource, } @@ -79,12 +104,12 @@ pub struct TrackPublishOptions { impl Default for TrackPublishOptions { fn default() -> Self { Self { - dynacast: false, + video_encoding: None, + audio_encoding: None, video_codec: VideoCodec::VP8, dtx: true, red: true, simulcast: true, - screenshare: false, name: "unnamed track".to_owned(), source: TrackSource::Unknown, } @@ -120,7 +145,8 @@ pub fn compute_video_encodings( height: u32, options: &TrackPublishOptions, ) -> Vec { - let encoding = compute_appropriate_encoding(options.screenshare, width, height); + let screenshare = options.source == TrackSource::Screenshare; + let encoding = compute_appropriate_encoding(screenshare, width, height); let initial_preset = VideoPreset { width, @@ -135,8 +161,7 @@ pub fn compute_video_encodings( return into_rtp_encodings(width, height, &[initial_preset]); } - let mut simulcast_presets = - compute_default_simulcast_presets(options.screenshare, &initial_preset); + let mut simulcast_presets = compute_default_simulcast_presets(screenshare, &initial_preset); let mid_preset = simulcast_presets.pop(); let low_preset = simulcast_presets.pop(); diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs index fceb405..a93c567 100644 --- a/livekit/src/room/participant/local_participant.rs +++ b/livekit/src/room/participant/local_participant.rs @@ -1,10 +1,12 @@ use super::{ConnectionQuality, ParticipantInner}; +use crate::options; use crate::options::compute_video_encodings; use crate::options::video_layers_from_encodings; use crate::options::TrackPublishOptions; use crate::prelude::*; use crate::proto; use crate::rtc_engine::RtcEngine; +use livekit_webrtc::rtp_parameters::RtpEncodingParameters; use parking_lot::RwLockReadGuard; use std::collections::HashMap; use std::sync::Arc; @@ -53,13 +55,24 @@ impl LocalParticipant { // Get the video dimension // TODO(theomonnom): Use MediaStreamTrack::getSettings() on web let capture_options = video_track.capture_options(); - req.width = capture_options.preset.width; - req.height = capture_options.preset.height; + req.width = capture_options.resolution.width; + req.height = capture_options.resolution.height; encodings = compute_video_encodings(req.width, req.height, &options); req.layers = video_layers_from_encodings(req.width, req.height, &encodings); } - LocalTrack::Audio(_audio_track) => {} + LocalTrack::Audio(_audio_track) => { + // Setup audio encoding + let audio_encoding = options + .audio_encoding + .as_ref() + .unwrap_or(&options::audio::SPEECH.encoding); + + encodings.push(RtpEncodingParameters { + max_bitrate: Some(audio_encoding.max_bitrate), + ..Default::default() + }); + } } let track_info = self.rtc_engine.add_track(req).await?; diff --git a/livekit/src/room/track/local_audio_track.rs b/livekit/src/room/track/local_audio_track.rs index 09f08d2..e55c3d6 100644 --- a/livekit/src/room/track/local_audio_track.rs +++ b/livekit/src/room/track/local_audio_track.rs @@ -1,79 +1,100 @@ use super::TrackInner; +use crate::options::AudioCaptureOptions; use crate::prelude::*; use crate::proto; +use crate::rtc_engine::lk_runtime::LkRuntime; +use crate::webrtc::peer_connection_factory::native::PeerConnectionFactoryExt; use livekit_webrtc as rtc; +use parking_lot::Mutex; +use rtc::audio_source::native::NativeAudioSource; use std::sync::Arc; use tokio::sync::mpsc; +#[derive(Debug)] +pub struct LocalAudioTrackInner { + track_inner: TrackInner, + capture_options: Mutex, +} + #[derive(Clone, Debug)] pub struct LocalAudioTrack { - pub(crate) inner: Arc, + inner: Arc, } impl LocalAudioTrack { pub(crate) fn new( - sid: TrackSid, name: String, rtc_track: rtc::media_stream::RtcAudioTrack, + capture_options: AudioCaptureOptions, ) -> Self { Self { - inner: Arc::new(TrackInner::new( - sid, - name, - TrackKind::Audio, - rtc::media_stream::MediaStreamTrack::Audio(rtc_track), - )), + inner: Arc::new(LocalAudioTrackInner { + track_inner: TrackInner::new( + "unknown".to_string().into(), // sid + name, + TrackKind::Audio, + rtc::media_stream::MediaStreamTrack::Audio(rtc_track), + ), + capture_options: Mutex::new(capture_options), + }), } } + #[inline] + pub fn capture_options(&self) -> AudioCaptureOptions { + self.inner.capture_options.lock().clone() + } + #[inline] pub fn sid(&self) -> TrackSid { - self.inner.sid() + self.inner.track_inner.sid() } #[inline] pub fn name(&self) -> String { - self.inner.name() + self.inner.track_inner.name() } #[inline] pub fn kind(&self) -> TrackKind { - self.inner.kind() + self.inner.track_inner.kind() } #[inline] pub fn source(&self) -> TrackSource { - self.inner.source() + self.inner.track_inner.source() } #[inline] pub fn stream_state(&self) -> StreamState { - self.inner.stream_state() + self.inner.track_inner.stream_state() } #[inline] pub fn start(&self) { - self.inner.start() + self.inner.track_inner.start() } #[inline] pub fn stop(&self) { - self.inner.stop() + self.inner.track_inner.stop() } #[inline] pub fn muted(&self) -> bool { - self.inner.muted() + self.inner.track_inner.muted() } #[inline] pub fn set_muted(&self, muted: bool) { - self.inner.set_muted(muted) + self.inner.track_inner.set_muted(muted) } #[inline] pub fn rtc_track(&self) -> rtc::media_stream::RtcAudioTrack { - if let rtc::media_stream::MediaStreamTrack::Audio(audio) = self.inner.rtc_track() { + if let rtc::media_stream::MediaStreamTrack::Audio(audio) = + self.inner.track_inner.rtc_track() + { audio } else { unreachable!() @@ -82,12 +103,12 @@ impl LocalAudioTrack { #[inline] pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.inner.register_observer() + self.inner.track_inner.register_observer() } #[inline] pub(crate) fn transceiver(&self) -> Option { - self.inner.transceiver() + self.inner.track_inner.transceiver() } #[inline] @@ -95,11 +116,25 @@ impl LocalAudioTrack { &self, transceiver: Option, ) { - self.inner.update_transceiver(transceiver) + self.inner.track_inner.update_transceiver(transceiver) } #[inline] pub(crate) fn update_info(&self, info: proto::TrackInfo) { - self.inner.update_info(info) + self.inner.track_inner.update_info(info) + } +} + +impl LocalAudioTrack { + pub fn create_audio_track( + name: &str, + options: AudioCaptureOptions, + source: NativeAudioSource, + ) -> LocalAudioTrack { + let rtc_track = LkRuntime::instance() + .pc_factory + .create_audio_track(&rtc::native::create_random_uuid(), source); + + Self::new(name.to_string(), rtc_track, options) } } diff --git a/livekit/src/room/track/local_video_track.rs b/livekit/src/room/track/local_video_track.rs index 51064b3..2a427a3 100644 --- a/livekit/src/room/track/local_video_track.rs +++ b/livekit/src/room/track/local_video_track.rs @@ -39,6 +39,7 @@ impl LocalVideoTrack { } } + #[inline] pub fn capture_options(&self) -> VideoCaptureOptions { self.inner.capture_options.lock().clone() } diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index 86e40fc..f742fd9 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -1,5 +1,6 @@ use super::{rtc_events, EngineError, EngineResult, SimulateScenario}; use crate::options::TrackPublishOptions; +use crate::prelude::TrackKind; use crate::rtc_engine::lk_runtime::LkRuntime; use crate::rtc_engine::peer_transport::PeerTransport; use crate::rtc_engine::rtc_events::{RtcEvent, RtcEvents}; @@ -652,38 +653,38 @@ impl SessionInner { .peer_connection() .add_transceiver(track.rtc_track(), init)?; - let capabilities = LkRuntime::instance() - .pc_factory - .get_rtp_sender_capabilities(track.kind().into()); + if track.kind() == TrackKind::Video { + let capabilities = LkRuntime::instance() + .pc_factory + .get_rtp_sender_capabilities(track.kind().into()); - let mut matched = Vec::new(); - let mut partial_matched = Vec::new(); - let mut unmatched = Vec::new(); + let mut matched = Vec::new(); + let mut partial_matched = Vec::new(); + let mut unmatched = Vec::new(); - for codec in capabilities.codecs { - let mime_type = codec.mime_type.to_lowercase(); - if mime_type == "audio/opus" { - matched.push(codec); - } else if mime_type == format!("video/{}", options.video_codec.as_str()) { - if let Some(sdp_fmtp_line) = codec.sdp_fmtp_line.as_ref() { - // for h264 codecs that have sdpFmtpLine available, use only if the - // profile-level-id is 42e01f for cross-browser compatibility - if sdp_fmtp_line.contains("profile-level-id=42e01f") { - matched.push(codec); - continue; + for codec in capabilities.codecs { + let mime_type = codec.mime_type.to_lowercase(); + if mime_type == format!("video/{}", options.video_codec.as_str()) { + if let Some(sdp_fmtp_line) = codec.sdp_fmtp_line.as_ref() { + // for h264 codecs that have sdpFmtpLine available, use only if the + // profile-level-id is 42e01f for cross-browser compatibility + if sdp_fmtp_line.contains("profile-level-id=42e01f") { + matched.push(codec); + continue; + } } + partial_matched.push(codec); + } else { + unmatched.push(codec); } - partial_matched.push(codec); - } else { - unmatched.push(codec); } + + matched.append(&mut partial_matched); + matched.append(&mut unmatched); + + transceiver.set_codec_preferences(matched)?; } - matched.append(&mut partial_matched); - matched.append(&mut unmatched); - - transceiver.set_codec_preferences(matched)?; - Ok(transceiver) } diff --git a/webrtc-sys/build.rs b/webrtc-sys/build.rs index 9291961..7cae89c 100644 --- a/webrtc-sys/build.rs +++ b/webrtc-sys/build.rs @@ -153,6 +153,7 @@ fn main() { builder.file("src/video_frame_buffer.cpp"); builder.file("src/video_encoder_factory.cpp"); builder.file("src/video_decoder_factory.cpp"); + builder.file("src/audio_device.cpp"); for include in includes { builder.include(include); diff --git a/webrtc-sys/include/livekit/audio_device.h b/webrtc-sys/include/livekit/audio_device.h new file mode 100644 index 0000000..135e178 --- /dev/null +++ b/webrtc-sys/include/livekit/audio_device.h @@ -0,0 +1,129 @@ +/* + * Copyright 2023 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the “License”); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an “AS IS” BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include + +#include "api/task_queue/task_queue_factory.h" +#include "modules/audio_device/include/audio_device.h" +#include "rtc_base/synchronization/mutex.h" +#include "rtc_base/task_queue.h" +#include "rtc_base/task_utils/repeating_task.h" + +namespace livekit { + +class AudioDevice : public webrtc::AudioDeviceModule { + public: + AudioDevice(webrtc::TaskQueueFactory* task_queue_factory); + ~AudioDevice() override; + + int32_t ActiveAudioLayer(AudioLayer* audioLayer) const override; + int32_t RegisterAudioCallback(webrtc::AudioTransport* transport) override; + + int32_t Init() override; + int32_t Terminate() override; + bool Initialized() const override; + + int16_t PlayoutDevices() override; + int16_t RecordingDevices() override; + int32_t PlayoutDeviceName(uint16_t index, + char name[webrtc::kAdmMaxDeviceNameSize], + char guid[webrtc::kAdmMaxGuidSize]) override; + + int32_t RecordingDeviceName(uint16_t index, + char name[webrtc::kAdmMaxDeviceNameSize], + char guid[webrtc::kAdmMaxGuidSize]) override; + + int32_t SetPlayoutDevice(uint16_t index) override; + int32_t SetPlayoutDevice(WindowsDeviceType device) override; + int32_t SetRecordingDevice(uint16_t index) override; + int32_t SetRecordingDevice(WindowsDeviceType device) override; + + int32_t PlayoutIsAvailable(bool* available) override; + int32_t InitPlayout() override; + bool PlayoutIsInitialized() const override; + int32_t RecordingIsAvailable(bool* available) override; + int32_t InitRecording() override; + bool RecordingIsInitialized() const override; + + int32_t StartPlayout() override; + int32_t StopPlayout() override; + bool Playing() const override; + int32_t StartRecording() override; + int32_t StopRecording() override; + bool Recording() const override; + + int32_t InitSpeaker() override; + bool SpeakerIsInitialized() const override; + int32_t InitMicrophone() override; + bool MicrophoneIsInitialized() const override; + + int32_t SpeakerVolumeIsAvailable(bool* available) override; + int32_t SetSpeakerVolume(uint32_t volume) override; + int32_t SpeakerVolume(uint32_t* volume) const override; + int32_t MaxSpeakerVolume(uint32_t* maxVolume) const override; + int32_t MinSpeakerVolume(uint32_t* minVolume) const override; + + int32_t MicrophoneVolumeIsAvailable(bool* available) override; + int32_t SetMicrophoneVolume(uint32_t volume) override; + int32_t MicrophoneVolume(uint32_t* volume) const override; + int32_t MaxMicrophoneVolume(uint32_t* maxVolume) const override; + int32_t MinMicrophoneVolume(uint32_t* minVolume) const override; + + int32_t SpeakerMuteIsAvailable(bool* available) override; + int32_t SetSpeakerMute(bool enable) override; + int32_t SpeakerMute(bool* enabled) const override; + + int32_t MicrophoneMuteIsAvailable(bool* available) override; + int32_t SetMicrophoneMute(bool enable) override; + int32_t MicrophoneMute(bool* enabled) const override; + + int32_t StereoPlayoutIsAvailable(bool* available) const override; + int32_t SetStereoPlayout(bool enable) override; + int32_t StereoPlayout(bool* enabled) const override; + int32_t StereoRecordingIsAvailable(bool* available) const override; + int32_t SetStereoRecording(bool enable) override; + int32_t StereoRecording(bool* enabled) const override; + + int32_t PlayoutDelay(uint16_t* delayMS) const override; + + bool BuiltInAECIsAvailable() const override; + bool BuiltInAGCIsAvailable() const override; + bool BuiltInNSIsAvailable() const override; + + int32_t EnableBuiltInAEC(bool enable) override; + int32_t EnableBuiltInAGC(bool enable) override; + int32_t EnableBuiltInNS(bool enable) override; + +#if defined(WEBRTC_IOS) + int GetPlayoutAudioParameters(AudioParameters* params) const override; + int GetRecordAudioParameters(AudioParameters* params) const override; +#endif // WEBRTC_IOS + + int32_t SetAudioDeviceSink(webrtc::AudioDeviceSink* sink) const override; + + private: + mutable webrtc::Mutex mutex_; + webrtc::TaskQueueFactory* task_queue_factory_; + std::unique_ptr audio_queue_; + webrtc::RepeatingTaskHandle audio_task_; + std::vector data_; + webrtc::AudioTransport* audio_transport_; + std::atomic playing_{false}; + std::atomic initialized_{false}; +}; +} // namespace livekit diff --git a/webrtc-sys/include/livekit/media_stream.h b/webrtc-sys/include/livekit/media_stream.h index 8960791..6d94bfc 100644 --- a/webrtc-sys/include/livekit/media_stream.h +++ b/webrtc-sys/include/livekit/media_stream.h @@ -20,9 +20,12 @@ #include "api/media_stream_interface.h" #include "api/video/video_frame.h" +#include "common_audio/resampler/include/push_resampler.h" +#include "common_audio/ring_buffer.h" #include "livekit/helper.h" #include "livekit/video_frame.h" #include "media/base/adapted_video_track_source.h" +#include "pc/local_audio_source.h" #include "rtc_base/synchronization/mutex.h" #include "rtc_base/timestamp_aligner.h" #include "rust/cxx.h" @@ -34,6 +37,8 @@ class MediaStreamTrack; class VideoTrack; class AudioTrack; class NativeVideoFrameSink; +class NativeAudioSink; +class AudioTrackSource; class AdaptedVideoTrackSource; } // namespace livekit #include "webrtc-sys/src/media_stream.rs.h" @@ -86,8 +91,73 @@ class MediaStreamTrack { class AudioTrack : public MediaStreamTrack { public: explicit AudioTrack(rtc::scoped_refptr track); + + void add_sink(NativeAudioSink& sink) const; + void remove_sink(NativeAudioSink& sink) const; + + private: + webrtc::AudioTrackInterface* track() const { + return static_cast(track_.get()); + } }; +class NativeAudioSink : public webrtc::AudioTrackSinkInterface { + public: + explicit NativeAudioSink(rust::Box observer); + void OnData(const void* audio_data, + int bits_per_sample, + int sample_rate, + size_t number_of_channels, + size_t number_of_frames) override; + + private: + rust::Box observer_; +}; + +std::unique_ptr new_native_audio_sink( + rust::Box observer); + +class NativeAudioTrackSource : public webrtc::LocalAudioSource { + public: + NativeAudioTrackSource(); + + SourceState state() const override; + bool remote() const override; + + const cricket::AudioOptions options() const override; + + void AddSink(webrtc::AudioTrackSinkInterface* sink) override; + void RemoveSink(webrtc::AudioTrackSinkInterface* sink) override; + + // AudioFrame should always contain 10 ms worth of data (see index.md of acm) + void on_captured_frame(const int16_t* audio_data, + int sample_rate, + size_t number_of_channels, + size_t number_of_frames); + + private: + webrtc::Mutex mutex_; + std::vector sinks_; + cricket::AudioOptions options_{}; +}; + +class AudioTrackSource { + public: + AudioTrackSource(rtc::scoped_refptr source); + + void on_captured_frame(const int16_t* audio_data, + int sample_rate, + size_t number_of_channels, + size_t number_of_frames) const; + + rtc::scoped_refptr get() const; + + private: + rtc::scoped_refptr source_; +}; + +std::shared_ptr new_audio_track_source(); + class VideoTrack : public MediaStreamTrack { public: explicit VideoTrack(rtc::scoped_refptr track); @@ -131,7 +201,7 @@ class NativeVideoTrackSource : public rtc::AdaptedVideoTrackSource { bool is_screencast() const override; absl::optional needs_denoising() const override; - webrtc::MediaSourceInterface::SourceState state() const override; + SourceState state() const override; bool remote() const override; bool on_captured_frame(const webrtc::VideoFrame& frame); diff --git a/webrtc-sys/include/livekit/peer_connection_factory.h b/webrtc-sys/include/livekit/peer_connection_factory.h index 5e7fcba..884bcef 100644 --- a/webrtc-sys/include/livekit/peer_connection_factory.h +++ b/webrtc-sys/include/livekit/peer_connection_factory.h @@ -46,6 +46,10 @@ class PeerConnectionFactory { rust::String label, std::shared_ptr source) const; + std::shared_ptr create_audio_track( + rust::String label, + std::shared_ptr source) const; + RtpCapabilities get_rtp_sender_capabilities(MediaType type) const; RtpCapabilities get_rtp_receiver_capabilities(MediaType type) const; diff --git a/webrtc-sys/src/audio_device.cpp b/webrtc-sys/src/audio_device.cpp new file mode 100644 index 0000000..7bc35c5 --- /dev/null +++ b/webrtc-sys/src/audio_device.cpp @@ -0,0 +1,323 @@ +/* + * Copyright 2023 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the “License”); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an “AS IS” BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "livekit/audio_device.h" + +const int kBitsPerSample = 16; +const int kSampleRate = 48000; +const int kChannels = 2; +const int kSamplesPer10Ms = kSampleRate / 100; + +namespace livekit { + +AudioDevice::AudioDevice(webrtc::TaskQueueFactory* task_queue_factory) + : task_queue_factory_(task_queue_factory), + data_(kSamplesPer10Ms * kChannels) {} + +AudioDevice::~AudioDevice() { + Terminate(); +} + +int32_t AudioDevice::ActiveAudioLayer(AudioLayer* audioLayer) const { + *audioLayer = AudioLayer::kDummyAudio; + return 0; +} + +int32_t AudioDevice::RegisterAudioCallback(webrtc::AudioTransport* transport) { + webrtc::MutexLock lock(&mutex_); + audio_transport_ = transport; + return 0; +} + +int32_t AudioDevice::Init() { + audio_queue_ = + std::make_unique(task_queue_factory_->CreateTaskQueue( + "AudioDevice", webrtc::TaskQueueFactory::Priority::NORMAL)); + + audio_task_ = + webrtc::RepeatingTaskHandle::Start(audio_queue_->Get(), [this]() { + webrtc::MutexLock lock(&mutex_); + + if (playing_) { + int64_t elapsed_time_ms = -1; + int64_t ntp_time_ms = -1; + void* data = data_.data(); + + // Request the AudioData, otherwise WebRTC will ignore the packets. + // 10ms of audio data. + audio_transport_->PullRenderData(kBitsPerSample, kSampleRate, + kChannels, kSamplesPer10Ms, data, + &elapsed_time_ms, &ntp_time_ms); + } + + return webrtc::TimeDelta::Millis(10); + }); + + initialized_ = true; + return 0; +} + +int32_t AudioDevice::Terminate() { + if (!initialized_) + return 0; + + initialized_ = false; + + audio_queue_->PostTask([this] { audio_task_.Stop(); }); + + StopRecording(); + StopPlayout(); + return 0; +} + +bool AudioDevice::Initialized() const { + return initialized_; +} + +int16_t AudioDevice::PlayoutDevices() { + return 0; +} + +int16_t AudioDevice::RecordingDevices() { + return 0; +} + +int32_t AudioDevice::PlayoutDeviceName(uint16_t index, + char name[webrtc::kAdmMaxDeviceNameSize], + char guid[webrtc::kAdmMaxGuidSize]) { + return 0; +} + +int32_t AudioDevice::RecordingDeviceName( + uint16_t index, + char name[webrtc::kAdmMaxDeviceNameSize], + char guid[webrtc::kAdmMaxGuidSize]) { + return 0; +} + +int32_t AudioDevice::SetPlayoutDevice(uint16_t index) { + return 0; +} + +int32_t AudioDevice::SetPlayoutDevice(WindowsDeviceType device) { + return 0; +} + +int32_t AudioDevice::SetRecordingDevice(uint16_t index) { + return 0; +} + +int32_t AudioDevice::SetRecordingDevice(WindowsDeviceType device) { + return 0; +} + +int32_t AudioDevice::PlayoutIsAvailable(bool* available) { + return 0; +} + +int32_t AudioDevice::InitPlayout() { + return 0; +} + +bool AudioDevice::PlayoutIsInitialized() const { + return false; +} + +int32_t AudioDevice::RecordingIsAvailable(bool* available) { + return 0; +} + +int32_t AudioDevice::InitRecording() { + return 0; +} + +bool AudioDevice::RecordingIsInitialized() const { + return false; +} + +int32_t AudioDevice::StartPlayout() { + playing_ = true; + return 0; +} + +int32_t AudioDevice::StopPlayout() { + playing_ = false; + return 0; +} + +bool AudioDevice::Playing() const { + return false; +} + +int32_t AudioDevice::StartRecording() { + return 0; +} + +int32_t AudioDevice::StopRecording() { + return 0; +} + +bool AudioDevice::Recording() const { + return false; +} + +int32_t AudioDevice::InitSpeaker() { + return 0; +} + +bool AudioDevice::SpeakerIsInitialized() const { + return false; +} + +int32_t AudioDevice::InitMicrophone() { + return 0; +} + +bool AudioDevice::MicrophoneIsInitialized() const { + return false; +} + +int32_t AudioDevice::SpeakerVolumeIsAvailable(bool* available) { + return 0; +} + +int32_t AudioDevice::SetSpeakerVolume(uint32_t volume) { + return 0; +} + +int32_t AudioDevice::SpeakerVolume(uint32_t* volume) const { + return 0; +} + +int32_t AudioDevice::MaxSpeakerVolume(uint32_t* maxVolume) const { + return 0; +} + +int32_t AudioDevice::MinSpeakerVolume(uint32_t* minVolume) const { + return 0; +} + +int32_t AudioDevice::MicrophoneVolumeIsAvailable(bool* available) { + return 0; +} + +int32_t AudioDevice::SetMicrophoneVolume(uint32_t volume) { + return 0; +} + +int32_t AudioDevice::MicrophoneVolume(uint32_t* volume) const { + return 0; +} + +int32_t AudioDevice::MaxMicrophoneVolume(uint32_t* maxVolume) const { + return 0; +} + +int32_t AudioDevice::MinMicrophoneVolume(uint32_t* minVolume) const { + return 0; +} + +int32_t AudioDevice::SpeakerMuteIsAvailable(bool* available) { + return 0; +} + +int32_t AudioDevice::SetSpeakerMute(bool enable) { + return 0; +} + +int32_t AudioDevice::SpeakerMute(bool* enabled) const { + return 0; +} + +int32_t AudioDevice::MicrophoneMuteIsAvailable(bool* available) { + return 0; +} + +int32_t AudioDevice::SetMicrophoneMute(bool enable) { + return 0; +} + +int32_t AudioDevice::MicrophoneMute(bool* enabled) const { + return 0; +} + +int32_t AudioDevice::StereoPlayoutIsAvailable(bool* available) const { + return 0; +} + +int32_t AudioDevice::SetStereoPlayout(bool enable) { + return 0; +} + +int32_t AudioDevice::StereoPlayout(bool* enabled) const { + return 0; +} + +int32_t AudioDevice::StereoRecordingIsAvailable(bool* available) const { + return 0; +} + +int32_t AudioDevice::SetStereoRecording(bool enable) { + return 0; +} + +int32_t AudioDevice::StereoRecording(bool* enabled) const { + return 0; +} + +int32_t AudioDevice::PlayoutDelay(uint16_t* delayMS) const { + return 0; +} + +bool AudioDevice::BuiltInAECIsAvailable() const { + return false; +} + +bool AudioDevice::BuiltInAGCIsAvailable() const { + return false; +} + +bool AudioDevice::BuiltInNSIsAvailable() const { + return false; +} + +int32_t AudioDevice::EnableBuiltInAEC(bool enable) { + return 0; +} + +int32_t AudioDevice::EnableBuiltInAGC(bool enable) { + return 0; +} + +int32_t AudioDevice::EnableBuiltInNS(bool enable) { + return 0; +} + +#if defined(WEBRTC_IOS) +int AudioDevice::GetPlayoutAudioParameters(AudioParameters* params) const { + return 0; +} + +int AudioDevice::GetRecordAudioParameters(AudioParameters* params) const { + return 0; +} +#endif // WEBRTC_IOS + +int32_t AudioDevice::SetAudioDeviceSink(webrtc::AudioDeviceSink* sink) const { + return 0; +} + +} // namespace livekit diff --git a/webrtc-sys/src/media_stream.cpp b/webrtc-sys/src/media_stream.cpp index a0839a0..cbd8c96 100644 --- a/webrtc-sys/src/media_stream.cpp +++ b/webrtc-sys/src/media_stream.cpp @@ -23,6 +23,8 @@ #include "api/media_stream_interface.h" #include "api/video/video_frame.h" #include "api/video/video_rotation.h" +#include "audio/remix_resample.h" +#include "common_audio/include/audio_util.h" #include "rtc_base/logging.h" #include "rtc_base/ref_counted_object.h" #include "rtc_base/time_utils.h" @@ -129,6 +131,93 @@ TrackState MediaStreamTrack::state() const { AudioTrack::AudioTrack(rtc::scoped_refptr track) : MediaStreamTrack(std::move(track)) {} +void AudioTrack::add_sink(NativeAudioSink& sink) const { + track()->AddSink(&sink); +} + +void AudioTrack::remove_sink(NativeAudioSink& sink) const { + track()->RemoveSink(&sink); +} + +NativeAudioSink::NativeAudioSink(rust::Box observer) + : observer_(std::move(observer)) {} + +void NativeAudioSink::OnData(const void* audio_data, + int bits_per_sample, + int sample_rate, + size_t number_of_channels, + size_t number_of_frames) { + RTC_CHECK_EQ(16, bits_per_sample); + + observer_->on_data(static_cast(audio_data), sample_rate, + number_of_channels, number_of_frames); +} + +std::unique_ptr new_native_audio_sink( + rust::Box observer) { + return std::make_unique(std::move(observer)); +} + +NativeAudioTrackSource::NativeAudioTrackSource() { + options_.echo_cancellation = false; + options_.auto_gain_control = false; + options_.noise_suppression = false; +} + +webrtc::MediaSourceInterface::SourceState NativeAudioTrackSource::state() + const { + return webrtc::MediaSourceInterface::SourceState::kLive; +} + +bool NativeAudioTrackSource::remote() const { + return false; +} + +const cricket::AudioOptions NativeAudioTrackSource::options() const { + return options_; +} + +void NativeAudioTrackSource::AddSink(webrtc::AudioTrackSinkInterface* sink) { + webrtc::MutexLock lock(&mutex_); + sinks_.push_back(sink); +} + +void NativeAudioTrackSource::RemoveSink(webrtc::AudioTrackSinkInterface* sink) { + webrtc::MutexLock lock(&mutex_); + sinks_.erase(std::remove(sinks_.begin(), sinks_.end(), sink), sinks_.end()); +} + +void NativeAudioTrackSource::on_captured_frame(const int16_t* data, + int sample_rate, + size_t number_of_channels, + size_t number_of_frames) { + webrtc::MutexLock lock(&mutex_); + for (auto sink : sinks_) { + sink->OnData(data, 16, sample_rate, number_of_channels, number_of_frames); + } +} + +AudioTrackSource::AudioTrackSource( + rtc::scoped_refptr source) + : source_(std::move(source)) {} + +void AudioTrackSource::on_captured_frame(const int16_t* audio_data, + int sample_rate, + size_t number_of_channels, + size_t number_of_frames) const { + source_->on_captured_frame(audio_data, sample_rate, number_of_channels, + number_of_frames); +} + +rtc::scoped_refptr AudioTrackSource::get() const { + return source_; +} + +std::shared_ptr new_audio_track_source() { + return std::make_shared( + rtc::make_ref_counted()); +} + VideoTrack::VideoTrack(rtc::scoped_refptr track) : MediaStreamTrack(std::move(track)) {} diff --git a/webrtc-sys/src/media_stream.rs b/webrtc-sys/src/media_stream.rs index 2afbd33..77fcabf 100644 --- a/webrtc-sys/src/media_stream.rs +++ b/webrtc-sys/src/media_stream.rs @@ -40,11 +40,13 @@ pub mod ffi { unsafe extern "C++" { include!("livekit/media_stream.h"); + type NativeAudioSink; type NativeVideoFrameSink; type MediaStreamTrack; type MediaStream; type AudioTrack; type VideoTrack; + type AudioTrackSource; type AdaptedVideoTrackSource; fn id(self: &MediaStream) -> String; @@ -61,6 +63,21 @@ pub mod ffi { fn set_enabled(self: &MediaStreamTrack, enable: bool) -> bool; fn state(self: &MediaStreamTrack) -> TrackState; + unsafe fn add_sink(self: &AudioTrack, sink: Pin<&mut NativeAudioSink>); + unsafe fn remove_sink(self: &AudioTrack, sink: Pin<&mut NativeAudioSink>); + + fn new_native_audio_sink(observer: Box) -> UniquePtr; + + unsafe fn on_captured_frame( + self: &AudioTrackSource, + data: *const i16, + sample_rate: i32, + nb_channels: usize, + nb_frames: usize, + ); + + fn new_audio_track_source() -> SharedPtr; + unsafe fn add_sink(self: &VideoTrack, sink: Pin<&mut NativeVideoFrameSink>); unsafe fn remove_sink(self: &VideoTrack, sink: Pin<&mut NativeVideoFrameSink>); @@ -90,8 +107,17 @@ pub mod ffi { } extern "Rust" { + type AudioSinkWrapper; type VideoFrameSinkWrapper; + unsafe fn on_data( + self: &AudioSinkWrapper, + data: *const i16, + sample_rate: i32, + nb_channels: usize, + nb_frames: usize, + ); + fn on_frame(self: &VideoFrameSinkWrapper, frame: UniquePtr); fn on_discarded_frame(self: &VideoFrameSinkWrapper); fn on_constraints_changed( @@ -106,8 +132,33 @@ impl_thread_safety!(ffi::MediaStream, Send + Sync); impl_thread_safety!(ffi::AudioTrack, Send + Sync); impl_thread_safety!(ffi::VideoTrack, Send + Sync); impl_thread_safety!(ffi::NativeVideoFrameSink, Send + Sync); +impl_thread_safety!(ffi::NativeAudioSink, Send + Sync); +impl_thread_safety!(ffi::AudioTrackSource, Send + Sync); impl_thread_safety!(ffi::AdaptedVideoTrackSource, Send + Sync); +pub trait AudioSink: Send { + fn on_data(&self, data: &[i16], sample_rate: i32, nb_channels: usize, nb_frames: usize); +} + +pub struct AudioSinkWrapper { + observer: *mut dyn AudioSink, +} + +impl AudioSinkWrapper { + /// # Safety + /// AudioSink must lives as long as AudioSinkWrapper does + pub unsafe fn new(observer: *mut dyn AudioSink) -> Self { + Self { observer } + } + + fn on_data(&self, data: *const i16, sample_rate: i32, nb_channels: usize, nb_frames: usize) { + unsafe { + let data = std::slice::from_raw_parts(data, nb_channels * nb_frames); + (*self.observer).on_data(data, sample_rate, nb_channels, nb_frames); + } + } +} + pub trait VideoFrameSink: Send { fn on_frame(&self, frame: UniquePtr); fn on_discarded_frame(&self); diff --git a/webrtc-sys/src/peer_connection_factory.cpp b/webrtc-sys/src/peer_connection_factory.cpp index 6a2281c..9e863bc 100644 --- a/webrtc-sys/src/peer_connection_factory.cpp +++ b/webrtc-sys/src/peer_connection_factory.cpp @@ -25,11 +25,14 @@ #include "api/task_queue/default_task_queue_factory.h" #include "api/video_codecs/builtin_video_decoder_factory.h" #include "api/video_codecs/builtin_video_encoder_factory.h" +#include "livekit/audio_device.h" #include "livekit/rtc_error.h" #include "livekit/rtp_parameters.h" #include "livekit/video_decoder_factory.h" #include "livekit/video_encoder_factory.h" #include "media/engine/webrtc_media_engine.h" +#include "rtc_base/location.h" +#include "rtc_base/thread.h" namespace livekit { @@ -51,6 +54,14 @@ PeerConnectionFactory::PeerConnectionFactory( cricket::MediaEngineDependencies media_deps; media_deps.task_queue_factory = dependencies.task_queue_factory.get(); + + media_deps.adm = rtc_runtime_->worker_thread() + ->Invoke>( + RTC_FROM_HERE, [&] { + return rtc::make_ref_counted( + media_deps.task_queue_factory); + }); + media_deps.video_encoder_factory = std::move(std::make_unique()); media_deps.video_decoder_factory = @@ -96,6 +107,13 @@ std::shared_ptr PeerConnectionFactory::create_video_track( peer_factory_->CreateVideoTrack(label.c_str(), source->get().get())); } +std::shared_ptr PeerConnectionFactory::create_audio_track( + rust::String label, + std::shared_ptr source) const { + return std::make_shared( + peer_factory_->CreateAudioTrack(label.c_str(), source->get().get())); +} + RtpCapabilities PeerConnectionFactory::get_rtp_sender_capabilities( MediaType type) const { return to_rust_rtp_capabilities(peer_factory_->GetRtpSenderCapabilities( diff --git a/webrtc-sys/src/peer_connection_factory.rs b/webrtc-sys/src/peer_connection_factory.rs index a099c64..3561fde 100644 --- a/webrtc-sys/src/peer_connection_factory.rs +++ b/webrtc-sys/src/peer_connection_factory.rs @@ -37,7 +37,9 @@ pub mod ffi { include!("livekit/webrtc.h"); include!("livekit/rtp_parameters.h"); + type AudioTrackSource = crate::media_stream::ffi::AudioTrackSource; type AdaptedVideoTrackSource = crate::media_stream::ffi::AdaptedVideoTrackSource; + type AudioTrack = crate::media_stream::ffi::AudioTrack; type VideoTrack = crate::media_stream::ffi::VideoTrack; type RtpCapabilities = crate::rtp_parameters::ffi::RtpCapabilities; type MediaType = crate::webrtc::ffi::MediaType; @@ -72,6 +74,12 @@ pub mod ffi { source: SharedPtr, ) -> SharedPtr; + fn create_audio_track( + self: &PeerConnectionFactory, + label: String, + source: SharedPtr, + ) -> SharedPtr; + fn get_rtp_sender_capabilities( self: &PeerConnectionFactory, kind: MediaType,