Merge pull request #4 from livekit/theo/temp1

Working DataChannel

crates/livekit-core/src/rtc_engine.rs

@@ -24,7 +24,6 @@ impl RTCEngine {
         Ok(())
     }
 
-
     pub fn update(&self) {}
 
     async fn handle_rtc(mut signal_receiver: broadcast::Receiver<signal_response::Message>) {

crates/livekit-webrtc/libwebrtc-sys/build.rs

@@ -1,9 +1,9 @@
+use regex::Regex;
 use std::env;
 use std::fs;
 use std::io::Write;
 use std::path;
 use std::process::Command;
-use regex::Regex;
 
 const MAC_SDKS: &str =
     "/Applications/Xcode.app/Contents/Developer/Platforms/MacOSX.platform/Developer/SDKs";
@@ -78,6 +78,7 @@ fn main() {
         "src/rtp_receiver.rs",
         "src/rtp_transceiver.rs",
         "src/rtc_error.rs",
+        "src/webrtc.rs",
     ]);
 
     builder.file("src/peer_connection.cpp");
@@ -89,6 +90,7 @@ fn main() {
     builder.file("src/rtp_receiver.cpp");
     builder.file("src/rtp_transceiver.cpp");
     builder.file("src/rtc_error.cpp");
+    builder.file("src/webrtc.cpp");
 
     for include in includes {
         builder.include(include);
@@ -241,16 +243,10 @@ fn main() {
     builder.warnings(false).compile("lkwebrtc");
 
     for entry in glob::glob("./src/**/*.cpp").unwrap() {
-        println!(
-            "cargo:rerun-if-changed={}",
-            entry.unwrap().display()
-        );
+        println!("cargo:rerun-if-changed={}", entry.unwrap().display());
     }
 
     for entry in glob::glob("./include/**/*.h").unwrap() {
-        println!(
-            "cargo:rerun-if-changed={}",
-            entry.unwrap().display()
-        );
+        println!("cargo:rerun-if-changed={}", entry.unwrap().display());
     }
 }

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

@@ -7,20 +7,44 @@
 
 #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(NativeDataChannelObserver &observer);
+        void unregister_observer();
+        bool send(const DataBuffer& buffer);
+        rust::String label() const;
+        void close();
     private:
         rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel_;
     };
 
+    std::unique_ptr<NativeDataChannelInit> create_data_channel_init(DataChannelInit init);
+
     static std::unique_ptr<DataChannel> _unique_data_channel(){
         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_data_channel_observer(rust::Box<DataChannelObserverWrapper> observer);
 } // livekit
 
 #endif //CLIENT_SDK_NATIVE_DATA_CHANNEL_H

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

@@ -17,6 +17,7 @@ namespace livekit {
     public:
         explicit IceCandidate(std::unique_ptr<webrtc::IceCandidateInterface> ice_candidate);
 
+        std::unique_ptr<webrtc::IceCandidateInterface> release();
     private:
         std::unique_ptr<webrtc::IceCandidateInterface> ice_candidate_;
     };
@@ -29,6 +30,7 @@ namespace livekit {
     public:
         explicit SessionDescription(std::unique_ptr<webrtc::SessionDescriptionInterface> session_description);
 
+        rust::String stringify() const;
         std::unique_ptr<SessionDescription> clone() const;
         std::unique_ptr<webrtc::SessionDescriptionInterface> release();
 

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

@@ -13,33 +13,43 @@
 #include "rust_types.h"
 
 namespace livekit {
-    class NativePeerConnectionObserver;
+    class NativeAddIceCandidateObserver;
 
     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(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 add_ice_candidate(std::unique_ptr<IceCandidate> candidate, NativeAddIceCandidateObserver &observer);
         void close();
-        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);
 
     private:
         rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
-        std::unique_ptr<NativePeerConnectionObserver> observer_;
     };
 
     static std::unique_ptr<PeerConnection> _unique_peer_connection() {
         return nullptr; // Ignore
     }
 
+    class NativeAddIceCandidateObserver {
+    public:
+        explicit NativeAddIceCandidateObserver(rust::Box<AddIceCandidateObserverWrapper> observer);
+
+        void OnComplete(const RTCError &error);
+    private:
+        rust::Box<AddIceCandidateObserverWrapper> observer_;
+    };
+
+    std::unique_ptr<NativeAddIceCandidateObserver> create_native_add_ice_candidate_observer(rust::Box<AddIceCandidateObserverWrapper> observer);
+
     class NativePeerConnectionObserver : public webrtc::PeerConnectionObserver {
     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/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/include/livekit/rust_types.h

@@ -14,10 +14,14 @@ namespace livekit {
     struct CreateSdpObserverWrapper;
     struct SetLocalSdpObserverWrapper;
     struct SetRemoteSdpObserverWrapper;
+    struct DataChannelObserverWrapper;
+    struct AddIceCandidateObserverWrapper;
 
     // Shared types
     struct RTCOfferAnswerOptions;
     struct RTCError;
+    struct DataChannelInit;
+    struct DataBuffer;
 }
 
 #endif //RUST_TYPES_H

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

@@ -0,0 +1,32 @@
+//
+// Created by theom on 18/09/2022.
+//
+
+#ifndef LIVEKIT_WEBRTC_WEBRTC_H
+#define LIVEKIT_WEBRTC_WEBRTC_H
+
+#include "rtc_base/ssl_adapter.h"
+#include "rtc_base/physical_socket_server.h"
+
+#ifdef WEBRTC_WIN
+#include "rtc_base/win32_socket_init.h"
+#endif
+
+namespace livekit {
+
+    class RTCRuntime {
+    public:
+        RTCRuntime();
+        ~RTCRuntime();
+
+        RTCRuntime(const RTCRuntime&) = delete;
+        RTCRuntime& operator=(const RTCRuntime&) = delete;
+    private:
+        rtc::WinsockInitializer winsock_;
+    };
+
+    std::unique_ptr<RTCRuntime> create_rtc_runtime();
+
+} // livekit
+
+#endif //LIVEKIT_WEBRTC_WEBRTC_H

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

@@ -6,7 +6,7 @@ pub mod ffi {
         include!("livekit/candidate.h");
 
         type Candidate; // cricket::Candidate
- 
+
         fn _unique_candidate() -> UniquePtr<Candidate>; // Ignore
     }
 }

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

@@ -2,11 +2,78 @@
 // Created by Théo Monnom on 01/09/2022.
 //
 
+#include <utility>
+
 #include "livekit/data_channel.h"
+#include "libwebrtc-sys/src/data_channel.rs.h"
 
 namespace livekit {
 
-    DataChannel::DataChannel(rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) : data_channel_(data_channel) {
+    DataChannel::DataChannel(rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) : data_channel_(std::move(data_channel)) {
 
     }
+
+    void DataChannel::register_observer(NativeDataChannelObserver &observer) {
+        data_channel_->RegisterObserver(&observer);
+    }
+
+    void DataChannel::unregister_observer() {
+        data_channel_->UnregisterObserver();
+    }
+
+    bool DataChannel::send(const DataBuffer &buffer) {
+        return data_channel_->Send(webrtc::DataBuffer{rtc::CopyOnWriteBuffer(buffer.ptr, buffer.len), buffer.binary });
+    }
+
+    rust::String DataChannel::label() const{
+        return data_channel_->label();
+    }
+
+    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;
+        rtc_init->negotiated = init.negotiated;
+        rtc_init->ordered = init.ordered;
+        rtc_init->protocol = init.protocol.c_str();
+        rtc_init->reliable = init.reliable;
+
+        if(init.has_max_retransmit_time)
+            rtc_init->maxRetransmitTime = init.max_retransmit_time;
+
+        if(init.has_max_retransmits)
+            rtc_init->maxRetransmits = init.max_retransmits;
+
+        if(init.has_priority)
+            rtc_init->priority = static_cast<webrtc::Priority>(init.priority);
+
+        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.ptr = 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_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

@@ -1,12 +1,123 @@
 use cxx::UniquePtr;
+use std::slice;
 
 #[cxx::bridge(namespace = "livekit")]
 pub mod ffi {
+
+    #[derive(Debug)]
+    #[repr(u32)]
+    pub enum Priority {
+        VeryLow,
+        Low,
+        Medium,
+        High,
+    }
+
+    #[derive(Debug)]
+    pub struct DataChannelInit {
+        #[allow(deprecated)]
+        #[deprecated]
+        pub reliable: bool,
+        pub ordered: bool,
+        pub has_max_retransmit_time: bool,
+        pub max_retransmit_time: i32,
+        pub has_max_retransmits: bool,
+        pub max_retransmits: i32,
+        pub protocol: String,
+        pub negotiated: bool,
+        pub id: i32,
+        pub has_priority: bool,
+        pub priority: Priority,
+    }
+
+    #[derive(Debug)]
+    pub struct DataBuffer {
+        pub ptr: *const u8,
+        pub len: usize,
+        pub binary: bool,
+    }
+
+    #[derive(Debug)]
+    #[repr(u32)]
+    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++" {
         include!("livekit/data_channel.h");
 
         type DataChannel;
+        type NativeDataChannelInit;
+        type NativeDataChannelObserver;
+
+        /// SAFETY
+        /// The observer must live as the datachannel uses it
+        unsafe fn register_observer(
+            self: Pin<&mut DataChannel>,
+            observer: Pin<&mut NativeDataChannelObserver>,
+        );
+
+        fn unregister_observer(self: Pin<&mut DataChannel>);
+        fn send(self: Pin<&mut DataChannel>, data: &DataBuffer) -> bool;
+        fn label(self: &DataChannel) -> String;
+        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
     }
 }
+
+unsafe impl Send for ffi::DataChannel {}
+unsafe impl Send for ffi::NativeDataChannelObserver {}
+
+// 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: *mut dyn DataChannelObserver,
+}
+
+impl DataChannelObserverWrapper {
+    /// SAFETY
+    /// DataChannelObserver must lives as long as DataChannelObserverWrapper does
+    pub unsafe fn new(observer: *mut dyn DataChannelObserver) -> Self {
+        Self { observer }
+    }
+
+    fn on_state_change(&self) {
+        unsafe {
+            (*self.observer).on_state_change();
+        }
+    }
+
+    fn on_message(&self, buffer: ffi::DataBuffer) {
+        unsafe {
+            let data = 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) {
+        unsafe { (*self.observer).on_buffered_amount_change(sent_data_size) };
+    }
+}

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

@@ -15,10 +15,20 @@ namespace livekit {
 
     }
 
+    std::unique_ptr<webrtc::IceCandidateInterface> IceCandidate::release() {
+        return std::move(ice_candidate_);
+    }
+
     SessionDescription::SessionDescription(std::unique_ptr<webrtc::SessionDescriptionInterface> session_description) : session_description_(std::move(session_description)){
 
     }
 
+    rust::String SessionDescription::stringify() const {
+        std::string str;
+        session_description_->ToString(&str);
+        return rust::String{str};
+    }
+
     std::unique_ptr<SessionDescription> SessionDescription::clone() const {
         return std::make_unique<SessionDescription>(session_description_->Clone());
     }

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

@@ -1,5 +1,5 @@
 use cxx::UniquePtr;
-use cxx::{type_id, ExternType};
+use std::fmt::{Debug, Formatter};
 
 use crate::rtc_error::ffi::RTCError;
 
@@ -8,7 +8,10 @@ pub mod ffi {
 
     extern "Rust" {
         type CreateSdpObserverWrapper;
-        fn on_success(self: &CreateSdpObserverWrapper, session_description: UniquePtr<SessionDescription>);
+        fn on_success(
+            self: &CreateSdpObserverWrapper,
+            session_description: UniquePtr<SessionDescription>,
+        );
         fn on_failure(self: &CreateSdpObserverWrapper, error: RTCError);
 
         type SetLocalSdpObserverWrapper;
@@ -29,18 +32,43 @@ pub mod ffi {
         type NativeSetLocalSdpObserverHandle;
         type NativeSetRemoteSdpObserverHandle;
 
-        fn create_native_create_sdp_observer(observer: Box<CreateSdpObserverWrapper>) -> UniquePtr<NativeCreateSdpObserverHandle>;
-        fn create_native_set_local_sdp_observer(observer: Box<SetLocalSdpObserverWrapper>) -> UniquePtr<NativeSetLocalSdpObserverHandle>;
-        fn create_native_set_remote_sdp_observer(observer: Box<SetRemoteSdpObserverWrapper>) -> UniquePtr<NativeSetRemoteSdpObserverHandle>;
+        fn stringify(self: &SessionDescription) -> String;
+        fn clone(self: &SessionDescription) -> UniquePtr<SessionDescription>;
+
+        fn create_native_create_sdp_observer(
+            observer: Box<CreateSdpObserverWrapper>,
+        ) -> UniquePtr<NativeCreateSdpObserverHandle>;
+        fn create_native_set_local_sdp_observer(
+            observer: Box<SetLocalSdpObserverWrapper>,
+        ) -> UniquePtr<NativeSetLocalSdpObserverHandle>;
+        fn create_native_set_remote_sdp_observer(
+            observer: Box<SetRemoteSdpObserverWrapper>,
+        ) -> UniquePtr<NativeSetRemoteSdpObserverHandle>;
 
         fn _unique_ice_candidate() -> UniquePtr<IceCandidate>; // Ignore
         fn _unique_session_description() -> UniquePtr<SessionDescription>; // Ignore
     }
 }
 
+impl Debug for ffi::SessionDescription {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "{}", self.stringify())
+    }
+}
+
+unsafe impl Send for ffi::SessionDescription {}
+
+impl Debug for ffi::IceCandidate {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "TODO") // TODO(theomonnom)
+    }
+}
+
+unsafe impl Send for ffi::IceCandidate {}
+
 // CreateSdpObserver
 
