dc progress

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

@@ -8,14 +8,19 @@
 #include <memory>
 #include "api/data_channel_interface.h"
 #include "rust_types.h"
+#include "rust/cxx.h"
 
 namespace livekit {
     using NativeDataChannelInit = webrtc::DataChannelInit;
+    class NativeDataChannelObserver;
 
     class DataChannel {
     public:
         explicit DataChannel(rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel);
 
+        void register_observer(std::unique_ptr<NativeDataChannelObserver> observer);
+        void unregister_observer();
+        void close();
     private:
         rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel_;
     };
@@ -26,6 +31,18 @@ namespace livekit {
         return nullptr; // Ignore
     }
 
+    class NativeDataChannelObserver : public webrtc::DataChannelObserver {
+    public:
+        explicit NativeDataChannelObserver(rust::Box<DataChannelObserverWrapper> observer);
+
+        void OnStateChange() override;
+        void OnMessage(const webrtc::DataBuffer& buffer) override;
+        void OnBufferedAmountChange(uint64_t sent_data_size) override;
+    private:
+        rust::Box<DataChannelObserverWrapper> observer_;
+    };
+
+    std::unique_ptr<NativeDataChannelObserver> create_native_peer_connection_observer(rust::Box<DataChannelObserverWrapper> observer);
 } // livekit
 
 #endif //CLIENT_SDK_NATIVE_DATA_CHANNEL_H

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

