diff --git a/crates/livekit-core/src/rtc_engine.rs b/crates/livekit-core/src/rtc_engine.rs index 0354acb..fa61959 100644 --- a/crates/livekit-core/src/rtc_engine.rs +++ b/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) { diff --git a/crates/livekit-webrtc/libwebrtc-sys/build.rs b/crates/livekit-webrtc/libwebrtc-sys/build.rs index f02d219..909f297 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/build.rs +++ b/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()); } } diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h index 0c50656..52c9487 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h @@ -7,20 +7,44 @@ #include #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 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 data_channel_; }; + std::unique_ptr create_data_channel_init(DataChannelInit init); + static std::unique_ptr _unique_data_channel(){ return nullptr; // Ignore } + + class NativeDataChannelObserver : public webrtc::DataChannelObserver { + public: + explicit NativeDataChannelObserver(rust::Box observer); + + void OnStateChange() override; + void OnMessage(const webrtc::DataBuffer& buffer) override; + void OnBufferedAmountChange(uint64_t sent_data_size) override; + private: + rust::Box observer_; + }; + + std::unique_ptr create_native_data_channel_observer(rust::Box observer); } // livekit #endif //CLIENT_SDK_NATIVE_DATA_CHANNEL_H diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h index a7ffab3..8ca39ce 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h @@ -17,6 +17,7 @@ namespace livekit { public: explicit IceCandidate(std::unique_ptr ice_candidate); + std::unique_ptr release(); private: std::unique_ptr ice_candidate_; }; @@ -29,6 +30,7 @@ namespace livekit { public: explicit SessionDescription(std::unique_ptr session_description); + rust::String stringify() const; std::unique_ptr clone() const; std::unique_ptr release(); diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h index ed79f0d..ea47b61 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h +++ b/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 peer_connection, std::unique_ptr observer); + explicit PeerConnection(rtc::scoped_refptr peer_connection); + void create_offer(NativeCreateSdpObserverHandle &observer, RTCOfferAnswerOptions options); + void create_answer(NativeCreateSdpObserverHandle &observer, RTCOfferAnswerOptions options); + void set_local_description(std::unique_ptr desc, NativeSetLocalSdpObserverHandle &observer); + void set_remote_description(std::unique_ptr desc, NativeSetRemoteSdpObserverHandle &observer); + std::unique_ptr create_data_channel(rust::String label, std::unique_ptr init); + void add_ice_candidate(std::unique_ptr candidate, NativeAddIceCandidateObserver &observer); void close(); - void create_offer(std::unique_ptr observer, RTCOfferAnswerOptions options); - void create_answer(std::unique_ptr observer, RTCOfferAnswerOptions options); - void set_local_description(std::unique_ptr desc, std::unique_ptr observer); - void set_remote_description(std::unique_ptr desc, std::unique_ptr observer); private: rtc::scoped_refptr peer_connection_; - std::unique_ptr observer_; }; static std::unique_ptr _unique_peer_connection() { return nullptr; // Ignore } + class NativeAddIceCandidateObserver { + public: + explicit NativeAddIceCandidateObserver(rust::Box observer); + + void OnComplete(const RTCError &error); + private: + rust::Box observer_; + }; + + std::unique_ptr create_native_add_ice_candidate_observer(rust::Box observer); + class NativePeerConnectionObserver : public webrtc::PeerConnectionObserver { public: explicit NativePeerConnectionObserver(rust::Box observer); - ~NativePeerConnectionObserver() override = default; - void OnSignalingChange(webrtc::PeerConnectionInterface::SignalingState new_state) override; void OnAddStream(rtc::scoped_refptr stream) override; void OnRemoveStream(rtc::scoped_refptr stream) override; diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h index 3ffb146..a199e60 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h @@ -17,7 +17,7 @@ namespace livekit { public: PeerConnectionFactory(); - std::unique_ptr create_peer_connection(std::unique_ptr config, std::unique_ptr observer) const; + std::unique_ptr create_peer_connection(std::unique_ptr config, NativePeerConnectionObserver &observer) const; private: std::unique_ptr network_thread_; diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h index eee4ef9..9e42d9d 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h +++ b/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 diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/webrtc.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/webrtc.h new file mode 100644 index 0000000..2dc3120 --- /dev/null +++ b/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 create_rtc_runtime(); + +} // livekit + +#endif //LIVEKIT_WEBRTC_WEBRTC_H diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/candidate.rs b/crates/livekit-webrtc/libwebrtc-sys/src/candidate.rs index daa49c5..8e442b8 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/candidate.rs +++ b/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; // Ignore } } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp index ba6ffb8..ad031ec 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp @@ -2,11 +2,78 @@ // Created by Théo Monnom on 01/09/2022. // +#include + #include "livekit/data_channel.h" +#include "libwebrtc-sys/src/data_channel.rs.h" namespace livekit { - DataChannel::DataChannel(rtc::scoped_refptr data_channel) : data_channel_(data_channel) { + DataChannel::DataChannel(rtc::scoped_refptr 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 create_data_channel_init(DataChannelInit init) { + auto rtc_init = std::make_unique(); + 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(init.priority); + + return rtc_init; + } + + NativeDataChannelObserver::NativeDataChannelObserver(rust::Box 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 create_native_data_channel_observer(rust::Box observer){ + return std::make_unique(std::move(observer)); + } } // livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs index a349944..fc1854b 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs +++ b/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; + fn create_native_data_channel_observer( + observer: Box, + ) -> UniquePtr; fn _unique_data_channel() -> UniquePtr; // 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) }; + } +} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp index 06e80b3..b607eb4 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp @@ -15,10 +15,20 @@ namespace livekit { } + std::unique_ptr IceCandidate::release() { + return std::move(ice_candidate_); + } + SessionDescription::SessionDescription(std::unique_ptr 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::clone() const { return std::make_unique(session_description_->Clone()); } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs index 0b80e47..1388e8c 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs +++ b/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); + fn on_success( + self: &CreateSdpObserverWrapper, + session_description: UniquePtr, + ); 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) -> UniquePtr; - fn create_native_set_local_sdp_observer(observer: Box) -> UniquePtr; - fn create_native_set_remote_sdp_observer(observer: Box) -> UniquePtr; + fn stringify(self: &SessionDescription) -> String; + fn clone(self: &SessionDescription) -> UniquePtr; + + fn create_native_create_sdp_observer( + observer: Box, + ) -> UniquePtr; + fn create_native_set_local_sdp_observer( + observer: Box, + ) -> UniquePtr; + fn create_native_set_remote_sdp_observer( + observer: Box, + ) -> UniquePtr; fn _unique_ice_candidate() -> UniquePtr; // Ignore fn _unique_session_description() -> UniquePtr; // 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); fn on_failure(&self, error: RTCError); } @@ -51,9 +79,7 @@ pub struct CreateSdpObserverWrapper { impl CreateSdpObserverWrapper { pub fn new(observer: Box) -> Self { - Self { - observer - } + Self { observer } } fn on_success(&self, session_description: UniquePtr) { @@ -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) -> 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) -> Self { - Self { - observer - } + Self { observer } } fn on_set_remote_description_complete(&self, error: RTCError) { diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs b/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs index 19ff3ad..9959239 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs +++ b/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; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp index 94487ea..c8c9a5a 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp @@ -6,7 +6,7 @@ namespace livekit { - MediaStreamInterface::MediaStreamInterface(rtc::scoped_refptr stream) : media_stream_(stream) { + MediaStreamInterface::MediaStreamInterface(rtc::scoped_refptr stream) : media_stream_(std::move(stream)) { } } // livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs index 7242ba7..5ae33c8 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs @@ -9,5 +9,4 @@ pub mod ffi { fn _unique_media_stream() -> UniquePtr; // Ignore } - } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp index b762444..732f52b 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp +++ b/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 peer_connection, - std::unique_ptr observer) : peer_connection_( - std::move(peer_connection)), observer_(std::move(observer)) { + PeerConnection::PeerConnection(rtc::scoped_refptr peer_connection) : peer_connection_(std::move(peer_connection)) { } + void PeerConnection::create_offer(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 desc, NativeSetLocalSdpObserverHandle &observer) { + peer_connection_->SetLocalDescription(desc->clone()->release(), observer.observer); + } + + void PeerConnection::set_remote_description(std::unique_ptr desc, NativeSetRemoteSdpObserverHandle &observer) { + peer_connection_->SetRemoteDescription(desc->clone()->release(), observer.observer); + } + + std::unique_ptr PeerConnection::create_data_channel(rust::String label, std::unique_ptr init) { + 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(result.value()); + } + + void PeerConnection::add_ice_candidate(std::unique_ptr 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 observer_handle, RTCOfferAnswerOptions options) { - peer_connection_->CreateOffer(observer_handle->observer.get(), toNativeOfferAnswerOptions(options)); + // AddIceCandidateObserver + + NativeAddIceCandidateObserver::NativeAddIceCandidateObserver(rust::Box observer) : observer_(std::move(observer)) { + } - void PeerConnection::create_answer(std::unique_ptr 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 desc, std::unique_ptr observer) { - peer_connection_->SetLocalDescription(desc->clone()->release(), observer->observer); + std::unique_ptr create_native_add_ice_candidate_observer(rust::Box observer) { + return std::make_unique(std::move(observer)); } - void PeerConnection::set_remote_description(std::unique_ptr desc, std::unique_ptr observer) { - peer_connection_->SetRemoteDescription(desc->clone()->release(), observer->observer); - } - - /* Observer */ + // PeerConnectionObserver NativePeerConnectionObserver::NativePeerConnectionObserver(rust::Box observer) : observer_(std::move(observer)) { diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index fbfac94..72f6424 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/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 + pub ptr: UniquePtr, } - struct CandidatePtr { - pub ptr: UniquePtr + struct CandidatePtr { + pub ptr: UniquePtr, } 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, options: RTCOfferAnswerOptions); - fn create_answer(self: Pin<&mut PeerConnection>, observer: UniquePtr, options: RTCOfferAnswerOptions); - fn set_local_description(self: Pin<&mut PeerConnection>, desc: UniquePtr, observer: UniquePtr); - fn set_remote_description(self: Pin<&mut PeerConnection>, desc: UniquePtr, observer: UniquePtr); + /// 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, + 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, + observer: Pin<&mut NativeSetRemoteSdpObserverHandle>, + ); + + fn create_data_channel( + self: Pin<&mut PeerConnection>, + label: String, + init: UniquePtr, + ) -> Result>; + + fn add_ice_candidate( + self: Pin<&mut PeerConnection>, + candidate: UniquePtr, + observer: Pin<&mut NativeAddIceCandidateObserver>, + ); + fn close(self: Pin<&mut PeerConnection>); - fn create_native_peer_connection_observer(observer: Box) -> UniquePtr; + fn create_native_peer_connection_observer( + observer: Box, + ) -> UniquePtr; + + fn create_native_add_ice_candidate_observer( + observer: Box, + ) -> UniquePtr; fn _unique_peer_connection() -> UniquePtr; // 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); - fn on_remove_stream(self: &mut PeerConnectionObserverWrapper, stream: UniquePtr); - fn on_data_channel(self: &mut PeerConnectionObserverWrapper, data_channel: UniquePtr); - 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); - 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); - 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, streams: Vec); - fn on_track(self: &mut PeerConnectionObserverWrapper, transceiver: UniquePtr); - fn on_remove_track(self: &mut PeerConnectionObserverWrapper, receiver: UniquePtr); - 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, + ); + fn on_remove_stream( + self: &PeerConnectionObserverWrapper, + stream: UniquePtr, + ); + fn on_data_channel( + self: &PeerConnectionObserverWrapper, + data_channel: UniquePtr, + ); + 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, + ); + 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, + ); + 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, + streams: Vec, + ); + fn on_track(self: &PeerConnectionObserverWrapper, transceiver: UniquePtr); + fn on_remove_track(self: &PeerConnectionObserverWrapper, receiver: UniquePtr); + 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, +} + +impl AddIceCandidateObserverWrapper { + pub fn new(observer: Box) -> 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); - fn on_remove_stream(&mut self, stream: UniquePtr); - fn on_data_channel(&mut self, data_channel: UniquePtr); - 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); - 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>); - 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, streams: Vec>); - fn on_track(&mut self, transceiver: UniquePtr); - fn on_remove_track(&mut self, receiver: UniquePtr); - 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); + fn on_remove_stream(&self, stream: UniquePtr); + fn on_data_channel(&self, data_channel: UniquePtr); + 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); + 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>); + 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, + streams: Vec>, + ); + fn on_track(&self, transceiver: UniquePtr); + fn on_remove_track(&self, receiver: UniquePtr); + fn on_interesting_usage(&self, usage_pattern: i32); } +// Thread safety is handled inside PeerConnectionObserver pub struct PeerConnectionObserverWrapper { - observer: Box, + observer: *mut dyn PeerConnectionObserver, } impl PeerConnectionObserverWrapper { - pub fn new(observer: Box) -> 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) { + unsafe { + (*self.observer).on_add_stream(stream); + } } - fn on_add_stream(&mut self, stream: UniquePtr) { - self.observer.on_add_stream(stream); + fn on_remove_stream(&self, stream: UniquePtr) { + unsafe { + (*self.observer).on_remove_stream(stream); + } } - fn on_remove_stream(&mut self, stream: UniquePtr) { - self.observer.on_remove_stream(stream); + fn on_data_channel(&self, data_channel: UniquePtr) { + unsafe { + (*self.observer).on_data_channel(data_channel); + } } - fn on_data_channel(&mut self, data_channel: UniquePtr) { - 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) { + unsafe { + (*self.observer).on_ice_candidate(candidate); + } } - fn on_ice_candidate(&mut self, candidate: UniquePtr) { - 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) { + fn on_ice_candidates_removed(&self, removed: Vec) { 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, streams: Vec) { + fn on_add_track(&self, receiver: UniquePtr, streams: Vec) { 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) { - self.observer.on_track(transceiver); + fn on_track(&self, transceiver: UniquePtr) { + unsafe { + (*self.observer).on_track(transceiver); + } } - fn on_remove_track(&mut self, receiver: UniquePtr) { - self.observer.on_remove_track(receiver); + fn on_remove_track(&self, receiver: UniquePtr) { + 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); + } } } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp index 94eba75..0b3f9d6 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp @@ -53,15 +53,15 @@ namespace livekit{ } } - std::unique_ptr PeerConnectionFactory::create_peer_connection(std::unique_ptr config, std::unique_ptr observer) const { - webrtc::PeerConnectionDependencies deps{observer.get()}; + std::unique_ptr PeerConnectionFactory::create_peer_connection(std::unique_ptr config, NativePeerConnectionObserver &observer) const { + webrtc::PeerConnectionDependencies deps{&observer}; auto result = peer_factory_->CreatePeerConnectionOrError(*config, std::move(deps)); - if(!result.ok()){ + if(!result.ok()) { throw std::runtime_error(serialize_error(to_error(result.error()))); } - return std::make_unique(std::move(result.value()), std::move(observer)); + return std::make_unique(result.value()); } std::unique_ptr create_peer_connection_factory() { @@ -70,14 +70,13 @@ namespace livekit{ std::unique_ptr create_rtc_configuration(RTCConfiguration conf){ auto rtc = std::make_unique(); - 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); diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs index 0c4e2a1..5252078 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs +++ b/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, - username: String, - password: String, + pub urls: Vec, + pub username: String, + pub password: String, } - #[derive(Debug)] + #[derive(Debug, Clone)] pub struct RTCConfiguration { pub ice_servers: Vec, } @@ -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; fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr; - unsafe fn create_peer_connection(self: &PeerConnectionFactory, config: UniquePtr, observer: UniquePtr) -> Result>; - } - } - - - - - - - -/* - - - - - - -#[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) { - todo!() - } - - fn on_remove_stream(&self, stream: UniquePtr) { - todo!() - } - - fn on_data_channel(&self, data_channel: UniquePtr) { - 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) { - 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>) { - 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, streams: Vec>) { - todo!() - } - - fn on_track(&self, transceiver: UniquePtr) { - todo!() - } - - fn on_remove_track(&self, receiver: UniquePtr) { - todo!() - } - - fn on_interesting_usage(&self, usage_pattern: i32) { - todo!() - } - } - - struct SessionObserver { - - } - - impl CreateSdpObserver for SessionObserver { - fn on_success(&self, session_description: UniquePtr) { - info!("on_success"); - } - - fn on_failure(&self, error: UniquePtr) { - 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, + observer: Pin<&mut NativePeerConnectionObserver>, + ) -> Result>; } } - -*/ - diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.cpp index f9eb1bc..5ea6217 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.cpp @@ -7,7 +7,6 @@ #include #include - namespace livekit { RTCError to_error(const webrtc::RTCError &error) { @@ -15,7 +14,7 @@ namespace livekit { lk_error.error_detail = static_cast(error.error_detail()); lk_error.error_type = static_cast(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; } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs b/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs index 68623bb..8ac3bc0 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs +++ b/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!"); } -} \ No newline at end of file +} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp new file mode 100644 index 0000000..b435694 --- /dev/null +++ b/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 create_rtc_runtime(){ + return std::make_unique(); + } +} // livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs new file mode 100644 index 0000000..d890fc9 --- /dev/null +++ b/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; + } +} diff --git a/crates/livekit-webrtc/src/data_channel.rs b/crates/livekit-webrtc/src/data_channel.rs index bd017b3..7819c17 100644 --- a/crates/livekit-webrtc/src/data_channel.rs +++ b/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, + observer: Box, + // Keep alive for C++ + native_observer: UniquePtr, +} -} \ No newline at end of file +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) -> 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; +pub type OnMessageHandler = Box; // data, is_binary +pub type OnBufferedAmountChangeHandler = Box; + +struct InternalDataChannelObserver { + on_state_change_handler: Arc>>, + on_message_handler: Arc>>, + on_buffered_amount_change_handler: Arc>>, +} + +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, + max_retransmits: Option, + protocol: String, + negotiated: bool, + id: i32, + priority: Option, +} + +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 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), + } + } +} diff --git a/crates/livekit-webrtc/src/ice_candidate.rs b/crates/livekit-webrtc/src/ice_candidate.rs deleted file mode 100644 index f9c307c..0000000 --- a/crates/livekit-webrtc/src/ice_candidate.rs +++ /dev/null @@ -1,4 +0,0 @@ - -pub struct IceCandidate { - -} \ No newline at end of file diff --git a/crates/livekit-webrtc/src/jsep.rs b/crates/livekit-webrtc/src/jsep.rs new file mode 100644 index 0000000..3d66eb6 --- /dev/null +++ b/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, +} + +impl IceCandidate { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } + + pub(crate) fn release(self) -> UniquePtr { + self.cxx_handle + } +} + +#[derive(Debug)] +pub struct SessionDescription { + cxx_handle: UniquePtr, +} + +impl SessionDescription { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } + + pub(crate) fn release(self) -> UniquePtr { + self.cxx_handle + } +} + +impl Clone for SessionDescription { + fn clone(&self) -> Self { + SessionDescription::new(self.cxx_handle.clone()) + } +} diff --git a/crates/livekit-webrtc/src/lib.rs b/crates/livekit-webrtc/src/lib.rs index 8ff3aaf..4bcbcfc 100644 --- a/crates/livekit-webrtc/src/lib.rs +++ b/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; \ No newline at end of file +pub mod webrtc; diff --git a/crates/livekit-webrtc/src/media_stream.rs b/crates/livekit-webrtc/src/media_stream.rs index f4adade..9a693e7 100644 --- a/crates/livekit-webrtc/src/media_stream.rs +++ b/crates/livekit-webrtc/src/media_stream.rs @@ -1,4 +1,2 @@ - -pub struct MediaStream { - -} \ No newline at end of file +#[derive(Debug)] +pub struct MediaStream {} diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index 580ac93..f03f123 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/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, - observer: InternalObserver + observer: Box, + + // Keep alive for C++ + native_observer: UniquePtr, } impl PeerConnection { - pub fn new(cxx_handle: UniquePtr) -> Self { + pub(crate) fn new( + cxx_handle: UniquePtr, + observer: Box, + native_observer: UniquePtr, + ) -> 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 { 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 { 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 { + 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> + tx: mpsc::Sender>, } impl sys_jsep::CreateSdpObserver for InternalCreateSdpObserver { - fn on_success(&self, session_description: UniquePtr) { - self.tx.blocking_send(Ok(SessionDescription{})).unwrap(); // TODO + fn on_success( + &self, + session_description: UniquePtr, + ) { + 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> + tx: mpsc::Sender>, } 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>, } 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; pub type OnRenegotiationNeededHandler = Box; pub type OnNegotiationNeededEventHandler = Box; pub type OnIceConnectionChangeHandler = Box; -pub type OnStandardizedIceConnectionChangeHandler = Box; +pub type OnStandardizedIceConnectionChangeHandler = + Box; pub type OnConnectionChangeHandler = Box; pub type OnIceGatheringChangeHandler = Box; pub type OnIceCandidateHandler = Box; -pub type OnIceCandidateErrorHandler = Box; +pub type OnIceCandidateErrorHandler = + Box; pub type OnIceCandidatesRemovedHandler = Box) + Send + Sync>; pub type OnIceConnectionReceivingChangeHandler = Box; -pub type OnIceSelectedCandidatePairChangedHandler = Box; +pub type OnIceSelectedCandidatePairChangedHandler = + Box; pub type OnAddTrackHandler = Box) + Send + Sync>; pub type OnTrackHandler = Box; pub type OnRemoveTrackHandler = Box; pub type OnInterestingUsageHandler = Box; -struct InternalObserver { +pub(crate) struct InternalObserver { on_signaling_change_handler: Arc>>, on_add_stream_handler: Arc>>, on_remove_stream_handler: Arc>>, @@ -244,152 +366,321 @@ struct InternalObserver { on_renegotiation_needed_handler: Arc>>, on_negotiation_needed_event_handler: Arc>>, on_ice_connection_change_handler: Arc>>, - on_standardized_ice_connection_change_handler: Arc>>, + on_standardized_ice_connection_change_handler: + Arc>>, on_connection_change_handler: Arc>>, on_ice_gathering_change_handler: Arc>>, on_ice_candidate_handler: Arc>>, on_ice_candidate_error_handler: Arc>>, on_ice_candidates_removed_handler: Arc>>, - on_ice_connection_receiving_change_handler: Arc>>, - on_ice_selected_candidate_pair_changed_handler: Arc>>, + on_ice_connection_receiving_change_handler: + Arc>>, + on_ice_selected_candidate_pair_changed_handler: + Arc>>, on_add_track_handler: Arc>>, on_track_handler: Arc>>, on_remove_track_handler: Arc>>, - on_interesting_usage_handler: Arc>> + on_interesting_usage_handler: Arc>>, +} + +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) { + fn on_add_stream( + &self, + stream: UniquePtr, + ) { + 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) { + fn on_remove_stream( + &self, + stream: UniquePtr, + ) { + 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) { + fn on_data_channel( + &self, + data_channel: UniquePtr, + ) { + 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) { + fn on_ice_candidate(&self, candidate: UniquePtr) { + 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>) { + fn on_ice_candidates_removed( + &self, + removed: Vec>, + ) { + 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, streams: Vec>) { + fn on_add_track( + &self, + receiver: UniquePtr, + streams: Vec>, + ) { + 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) { + fn on_track( + &self, + transceiver: UniquePtr, + ) { + 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) { + fn on_remove_track(&self, receiver: UniquePtr) { + trace!("on_remove_track"); let mut handler = self.on_remove_track_handler.lock().unwrap(); if let Some(f) = handler.as_mut() { // 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); } } -} \ No newline at end of file +} + +#[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::(16); + let (alice_ice_tx, mut alice_ice_rx) = mpsc::channel::(16); + let (alice_dc_tx, mut alice_dc_rx) = mpsc::channel::(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::(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(); + } +} diff --git a/crates/livekit-webrtc/src/peer_connection_factory.rs b/crates/livekit-webrtc/src/peer_connection_factory.rs index 8e136bc..2724bf5 100644 --- a/crates/livekit-webrtc/src/peer_connection_factory.rs +++ b/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 + cxx_handle: UniquePtr, } 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) -> Result { + pub fn create_peer_connection( + &self, + config: RTCConfiguration, + ) -> Result { 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, 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(())); - } -} \ No newline at end of file diff --git a/crates/livekit-webrtc/src/rtc_error.rs b/crates/livekit-webrtc/src/rtc_error.rs index 282b2cd..e7aa9c0 100644 --- a/crates/livekit-webrtc/src/rtc_error.rs +++ b/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; - diff --git a/crates/livekit-webrtc/src/rtp_receiver.rs b/crates/livekit-webrtc/src/rtp_receiver.rs index 6ab1a0f..d7f431b 100644 --- a/crates/livekit-webrtc/src/rtp_receiver.rs +++ b/crates/livekit-webrtc/src/rtp_receiver.rs @@ -1,5 +1,2 @@ - - -pub struct RtpReceiver { - -} \ No newline at end of file +#[derive(Debug)] +pub struct RtpReceiver {} diff --git a/crates/livekit-webrtc/src/rtp_transceiver.rs b/crates/livekit-webrtc/src/rtp_transceiver.rs index dc12234..bb7598c 100644 --- a/crates/livekit-webrtc/src/rtp_transceiver.rs +++ b/crates/livekit-webrtc/src/rtp_transceiver.rs @@ -1,5 +1,2 @@ - - -pub struct RtpTransceiver { - -} \ No newline at end of file +#[derive(Debug)] +pub struct RtpTransceiver {} diff --git a/crates/livekit-webrtc/src/session_description.rs b/crates/livekit-webrtc/src/session_description.rs deleted file mode 100644 index 14eaa30..0000000 --- a/crates/livekit-webrtc/src/session_description.rs +++ /dev/null @@ -1,5 +0,0 @@ - -#[derive(Debug)] -pub struct SessionDescription { - -} \ No newline at end of file diff --git a/crates/livekit-webrtc/src/webrtc.rs b/crates/livekit-webrtc/src/webrtc.rs new file mode 100644 index 0000000..80484ae --- /dev/null +++ b/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, +} + +impl RTCRuntime { + pub fn new() -> Self { + Self { + cxx_handle: sys_rtc::ffi::create_rtc_runtime(), + } + } +}