-pub trait CreateSdpObserver: Send + Sync {
+pub trait CreateSdpObserver: Send {
     fn on_success(&self, session_description: UniquePtr<ffi::SessionDescription>);
     fn on_failure(&self, error: RTCError);
 }
@@ -51,9 +79,7 @@ pub struct CreateSdpObserverWrapper {
 
 impl CreateSdpObserverWrapper {
     pub fn new(observer: Box<dyn CreateSdpObserver>) -> Self {
-        Self {
-            observer
-        }
+        Self { observer }
     }
 
     fn on_success(&self, session_description: UniquePtr<ffi::SessionDescription>) {
@@ -67,7 +93,7 @@ impl CreateSdpObserverWrapper {
 
 // SetLocalSdpObserver
 
-pub trait SetLocalSdpObserver: Send + Sync {
+pub trait SetLocalSdpObserver: Send {
     fn on_set_local_description_complete(&self, error: RTCError);
 }
 
@@ -77,9 +103,7 @@ pub struct SetLocalSdpObserverWrapper {
 
 impl SetLocalSdpObserverWrapper {
     pub fn new(observer: Box<dyn SetLocalSdpObserver>) -> Self {
-        Self {
-            observer
-        }
+        Self { observer }
     }
 
     fn on_set_local_description_complete(&self, error: RTCError) {
@@ -89,7 +113,7 @@ impl SetLocalSdpObserverWrapper {
 
 // SetRemoteSdpObserver
 
-pub trait SetRemoteSdpObserver: Send + Sync {
+pub trait SetRemoteSdpObserver: Send {
     fn on_set_remote_description_complete(&self, error: RTCError);
 }
 
@@ -99,9 +123,7 @@ pub struct SetRemoteSdpObserverWrapper {
 
 impl SetRemoteSdpObserverWrapper {
     pub fn new(observer: Box<dyn SetRemoteSdpObserver>) -> Self {
-        Self {
-            observer
-        }
+        Self { observer }
     }
 
     fn on_set_remote_description_complete(&self, error: RTCError) {

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

@@ -1,10 +1,10 @@
+pub mod candidate;
+pub mod data_channel;
+pub mod jsep;
 pub mod media_stream_interface;
 pub mod peer_connection;
 pub mod peer_connection_factory;
-pub mod data_channel;
-pub mod jsep;
-pub mod candidate;
+pub mod rtc_error;
 pub mod rtp_receiver;
 pub mod rtp_transceiver;
-pub mod rtc_error;
-
+pub mod webrtc;

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

@@ -6,7 +6,7 @@
 
 namespace livekit {
 
-    MediaStreamInterface::MediaStreamInterface(rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) : media_stream_(stream) {
+    MediaStreamInterface::MediaStreamInterface(rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) : media_stream_(std::move(stream)) {
 
     }
 } // livekit

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

@@ -9,5 +9,4 @@ pub mod ffi {
 
         fn _unique_media_stream() -> UniquePtr<MediaStreamInterface>; // Ignore
     }
-
 }

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

@@ -4,6 +4,7 @@
 
 #include "livekit/peer_connection.h"
 #include "libwebrtc-sys/src/peer_connection.rs.h"
+#include "livekit/rtc_error.h"
 
 namespace livekit {
 
@@ -20,33 +21,61 @@ 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(NativeCreateSdpObserverHandle &observer_handle, RTCOfferAnswerOptions options) {
+        peer_connection_->CreateOffer(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, NativeSetLocalSdpObserverHandle &observer) {
+        peer_connection_->SetLocalDescription(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) {
+        auto result = peer_connection_->CreateDataChannelOrError(label.c_str(), init.get());
+
+        if(!result.ok()) {
+            throw std::runtime_error(serialize_error(to_error(result.error())));
+        }
+
+        return std::make_unique<DataChannel>(result.value());
+    }
+
+    void PeerConnection::add_ice_candidate(std::unique_ptr<IceCandidate> candidate, NativeAddIceCandidateObserver &observer){
+        peer_connection_->AddIceCandidate(candidate->release(), [&](const webrtc::RTCError& err){
+            observer.OnComplete(to_error(err));
+        });
+    }
+
     void PeerConnection::close() {
         peer_connection_->Close();
     }
 
-    void PeerConnection::create_offer(std::unique_ptr<NativeCreateSdpObserverHandle> observer_handle, RTCOfferAnswerOptions options) {
-        peer_connection_->CreateOffer(observer_handle->observer.get(), toNativeOfferAnswerOptions(options));
+    // AddIceCandidateObserver
+
+    NativeAddIceCandidateObserver::NativeAddIceCandidateObserver(rust::Box<AddIceCandidateObserverWrapper> observer) : observer_(std::move(observer)) {
+
     }
 
-    void PeerConnection::create_answer(std::unique_ptr<NativeCreateSdpObserverHandle> observer_handle, RTCOfferAnswerOptions options) {
-        peer_connection_->CreateAnswer(observer_handle->observer.get(), toNativeOfferAnswerOptions(options));
+    void NativeAddIceCandidateObserver::OnComplete(const RTCError &error) {
+        observer_->on_complete(error);
     }
 
-    void PeerConnection::set_local_description(std::unique_ptr<SessionDescription> desc, std::unique_ptr<NativeSetLocalSdpObserverHandle> observer) {
-        peer_connection_->SetLocalDescription(desc->clone()->release(), observer->observer);
+    std::unique_ptr<NativeAddIceCandidateObserver> create_native_add_ice_candidate_observer(rust::Box<AddIceCandidateObserverWrapper> observer) {
+        return std::make_unique<NativeAddIceCandidateObserver>(std::move(observer));
     }
 
-    void PeerConnection::set_remote_description(std::unique_ptr<SessionDescription> desc, std::unique_ptr<NativeSetRemoteSdpObserverHandle> observer) {
-        peer_connection_->SetRemoteDescription(desc->clone()->release(), observer->observer);
-    }
-
-    /* Observer */
+    // PeerConnectionObserver
 
     NativePeerConnectionObserver::NativePeerConnectionObserver(rust::Box<PeerConnectionObserverWrapper> observer) : observer_(std::move(observer)) {
 

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

@@ -1,10 +1,11 @@
-use cxx::UniquePtr;
 use crate::candidate::ffi::Candidate;
 use crate::data_channel::ffi::DataChannel;
 use crate::jsep::ffi::IceCandidate;
 use crate::media_stream_interface::ffi::MediaStreamInterface;
+use crate::rtc_error::ffi::RTCError;
 use crate::rtp_receiver::ffi::RtpReceiver;
 use crate::rtp_transceiver::ffi::RtpTransceiver;
+use cxx::UniquePtr;
 
 #[cxx::bridge(namespace = "livekit")]
 pub mod ffi {
@@ -61,7 +62,7 @@ pub mod ffi {
     pub enum IceGatheringState {
         IceGatheringNew,
         IceGatheringGathering,
-        IceGatheringComplete
+        IceGatheringComplete,
     }
 
     #[derive(Debug)]
@@ -79,11 +80,11 @@ pub mod ffi {
     // Wrapper to opaque C++ objects
     // https://github.com/dtolnay/cxx/issues/741
     struct MediaStreamPtr {
-        pub ptr: UniquePtr<MediaStreamInterface>
+        pub ptr: UniquePtr<MediaStreamInterface>,
     }
 
-    struct CandidatePtr  {
-        pub ptr: UniquePtr<Candidate>
+    struct CandidatePtr {
+        pub ptr: UniquePtr<Candidate>,
     }
 
     unsafe extern "C++" {
@@ -94,7 +95,9 @@ pub mod ffi {
         include!("livekit/rtp_transceiver.h");
         include!("livekit/media_stream_interface.h");
         include!("livekit/candidate.h");
+        include!("libwebrtc-sys/src/rtc_error.rs.h");
 
+        type RTCError = crate::rtc_error::ffi::RTCError;
         type Candidate = crate::candidate::ffi::Candidate;
         type IceCandidate = crate::jsep::ffi::IceCandidate;
         type DataChannel = crate::data_channel::ffi::DataChannel;
@@ -104,44 +107,140 @@ pub mod ffi {
         type NativeCreateSdpObserverHandle = crate::jsep::ffi::NativeCreateSdpObserverHandle;
         type NativeSetLocalSdpObserverHandle = crate::jsep::ffi::NativeSetLocalSdpObserverHandle;
         type NativeSetRemoteSdpObserverHandle = crate::jsep::ffi::NativeSetRemoteSdpObserverHandle;
+        type NativeDataChannelInit = crate::data_channel::ffi::NativeDataChannelInit;
         type SessionDescription = crate::jsep::ffi::SessionDescription;
 
+        type NativeAddIceCandidateObserver;
         type NativePeerConnectionObserver;
         type PeerConnection;
 
-        fn create_offer(self: Pin<&mut PeerConnection>, observer: UniquePtr<NativeCreateSdpObserverHandle>, options: RTCOfferAnswerOptions);
-        fn create_answer(self: Pin<&mut PeerConnection>, observer: UniquePtr<NativeCreateSdpObserverHandle>, options: RTCOfferAnswerOptions);
-        fn set_local_description(self: Pin<&mut PeerConnection>, desc: UniquePtr<SessionDescription>, observer: UniquePtr<NativeSetLocalSdpObserverHandle>);
-        fn set_remote_description(self: Pin<&mut PeerConnection>, desc: UniquePtr<SessionDescription>, observer: UniquePtr<NativeSetRemoteSdpObserverHandle>);
+        /// SAFETY
+        /// The observer must live as long as the operation ends
+        unsafe fn create_offer(
+            self: Pin<&mut PeerConnection>,
+            observer: Pin<&mut NativeCreateSdpObserverHandle>,
+            options: RTCOfferAnswerOptions,
+        );
+
+        /// SAFETY
+        /// The observer must live as long as the operation ends
+        unsafe fn create_answer(
+            self: Pin<&mut PeerConnection>,
+            observer: Pin<&mut NativeCreateSdpObserverHandle>,
+            options: RTCOfferAnswerOptions,
+        );
+
+        /// SAFETY
+        /// The observer must live as long as the operation ends
+        unsafe fn set_local_description(
+            self: Pin<&mut PeerConnection>,
+            desc: UniquePtr<SessionDescription>,
+            observer: Pin<&mut NativeSetLocalSdpObserverHandle>,
+        );
+
+        /// SAFETY
+        /// The observer must live as long as the operation ends
+        unsafe fn set_remote_description(
+            self: Pin<&mut PeerConnection>,
+            desc: UniquePtr<SessionDescription>,
+            observer: Pin<&mut NativeSetRemoteSdpObserverHandle>,
+        );
+
+        fn create_data_channel(
+            self: Pin<&mut PeerConnection>,
+            label: String,
+            init: UniquePtr<NativeDataChannelInit>,
+        ) -> Result<UniquePtr<DataChannel>>;
+
+        fn add_ice_candidate(
+            self: Pin<&mut PeerConnection>,
+            candidate: UniquePtr<IceCandidate>,
+            observer: Pin<&mut NativeAddIceCandidateObserver>,
+        );
+
         fn close(self: Pin<&mut PeerConnection>);
 
-        fn create_native_peer_connection_observer(observer: Box<PeerConnectionObserverWrapper>) -> UniquePtr<NativePeerConnectionObserver>;
+        fn create_native_peer_connection_observer(
+            observer: Box<PeerConnectionObserverWrapper>,
+        ) -> UniquePtr<NativePeerConnectionObserver>;
+
+        fn create_native_add_ice_candidate_observer(
+            observer: Box<AddIceCandidateObserverWrapper>,
+        ) -> UniquePtr<NativeAddIceCandidateObserver>;
 
         fn _unique_peer_connection() -> UniquePtr<PeerConnection>; // Ignore
     }
 
     extern "Rust" {
+        type AddIceCandidateObserverWrapper;
+
+        fn on_complete(self: &AddIceCandidateObserverWrapper, error: RTCError);
+
         type PeerConnectionObserverWrapper;
 
-        fn on_signaling_change(self: &mut PeerConnectionObserverWrapper, new_state: SignalingState);
-        fn on_add_stream(self: &mut PeerConnectionObserverWrapper, stream: UniquePtr<MediaStreamInterface>);
-        fn on_remove_stream(self: &mut PeerConnectionObserverWrapper, stream: UniquePtr<MediaStreamInterface>);
-        fn on_data_channel(self: &mut PeerConnectionObserverWrapper, data_channel: UniquePtr<DataChannel>);
-        fn on_renegotiation_needed(self: &mut PeerConnectionObserverWrapper);
-        fn on_negotiation_needed_event(self: &mut PeerConnectionObserverWrapper, event: u32);
-        fn on_ice_connection_change(self: &mut PeerConnectionObserverWrapper, new_state: IceConnectionState);
-        fn on_standardized_ice_connection_change(self: &mut PeerConnectionObserverWrapper, new_state: IceConnectionState);
-        fn on_connection_change(self: &mut PeerConnectionObserverWrapper, new_state: PeerConnectionState);
-        fn on_ice_gathering_change(self: &mut PeerConnectionObserverWrapper, new_state: IceGatheringState);
-        fn on_ice_candidate(self: &mut PeerConnectionObserverWrapper, candidate: UniquePtr<IceCandidate>);
-        fn on_ice_candidate_error(self: &mut PeerConnectionObserverWrapper, address: String, port: i32, url: String, error_code: i32, error_text: String);
-        fn on_ice_candidates_removed(self: &mut PeerConnectionObserverWrapper, removed: Vec<CandidatePtr>);
-        fn on_ice_connection_receiving_change(self: &mut PeerConnectionObserverWrapper, receiving: bool);
-        fn on_ice_selected_candidate_pair_changed(self: &mut PeerConnectionObserverWrapper, event: CandidatePairChangeEvent);
-        fn on_add_track(self: &mut PeerConnectionObserverWrapper, receiver: UniquePtr<RtpReceiver>, streams: Vec<MediaStreamPtr>);
-        fn on_track(self: &mut PeerConnectionObserverWrapper, transceiver: UniquePtr<RtpTransceiver>);
-        fn on_remove_track(self: &mut PeerConnectionObserverWrapper, receiver: UniquePtr<RtpReceiver>);
-        fn on_interesting_usage(self: &mut PeerConnectionObserverWrapper, usage_pattern: i32);
+        fn on_signaling_change(self: &PeerConnectionObserverWrapper, new_state: SignalingState);
+        fn on_add_stream(
+            self: &PeerConnectionObserverWrapper,
+            stream: UniquePtr<MediaStreamInterface>,
+        );
+        fn on_remove_stream(
+            self: &PeerConnectionObserverWrapper,
+            stream: UniquePtr<MediaStreamInterface>,
+        );
+        fn on_data_channel(
+            self: &PeerConnectionObserverWrapper,
+            data_channel: UniquePtr<DataChannel>,
+        );
+        fn on_renegotiation_needed(self: &PeerConnectionObserverWrapper);
+        fn on_negotiation_needed_event(self: &PeerConnectionObserverWrapper, event: u32);
+        fn on_ice_connection_change(
+            self: &PeerConnectionObserverWrapper,
+            new_state: IceConnectionState,
+        );
+        fn on_standardized_ice_connection_change(
+            self: &PeerConnectionObserverWrapper,
+            new_state: IceConnectionState,
+        );
+        fn on_connection_change(
+            self: &PeerConnectionObserverWrapper,
+            new_state: PeerConnectionState,
+        );
+        fn on_ice_gathering_change(
+            self: &PeerConnectionObserverWrapper,
+            new_state: IceGatheringState,
+        );
+        fn on_ice_candidate(
+            self: &PeerConnectionObserverWrapper,
+            candidate: UniquePtr<IceCandidate>,
+        );
+        fn on_ice_candidate_error(
+            self: &PeerConnectionObserverWrapper,
+            address: String,
+            port: i32,
+            url: String,
+            error_code: i32,
+            error_text: String,
+        );
+        fn on_ice_candidates_removed(
+            self: &PeerConnectionObserverWrapper,
+            removed: Vec<CandidatePtr>,
+        );
+        fn on_ice_connection_receiving_change(
+            self: &PeerConnectionObserverWrapper,
+            receiving: bool,
+        );
+        fn on_ice_selected_candidate_pair_changed(
+            self: &PeerConnectionObserverWrapper,
+            event: CandidatePairChangeEvent,
+        );
+        fn on_add_track(
+            self: &PeerConnectionObserverWrapper,
+            receiver: UniquePtr<RtpReceiver>,
+            streams: Vec<MediaStreamPtr>,
+        );
+        fn on_track(self: &PeerConnectionObserverWrapper, transceiver: UniquePtr<RtpTransceiver>);
+        fn on_remove_track(self: &PeerConnectionObserverWrapper, receiver: UniquePtr<RtpReceiver>);
+        fn on_interesting_usage(self: &PeerConnectionObserverWrapper, usage_pattern: i32);
     }
 }
 
@@ -149,12 +248,14 @@ pub mod ffi {
 unsafe impl Sync for ffi::PeerConnection {}
 unsafe impl Send for ffi::PeerConnection {}
 
+unsafe impl Send for ffi::NativePeerConnectionObserver {}
+
 impl Default for ffi::RTCOfferAnswerOptions {
     /*
-        static const int kUndefined = -1;
-        static const int kMaxOfferToReceiveMedia = 1;
-        static const int kOfferToReceiveMediaTrue = 1;
-     */
+       static const int kUndefined = -1;
+       static const int kMaxOfferToReceiveMedia = 1;
+       static const int kOfferToReceiveMediaTrue = 1;
+    */
 
     fn default() -> Self {
         Self {
@@ -165,129 +266,200 @@ impl Default for ffi::RTCOfferAnswerOptions {
             use_rtp_mux: true,
             raw_packetization_for_video: false,
             num_simulcast_layers: 1,
-            use_obsolete_sctp_sdp: false
+            use_obsolete_sctp_sdp: false,
         }
     }
 }
 
+pub struct AddIceCandidateObserverWrapper {
+    observer: Box<dyn Fn(RTCError) + Send>,
+}
+
+impl AddIceCandidateObserverWrapper {
+    pub fn new(observer: Box<dyn Fn(RTCError) + Send>) -> Self {
+        Self { observer }
+    }
+
+    fn on_complete(&self, error: RTCError) {
+        (self.observer)(error);
+    }
+}
+
 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_ice_candidate_error(&mut 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_add_track(&mut 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_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(
+        &self,
+        address: String,
+        port: i32,
+        url: String,
+        error_code: i32,
+        error_text: String,
+    );
+    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(
+        &self,
+        receiver: UniquePtr<RtpReceiver>,
+        streams: Vec<UniquePtr<MediaStreamInterface>>,
+    );
+    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
 pub struct PeerConnectionObserverWrapper {
-    observer: Box<dyn PeerConnectionObserver>,
+    observer: *mut dyn PeerConnectionObserver,
 }
 
 impl PeerConnectionObserverWrapper {
-    pub fn new(observer: Box<dyn PeerConnectionObserver>) -> Self {
-        Self {
-            observer
+    /// SAFETY
+    /// PeerConnectionObserver must lives as long as PeerConnectionObserverWrapper does
+    pub unsafe fn new(observer: *mut dyn PeerConnectionObserver) -> Self {
+        Self { observer }
+    }
+
+    fn on_signaling_change(&self, new_state: ffi::SignalingState) {
+        unsafe {
+            (*self.observer).on_signaling_change(new_state);
         }
     }
 
-    fn on_signaling_change(&mut self, new_state: ffi::SignalingState) {
-        self.observer.on_signaling_change(new_state);
+    fn on_add_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
+        unsafe {
+            (*self.observer).on_add_stream(stream);
+        }
     }
 
-    fn on_add_stream(&mut self, stream: UniquePtr<MediaStreamInterface>) {
-        self.observer.on_add_stream(stream);
+    fn on_remove_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
+        unsafe {
+            (*self.observer).on_remove_stream(stream);
+        }
     }
 
-    fn on_remove_stream(&mut self, stream: UniquePtr<MediaStreamInterface>) {
-        self.observer.on_remove_stream(stream);
+    fn on_data_channel(&self, data_channel: UniquePtr<DataChannel>) {
+        unsafe {
+            (*self.observer).on_data_channel(data_channel);
+        }
     }
 
-    fn on_data_channel(&mut self, data_channel: UniquePtr<DataChannel>) {
-        self.observer.on_data_channel(data_channel);
+    fn on_renegotiation_needed(&self) {
+        unsafe {
+            (*self.observer).on_renegotiation_needed();
+        }
     }
 
-    fn on_renegotiation_needed(&mut self) {
-        self.observer.on_renegotiation_needed();
+    fn on_negotiation_needed_event(&self, event: u32) {
+        unsafe {
+            (*self.observer).on_negotiation_needed_event(event);
+        }
     }
 
-    fn on_negotiation_needed_event(&mut self, event: u32) {
-        self.observer.on_negotiation_needed_event(event);
+    fn on_ice_connection_change(&self, new_state: ffi::IceConnectionState) {
+        unsafe {
+            (*self.observer).on_ice_connection_change(new_state);
+        }
     }
 
-    fn on_ice_connection_change(&mut self, new_state: ffi::IceConnectionState) {
-        self.observer.on_ice_connection_change(new_state);
+    fn on_standardized_ice_connection_change(&self, new_state: ffi::IceConnectionState) {
+        unsafe {
+            (*self.observer).on_standardized_ice_connection_change(new_state);
+        }
     }
 
-    fn on_standardized_ice_connection_change(&mut self, new_state: ffi::IceConnectionState) {
-        self.observer.on_standardized_ice_connection_change(new_state);
+    fn on_connection_change(&self, new_state: ffi::PeerConnectionState) {
+        unsafe {
+            (*self.observer).on_connection_change(new_state);
+        }
     }
 
-    fn on_connection_change(&mut self, new_state: ffi::PeerConnectionState) {
-        self.observer.on_connection_change(new_state);
+    fn on_ice_gathering_change(&self, new_state: ffi::IceGatheringState) {
+        unsafe {
+            (*self.observer).on_ice_gathering_change(new_state);
+        }
     }
 
-    fn on_ice_gathering_change(&mut self, new_state: ffi::IceGatheringState) {
-        self.observer.on_ice_gathering_change(new_state);
+    fn on_ice_candidate(&self, candidate: UniquePtr<IceCandidate>) {
+        unsafe {
+            (*self.observer).on_ice_candidate(candidate);
+        }
     }
 
-    fn on_ice_candidate(&mut self, candidate: UniquePtr<IceCandidate>) {
-        self.observer.on_ice_candidate(candidate);
+    fn on_ice_candidate_error(
+        &self,
+        address: String,
+        port: i32,
+        url: String,
+        error_code: i32,
+        error_text: String,
+    ) {
+        unsafe {
+            (*self.observer).on_ice_candidate_error(address, port, url, error_code, error_text);
+        }
     }
 
-    fn on_ice_candidate_error(&mut self, address: String, port: i32, url: String, error_code: i32, error_text: String) {
-        self.observer.on_ice_candidate_error(address, port, url, error_code, error_text);
-    }
-
-    fn on_ice_candidates_removed(&mut self, removed: Vec<ffi::CandidatePtr>) {
+    fn on_ice_candidates_removed(&self, removed: Vec<ffi::CandidatePtr>) {
         let mut vec = Vec::new();
 
         for v in removed {
             vec.push(v.ptr);
         }
 
-        self.observer.on_ice_candidates_removed(vec);
+        unsafe {
+            (*self.observer).on_ice_candidates_removed(vec);
+        }
     }
 
-    fn on_ice_connection_receiving_change(&mut self, receiving: bool) {
-        self.observer.on_ice_connection_receiving_change(receiving);
+    fn on_ice_connection_receiving_change(&self, receiving: bool) {
+        unsafe {
+            (*self.observer).on_ice_connection_receiving_change(receiving);
+        }
     }
 
-    fn on_ice_selected_candidate_pair_changed(&mut self, event: ffi::CandidatePairChangeEvent) {
-        self.observer.on_ice_selected_candidate_pair_changed(event);
+    fn on_ice_selected_candidate_pair_changed(&self, event: ffi::CandidatePairChangeEvent) {
+        unsafe {
+            (*self.observer).on_ice_selected_candidate_pair_changed(event);
+        }
     }
 
-    fn on_add_track(&mut self, receiver: UniquePtr<RtpReceiver>, streams: Vec<ffi::MediaStreamPtr>) {
+    fn on_add_track(&self, receiver: UniquePtr<RtpReceiver>, streams: Vec<ffi::MediaStreamPtr>) {
         let mut vec = Vec::new();
 
         for v in streams {
             vec.push(v.ptr);
         }
 
-        self.observer.on_add_track(receiver, vec);
+        unsafe {
+            (*self.observer).on_add_track(receiver, vec);
+        }
     }
 
-    fn on_track(&mut self, transceiver: UniquePtr<RtpTransceiver>) {
-        self.observer.on_track(transceiver);
+    fn on_track(&self, transceiver: UniquePtr<RtpTransceiver>) {
+        unsafe {
+            (*self.observer).on_track(transceiver);
+        }
     }
 
-    fn on_remove_track(&mut self, receiver: UniquePtr<RtpReceiver>) {
-        self.observer.on_remove_track(receiver);
+    fn on_remove_track(&self, receiver: UniquePtr<RtpReceiver>) {
+        unsafe {
+            (*self.observer).on_remove_track(receiver);
+        }
     }
 
-    fn on_interesting_usage(&mut self, usage_pattern: i32) {
-        self.observer.on_interesting_usage(usage_pattern);
+    fn on_interesting_usage(&self, usage_pattern: i32) {
+        unsafe {
+            (*self.observer).on_interesting_usage(usage_pattern);
+        }
     }
 }

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()){
+        if(!result.ok()) {
             throw std::runtime_error(serialize_error(to_error(result.error())));
         }
 
-        return std::make_unique<PeerConnection>(std::move(result.value()), std::move(observer));
+        return std::make_unique<PeerConnection>(result.value());
     }
 
     std::unique_ptr<PeerConnectionFactory> create_peer_connection_factory() {
@@ -70,14 +70,13 @@ namespace livekit{
 
     std::unique_ptr<NativeRTCConfiguration> create_rtc_configuration(RTCConfiguration conf){
         auto rtc = std::make_unique<webrtc::PeerConnectionInterface::RTCConfiguration>();
-
         for (auto &item: conf.ice_servers){
             webrtc::PeerConnectionInterface::IceServer ice_server;
             ice_server.username = item.username.c_str();
             ice_server.password = item.password.c_str();
 
             for (auto &url: item.urls){
-                ice_server.urls.push_back(url.c_str());
+                ice_server.urls.emplace_back(url.c_str());
             }
 
             rtc->servers.push_back(ice_server);

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

@@ -1,30 +1,33 @@
-use std::any::Any;
-use std::thread::sleep;
-use std::time::Duration;
-use cxx::UniquePtr;
-use log::info;
 use crate::candidate::ffi::Candidate;
 use crate::data_channel::ffi::DataChannel;
 use crate::jsep::ffi::IceCandidate;
-use crate::media_stream_interface::ffi::MediaStreamInterface;
-use crate::{jsep, peer_connection};
 use crate::jsep::CreateSdpObserver;
-use crate::peer_connection::ffi::{CandidatePairChangeEvent, IceConnectionState, IceGatheringState, PeerConnectionState, SignalingState};
+use crate::media_stream_interface::ffi::MediaStreamInterface;
+use crate::peer_connection::ffi::{
+    CandidatePairChangeEvent, IceConnectionState, IceGatheringState, PeerConnectionState,
+    SignalingState,
+};
 use crate::peer_connection::PeerConnectionObserver;
 use crate::rtp_receiver::ffi::RtpReceiver;
 use crate::rtp_transceiver::ffi::RtpTransceiver;
+use crate::{jsep, peer_connection};
+use cxx::UniquePtr;
+use log::info;
+use std::any::Any;
+use std::thread::sleep;
+use std::time::Duration;
 
 #[cxx::bridge(namespace = "livekit")]
 pub mod ffi {
 
     #[derive(Debug, Clone)]
     pub struct ICEServer {
-        urls: Vec<String>,
-        username: String,
-        password: String,
+        pub urls: Vec<String>,
+        pub username: String,
+        pub password: String,
     }
 
-    #[derive(Debug)]
+    #[derive(Debug, Clone)]
     pub struct RTCConfiguration {
         pub ice_servers: Vec<ICEServer>,
     }
@@ -33,154 +36,20 @@ pub mod ffi {
         include!("livekit/peer_connection_factory.h");
 
         type PeerConnection = crate::peer_connection::ffi::PeerConnection;
-        type NativePeerConnectionObserver = crate::peer_connection::ffi::NativePeerConnectionObserver;
+        type NativePeerConnectionObserver =
+            crate::peer_connection::ffi::NativePeerConnectionObserver;
         type PeerConnectionFactory;
         type NativeRTCConfiguration;
 
         fn create_peer_connection_factory() -> UniquePtr<PeerConnectionFactory>;
         fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr<NativeRTCConfiguration>;
 
-        unsafe fn create_peer_connection(self: &PeerConnectionFactory, config: UniquePtr<NativeRTCConfiguration>, observer: UniquePtr<NativePeerConnectionObserver>) -> Result<UniquePtr<PeerConnection>>;
-    }
-  }
-
-
-
-
-
-
-
-/*
-
-
-
-
-
-
-#[cfg(test)]
-mod test {
-
-    struct TestObserver {
-
-    }
-
-    impl PeerConnectionObserver for TestObserver {
-        fn on_signaling_change(&self, new_state: SignalingState) {
-            log::debug!("Signaling state changed: {:?}", new_state);
-        }
-
-        fn on_add_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
-            todo!()
-        }
-
-        fn on_remove_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
-            todo!()
-        }
-
-        fn on_data_channel(&self, data_channel: UniquePtr<DataChannel>) {
-            todo!()
-        }
-
-        fn on_renegotiation_needed(&self) {
-            todo!()
-        }
-
-        fn on_negotiation_needed_event(&self, event: u32) {
-            todo!()
-        }
-
-        fn on_ice_connection_change(&self, new_state: IceConnectionState) {
-            log::debug!("ICE connection state changed: {:?}", new_state);
-        }
-
-        fn on_standardized_ice_connection_change(&self, new_state: IceConnectionState) {
-            todo!()
-        }
-
-        fn on_connection_change(&self, new_state: PeerConnectionState) {
-            log::debug!("PeerConnection state changed: {:?}", new_state);
-        }
-
-        fn on_ice_gathering_change(&self, new_state: IceGatheringState) {
-            todo!()
-        }
-
-        fn on_ice_candidate(&self, candidate: UniquePtr<IceCandidate>) {
-            todo!()
-        }
-
-        fn on_ice_candidate_error(&self, address: String, port: i32, url: String, error_code: i32, error_text: String) {
-            todo!()
-        }
-
-        fn on_ice_candidates_removed(&self, removed: Vec<UniquePtr<Candidate>>) {
-            todo!()
-        }
-
-        fn on_ice_connection_receiving_change(&self, receiving: bool) {
-            todo!()
-        }
-
-        fn on_ice_selected_candidate_pair_changed(&self, event: CandidatePairChangeEvent) {
-            todo!()
-        }
-
-        fn on_add_track(&self, receiver: UniquePtr<RtpReceiver>, streams: Vec<UniquePtr<MediaStreamInterface>>) {
-            todo!()
-        }
-
-        fn on_track(&self, transceiver: UniquePtr<RtpTransceiver>) {
-            todo!()
-        }
-
-        fn on_remove_track(&self, receiver: UniquePtr<RtpReceiver>) {
-            todo!()
-        }
-
-        fn on_interesting_usage(&self, usage_pattern: i32) {
-            todo!()
-        }
-    }
-
-    struct SessionObserver {
-
-    }
-
-    impl CreateSdpObserver for SessionObserver {
-        fn on_success(&self, session_description: UniquePtr<crate::jsep::ffi::SessionDescription>) {
-            info!("on_success");
-        }
-
-        fn on_failure(&self, error: UniquePtr<crate::rtc_error::ffi::RTCError>) {
-            info!("on_failure");
-        }
-    }
-
-    #[test]
-    fn create_pc_test() {
-        env_logger::init();
-        let factory = ffi::create_peer_connection_factory(); // Default factory config is defined on the c++ side atm
-        unsafe {
-            let mut pc = factory.create_peer_connection(ffi::create_rtc_configuration(ffi::RTCConfiguration {
-                ice_servers: vec![ffi::ICEServer {
-                    urls: vec!["stun:stun.l.google.com:19302".to_string()],
-                    username: "".to_string(),
-                    password: "".to_string(),
-                }],
-            }),  peer_connection::ffi::create_native_peer_connection_observer(Box::new(peer_connection::PeerConnectionObserverWrapper::new(Box::new(TestObserver{}))))).unwrap();
-
-
-            let options = peer_connection::ffi::RTCOfferAnswerOptions::default();
-
-            let sdp_observer = jsep::ffi::create_native_create_sdp_observer(Box::new(jsep::CreateSdpObserverWrapper::new(Box::new(SessionObserver{}))));
-            pc.pin_mut().create_offer(sdp_observer, options);
-
-            sleep(Duration::from_secs(2));
-
-            pc.pin_mut().close();
-        }
+        /// SAFETY
+        /// The observer must live as long as the PeerConnection
+        unsafe fn create_peer_connection(
+            self: &PeerConnectionFactory,
+            config: UniquePtr<NativeRTCConfiguration>,
+            observer: Pin<&mut NativePeerConnectionObserver>,
+        ) -> Result<UniquePtr<PeerConnection>>;
     }
 }
-
-*/
-

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

@@ -7,7 +7,6 @@
 #include <sstream>
 #include <string>
 
-
 namespace livekit {
 
     RTCError to_error(const webrtc::RTCError &error) {
@@ -15,7 +14,7 @@ namespace livekit {
         lk_error.error_detail = static_cast<RTCErrorDetailType>(error.error_detail());
         lk_error.error_type = static_cast<RTCErrorType>(error.type());
         lk_error.has_sctp_cause_code = error.sctp_cause_code().has_value();
-        lk_error.sctp_cause_code = error.sctp_cause_code().value();
+        lk_error.sctp_cause_code = error.sctp_cause_code().value_or(0);
         lk_error.message = error.message();
         return lk_error;
     }

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

@@ -1,5 +1,8 @@
-use std::fmt::{Display, Formatter};
+use crate::rtc_error::ffi::RTCErrorType;
 use std::error::Error;
+use std::fmt::{Display, Formatter};
+
+// cxx doesn't support custom Exception type, so we serialize RTCError inside the cxx::Exception "what" string
 
 #[cxx::bridge(namespace = "livekit")]
 pub mod ffi {
@@ -40,7 +43,7 @@ pub mod ffi {
         pub message: String,
         pub error_detail: RTCErrorDetailType,
         pub has_sctp_cause_code: bool, // cxx doesn't support the Option trait
-        pub sctp_cause_code: u16
+        pub sctp_cause_code: u16,
     }
 }
 
@@ -63,15 +66,21 @@ impl ffi::RTCError {
             message,
         }
     }
+
+    pub fn ok(&self) -> bool {
+        return self.error_type == RTCErrorType::None;
+    }
 }
 
-impl Error for ffi::RTCError {
-
-}
+impl Error for ffi::RTCError {}
 
 impl Display for ffi::RTCError {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "RtcError occurred {:?}: {}", self.error_type, self.message)
+        write!(
+            f,
+            "RtcError occurred {:?}: {}",
+            self.error_type, self.message
+        )
     }
 }
 
@@ -90,7 +99,7 @@ mod tests {
     }
 
     #[test]
-    fn serialize_deserialize(){
+    fn serialize_deserialize() {
         let str = ffi::serialize_deserialize();
         let error = unsafe { RTCError::from(&str) };
 
@@ -98,11 +107,14 @@ mod tests {
         assert_eq!(error.error_detail, RTCErrorDetailType::DataChannelFailure);
         assert_eq!(error.has_sctp_cause_code, true);
         assert_eq!(error.sctp_cause_code, 24);
-        assert_eq!(error.message, "this is not a test, I repeat, this is not a test");
+        assert_eq!(
+            error.message,
+            "this is not a test, I repeat, this is not a test"
+        );
     }
 
     #[test]
-    fn throw_error(){
+    fn throw_error() {
         let exc: cxx::Exception = ffi::throw_error().err().unwrap();
         let error = unsafe { RTCError::from(exc.what()) };
 
@@ -112,4 +124,4 @@ mod tests {
         assert_eq!(error.sctp_cause_code, 0);
         assert_eq!(error.message, "exception is thrown!");
     }
-}
+}

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

@@ -0,0 +1,22 @@
+//
+// Created by theom on 18/09/2022.
+//
+
+#include "livekit/webrtc.h"
+#include "rtc_base/logging.h"
+
+namespace livekit {
+    RTCRuntime::RTCRuntime() {
+        RTC_LOG(LS_INFO) << "RTCRuntime()";
+        RTC_CHECK(rtc::InitializeSSL()) << "Failed to InitializeSSL()";
+    }
+
+    RTCRuntime::~RTCRuntime() {
+        RTC_LOG(LS_INFO) << "~RTCRuntime()";
+        RTC_CHECK(rtc::CleanupSSL()) << "Failed to CleanupSSL()";
+    }
+
+    std::unique_ptr<RTCRuntime> create_rtc_runtime(){
+        return std::make_unique<RTCRuntime>();
+    }
+} // livekit

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

@@ -0,0 +1,12 @@
+use cxx::UniquePtr;
+
+#[cxx::bridge(namespace = "livekit")]
+pub mod ffi {
+    unsafe extern "C++" {
+        include!("livekit/webrtc.h");
+
+        type RTCRuntime;
+
+        fn create_rtc_runtime() -> UniquePtr<RTCRuntime>;
+    }
+}

crates/livekit-webrtc/src/data_channel.rs

@@ -1,6 +1,170 @@
+use cxx::UniquePtr;
+use libwebrtc_sys::data_channel as sys_dc;
+use log::trace;
+use std::fmt::{Debug, Formatter};
+use std::sync::{Arc, Mutex};
 
+pub use sys_dc::ffi::Priority;
 
 pub struct DataChannel {
+    cxx_handle: UniquePtr<sys_dc::ffi::DataChannel>,
+    observer: Box<InternalDataChannelObserver>,
 
+    // Keep alive for C++
+    native_observer: UniquePtr<sys_dc::ffi::NativeDataChannelObserver>,
+}
 
-}
+impl Debug for DataChannel {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "DataChannel [{:?}]", self.label())
+    }
+}
+
+impl DataChannel {
+    pub(crate) fn new(cxx_handle: UniquePtr<sys_dc::ffi::DataChannel>) -> Self {
+        let mut observer = Box::new(InternalDataChannelObserver::default());
+
+        let mut dc = unsafe {
+            Self {
+                cxx_handle,
+                native_observer: sys_dc::ffi::create_native_data_channel_observer(Box::new(
+                    sys_dc::DataChannelObserverWrapper::new(&mut *observer),
+                )),
+                observer,
+            }
+        };
+
+        unsafe {
+            dc.cxx_handle
+                .pin_mut()
+                .register_observer(dc.native_observer.pin_mut());
+        }
+
+        dc
+    }
+
+    pub fn send(&mut self, data: &[u8], binary: bool) -> bool {
+        let buffer = sys_dc::ffi::DataBuffer {
+            ptr: data.as_ptr(),
+            len: data.len(),
+            binary,
+        };
+        self.cxx_handle.pin_mut().send(&buffer)
+    }
+
+    pub fn label(&self) -> String {
+        self.cxx_handle.label()
+    }
+
+    pub fn close(&mut self) {
+        self.cxx_handle.pin_mut().close();
+    }
+
+    pub fn on_state_change(&mut self, handler: OnStateChangeHandler) {
+        *self.observer.on_state_change_handler.lock().unwrap() = Some(handler);
+    }
+
+    pub fn on_message(&mut self, handler: OnMessageHandler) {
+        *self.observer.on_message_handler.lock().unwrap() = Some(handler);
+    }
+
+    pub fn on_buffer(&mut self, handler: OnBufferedAmountChangeHandler) {
+        *self
+            .observer
+            .on_buffered_amount_change_handler
+            .lock()
+            .unwrap() = Some(handler);
+    }
+}
+
+pub type OnStateChangeHandler = Box<dyn FnMut() + Send + Sync>;
+pub type OnMessageHandler = Box<dyn FnMut(&[u8], bool) + Send + Sync>; // data, is_binary
+pub type OnBufferedAmountChangeHandler = Box<dyn FnMut(u64) + Send + Sync>;
+
+struct InternalDataChannelObserver {
+    on_state_change_handler: Arc<Mutex<Option<OnStateChangeHandler>>>,
+    on_message_handler: Arc<Mutex<Option<OnMessageHandler>>>,
+    on_buffered_amount_change_handler: Arc<Mutex<Option<OnBufferedAmountChangeHandler>>>,
+}
+
+impl sys_dc::DataChannelObserver for InternalDataChannelObserver {
+    fn on_state_change(&self) {
+        trace!("DataChannel: on_state_change");
+        let mut handler = self.on_state_change_handler.lock().unwrap();
+        if let Some(f) = handler.as_mut() {
+            f();
+        }
+    }
+
+    fn on_message(&self, data: &[u8], is_binary: bool) {
+        trace!("DataChannel: on_message");
+        let mut handler = self.on_message_handler.lock().unwrap();
+        if let Some(f) = handler.as_mut() {
+            f(data, is_binary);
+        }
+    }
+
+    fn on_buffered_amount_change(&self, sent_data_size: u64) {
+        trace!("DataChannel: on_buffered_amount_change");
+        let mut handler = self.on_buffered_amount_change_handler.lock().unwrap();
+        if let Some(f) = handler.as_mut() {
+            f(sent_data_size);
+        }
+    }
+}
+
+impl Default for InternalDataChannelObserver {
+    fn default() -> Self {
+        Self {
+            on_state_change_handler: Arc::new(Default::default()),
+            on_message_handler: Arc::new(Default::default()),
+            on_buffered_amount_change_handler: Arc::new(Default::default()),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct DataChannelInit {
+    #[deprecated]
+    reliable: bool,
+    ordered: bool,
+    max_retransmit_time: Option<i32>,
+    max_retransmits: Option<i32>,
+    protocol: String,
+    negotiated: bool,
+    id: i32,
+    priority: Option<Priority>,
+}
+
+impl Default for DataChannelInit {
+    fn default() -> Self {
+        Self {
+            reliable: false,
+            ordered: true,
+            max_retransmit_time: None,
+            max_retransmits: None,
+            protocol: "".to_string(),
+            negotiated: false,
+            id: -1,
+            priority: None,
+        }
+    }
+}
+
+impl From<DataChannelInit> for sys_dc::ffi::DataChannelInit {
+    fn from(init: DataChannelInit) -> Self {
+        Self {
+            reliable: init.reliable,
+            ordered: init.ordered,
+            has_max_retransmit_time: init.max_retransmit_time.is_some(),
+            max_retransmit_time: init.max_retransmit_time.unwrap_or_default(),
+            has_max_retransmits: init.max_retransmits.is_some(),
+            max_retransmits: init.max_retransmits.unwrap_or_default(),
+            protocol: init.protocol,
+            negotiated: init.negotiated,
+            id: init.id,
+            has_priority: init.priority.is_some(),
+            priority: init.priority.unwrap_or(Priority::Low),
+        }
+    }
+}

crates/livekit-webrtc/src/ice_candidate.rs

@@ -1,4 +0,0 @@
-
-pub struct IceCandidate {
-
-}

crates/livekit-webrtc/src/jsep.rs

@@ -0,0 +1,38 @@
+use cxx::{SharedPtr, UniquePtr};
+use libwebrtc_sys::jsep as sys_jsep;
+
+#[derive(Debug)]
+pub struct IceCandidate {
+    cxx_handle: UniquePtr<sys_jsep::ffi::IceCandidate>,
+}
+
+impl IceCandidate {
+    pub(crate) fn new(cxx_handle: UniquePtr<sys_jsep::ffi::IceCandidate>) -> Self {
+        Self { cxx_handle }
+    }
+
+    pub(crate) fn release(self) -> UniquePtr<sys_jsep::ffi::IceCandidate> {
+        self.cxx_handle
+    }
+}
+
+#[derive(Debug)]
+pub struct SessionDescription {
+    cxx_handle: UniquePtr<sys_jsep::ffi::SessionDescription>,
+}
+
+impl SessionDescription {
+    pub(crate) fn new(cxx_handle: UniquePtr<sys_jsep::ffi::SessionDescription>) -> Self {
+        Self { cxx_handle }
+    }
+
+    pub(crate) fn release(self) -> UniquePtr<sys_jsep::ffi::SessionDescription> {
+        self.cxx_handle
+    }
+}
+
+impl Clone for SessionDescription {
+    fn clone(&self) -> Self {
+        SessionDescription::new(self.cxx_handle.clone())
+    }
+}

crates/livekit-webrtc/src/lib.rs

@@ -1,9 +1,9 @@
-pub mod peer_connection_factory;
-pub mod peer_connection;
-pub mod rtc_error;
 pub mod data_channel;
+pub mod jsep;
 pub mod media_stream;
+pub mod peer_connection;
+pub mod peer_connection_factory;
+pub mod rtc_error;
 pub mod rtp_receiver;
 pub mod rtp_transceiver;
-pub mod ice_candidate;
-pub mod session_description;
+pub mod webrtc;

crates/livekit-webrtc/src/media_stream.rs

@@ -1,4 +1,2 @@
-
-pub struct MediaStream {
-
-}
+#[derive(Debug)]
+pub struct MediaStream {}

crates/livekit-webrtc/src/peer_connection.rs

@@ -1,23 +1,25 @@
-use std::sync::{Arc, Mutex};
 use cxx::UniquePtr;
-use tokio::sync::{mpsc, oneshot};
-use libwebrtc_sys::peer_connection as sys_pc;
+use libwebrtc_sys::data_channel as sys_dc;
 use libwebrtc_sys::jsep as sys_jsep;
+use libwebrtc_sys::peer_connection as sys_pc;
+use log::trace;
+use std::fmt::{Debug, Formatter};
+use std::sync::{Arc, Mutex};
 use thiserror::Error;
+use tokio::sync::{mpsc, oneshot};
 
-use crate::data_channel::DataChannel;
+use crate::data_channel::{DataChannel, DataChannelInit};
+use crate::jsep::{IceCandidate, SessionDescription};
 use crate::media_stream::MediaStream;
-use crate::ice_candidate::IceCandidate;
+use crate::rtc_error::RTCError;
 use crate::rtp_receiver::RtpReceiver;
 use crate::rtp_transceiver::RtpTransceiver;
-use crate::session_description::SessionDescription;
-use crate::rtc_error::RTCError;
 
-pub use libwebrtc_sys::peer_connection::ffi::PeerConnectionState;
-pub use libwebrtc_sys::peer_connection::ffi::SignalingState;
 pub use libwebrtc_sys::peer_connection::ffi::IceConnectionState;
 pub use libwebrtc_sys::peer_connection::ffi::IceGatheringState;
+pub use libwebrtc_sys::peer_connection::ffi::PeerConnectionState;
 pub use libwebrtc_sys::peer_connection::ffi::RTCOfferAnswerOptions;
+pub use libwebrtc_sys::peer_connection::ffi::SignalingState;
 
 #[derive(Error, Debug)]
 pub enum SdpError {
@@ -29,44 +31,38 @@ pub enum SdpError {
 
 pub struct PeerConnection {
     cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>,
-    observer: InternalObserver
+    observer: Box<InternalObserver>,
+
+    // Keep alive for C++
+    native_observer: UniquePtr<sys_pc::ffi::NativePeerConnectionObserver>,
 }
 
 impl PeerConnection {
-    pub fn new(cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>) -> Self {
+    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: InternalObserver {
-                on_signaling_change_handler: Arc::new(Default::default()),
-                on_add_stream_handler: Arc::new(Default::default()),
-                on_remove_stream_handler: Arc::new(Default::default()),
-                on_data_channel_handler: Arc::new(Default::default()),
-                on_renegotiation_needed_handler: Arc::new(Default::default()),
-                on_negotiation_needed_event_handler: Arc::new(Default::default()),
-                on_ice_connection_change_handler: Arc::new(Default::default()),
-                on_standardized_ice_connection_change_handler: Arc::new(Default::default()),
-                on_connection_change_handler: Arc::new(Default::default()),
-                on_ice_gathering_change_handler: Arc::new(Default::default()),
-                on_ice_candidate_handler: Arc::new(Default::default()),
-                on_ice_candidate_error_handler: Arc::new(Default::default()),
-                on_ice_candidates_removed_handler: Arc::new(Default::default()),
-                on_ice_connection_receiving_change_handler: Arc::new(Default::default()),
-                on_ice_selected_candidate_pair_changed_handler: Arc::new(Default::default()),
-                on_add_track_handler: Arc::new(Default::default()),
-                on_track_handler: Arc::new(Default::default()),
-                on_remove_track_handler: Arc::new(Default::default()),
-                on_interesting_usage_handler: Arc::new(Default::default())
-            }
+            observer,
+            native_observer,
         }
     }
 
     pub async fn create_offer(&mut self) -> Result<SessionDescription, SdpError> {
         let (tx, mut rx) = mpsc::channel(1);
 
-        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 wrapper =
+            sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx }));
+        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),
@@ -77,10 +73,16 @@ impl PeerConnection {
     pub async fn create_answer(&mut self) -> Result<SessionDescription, SdpError> {
         let (tx, mut rx) = mpsc::channel(1);
 
-        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 wrapper =
+            sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx }));
+        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),
@@ -88,15 +90,89 @@ impl PeerConnection {
         }
     }
 
-    pub async fn set_local_description(&mut self, desc: SessionDescription) -> Result<(), SdpError> {
+    pub async fn set_local_description(
+        &mut self,
+        desc: SessionDescription,
+    ) -> Result<(), SdpError> {
+        let (tx, mut rx) = mpsc::channel(1);
+        let wrapper =
+            sys_jsep::SetLocalSdpObserverWrapper::new(Box::new(InternalSetLocalSdpObserver { tx }));
+        let mut native_wrapper =
+            sys_jsep::ffi::create_native_set_local_sdp_observer(Box::new(wrapper));
 
+        unsafe {
+            self.cxx_handle
+                .pin_mut()
+                .set_local_description(desc.release(), native_wrapper.pin_mut());
+        }
 
-        Ok(())
+        match rx.recv().await {
+            Some(value) => value.map_err(Into::into),
+            None => Err(SdpError::RecvError("channel closed".to_string())),
+        }
     }
 
-    pub async fn set_remote_description(&mut self, desc: SessionDescription) -> Result<(), SdpError> {
+    pub async fn set_remote_description(
+        &mut self,
+        desc: SessionDescription,
+    ) -> Result<(), SdpError> {
+        let (tx, mut rx) = mpsc::channel(1);
+        let wrapper =
+            sys_jsep::SetRemoteSdpObserverWrapper::new(Box::new(InternalSetRemoteSdpObserver {
+                tx,
+            }));
+        let mut native_wrapper =
+            sys_jsep::ffi::create_native_set_remote_sdp_observer(Box::new(wrapper));
 
-        Ok(())
+        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),
+            None => Err(SdpError::RecvError("channel closed".to_string())),
+        }
+    }
+
+    pub fn create_data_channel(
+        &mut self,
+        label: &str,
+        init: DataChannelInit,
+    ) -> Result<DataChannel, RTCError> {
+        let native_init = sys_dc::ffi::create_data_channel_init(init.into());
+        let res = self
+            .cxx_handle
+            .pin_mut()
+            .create_data_channel(label.to_string(), native_init);
+
+        match res {
+            Ok(cxx_handle) => Ok(DataChannel::new(cxx_handle)),
+            Err(e) => Err(unsafe { RTCError::from(e.what()) }),
+        }
+    }
+
+    pub async fn add_ice_candidate(&mut self, candidate: IceCandidate) -> Result<(), SdpError> {
+        let (tx, mut rx) = mpsc::channel(1);
+        let observer = sys_pc::AddIceCandidateObserverWrapper::new(Box::new(move |error| {
+            tx.blocking_send(error).unwrap();
+        }));
+
+        let mut native_observer =
+            sys_pc::ffi::create_native_add_ice_candidate_observer(Box::new(observer));
+        self.cxx_handle
+            .pin_mut()
+            .add_ice_candidate(candidate.release(), native_observer.pin_mut());
+
+        match rx.recv().await {
+            Some(value) => Ok(()),
+            None => Err(SdpError::RecvError("channel closed".to_string())),
+        }
+    }
+
+    pub fn close(&mut self) {
+        self.cxx_handle.pin_mut().close();
     }
 
     pub fn on_signaling_change(&mut self, handler: OnSignalingChangeHandler) {
@@ -116,15 +192,30 @@ impl PeerConnection {
     }
 
     pub fn on_renegotiation_needed(&mut self, handler: OnRenegotiationNeededHandler) {
-        *self.observer.on_renegotiation_needed_handler.lock().unwrap() = Some(handler);
+        *self
+            .observer
+            .on_renegotiation_needed_handler
+            .lock()
+            .unwrap() = Some(handler);
     }
 
     pub fn on_ice_connection_change(&mut self, handler: OnIceConnectionChangeHandler) {
-        *self.observer.on_ice_connection_change_handler.lock().unwrap() = Some(handler);
+        *self
+            .observer
+            .on_ice_connection_change_handler
+            .lock()
+            .unwrap() = Some(handler);
     }
 
-    pub fn on_standardized_ice_connection_change(&mut self, handler: OnStandardizedIceConnectionChangeHandler) {
-        *self.observer.on_standardized_ice_connection_change_handler.lock().unwrap() = Some(handler);
+    pub fn on_standardized_ice_connection_change(
+        &mut self,
+        handler: OnStandardizedIceConnectionChangeHandler,
+    ) {
+        *self
+            .observer
+            .on_standardized_ice_connection_change_handler
+            .lock()
+            .unwrap() = Some(handler);
     }
 
     pub fn on_connection_change(&mut self, handler: OnConnectionChangeHandler) {
@@ -132,7 +223,11 @@ impl PeerConnection {
     }
 
     pub fn on_ice_gathering_change(&mut self, handler: OnIceGatheringChangeHandler) {
-        *self.observer.on_ice_gathering_change_handler.lock().unwrap() = Some(handler);
+        *self
+            .observer
+            .on_ice_gathering_change_handler
+            .lock()
+            .unwrap() = Some(handler);
     }
 
     pub fn on_ice_candidate(&mut self, handler: OnIceCandidateHandler) {
@@ -144,15 +239,33 @@ impl PeerConnection {
     }
 
     pub fn on_ice_candidates_removed(&mut self, handler: OnIceCandidatesRemovedHandler) {
-        *self.observer.on_ice_candidates_removed_handler.lock().unwrap() = Some(handler);
+        *self
+            .observer
+            .on_ice_candidates_removed_handler
+            .lock()
+            .unwrap() = Some(handler);
     }
 
-    pub fn on_ice_connection_receiving_change(&mut self, handler: OnIceConnectionReceivingChangeHandler) {
-        *self.observer.on_ice_connection_receiving_change_handler.lock().unwrap() = Some(handler);
+    pub fn on_ice_connection_receiving_change(
+        &mut self,
+        handler: OnIceConnectionReceivingChangeHandler,
+    ) {
+        *self
+            .observer
+            .on_ice_connection_receiving_change_handler
+            .lock()
+            .unwrap() = Some(handler);
     }
 
-    pub fn on_ice_selected_candidate_pair_changed(&mut self, handler: OnIceSelectedCandidatePairChangedHandler) {
-        *self.observer.on_ice_selected_candidate_pair_changed_handler.lock().unwrap() = Some(handler);
+    pub fn on_ice_selected_candidate_pair_changed(
+        &mut self,
+        handler: OnIceSelectedCandidatePairChangedHandler,
+    ) {
+        *self
+            .observer
+            .on_ice_selected_candidate_pair_changed_handler
+            .lock()
+            .unwrap() = Some(handler);
     }
 
     pub fn on_add_track(&mut self, handler: OnAddTrackHandler) {
@@ -175,44 +288,50 @@ impl PeerConnection {
 // CreateSdpObserver
 
 struct InternalCreateSdpObserver {
-    tx: mpsc::Sender<Result<SessionDescription, RTCError>>
+    tx: mpsc::Sender<Result<SessionDescription, RTCError>>,
 }
 
 impl sys_jsep::CreateSdpObserver for InternalCreateSdpObserver {
-    fn on_success(&self, session_description: UniquePtr<libwebrtc_sys::jsep::ffi::SessionDescription>) {
-        self.tx.blocking_send(Ok(SessionDescription{})).unwrap(); // TODO
+    fn on_success(
+        &self,
+        session_description: UniquePtr<libwebrtc_sys::jsep::ffi::SessionDescription>,
+    ) {
+        self.tx
+            .blocking_send(Ok(SessionDescription::new(session_description)))
+            .unwrap();
     }
 
     fn on_failure(&self, error: RTCError) {
-        self.tx.blocking_send(Err(error)).unwrap(); // TODO
+        self.tx.blocking_send(Err(error)).unwrap();
     }
 }
 
 // SetLocalSdpObserver
 
 struct InternalSetLocalSdpObserver {
-    tx: mpsc::Sender<Result<(), RTCError>>
+    tx: mpsc::Sender<Result<(), RTCError>>,
 }
 
 impl sys_jsep::SetLocalSdpObserver for InternalSetLocalSdpObserver {
     fn on_set_local_description_complete(&self, error: RTCError) {
-        self.tx.blocking_send(Ok(())).unwrap();
+        let res = if error.ok() { Ok(()) } else { Err(error) };
+        self.tx.blocking_send(res).unwrap();
     }
 }
 
 // SetRemoteSdpObserver
 
 struct InternalSetRemoteSdpObserver {
-
+    tx: mpsc::Sender<Result<(), RTCError>>,
 }
 
 impl sys_jsep::SetRemoteSdpObserver for InternalSetRemoteSdpObserver {
     fn on_set_remote_description_complete(&self, error: RTCError) {
-        todo!()
+        let res = if error.ok() { Ok(()) } else { Err(error) };
+        self.tx.blocking_send(res).unwrap();
     }
 }
 
-
 // PeerConnectionObserver
 
 // TODO(theomonnom) Should we return futures?
@@ -223,20 +342,23 @@ pub type OnDataChannelHandler = Box<dyn FnMut(DataChannel) + Send + Sync>;
 pub type OnRenegotiationNeededHandler = Box<dyn FnMut() + Send + Sync>;
 pub type OnNegotiationNeededEventHandler = Box<dyn FnMut(u32) + Send + Sync>;
 pub type OnIceConnectionChangeHandler = Box<dyn FnMut(IceConnectionState) + Send + Sync>;
-pub type OnStandardizedIceConnectionChangeHandler = Box<dyn FnMut(IceConnectionState) + Send + Sync>;
+pub type OnStandardizedIceConnectionChangeHandler =
+    Box<dyn FnMut(IceConnectionState) + Send + Sync>;
 pub type OnConnectionChangeHandler = Box<dyn FnMut(PeerConnectionState) + Send + Sync>;
 pub type OnIceGatheringChangeHandler = Box<dyn FnMut(IceGatheringState) + Send + Sync>;
 pub type OnIceCandidateHandler = Box<dyn FnMut(IceCandidate) + Send + Sync>;
-pub type OnIceCandidateErrorHandler = Box<dyn FnMut(String, i32, String, i32, String) + Send + Sync>;
+pub type OnIceCandidateErrorHandler =
+    Box<dyn FnMut(String, i32, String, i32, String) + Send + Sync>;
 pub type OnIceCandidatesRemovedHandler = Box<dyn FnMut(Vec<IceCandidate>) + Send + Sync>;
 pub type OnIceConnectionReceivingChangeHandler = Box<dyn FnMut(bool) + Send + Sync>;
-pub type OnIceSelectedCandidatePairChangedHandler = Box<dyn FnMut(libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent) + Send + Sync>;
+pub type OnIceSelectedCandidatePairChangedHandler =
+    Box<dyn FnMut(libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent) + Send + Sync>;
 pub type OnAddTrackHandler = Box<dyn FnMut(RtpReceiver, Vec<MediaStream>) + Send + Sync>;
 pub type OnTrackHandler = Box<dyn FnMut(RtpTransceiver) + Send + Sync>;
 pub type OnRemoveTrackHandler = Box<dyn FnMut(RtpReceiver) + Send + Sync>;
 pub type OnInterestingUsageHandler = Box<dyn FnMut(i32) + Send + Sync>;
 
-struct InternalObserver {
+pub(crate) struct InternalObserver {
     on_signaling_change_handler: Arc<Mutex<Option<OnSignalingChangeHandler>>>,
     on_add_stream_handler: Arc<Mutex<Option<OnAddStreamHandler>>>,
     on_remove_stream_handler: Arc<Mutex<Option<OnRemoveStreamHandler>>>,
@@ -244,152 +366,321 @@ struct InternalObserver {
     on_renegotiation_needed_handler: Arc<Mutex<Option<OnRenegotiationNeededHandler>>>,
     on_negotiation_needed_event_handler: Arc<Mutex<Option<OnNegotiationNeededEventHandler>>>,
     on_ice_connection_change_handler: Arc<Mutex<Option<OnIceConnectionChangeHandler>>>,
-    on_standardized_ice_connection_change_handler: Arc<Mutex<Option<OnStandardizedIceConnectionChangeHandler>>>,
+    on_standardized_ice_connection_change_handler:
+        Arc<Mutex<Option<OnStandardizedIceConnectionChangeHandler>>>,
     on_connection_change_handler: Arc<Mutex<Option<OnConnectionChangeHandler>>>,
     on_ice_gathering_change_handler: Arc<Mutex<Option<OnIceGatheringChangeHandler>>>,
     on_ice_candidate_handler: Arc<Mutex<Option<OnIceCandidateHandler>>>,
     on_ice_candidate_error_handler: Arc<Mutex<Option<OnIceCandidateErrorHandler>>>,
     on_ice_candidates_removed_handler: Arc<Mutex<Option<OnIceCandidatesRemovedHandler>>>,
-    on_ice_connection_receiving_change_handler: Arc<Mutex<Option<OnIceConnectionReceivingChangeHandler>>>,
-    on_ice_selected_candidate_pair_changed_handler: Arc<Mutex<Option<OnIceSelectedCandidatePairChangedHandler>>>,
+    on_ice_connection_receiving_change_handler:
+        Arc<Mutex<Option<OnIceConnectionReceivingChangeHandler>>>,
+    on_ice_selected_candidate_pair_changed_handler:
+        Arc<Mutex<Option<OnIceSelectedCandidatePairChangedHandler>>>,
     on_add_track_handler: Arc<Mutex<Option<OnAddTrackHandler>>>,
     on_track_handler: Arc<Mutex<Option<OnTrackHandler>>>,
     on_remove_track_handler: Arc<Mutex<Option<OnRemoveTrackHandler>>>,
-    on_interesting_usage_handler: Arc<Mutex<Option<OnInterestingUsageHandler>>>
+    on_interesting_usage_handler: Arc<Mutex<Option<OnInterestingUsageHandler>>>,
+}
+
+impl Default for InternalObserver {
+    fn default() -> Self {
+        Self {
+            on_signaling_change_handler: Arc::new(Default::default()),
+            on_add_stream_handler: Arc::new(Default::default()),
+            on_remove_stream_handler: Arc::new(Default::default()),
+            on_data_channel_handler: Arc::new(Default::default()),
+            on_renegotiation_needed_handler: Arc::new(Default::default()),
+            on_negotiation_needed_event_handler: Arc::new(Default::default()),
+            on_ice_connection_change_handler: Arc::new(Default::default()),
+            on_standardized_ice_connection_change_handler: Arc::new(Default::default()),
+            on_connection_change_handler: Arc::new(Default::default()),
+            on_ice_gathering_change_handler: Arc::new(Default::default()),
+            on_ice_candidate_handler: Arc::new(Default::default()),
+            on_ice_candidate_error_handler: Arc::new(Default::default()),
+            on_ice_candidates_removed_handler: Arc::new(Default::default()),
+            on_ice_connection_receiving_change_handler: Arc::new(Default::default()),
+            on_ice_selected_candidate_pair_changed_handler: Arc::new(Default::default()),
+            on_add_track_handler: Arc::new(Default::default()),
+            on_track_handler: Arc::new(Default::default()),
+            on_remove_track_handler: Arc::new(Default::default()),
+            on_interesting_usage_handler: Arc::new(Default::default()),
+        }
+    }
 }
 
 // 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() {
             f(new_state);
         }
     }
 
-    fn on_add_stream(&mut self, stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>) {
+    fn on_add_stream(
+        &self,
+        stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
+    ) {
+        trace!("on_add_stream");
         let mut handler = self.on_add_stream_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
             // TODO(theomonnom)
         }
     }
 
-    fn on_remove_stream(&mut self, stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>) {
+    fn on_remove_stream(
+        &self,
+        stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
+    ) {
+        trace!("on_remove_stream");
         let mut handler = self.on_remove_stream_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
             // TODO(theomonnom)
         }
     }
 
-    fn on_data_channel(&mut self, data_channel: UniquePtr<libwebrtc_sys::data_channel::ffi::DataChannel>) {
+    fn on_data_channel(
+        &self,
+        data_channel: UniquePtr<libwebrtc_sys::data_channel::ffi::DataChannel>,
+    ) {
+        trace!("on_data_channel");
         let mut handler = self.on_data_channel_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
-            // TODO(theomonnom)
+            f(DataChannel::new(data_channel));
         }
     }
 
-    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() {
             f();
         }
     }
 
-    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() {
             f(event);
         }
     }
 
-    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() {
             f(new_state);
         }
     }
 
-    fn on_standardized_ice_connection_change(&mut self, new_state: IceConnectionState) {
-        let mut handler = self.on_standardized_ice_connection_change_handler.lock().unwrap();
+    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
+            .lock()
+            .unwrap();
         if let Some(f) = handler.as_mut() {
             f(new_state);
         }
     }
 
-    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() {
             f(new_state);
         }
     }
 
-    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() {
             f(new_state);
         }
     }
 
-    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() {
-            // TODO(theomonnom)
+            f(IceCandidate::new(candidate));
         }
     }
 
-    fn on_ice_candidate_error(&mut self, address: String, port: i32, url: String, error_code: i32, error_text: String) {
+    fn on_ice_candidate_error(
+        &self,
+        address: String,
+        port: i32,
+        url: String,
+        error_code: i32,
+        error_text: String,
+    ) {
+        trace!("on_ice_candidate_error");
         let mut handler = self.on_ice_candidate_error_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
             f(address, port, url, error_code, error_text);
         }
     }
 
-    fn on_ice_candidates_removed(&mut self, removed: Vec<UniquePtr<libwebrtc_sys::candidate::ffi::Candidate>>) {
+    fn on_ice_candidates_removed(
+        &self,
+        removed: Vec<UniquePtr<libwebrtc_sys::candidate::ffi::Candidate>>,
+    ) {
+        trace!("on_ice_candidates_removed");
         let mut handler = self.on_ice_candidates_removed_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
             // TODO(theomonnom)
         }
     }
 
-    fn on_ice_connection_receiving_change(&mut self, receiving: bool) {
-        let mut handler = self.on_ice_connection_receiving_change_handler.lock().unwrap();
+    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
+            .lock()
+            .unwrap();
         if let Some(f) = handler.as_mut() {
             f(receiving);
         }
     }
 
-    fn on_ice_selected_candidate_pair_changed(&mut self, event: libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent) {
-        let mut handler = self.on_ice_selected_candidate_pair_changed_handler.lock().unwrap();
+    fn on_ice_selected_candidate_pair_changed(
+        &self,
+        event: libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent,
+    ) {
+        trace!("on_ice_selected_candidate_pair_changed");
+        let mut handler = self
+            .on_ice_selected_candidate_pair_changed_handler
+            .lock()
+            .unwrap();
         if let Some(f) = handler.as_mut() {
             f(event);
         }
     }
 
-    fn on_add_track(&mut self, receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>, streams: Vec<UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>>) {
+    fn on_add_track(
+        &self,
+        receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>,
+        streams: Vec<UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>>,
+    ) {
+        trace!("on_add_track");
         let mut handler = self.on_add_track_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
             // TODO(theomonnom)
         }
     }
 
-    fn on_track(&mut self, transceiver: UniquePtr<libwebrtc_sys::rtp_transceiver::ffi::RtpTransceiver>) {
+    fn on_track(
+        &self,
+        transceiver: UniquePtr<libwebrtc_sys::rtp_transceiver::ffi::RtpTransceiver>,
+    ) {
+        trace!("on_track");
         let mut handler = self.on_track_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
             // TODO(theomonnom)
         }
     }
 
-    fn on_remove_track(&mut 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() {
             // TODO(theomonnom)
         }
     }
 
-    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() {
             f(usage_pattern);
         }
     }
-}
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::data_channel::{DataChannel, DataChannelInit};
+    use crate::jsep::IceCandidate;
+    use crate::peer_connection_factory::{ICEServer, PeerConnectionFactory, RTCConfiguration};
+    use crate::webrtc::RTCRuntime;
+    use log::trace;
+    use tokio::sync::mpsc;
+
+    fn init_log() {
+        let _ = env_logger::builder().is_test(true).try_init();
+    }
+
+    #[tokio::test]
+    async fn create_pc() {
+        init_log();
+
+        let test = RTCRuntime::new();
+
+        let factory = PeerConnectionFactory::new();
+        let config = RTCConfiguration {
+            ice_servers: vec![ICEServer {
+                urls: vec!["stun:stun1.l.google.com:19302".to_string()],
+                username: "".into(),
+                password: "".into(),
+            }],
+        };
+
+        let mut bob = factory.create_peer_connection(config.clone()).unwrap();
+        let mut alice = factory.create_peer_connection(config.clone()).unwrap();
+
+        let (bob_ice_tx, mut bob_ice_rx) = mpsc::channel::<IceCandidate>(16);
+        let (alice_ice_tx, mut alice_ice_rx) = mpsc::channel::<IceCandidate>(16);
+        let (alice_dc_tx, mut alice_dc_rx) = mpsc::channel::<DataChannel>(16);
+
+        bob.on_ice_candidate(Box::new(move |candidate| {
+            bob_ice_tx.blocking_send(candidate).unwrap();
+        }));
+
+        alice.on_ice_candidate(Box::new(move |candidate| {
+            alice_ice_tx.blocking_send(candidate).unwrap();
+        }));
+
+        alice.on_data_channel(Box::new(move |dc| {
+            alice_dc_tx.blocking_send(dc).unwrap();
+        }));
+
+        let mut bob_dc = bob
+            .create_data_channel("test_dc", DataChannelInit::default())
+            .unwrap();
+
+        let offer = bob.create_offer().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().await.unwrap();
+        trace!("Alice answer: {:?}", answer);
+        alice.set_local_description(answer.clone()).await.unwrap();
+        bob.set_remote_description(answer).await.unwrap();
+
+        let bob_ice = bob_ice_rx.recv().await.unwrap();
+        let alice_ice = alice_ice_rx.recv().await.unwrap();
+
+        bob.add_ice_candidate(alice_ice).await.unwrap();
+        alice.add_ice_candidate(bob_ice).await.unwrap();
+
+        let (data_tx, mut data_rx) = mpsc::channel::<String>(1);
+        let mut alice_dc = alice_dc_rx.recv().await.unwrap();
+        alice_dc.on_message(Box::new(move |data, is_binary| {
+            data_tx
+                .blocking_send(String::from_utf8_lossy(data).to_string())
+                .unwrap();
+        }));
+
+        assert!(bob_dc.send(b"This is a test", true));
+        assert_eq!(data_rx.recv().await.unwrap(), "This is a test");
+
+        alice.close();
+        bob.close();
+    }
+}

crates/livekit-webrtc/src/peer_connection_factory.rs

@@ -1,53 +1,43 @@
 use cxx::UniquePtr;
-use libwebrtc_sys::peer_connection_factory as sys_factory;
 use libwebrtc_sys::peer_connection as sys_pc;
+use libwebrtc_sys::peer_connection_factory as sys_factory;
 
-use crate::peer_connection::PeerConnection;
+use crate::peer_connection::{InternalObserver, PeerConnection};
 use crate::rtc_error::RTCError;
 
-pub use sys_factory::ffi::{RTCConfiguration, ICEServer};
+pub use sys_factory::ffi::{ICEServer, RTCConfiguration};
 
 pub struct PeerConnectionFactory {
-    cxx_handle: UniquePtr<sys_factory::ffi::PeerConnectionFactory>
+    cxx_handle: UniquePtr<sys_factory::ffi::PeerConnectionFactory>,
 }
 
 impl PeerConnectionFactory {
     pub fn new() -> Self {
         Self {
-            cxx_handle: sys_factory::ffi::create_peer_connection_factory()
+            cxx_handle: sys_factory::ffi::create_peer_connection_factory(),
         }
     }
 
-    pub fn create_peer_connection(&self, config: RTCConfiguration, observer: Box<dyn sys_pc::PeerConnectionObserver>) -> Result<PeerConnection, RTCError> {
+    pub fn create_peer_connection(
+        &self,
+        config: RTCConfiguration,
+    ) -> Result<PeerConnection, RTCError> {
         let native_config = sys_factory::ffi::create_rtc_configuration(config);
-        let native_observer = sys_pc::ffi::create_native_peer_connection_observer(Box::new(sys_pc::PeerConnectionObserverWrapper::new(observer)));
 
-        let pc_result : Result<UniquePtr<sys_pc::ffi::PeerConnection>, cxx::Exception> = unsafe {
-            self.cxx_handle.create_peer_connection(native_config, native_observer)
-        };
+        unsafe {
+            let mut observer = Box::new(InternalObserver::default());
+            let mut native_observer = sys_pc::ffi::create_native_peer_connection_observer(
+                Box::new(sys_pc::PeerConnectionObserverWrapper::new(&mut *observer)),
+            );
 
-        match pc_result {
-            Ok(cxx_handle) => {
-                Ok(PeerConnection::new(cxx_handle))
-            }
-            Err(e) => {
-                Err(unsafe {RTCError::from(e.what()) }) // TODO
+            let res = self
+                .cxx_handle
+                .create_peer_connection(native_config, native_observer.pin_mut());
+
+            match res {
+                Ok(cxx_handle) => Ok(PeerConnection::new(cxx_handle, observer, native_observer)),
+                Err(e) => Err(RTCError::from(e.what())),
             }
         }
     }
 }
-
-
-#[cfg(test)]
-mod tests {
-    use crate::peer_connection_factory::PeerConnectionFactory;
-    use crate::peer_connection_factory::{RTCConfiguration, ICEServer};
-
-    #[test]
-    fn create_pc(){
-        let factory = PeerConnectionFactory::new();
-        let pc = factory.create_peer_connection(RTCConfiguration{
-            ice_servers: vec!()
-        }, Box::new(()));
-    }
-}

crates/livekit-webrtc/src/rtc_error.rs

@@ -1,4 +1,4 @@
 use std::fmt;
 
+// TODO(theomonnom) Wrap the RTCError ffi so we can use Option(u16)
 pub use libwebrtc_sys::rtc_error::ffi::RTCError;
-

crates/livekit-webrtc/src/rtp_receiver.rs

@@ -1,5 +1,2 @@
-
-
-pub struct RtpReceiver {
-
-}
+#[derive(Debug)]
+pub struct RtpReceiver {}

crates/livekit-webrtc/src/rtp_transceiver.rs

@@ -1,5 +1,2 @@
-
-
-pub struct RtpTransceiver {
-
-}
+#[derive(Debug)]
+pub struct RtpTransceiver {}

crates/livekit-webrtc/src/session_description.rs

@@ -1,5 +0,0 @@
-
-#[derive(Debug)]
-pub struct SessionDescription {
-
-}

crates/livekit-webrtc/src/webrtc.rs

@@ -0,0 +1,14 @@
+use cxx::UniquePtr;
+use libwebrtc_sys::webrtc as sys_rtc;
+
+pub struct RTCRuntime {
+    cxx_handle: UniquePtr<sys_rtc::ffi::RTCRuntime>,
+}
+
+impl RTCRuntime {
+    pub fn new() -> Self {
+        Self {
+            cxx_handle: sys_rtc::ffi::create_rtc_runtime(),
+        }
+    }
+}