From dca8cf7796512139f0fe85b7be961a046715a2c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sun, 18 Sep 2022 15:22:57 +0200 Subject: [PATCH] Keep ownership of observers on the Rust side --- .../include/livekit/data_channel.h | 4 +- .../include/livekit/peer_connection.h | 11 ++-- .../include/livekit/peer_connection_factory.h | 2 +- .../libwebrtc-sys/src/data_channel.cpp | 6 +-- .../libwebrtc-sys/src/data_channel.rs | 1 - .../libwebrtc-sys/src/peer_connection.cpp | 20 ++++--- .../libwebrtc-sys/src/peer_connection.rs | 38 +++++++------ .../src/peer_connection_factory.cpp | 6 +-- .../src/peer_connection_factory.rs | 6 ++- crates/livekit-webrtc/src/peer_connection.rs | 53 ++++++++++++------- .../src/peer_connection_factory.rs | 11 ++-- 11 files changed, 88 insertions(+), 70 deletions(-) diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h index 63032be..e1b2211 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h @@ -18,7 +18,7 @@ namespace livekit { public: explicit DataChannel(rtc::scoped_refptr data_channel); - void register_observer(std::unique_ptr observer); + void register_observer(NativeDataChannelObserver &observer); void unregister_observer(); void close(); private: @@ -42,7 +42,7 @@ namespace livekit { rust::Box observer_; }; - std::unique_ptr create_native_peer_connection_observer(rust::Box observer); + std::unique_ptr create_native_data_channel_observer(rust::Box observer); } // livekit #endif //CLIENT_SDK_NATIVE_DATA_CHANNEL_H diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h index f950c9f..91abb27 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h @@ -17,18 +17,17 @@ namespace livekit { class PeerConnection { public: - explicit PeerConnection(rtc::scoped_refptr peer_connection, std::unique_ptr observer); + explicit PeerConnection(rtc::scoped_refptr peer_connection); - void create_offer(std::unique_ptr observer, RTCOfferAnswerOptions options); - void create_answer(std::unique_ptr observer, RTCOfferAnswerOptions options); - void set_local_description(std::unique_ptr desc, std::unique_ptr observer); - void set_remote_description(std::unique_ptr desc, std::unique_ptr observer); + void create_offer(NativeCreateSdpObserverHandle &observer, RTCOfferAnswerOptions options); + void create_answer(NativeCreateSdpObserverHandle &observer, RTCOfferAnswerOptions options); + void set_local_description(std::unique_ptr desc, NativeSetLocalSdpObserverHandle &observer); + void set_remote_description(std::unique_ptr desc, NativeSetRemoteSdpObserverHandle &observer); std::unique_ptr create_data_channel(rust::String label, std::unique_ptr init); void close(); private: rtc::scoped_refptr peer_connection_; - std::unique_ptr observer_; }; static std::unique_ptr _unique_peer_connection() { diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h index 3ffb146..a199e60 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h @@ -17,7 +17,7 @@ namespace livekit { public: PeerConnectionFactory(); - std::unique_ptr create_peer_connection(std::unique_ptr config, std::unique_ptr observer) const; + std::unique_ptr create_peer_connection(std::unique_ptr config, NativePeerConnectionObserver &observer) const; private: std::unique_ptr network_thread_; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp index d1352bb..1da192e 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp @@ -13,8 +13,8 @@ namespace livekit { } - void DataChannel::register_observer(std::unique_ptr observer) { - data_channel_->RegisterObserver(observer.get()); + void DataChannel::register_observer(NativeDataChannelObserver &observer) { + data_channel_->RegisterObserver(&observer); } void DataChannel::unregister_observer() { @@ -65,7 +65,7 @@ namespace livekit { observer_->on_buffered_amount_change(sent_data_size); } - std::unique_ptr create_native_peer_connection_observer(rust::Box observer){ + std::unique_ptr create_native_data_channel_observer(rust::Box observer){ return std::make_unique(std::move(observer)); } } // livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs index 80743d6..4b3ac48 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs @@ -60,7 +60,6 @@ pub mod ffi { type NativeDataChannelInit; type NativeDataChannelObserver; - fn close(self: Pin<&mut DataChannel>); fn create_data_channel_init(init: DataChannelInit) -> UniquePtr; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp index 6ebbf59..a7a0bf3 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp @@ -21,26 +21,24 @@ namespace livekit { return rtc_options; } - PeerConnection::PeerConnection(rtc::scoped_refptr peer_connection, - std::unique_ptr observer) : peer_connection_( - std::move(peer_connection)), observer_(std::move(observer)) { + PeerConnection::PeerConnection(rtc::scoped_refptr peer_connection) : peer_connection_(std::move(peer_connection)) { } - void PeerConnection::create_offer(std::unique_ptr observer_handle, RTCOfferAnswerOptions options) { - peer_connection_->CreateOffer(observer_handle->observer.get(), toNativeOfferAnswerOptions(options)); + void PeerConnection::create_offer(NativeCreateSdpObserverHandle &observer_handle, RTCOfferAnswerOptions options) { + peer_connection_->CreateOffer(observer_handle.observer.get(), toNativeOfferAnswerOptions(options)); } - void PeerConnection::create_answer(std::unique_ptr observer_handle, RTCOfferAnswerOptions options) { - peer_connection_->CreateAnswer(observer_handle->observer.get(), toNativeOfferAnswerOptions(options)); + void PeerConnection::create_answer(NativeCreateSdpObserverHandle &observer_handle, RTCOfferAnswerOptions options) { + peer_connection_->CreateAnswer(observer_handle.observer.get(), toNativeOfferAnswerOptions(options)); } - void PeerConnection::set_local_description(std::unique_ptr desc, std::unique_ptr observer) { - peer_connection_->SetLocalDescription(desc->clone()->release(), observer->observer); + void PeerConnection::set_local_description(std::unique_ptr desc, NativeSetLocalSdpObserverHandle &observer) { + peer_connection_->SetLocalDescription(desc->clone()->release(), observer.observer); } - void PeerConnection::set_remote_description(std::unique_ptr desc, std::unique_ptr observer) { - peer_connection_->SetRemoteDescription(desc->clone()->release(), observer->observer); + void PeerConnection::set_remote_description(std::unique_ptr desc, NativeSetRemoteSdpObserverHandle &observer) { + peer_connection_->SetRemoteDescription(desc->clone()->release(), observer.observer); } std::unique_ptr PeerConnection::create_data_channel(rust::String label, std::unique_ptr init) { diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index feb43bb..b39e2e0 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs @@ -109,26 +109,38 @@ pub mod ffi { type NativePeerConnectionObserver; type PeerConnection; - fn create_offer( + /// SAFETY + /// The observer must live as long as the operation ends + unsafe fn create_offer( self: Pin<&mut PeerConnection>, - observer: UniquePtr, + observer: Pin<&mut NativeCreateSdpObserverHandle>, options: RTCOfferAnswerOptions, ); - fn create_answer( + + /// SAFETY + /// The observer must live as long as the operation ends + unsafe fn create_answer( self: Pin<&mut PeerConnection>, - observer: UniquePtr, + observer: Pin<&mut NativeCreateSdpObserverHandle>, options: RTCOfferAnswerOptions, ); - fn set_local_description( + + /// SAFETY + /// The observer must live as long as the operation ends + unsafe fn set_local_description( self: Pin<&mut PeerConnection>, desc: UniquePtr, - observer: UniquePtr, + observer: Pin<&mut NativeSetLocalSdpObserverHandle>, ); - fn set_remote_description( + + /// SAFETY + /// The observer must live as long as the operation ends + unsafe fn set_remote_description( self: Pin<&mut PeerConnection>, desc: UniquePtr, - observer: UniquePtr, + observer: Pin<&mut NativeSetRemoteSdpObserverHandle>, ); + fn close(self: Pin<&mut PeerConnection>); fn create_native_peer_connection_observer( @@ -212,10 +224,7 @@ pub mod ffi { self: &PeerConnectionObserverWrapper, receiver: UniquePtr, ); - unsafe fn on_interesting_usage( - self: &PeerConnectionObserverWrapper, - usage_pattern: i32, - ); + unsafe fn on_interesting_usage(self: &PeerConnectionObserverWrapper, usage_pattern: i32); } } @@ -358,10 +367,7 @@ impl PeerConnectionObserverWrapper { (*self.observer).on_ice_connection_receiving_change(receiving); } - unsafe fn on_ice_selected_candidate_pair_changed( - &self, - event: ffi::CandidatePairChangeEvent, - ) { + unsafe fn on_ice_selected_candidate_pair_changed(&self, event: ffi::CandidatePairChangeEvent) { (*self.observer).on_ice_selected_candidate_pair_changed(event); } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp index ee31583..38c0631 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp @@ -53,15 +53,15 @@ namespace livekit{ } } - std::unique_ptr PeerConnectionFactory::create_peer_connection(std::unique_ptr config, std::unique_ptr observer) const { - webrtc::PeerConnectionDependencies deps{observer.get()}; + std::unique_ptr PeerConnectionFactory::create_peer_connection(std::unique_ptr config, NativePeerConnectionObserver &observer) const { + webrtc::PeerConnectionDependencies deps{&observer}; auto result = peer_factory_->CreatePeerConnectionOrError(*config, std::move(deps)); if(!result.ok()) { throw std::runtime_error(serialize_error(to_error(result.error()))); } - return std::make_unique(result.value(), std::move(observer)); + return std::make_unique(result.value()); } std::unique_ptr create_peer_connection_factory() { diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs index df964b1..5252078 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs @@ -44,10 +44,12 @@ pub mod ffi { fn create_peer_connection_factory() -> UniquePtr; fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr; - fn create_peer_connection( + /// SAFETY + /// The observer must live as long as the PeerConnection + unsafe fn create_peer_connection( self: &PeerConnectionFactory, config: UniquePtr, - observer: UniquePtr, + observer: Pin<&mut NativePeerConnectionObserver>, ) -> Result>; } } diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index 1b623e1..e855d21 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -30,16 +30,21 @@ pub enum SdpError { pub struct PeerConnection { cxx_handle: UniquePtr, observer: Box, + + // Keep alive for C++ + native_observer: UniquePtr } impl PeerConnection { pub(crate) fn new( cxx_handle: UniquePtr, observer: Box, + native_observer: UniquePtr ) -> Self { Self { cxx_handle, observer, + native_observer } } @@ -48,11 +53,14 @@ impl PeerConnection { let wrapper = sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx })); - let native_wrapper = sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper)); + let mut native_wrapper = + sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper)); - self.cxx_handle - .pin_mut() - .create_offer(native_wrapper, RTCOfferAnswerOptions::default()); + unsafe { + self.cxx_handle + .pin_mut() + .create_offer(native_wrapper.pin_mut(), RTCOfferAnswerOptions::default()); + } match rx.recv().await { Some(value) => value.map_err(Into::into), @@ -65,11 +73,14 @@ impl PeerConnection { let wrapper = sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx })); - let native_wrapper = sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper)); + let mut native_wrapper = + sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper)); - self.cxx_handle - .pin_mut() - .create_answer(native_wrapper, RTCOfferAnswerOptions::default()); + unsafe { + self.cxx_handle + .pin_mut() + .create_answer(native_wrapper.pin_mut(), RTCOfferAnswerOptions::default()); + } match rx.recv().await { Some(value) => value.map_err(Into::into), @@ -84,11 +95,14 @@ impl PeerConnection { let (tx, mut rx) = mpsc::channel(1); let wrapper = sys_jsep::SetLocalSdpObserverWrapper::new(Box::new(InternalSetLocalSdpObserver { tx })); - let native_wrapper = sys_jsep::ffi::create_native_set_local_sdp_observer(Box::new(wrapper)); + let mut native_wrapper = + sys_jsep::ffi::create_native_set_local_sdp_observer(Box::new(wrapper)); - self.cxx_handle - .pin_mut() - .set_local_description(desc.release(), native_wrapper); + unsafe { + self.cxx_handle + .pin_mut() + .set_local_description(desc.release(), native_wrapper.pin_mut()); + } match rx.recv().await { Some(value) => value.map_err(Into::into), @@ -105,12 +119,14 @@ impl PeerConnection { sys_jsep::SetRemoteSdpObserverWrapper::new(Box::new(InternalSetRemoteSdpObserver { tx, })); - let native_wrapper = + let mut native_wrapper = sys_jsep::ffi::create_native_set_remote_sdp_observer(Box::new(wrapper)); - self.cxx_handle - .pin_mut() - .set_remote_description(desc.release(), native_wrapper); + unsafe { + self.cxx_handle + .pin_mut() + .set_remote_description(desc.release(), native_wrapper.pin_mut()); + } match rx.recv().await { Some(value) => value.map_err(Into::into), @@ -532,10 +548,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver { } } - fn on_remove_track( - &self, - receiver: UniquePtr, - ) { + fn on_remove_track(&self, receiver: UniquePtr) { trace!("on_remove_track"); let mut handler = self.on_remove_track_handler.lock().unwrap(); if let Some(f) = handler.as_mut() { diff --git a/crates/livekit-webrtc/src/peer_connection_factory.rs b/crates/livekit-webrtc/src/peer_connection_factory.rs index ed17686..2942b63 100644 --- a/crates/livekit-webrtc/src/peer_connection_factory.rs +++ b/crates/livekit-webrtc/src/peer_connection_factory.rs @@ -26,15 +26,16 @@ impl PeerConnectionFactory { unsafe { let mut observer = Box::new(InternalObserver::default()); - let observer_wrapper = sys_pc::PeerConnectionObserverWrapper::new(&mut *observer); - let native_observer = - sys_pc::ffi::create_native_peer_connection_observer(Box::new(observer_wrapper)); + let mut native_observer = sys_pc::ffi::create_native_peer_connection_observer( + Box::new(sys_pc::PeerConnectionObserverWrapper::new(&mut *observer)), + ); + let res = self .cxx_handle - .create_peer_connection(native_config, native_observer); + .create_peer_connection(native_config, native_observer.pin_mut()); match res { - Ok(cxx_handle) => Ok(PeerConnection::new(cxx_handle, observer)), + Ok(cxx_handle) => Ok(PeerConnection::new(cxx_handle, observer, native_observer)), Err(e) => { Err(RTCError::from(e.what())) // TODO }