From 410bbbb5ed2e53123c351ca7c2eb3e1bb084d789 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sat, 8 Oct 2022 20:58:40 +0200 Subject: [PATCH] use FnOnce if possible on libwebrtc callbacks --- .../livekit-webrtc/libwebrtc-sys/Cargo.toml | 1 - .../livekit-webrtc/libwebrtc-sys/src/jsep.rs | 53 ++---- .../libwebrtc-sys/src/peer_connection.rs | 17 +- crates/livekit-webrtc/src/peer_connection.rs | 174 ++++++++---------- 4 files changed, 95 insertions(+), 150 deletions(-) diff --git a/crates/livekit-webrtc/libwebrtc-sys/Cargo.toml b/crates/livekit-webrtc/libwebrtc-sys/Cargo.toml index 19635b1..cb2bc5b 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/Cargo.toml +++ b/crates/livekit-webrtc/libwebrtc-sys/Cargo.toml @@ -12,7 +12,6 @@ log = "0.4" cxx-build = "1.0" glob = "0.3.0" regex = "1.0" -cc = { version = "1.0", features = ["parallel"] } [dev-dependencies] env_logger = "0.9" diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs index b631665..7b15d29 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs @@ -1,5 +1,6 @@ use std::error::Error; use std::fmt::{Debug, Display, Formatter}; +use std::mem::ManuallyDrop; use std::str::FromStr; use cxx::UniquePtr; @@ -123,66 +124,46 @@ impl FromStr for ffi::SdpType { // CreateSdpObserver -pub trait CreateSdpObserver: Send { - fn on_success(&self, session_description: UniquePtr); - fn on_failure(&self, error: RTCError); -} - pub struct CreateSdpObserverWrapper { - observer: Box, + pub on_success: ManuallyDrop) + Send>>, + pub on_failure: ManuallyDrop>, } impl CreateSdpObserverWrapper { - pub fn new(observer: Box) -> Self { - Self { observer } - } - fn on_success(&self, session_description: UniquePtr) { - self.observer.on_success(session_description); + unsafe { + std::ptr::read(&*self.on_success)(session_description); + } } fn on_failure(&self, error: RTCError) { - self.observer.on_failure(error); + unsafe { + std::ptr::read(&*self.on_failure)(error); + } } } // SetLocalSdpObserver -pub trait SetLocalSdpObserver: Send { - fn on_set_local_description_complete(&self, error: RTCError); -} - -pub struct SetLocalSdpObserverWrapper { - observer: Box, -} +pub struct SetLocalSdpObserverWrapper(pub ManuallyDrop>); impl SetLocalSdpObserverWrapper { - pub fn new(observer: Box) -> Self { - Self { observer } - } - fn on_set_local_description_complete(&self, error: RTCError) { - self.observer.on_set_local_description_complete(error); + unsafe { + std::ptr::read(&*self.0)(error); + } } } // SetRemoteSdpObserver -pub trait SetRemoteSdpObserver: Send { - fn on_set_remote_description_complete(&self, error: RTCError); -} - -pub struct SetRemoteSdpObserverWrapper { - observer: Box, -} +pub struct SetRemoteSdpObserverWrapper(pub ManuallyDrop>); impl SetRemoteSdpObserverWrapper { - pub fn new(observer: Box) -> Self { - Self { observer } - } - fn on_set_remote_description_complete(&self, error: RTCError) { - self.observer.on_set_remote_description_complete(error); + unsafe { + std::ptr::read(&*self.0)(error); + } } } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index bb32da8..6b82f5f 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs @@ -1,4 +1,5 @@ use std::fmt::Debug; +use std::mem::ManuallyDrop; use cxx::UniquePtr; @@ -304,21 +305,13 @@ impl Default for ffi::RTCOfferAnswerOptions { } } -pub trait AddIceCandidateObserver: Send { - fn on_complete(&self, error: RTCError); -} - -pub struct AddIceCandidateObserverWrapper { - observer: Box, -} +pub struct AddIceCandidateObserverWrapper(pub ManuallyDrop>); impl AddIceCandidateObserverWrapper { - pub fn new(observer: Box) -> Self { - Self { observer } - } - fn on_complete(&self, error: RTCError) { - self.observer.on_complete(error); + unsafe { + std::ptr::read(&*self.0)(error); + } } } diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index 2d78e38..1b0dedd 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -1,14 +1,16 @@ +use std::mem::ManuallyDrop; use std::sync::{Arc, Mutex}; use cxx::UniquePtr; use log::trace; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, oneshot}; use libwebrtc_sys::data_channel as sys_dc; use libwebrtc_sys::jsep as sys_jsep; use libwebrtc_sys::peer_connection as sys_pc; pub use libwebrtc_sys::peer_connection::ffi::IceConnectionState; pub use libwebrtc_sys::peer_connection::ffi::IceGatheringState; +use libwebrtc_sys::peer_connection::ffi::NativeCreateSdpObserverHandle; pub use libwebrtc_sys::peer_connection::ffi::PeerConnectionState; pub use libwebrtc_sys::peer_connection::ffi::RTCOfferAnswerOptions; pub use libwebrtc_sys::peer_connection::ffi::SignalingState; @@ -42,13 +44,34 @@ impl PeerConnection { } } - pub async fn create_offer(&mut self, options: RTCOfferAnswerOptions) -> Result { - let (tx, mut rx) = mpsc::channel(1); + fn create_sdp_observer() -> ( + UniquePtr, + mpsc::Receiver>, + ) { + let (tx, rx) = mpsc::channel(1); + let wrapper = sys_jsep::CreateSdpObserverWrapper { + on_success: ManuallyDrop::new(Box::new({ + let tx = tx.clone(); + move |session_description| { + let _ = tx.blocking_send(Ok(SessionDescription::new(session_description))); + } + })), + on_failure: ManuallyDrop::new(Box::new(move |error| { + let _ = tx.blocking_send(Err(error)); + })), + }; - let wrapper = - sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx })); - let mut native_wrapper = - sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper)); + ( + sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper)), + rx, + ) + } + + pub async fn create_offer( + &mut self, + options: RTCOfferAnswerOptions, + ) -> Result { + let (mut native_wrapper, mut rx) = Self::create_sdp_observer(); unsafe { self.cxx_handle @@ -59,13 +82,11 @@ impl PeerConnection { rx.recv().await.unwrap() } - pub async fn create_answer(&mut self, options: RTCOfferAnswerOptions) -> Result { - let (tx, mut rx) = mpsc::channel(1); - - let wrapper = - sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx })); - let mut native_wrapper = - sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper)); + pub async fn create_answer( + &mut self, + options: RTCOfferAnswerOptions, + ) -> Result { + let (mut native_wrapper, mut rx) = Self::create_sdp_observer(); unsafe { self.cxx_handle @@ -80,9 +101,11 @@ impl PeerConnection { &mut self, desc: SessionDescription, ) -> Result<(), RTCError> { - let (tx, mut rx) = mpsc::channel(1); + let (tx, rx) = oneshot::channel(); let wrapper = - sys_jsep::SetLocalSdpObserverWrapper::new(Box::new(InternalSetLocalSdpObserver { tx })); + sys_jsep::SetLocalSdpObserverWrapper(ManuallyDrop::new(Box::new(move |error| { + let _ = tx.send(if error.ok() { Ok(()) } else { Err(error) }); + }))); let mut native_wrapper = sys_jsep::ffi::create_native_set_local_sdp_observer(Box::new(wrapper)); @@ -92,18 +115,18 @@ impl PeerConnection { .set_local_description(desc.release(), native_wrapper.pin_mut()); } - rx.recv().await.unwrap() + rx.await.unwrap() } pub async fn set_remote_description( &mut self, desc: SessionDescription, ) -> Result<(), RTCError> { - let (tx, mut rx) = mpsc::channel(1); + let (tx, rx) = oneshot::channel(); let wrapper = - sys_jsep::SetRemoteSdpObserverWrapper::new(Box::new(InternalSetRemoteSdpObserver { - tx, - })); + sys_jsep::SetRemoteSdpObserverWrapper(ManuallyDrop::new(Box::new(move |error| { + let _ = tx.send(if error.ok() { Ok(()) } else { Err(error) }); + }))); let mut native_wrapper = sys_jsep::ffi::create_native_set_remote_sdp_observer(Box::new(wrapper)); @@ -113,7 +136,7 @@ impl PeerConnection { .set_remote_description(desc.release(), native_wrapper.pin_mut()); } - rx.recv().await.unwrap() + rx.await.unwrap() } pub fn create_data_channel( @@ -135,10 +158,11 @@ impl PeerConnection { // TODO(theomonnom) Use IceCandidateInit instead of IceCandidate pub async fn add_ice_candidate(&mut self, candidate: IceCandidate) -> Result<(), RTCError> { - let (tx, mut rx) = mpsc::channel(1); - let observer = sys_pc::AddIceCandidateObserverWrapper::new(Box::new(InternalAddIceCandidateObserver { - tx, - })); + let (tx, rx) = oneshot::channel(); + let observer = + sys_pc::AddIceCandidateObserverWrapper(ManuallyDrop::new(Box::new(|error| { + let _ = tx.send(if error.ok() { Ok(()) } else { Err(error) }); + }))); let mut native_observer = sys_pc::ffi::create_native_add_ice_candidate_observer(Box::new(observer)); @@ -146,7 +170,7 @@ impl PeerConnection { .pin_mut() .add_ice_candidate(candidate.release(), native_observer.pin_mut()); - rx.recv().await.unwrap() + rx.await.unwrap() } pub fn local_description(&self) -> Option { @@ -293,67 +317,6 @@ impl PeerConnection { } } - -// SetLocalSdpObserver - -struct InternalAddIceCandidateObserver { - tx: mpsc::Sender>, -} - -impl sys_pc::AddIceCandidateObserver for InternalAddIceCandidateObserver { - fn on_complete(&self, error: RTCError) { - let res = if error.ok() { Ok(()) } else { Err(error) }; - let _ = self.tx.blocking_send(res); - } -} - -// CreateSdpObserver - -struct InternalCreateSdpObserver { - tx: mpsc::Sender>, -} - -impl sys_jsep::CreateSdpObserver for InternalCreateSdpObserver { - fn on_success( - &self, - session_description: UniquePtr, - ) { - let _ = self.tx.blocking_send(Ok(SessionDescription::new(session_description))); - } - - fn on_failure(&self, error: RTCError) { - let _ = self.tx.blocking_send(Err(error)); - } -} - -// SetLocalSdpObserver - -struct InternalSetLocalSdpObserver { - tx: mpsc::Sender>, -} - -impl sys_jsep::SetLocalSdpObserver for InternalSetLocalSdpObserver { - fn on_set_local_description_complete(&self, error: RTCError) { - let res = if error.ok() { Ok(()) } else { Err(error) }; - let _ = self.tx.blocking_send(res); - } -} - -// SetRemoteSdpObserver - -struct InternalSetRemoteSdpObserver { - tx: mpsc::Sender>, -} - -impl sys_jsep::SetRemoteSdpObserver for InternalSetRemoteSdpObserver { - fn on_set_remote_description_complete(&self, error: RTCError) { - let res = if error.ok() { Ok(()) } else { Err(error) }; - let _ = self.tx.blocking_send(res); - } -} - -// PeerConnectionObserver - // TODO(theomonnom) Should we return futures? pub type OnSignalingChangeHandler = Box; pub type OnAddStreamHandler = Box; @@ -363,16 +326,16 @@ pub type OnRenegotiationNeededHandler = Box; pub type OnNegotiationNeededEventHandler = Box; pub type OnIceConnectionChangeHandler = Box; pub type OnStandardizedIceConnectionChangeHandler = -Box; + Box; pub type OnConnectionChangeHandler = Box; pub type OnIceGatheringChangeHandler = Box; pub type OnIceCandidateHandler = Box; pub type OnIceCandidateErrorHandler = -Box; + Box; pub type OnIceCandidatesRemovedHandler = Box) + Send + Sync>; pub type OnIceConnectionReceivingChangeHandler = Box; pub type OnIceSelectedCandidatePairChangedHandler = -Box; + Box; pub type OnAddTrackHandler = Box) + Send + Sync>; pub type OnTrackHandler = Box; pub type OnRemoveTrackHandler = Box; @@ -387,16 +350,16 @@ pub(crate) struct InternalObserver { on_negotiation_needed_event_handler: Arc>>, on_ice_connection_change_handler: Arc>>, on_standardized_ice_connection_change_handler: - Arc>>, + Arc>>, on_connection_change_handler: Arc>>, on_ice_gathering_change_handler: Arc>>, on_ice_candidate_handler: Arc>>, on_ice_candidate_error_handler: Arc>>, on_ice_candidates_removed_handler: Arc>>, on_ice_connection_receiving_change_handler: - Arc>>, + Arc>>, on_ice_selected_candidate_pair_changed_handler: - Arc>>, + Arc>>, on_add_track_handler: Arc>>, on_track_handler: Arc>>, on_remove_track_handler: Arc>>, @@ -628,7 +591,9 @@ mod tests { use tokio::sync::mpsc; use libwebrtc_sys::peer_connection::ffi::RTCOfferAnswerOptions; - use libwebrtc_sys::peer_connection_factory::ffi::{ContinualGatheringPolicy, IceTransportsType}; + use libwebrtc_sys::peer_connection_factory::ffi::{ + ContinualGatheringPolicy, IceTransportsType, + }; use crate::data_channel::{DataChannel, DataChannelInit}; use crate::jsep::IceCandidate; @@ -643,9 +608,9 @@ mod tests { async fn create_pc() { init_log(); - let test = RTCRuntime::new(); + let rtc_runtime = RTCRuntime::new(); - let factory = PeerConnectionFactory::new(); + let factory = PeerConnectionFactory::new(rtc_runtime); let config = RTCConfiguration { ice_servers: vec![ICEServer { urls: vec!["stun:stun1.l.google.com:19302".to_string()], @@ -679,12 +644,19 @@ mod tests { .create_data_channel("test_dc", DataChannelInit::default()) .unwrap(); - let offer = bob.create_offer(RTCOfferAnswerOptions::default()).await.unwrap(); + let offer = bob + .create_offer(RTCOfferAnswerOptions::default()) + .await + .unwrap(); trace!("Bob offer: {:?}", offer); bob.set_local_description(offer.clone()).await.unwrap(); alice.set_remote_description(offer).await.unwrap(); - let answer = alice.create_answer(RTCOfferAnswerOptions::default()).await.unwrap(); + let answer = alice + .create_answer(RTCOfferAnswerOptions::default()) + .await + .unwrap(); + trace!("Alice answer: {:?}", answer); alice.set_local_description(answer.clone()).await.unwrap(); bob.set_remote_description(answer).await.unwrap(); @@ -697,13 +669,13 @@ mod tests { let (data_tx, mut data_rx) = mpsc::channel::(1); let mut alice_dc = alice_dc_rx.recv().await.unwrap(); - alice_dc.on_message(Box::new(move |data, is_binary| { + alice_dc.on_message(Box::new(move |data, _| { data_tx .blocking_send(String::from_utf8_lossy(data).to_string()) .unwrap(); })); - assert!(bob_dc.send(b"This is a test", true)); + bob_dc.send(b"This is a test", true).unwrap(); assert_eq!(data_rx.recv().await.unwrap(), "This is a test"); alice.close();