Keep ownership of observers on the Rust side

crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h

@@ -18,7 +18,7 @@ namespace livekit {
     public:
         explicit DataChannel(rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel);
 
-        void register_observer(std::unique_ptr<NativeDataChannelObserver> observer);
+        void register_observer(NativeDataChannelObserver &observer);
         void unregister_observer();
         void close();
     private:
@@ -42,7 +42,7 @@ namespace livekit {
         rust::Box<DataChannelObserverWrapper> observer_;
     };
 
-    std::unique_ptr<NativeDataChannelObserver> create_native_peer_connection_observer(rust::Box<DataChannelObserverWrapper> observer);
+    std::unique_ptr<NativeDataChannelObserver> create_native_data_channel_observer(rust::Box<DataChannelObserverWrapper> observer);
 } // livekit
 
 #endif //CLIENT_SDK_NATIVE_DATA_CHANNEL_H

crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h

@@ -17,18 +17,17 @@ namespace livekit {
 
     class PeerConnection {
     public:
-        explicit PeerConnection(rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection, std::unique_ptr<NativePeerConnectionObserver> observer);
+        explicit PeerConnection(rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection);
 
-        void create_offer(std::unique_ptr<NativeCreateSdpObserverHandle> observer, RTCOfferAnswerOptions options);
-        void create_answer(std::unique_ptr<NativeCreateSdpObserverHandle> observer, RTCOfferAnswerOptions options);
-        void set_local_description(std::unique_ptr<SessionDescription> desc, std::unique_ptr<NativeSetLocalSdpObserverHandle> observer);
-        void set_remote_description(std::unique_ptr<SessionDescription> desc, std::unique_ptr<NativeSetRemoteSdpObserverHandle> observer);
+        void create_offer(NativeCreateSdpObserverHandle &observer, RTCOfferAnswerOptions options);
+        void create_answer(NativeCreateSdpObserverHandle &observer, RTCOfferAnswerOptions options);
+        void set_local_description(std::unique_ptr<SessionDescription> desc, NativeSetLocalSdpObserverHandle &observer);
+        void set_remote_description(std::unique_ptr<SessionDescription> desc, NativeSetRemoteSdpObserverHandle &observer);
         std::unique_ptr<DataChannel> create_data_channel(rust::String label, std::unique_ptr<NativeDataChannelInit> init);
         void close();
 
     private:
         rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
-        std::unique_ptr<NativePeerConnectionObserver> observer_;
     };
 
     static std::unique_ptr<PeerConnection> _unique_peer_connection() {

crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h

@@ -17,7 +17,7 @@ namespace livekit {
     public:
         PeerConnectionFactory();
 
-        std::unique_ptr<PeerConnection> create_peer_connection(std::unique_ptr<NativeRTCConfiguration> config, std::unique_ptr<NativePeerConnectionObserver> observer) const;
+        std::unique_ptr<PeerConnection> create_peer_connection(std::unique_ptr<NativeRTCConfiguration> config, NativePeerConnectionObserver &observer) const;
 
     private:
         std::unique_ptr<rtc::Thread> network_thread_;

crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp

@@ -13,8 +13,8 @@ namespace livekit {
 
     }
 
-    void DataChannel::register_observer(std::unique_ptr<NativeDataChannelObserver> 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<NativeDataChannelObserver> create_native_peer_connection_observer(rust::Box<DataChannelObserverWrapper> observer){
+    std::unique_ptr<NativeDataChannelObserver> create_native_data_channel_observer(rust::Box<DataChannelObserverWrapper> observer){
         return std::make_unique<NativeDataChannelObserver>(std::move(observer));
     }
 } // livekit

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<NativeDataChannelInit>;

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp

@@ -21,26 +21,24 @@ namespace livekit {
         return rtc_options;
     }
 
-    PeerConnection::PeerConnection(rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection,
-                                   std::unique_ptr<NativePeerConnectionObserver> observer) : peer_connection_(
-            std::move(peer_connection)), observer_(std::move(observer)) {
+    PeerConnection::PeerConnection(rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection) : peer_connection_(std::move(peer_connection)) {
 
     }
 
-    void PeerConnection::create_offer(std::unique_ptr<NativeCreateSdpObserverHandle> 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<NativeCreateSdpObserverHandle> 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<SessionDescription> desc, std::unique_ptr<NativeSetLocalSdpObserverHandle> observer) {
-        peer_connection_->SetLocalDescription(desc->clone()->release(), observer->observer);
+    void PeerConnection::set_local_description(std::unique_ptr<SessionDescription> desc, NativeSetLocalSdpObserverHandle &observer) {
+        peer_connection_->SetLocalDescription(desc->clone()->release(), observer.observer);
     }
 
-    void PeerConnection::set_remote_description(std::unique_ptr<SessionDescription> desc, std::unique_ptr<NativeSetRemoteSdpObserverHandle> observer) {
-        peer_connection_->SetRemoteDescription(desc->clone()->release(), observer->observer);
+    void PeerConnection::set_remote_description(std::unique_ptr<SessionDescription> desc, NativeSetRemoteSdpObserverHandle &observer) {
+        peer_connection_->SetRemoteDescription(desc->clone()->release(), observer.observer);
     }
 
     std::unique_ptr<DataChannel> PeerConnection::create_data_channel(rust::String label, std::unique_ptr<NativeDataChannelInit> init) {

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<NativeCreateSdpObserverHandle>,
+            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<NativeCreateSdpObserverHandle>,
+            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<SessionDescription>,
-            observer: UniquePtr<NativeSetLocalSdpObserverHandle>,
+            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<SessionDescription>,
-            observer: UniquePtr<NativeSetRemoteSdpObserverHandle>,
+            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<RtpReceiver>,
         );
-        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);
     }
 

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp

@@ -53,15 +53,15 @@ namespace livekit{
         }
     }
 
-    std::unique_ptr<PeerConnection> PeerConnectionFactory::create_peer_connection(std::unique_ptr<webrtc::PeerConnectionInterface::RTCConfiguration> config, std::unique_ptr<NativePeerConnectionObserver> observer) const {
-        webrtc::PeerConnectionDependencies deps{observer.get()};
+    std::unique_ptr<PeerConnection> PeerConnectionFactory::create_peer_connection(std::unique_ptr<webrtc::PeerConnectionInterface::RTCConfiguration> 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<PeerConnection>(result.value(), std::move(observer));
+        return std::make_unique<PeerConnection>(result.value());
     }
 
     std::unique_ptr<PeerConnectionFactory> create_peer_connection_factory() {

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs

@@ -44,10 +44,12 @@ pub mod ffi {
         fn create_peer_connection_factory() -> UniquePtr<PeerConnectionFactory>;
         fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr<NativeRTCConfiguration>;
 
-        fn create_peer_connection(
+        /// SAFETY
+        /// The observer must live as long as the PeerConnection
+        unsafe fn create_peer_connection(
             self: &PeerConnectionFactory,
             config: UniquePtr<NativeRTCConfiguration>,
-            observer: UniquePtr<NativePeerConnectionObserver>,
+            observer: Pin<&mut NativePeerConnectionObserver>,
         ) -> Result<UniquePtr<PeerConnection>>;
     }
 }

crates/livekit-webrtc/src/peer_connection.rs

@@ -30,16 +30,21 @@ pub enum SdpError {
 pub struct PeerConnection {
     cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>,
     observer: Box<InternalObserver>,
+
+    // Keep alive for C++
+    native_observer: UniquePtr<sys_pc::ffi::NativePeerConnectionObserver>
 }
 
 impl PeerConnection {
     pub(crate) fn new(
         cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>,
         observer: Box<InternalObserver>,
+        native_observer: UniquePtr<sys_pc::ffi::NativePeerConnectionObserver>
     ) -> 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<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>,
-    ) {
+    fn on_remove_track(&self, receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>) {
         trace!("on_remove_track");
         let mut handler = self.on_remove_track_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {

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
                 }