@@ -39,8 +39,6 @@ namespace livekit {
     public:
         explicit NativePeerConnectionObserver(rust::Box<PeerConnectionObserverWrapper> observer);
 
-        ~NativePeerConnectionObserver() override = default;
-
         void OnSignalingChange(webrtc::PeerConnectionInterface::SignalingState new_state) override;
         void OnAddStream(rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) override;
         void OnRemoveStream(rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) override;

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

@@ -14,6 +14,7 @@ namespace livekit {
     struct CreateSdpObserverWrapper;
     struct SetLocalSdpObserverWrapper;
     struct SetRemoteSdpObserverWrapper;
+    struct DataChannelObserverWrapper;
 
     // Shared types
     struct RTCOfferAnswerOptions;

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

@@ -13,6 +13,18 @@ namespace livekit {
 
     }
 
+    void DataChannel::register_observer(std::unique_ptr<NativeDataChannelObserver> observer) {
+        data_channel_->RegisterObserver(observer.get());
+    }
+
+    void DataChannel::unregister_observer() {
+        data_channel_->UnregisterObserver();
+    }
+
+    void DataChannel::close() {
+        return data_channel_->Close();
+    }
+
     std::unique_ptr<NativeDataChannelInit> create_data_channel_init(DataChannelInit init) {
         auto rtc_init = std::make_unique<webrtc::DataChannelInit>();
         rtc_init->id = init.id;
@@ -33,4 +45,27 @@ namespace livekit {
         return rtc_init;
     }
 
+    NativeDataChannelObserver::NativeDataChannelObserver(rust::Box<DataChannelObserverWrapper> observer) : observer_(std::move(observer)){
+
+    }
+
+    void NativeDataChannelObserver::OnStateChange() {
+        observer_->on_state_change();
+    }
+
+    void NativeDataChannelObserver::OnMessage(const webrtc::DataBuffer &buffer) {
+        DataBuffer data{};
+        data.binary = buffer.data.data();
+        data.len = buffer.data.size();
+        data.binary = buffer.binary;
+        observer_->on_message(data);
+    }
+
+    void NativeDataChannelObserver::OnBufferedAmountChange(uint64_t sent_data_size) {
+        observer_->on_buffered_amount_change(sent_data_size);
+    }
+
+    std::unique_ptr<NativeDataChannelObserver> create_native_peer_connection_observer(rust::Box<DataChannelObserverWrapper> observer){
+        return std::make_unique<NativeDataChannelObserver>(std::move(observer));
+    }
 } // livekit

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

@@ -1,4 +1,5 @@
 use cxx::UniquePtr;
+use std::slice;
 
 #[cxx::bridge(namespace = "livekit")]
 pub mod ffi {
@@ -13,7 +14,7 @@ pub mod ffi {
     }
 
     #[derive(Debug)]
-    //#[allow(deprecated)]
+    #[allow(deprecated)]
     pub struct DataChannelInit {
         #[deprecated]
         reliable: bool,
@@ -26,7 +27,30 @@ pub mod ffi {
         negotiated: bool,
         id: i32,
         has_priority: bool,
-        priority: Priority
+        priority: Priority,
+    }
+
+    #[derive(Debug)]
+    pub struct DataBuffer {
+        pub ptr: *const u8,
+        pub len: usize,
+        pub binary: bool,
+    }
+
+    #[derive(Debug)]
+    pub enum DataState {
+        Connecting,
+        Open,
+        Closing,
+        Closed,
+    }
+
+    extern "Rust" {
+        type DataChannelObserverWrapper;
+
+        fn on_state_change(self: &DataChannelObserverWrapper);
+        fn on_message(self: &DataChannelObserverWrapper, buffer: DataBuffer);
+        fn on_buffered_amount_change(self: &DataChannelObserverWrapper, sent_data_size: u64);
     }
 
     unsafe extern "C++" {
@@ -34,9 +58,47 @@ pub mod ffi {
 
         type DataChannel;
         type NativeDataChannelInit;
+        type NativeDataChannelObserver;
+
+        
+        fn close(self: Pin<&mut DataChannel>);
 
         fn create_data_channel_init(init: DataChannelInit) -> UniquePtr<NativeDataChannelInit>;
+        fn create_native_data_channel_observer(
+            observer: Box<DataChannelObserverWrapper>,
+        ) -> UniquePtr<NativeDataChannelObserver>;
 
         fn _unique_data_channel() -> UniquePtr<DataChannel>; // Ignore
     }
 }
+
+// DataChannelObserver
+
+pub trait DataChannelObserver: Send {
+    fn on_state_change(&self);
+    fn on_message(&self, data: &[u8], is_binary: bool);
+    fn on_buffered_amount_change(&self, sent_data_size: u64);
+}
+
+pub struct DataChannelObserverWrapper {
+    observer: Box<dyn DataChannelObserver>,
+}
+
+impl DataChannelObserverWrapper {
+    pub fn new(observer: Box<dyn DataChannelObserver>) -> Self {
+        Self { observer }
+    }
+
+    fn on_state_change(&self) {
+        self.observer.on_state_change();
+    }
+
+    fn on_message(&self, buffer: ffi::DataBuffer) {
+        let data = unsafe { slice::from_raw_parts(buffer.ptr, buffer.len) };
+        self.observer.on_message(data, buffer.binary);
+    }
+
+    fn on_buffered_amount_change(&self, sent_data_size: u64) {
+        self.observer.on_buffered_amount_change(sent_data_size);
+    }
+}

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

@@ -142,45 +142,45 @@ pub mod ffi {
         type PeerConnectionObserverWrapper;
 
         unsafe fn on_signaling_change(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             new_state: SignalingState,
         );
         unsafe fn on_add_stream(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             stream: UniquePtr<MediaStreamInterface>,
         );
         unsafe fn on_remove_stream(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             stream: UniquePtr<MediaStreamInterface>,
         );
         unsafe fn on_data_channel(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             data_channel: UniquePtr<DataChannel>,
         );
-        unsafe fn on_renegotiation_needed(self: &mut PeerConnectionObserverWrapper);
-        unsafe fn on_negotiation_needed_event(self: &mut PeerConnectionObserverWrapper, event: u32);
+        unsafe fn on_renegotiation_needed(self: &PeerConnectionObserverWrapper);
+        unsafe fn on_negotiation_needed_event(self: &PeerConnectionObserverWrapper, event: u32);
         unsafe fn on_ice_connection_change(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             new_state: IceConnectionState,
         );
         unsafe fn on_standardized_ice_connection_change(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             new_state: IceConnectionState,
         );
         unsafe fn on_connection_change(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             new_state: PeerConnectionState,
         );
         unsafe fn on_ice_gathering_change(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             new_state: IceGatheringState,
         );
         unsafe fn on_ice_candidate(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             candidate: UniquePtr<IceCandidate>,
         );
         unsafe fn on_ice_candidate_error(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             address: String,
             port: i32,
             url: String,
@@ -188,32 +188,32 @@ pub mod ffi {
             error_text: String,
         );
         unsafe fn on_ice_candidates_removed(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             removed: Vec<CandidatePtr>,
         );
         unsafe fn on_ice_connection_receiving_change(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             receiving: bool,
         );
         unsafe fn on_ice_selected_candidate_pair_changed(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             event: CandidatePairChangeEvent,
         );
         unsafe fn on_add_track(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             receiver: UniquePtr<RtpReceiver>,
             streams: Vec<MediaStreamPtr>,
         );
         unsafe fn on_track(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             transceiver: UniquePtr<RtpTransceiver>,
         );
         unsafe fn on_remove_track(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             receiver: UniquePtr<RtpReceiver>,
         );
         unsafe fn on_interesting_usage(
-            self: &mut PeerConnectionObserverWrapper,
+            self: &PeerConnectionObserverWrapper,
             usage_pattern: i32,
         );
     }
@@ -245,36 +245,36 @@ impl Default for ffi::RTCOfferAnswerOptions {
 }
 
 pub trait PeerConnectionObserver: Send + Sync {
-    fn on_signaling_change(&mut self, new_state: ffi::SignalingState);
-    fn on_add_stream(&mut self, stream: UniquePtr<MediaStreamInterface>);
-    fn on_remove_stream(&mut self, stream: UniquePtr<MediaStreamInterface>);
-    fn on_data_channel(&mut self, data_channel: UniquePtr<DataChannel>);
-    fn on_renegotiation_needed(&mut self);
-    fn on_negotiation_needed_event(&mut self, event: u32);
-    fn on_ice_connection_change(&mut self, new_state: ffi::IceConnectionState);
-    fn on_standardized_ice_connection_change(&mut self, new_state: ffi::IceConnectionState);
-    fn on_connection_change(&mut self, new_state: ffi::PeerConnectionState);
-    fn on_ice_gathering_change(&mut self, new_state: ffi::IceGatheringState);
-    fn on_ice_candidate(&mut self, candidate: UniquePtr<IceCandidate>);
+    fn on_signaling_change(&self, new_state: ffi::SignalingState);
+    fn on_add_stream(&self, stream: UniquePtr<MediaStreamInterface>);
+    fn on_remove_stream(&self, stream: UniquePtr<MediaStreamInterface>);
+    fn on_data_channel(&self, data_channel: UniquePtr<DataChannel>);
+    fn on_renegotiation_needed(&self);
+    fn on_negotiation_needed_event(&self, event: u32);
+    fn on_ice_connection_change(&self, new_state: ffi::IceConnectionState);
+    fn on_standardized_ice_connection_change(&self, new_state: ffi::IceConnectionState);
+    fn on_connection_change(&self, new_state: ffi::PeerConnectionState);
+    fn on_ice_gathering_change(&self, new_state: ffi::IceGatheringState);
+    fn on_ice_candidate(&self, candidate: UniquePtr<IceCandidate>);
     fn on_ice_candidate_error(
-        &mut self,
+        &self,
         address: String,
         port: i32,
         url: String,
         error_code: i32,
         error_text: String,
     );
-    fn on_ice_candidates_removed(&mut self, removed: Vec<UniquePtr<Candidate>>);
-    fn on_ice_connection_receiving_change(&mut self, receiving: bool);
-    fn on_ice_selected_candidate_pair_changed(&mut self, event: ffi::CandidatePairChangeEvent);
+    fn on_ice_candidates_removed(&self, removed: Vec<UniquePtr<Candidate>>);
+    fn on_ice_connection_receiving_change(&self, receiving: bool);
+    fn on_ice_selected_candidate_pair_changed(&self, event: ffi::CandidatePairChangeEvent);
     fn on_add_track(
-        &mut self,
+        &self,
         receiver: UniquePtr<RtpReceiver>,
         streams: Vec<UniquePtr<MediaStreamInterface>>,
     );
-    fn on_track(&mut self, transceiver: UniquePtr<RtpTransceiver>);
-    fn on_remove_track(&mut self, receiver: UniquePtr<RtpReceiver>);
-    fn on_interesting_usage(&mut self, usage_pattern: i32);
+    fn on_track(&self, transceiver: UniquePtr<RtpTransceiver>);
+    fn on_remove_track(&self, receiver: UniquePtr<RtpReceiver>);
+    fn on_interesting_usage(&self, usage_pattern: i32);
 }
 
 // Thread safety is handled inside PeerConnectionObserver
@@ -289,52 +289,52 @@ impl PeerConnectionObserverWrapper {
         Self { observer }
     }
 
-    unsafe fn on_signaling_change(&mut self, new_state: ffi::SignalingState) {
+    unsafe fn on_signaling_change(&self, new_state: ffi::SignalingState) {
         (*self.observer).on_signaling_change(new_state);
     }
 
-    unsafe fn on_add_stream(&mut self, stream: UniquePtr<MediaStreamInterface>) {
+    unsafe fn on_add_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
         (*self.observer).on_add_stream(stream);
     }
 
-    unsafe fn on_remove_stream(&mut self, stream: UniquePtr<MediaStreamInterface>) {
+    unsafe fn on_remove_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
         (*self.observer).on_remove_stream(stream);
     }
 
-    unsafe fn on_data_channel(&mut self, data_channel: UniquePtr<DataChannel>) {
+    unsafe fn on_data_channel(&self, data_channel: UniquePtr<DataChannel>) {
         (*self.observer).on_data_channel(data_channel);
     }
 
-    unsafe fn on_renegotiation_needed(&mut self) {
+    unsafe fn on_renegotiation_needed(&self) {
         (*self.observer).on_renegotiation_needed();
     }
 
-    unsafe fn on_negotiation_needed_event(&mut self, event: u32) {
+    unsafe fn on_negotiation_needed_event(&self, event: u32) {
         (*self.observer).on_negotiation_needed_event(event);
     }
 
-    unsafe fn on_ice_connection_change(&mut self, new_state: ffi::IceConnectionState) {
+    unsafe fn on_ice_connection_change(&self, new_state: ffi::IceConnectionState) {
         (*self.observer).on_ice_connection_change(new_state);
     }
 
-    unsafe fn on_standardized_ice_connection_change(&mut self, new_state: ffi::IceConnectionState) {
+    unsafe fn on_standardized_ice_connection_change(&self, new_state: ffi::IceConnectionState) {
         (*self.observer).on_standardized_ice_connection_change(new_state);
     }
 
-    unsafe fn on_connection_change(&mut self, new_state: ffi::PeerConnectionState) {
+    unsafe fn on_connection_change(&self, new_state: ffi::PeerConnectionState) {
         (*self.observer).on_connection_change(new_state);
     }
 
-    unsafe fn on_ice_gathering_change(&mut self, new_state: ffi::IceGatheringState) {
+    unsafe fn on_ice_gathering_change(&self, new_state: ffi::IceGatheringState) {
         (*self.observer).on_ice_gathering_change(new_state);
     }
 
-    unsafe fn on_ice_candidate(&mut self, candidate: UniquePtr<IceCandidate>) {
+    unsafe fn on_ice_candidate(&self, candidate: UniquePtr<IceCandidate>) {
         (*self.observer).on_ice_candidate(candidate);
     }
 
     unsafe fn on_ice_candidate_error(
-        &mut self,
+        &self,
         address: String,
         port: i32,
         url: String,
@@ -344,7 +344,7 @@ impl PeerConnectionObserverWrapper {
         (*self.observer).on_ice_candidate_error(address, port, url, error_code, error_text);
     }
 
-    unsafe fn on_ice_candidates_removed(&mut self, removed: Vec<ffi::CandidatePtr>) {
+    unsafe fn on_ice_candidates_removed(&self, removed: Vec<ffi::CandidatePtr>) {
         let mut vec = Vec::new();
 
         for v in removed {
@@ -354,19 +354,19 @@ impl PeerConnectionObserverWrapper {
         (*self.observer).on_ice_candidates_removed(vec);
     }
 
-    unsafe fn on_ice_connection_receiving_change(&mut self, receiving: bool) {
+    unsafe fn on_ice_connection_receiving_change(&self, receiving: bool) {
         (*self.observer).on_ice_connection_receiving_change(receiving);
     }
 
     unsafe fn on_ice_selected_candidate_pair_changed(
-        &mut self,
+        &self,
         event: ffi::CandidatePairChangeEvent,
     ) {
         (*self.observer).on_ice_selected_candidate_pair_changed(event);
     }
 
     unsafe fn on_add_track(
-        &mut self,
+        &self,
         receiver: UniquePtr<RtpReceiver>,
         streams: Vec<ffi::MediaStreamPtr>,
     ) {
@@ -379,15 +379,15 @@ impl PeerConnectionObserverWrapper {
         (*self.observer).on_add_track(receiver, vec);
     }
 
-    unsafe fn on_track(&mut self, transceiver: UniquePtr<RtpTransceiver>) {
+    unsafe fn on_track(&self, transceiver: UniquePtr<RtpTransceiver>) {
         (*self.observer).on_track(transceiver);
     }
 
-    unsafe fn on_remove_track(&mut self, receiver: UniquePtr<RtpReceiver>) {
+    unsafe fn on_remove_track(&self, receiver: UniquePtr<RtpReceiver>) {
         (*self.observer).on_remove_track(receiver);
     }
 
-    unsafe fn on_interesting_usage(&mut self, usage_pattern: i32) {
+    unsafe fn on_interesting_usage(&self, usage_pattern: i32) {
         (*self.observer).on_interesting_usage(usage_pattern);
     }
 }

crates/livekit-webrtc/src/data_channel.rs

@@ -7,8 +7,6 @@ pub struct DataChannel {
 
 impl DataChannel {
     pub(crate) fn new(cxx_handle: UniquePtr<sys_dc::ffi::DataChannel>) -> Self {
-        Self {
-            cxx_handle,
-        }
+        Self { cxx_handle }
     }
-}
+}

crates/livekit-webrtc/src/jsep.rs

@@ -19,9 +19,8 @@ impl SessionDescription {
     }
 }
 
-
 impl Clone for SessionDescription {
     fn clone(&self) -> Self {
         SessionDescription::new(self.cxx_handle.clone())
     }
-}
+}

crates/livekit-webrtc/src/peer_connection.rs

@@ -1,8 +1,8 @@
 use cxx::UniquePtr;
 use libwebrtc_sys::jsep as sys_jsep;
 use libwebrtc_sys::peer_connection as sys_pc;
-use std::sync::{Arc, Mutex};
 use log::trace;
+use std::sync::{Arc, Mutex};
 use thiserror::Error;
 use tokio::sync::{mpsc, oneshot};
 
@@ -358,7 +358,7 @@ impl Default for InternalObserver {
 
 // Observers are being called on the Signaling Thread
 impl sys_pc::PeerConnectionObserver for InternalObserver {
-    fn on_signaling_change(&mut self, new_state: SignalingState) {
+    fn on_signaling_change(&self, new_state: SignalingState) {
         trace!("on_signaling_change, {:?}", new_state);
         let mut handler = self.on_signaling_change_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -367,7 +367,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_add_stream(
-        &mut self,
+        &self,
         stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
     ) {
         trace!("on_add_stream");
@@ -378,7 +378,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_remove_stream(
-        &mut self,
+        &self,
         stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
     ) {
         trace!("on_remove_stream");
@@ -389,7 +389,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_data_channel(
-        &mut self,
+        &self,
         data_channel: UniquePtr<libwebrtc_sys::data_channel::ffi::DataChannel>,
     ) {
         trace!("on_data_channel");
@@ -399,7 +399,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_renegotiation_needed(&mut self) {
+    fn on_renegotiation_needed(&self) {
         trace!("on_renegotiation_needed");
         let mut handler = self.on_renegotiation_needed_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -407,7 +407,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_negotiation_needed_event(&mut self, event: u32) {
+    fn on_negotiation_needed_event(&self, event: u32) {
         trace!("on_negotiation_needed_event");
         let mut handler = self.on_negotiation_needed_event_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -415,7 +415,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_ice_connection_change(&mut self, new_state: IceConnectionState) {
+    fn on_ice_connection_change(&self, new_state: IceConnectionState) {
         trace!("on_ice_connection_change");
         let mut handler = self.on_ice_connection_change_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -423,7 +423,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_standardized_ice_connection_change(&mut self, new_state: IceConnectionState) {
+    fn on_standardized_ice_connection_change(&self, new_state: IceConnectionState) {
         trace!("on_standardized_ice_connection_change");
         let mut handler = self
             .on_standardized_ice_connection_change_handler
@@ -434,7 +434,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_connection_change(&mut self, new_state: PeerConnectionState) {
+    fn on_connection_change(&self, new_state: PeerConnectionState) {
         trace!("on_connection_change");
         let mut handler = self.on_connection_change_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -442,7 +442,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_ice_gathering_change(&mut self, new_state: IceGatheringState) {
+    fn on_ice_gathering_change(&self, new_state: IceGatheringState) {
         trace!("on_ice_gathering_change");
         let mut handler = self.on_ice_gathering_change_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -450,7 +450,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_ice_candidate(&mut self, candidate: UniquePtr<libwebrtc_sys::jsep::ffi::IceCandidate>) {
+    fn on_ice_candidate(&self, candidate: UniquePtr<libwebrtc_sys::jsep::ffi::IceCandidate>) {
         trace!("on_ice_candidate");
         let mut handler = self.on_ice_candidate_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -459,7 +459,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_ice_candidate_error(
-        &mut self,
+        &self,
         address: String,
         port: i32,
         url: String,
@@ -474,7 +474,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_ice_candidates_removed(
-        &mut self,
+        &self,
         removed: Vec<UniquePtr<libwebrtc_sys::candidate::ffi::Candidate>>,
     ) {
         trace!("on_ice_candidates_removed");
@@ -484,7 +484,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_ice_connection_receiving_change(&mut self, receiving: bool) {
+    fn on_ice_connection_receiving_change(&self, receiving: bool) {
         trace!("on_ice_connection_receiving_change");
         let mut handler = self
             .on_ice_connection_receiving_change_handler
@@ -496,7 +496,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_ice_selected_candidate_pair_changed(
-        &mut self,
+        &self,
         event: libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent,
     ) {
         trace!("on_ice_selected_candidate_pair_changed");
@@ -510,7 +510,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_add_track(
-        &mut self,
+        &self,
         receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>,
         streams: Vec<UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>>,
     ) {
@@ -522,7 +522,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_track(
-        &mut self,
+        &self,
         transceiver: UniquePtr<libwebrtc_sys::rtp_transceiver::ffi::RtpTransceiver>,
     ) {
         trace!("on_track");
@@ -533,7 +533,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     }
 
     fn on_remove_track(
-        &mut self,
+        &self,
         receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>,
     ) {
         trace!("on_remove_track");
@@ -543,7 +543,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_interesting_usage(&mut self, usage_pattern: i32) {
+    fn on_interesting_usage(&self, usage_pattern: i32) {
         trace!("on_interesting_usage");
         let mut handler = self.on_interesting_usage_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -555,7 +555,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
 #[cfg(test)]
 mod tests {
     use crate::peer_connection_factory::PeerConnectionFactory;
-    use libwebrtc_sys::peer_connection_factory::ffi::{RTCConfiguration};
+    use libwebrtc_sys::peer_connection_factory::ffi::RTCConfiguration;
 
     fn init_log() {
         let _ = env_logger::builder().is_test(true).try_init();