From 4067a9add638c109761560a2e0f6ac8ba439e932 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sun, 25 Sep 2022 13:54:56 +0200 Subject: [PATCH] rtc_config --- Cargo.lock | 154 +++++++++ crates/{livekit-webrtc => }/.cargo/config | 0 crates/livekit-core/Cargo.toml | 4 +- crates/livekit-core/src/lib.rs | 2 + crates/livekit-core/src/lk_runtime.rs | 25 ++ crates/livekit-core/src/pc_transport.rs | 105 ++++++ crates/livekit-core/src/rtc_engine.rs | 311 +++++++++++++++--- crates/livekit-core/src/signal_client.rs | 216 ++++++++---- .../libwebrtc-sys/.cargo/config | 2 - .../libwebrtc-sys/include/livekit/jsep.h | 1 + .../include/livekit/peer_connection.h | 4 + .../include/livekit/rust_types.h | 4 + .../libwebrtc-sys/src/data_channel.rs | 4 +- .../livekit-webrtc/libwebrtc-sys/src/jsep.cpp | 6 + .../livekit-webrtc/libwebrtc-sys/src/jsep.rs | 8 +- .../libwebrtc-sys/src/peer_connection.cpp | 25 ++ .../libwebrtc-sys/src/peer_connection.rs | 16 +- .../src/peer_connection_factory.cpp | 7 +- .../src/peer_connection_factory.rs | 23 +- .../libwebrtc-sys/src/rtc_error.rs | 4 +- .../libwebrtc-sys/src/webrtc.rs | 3 + crates/livekit-webrtc/src/data_channel.rs | 16 +- crates/livekit-webrtc/src/jsep.rs | 13 + crates/livekit-webrtc/src/peer_connection.rs | 47 ++- .../src/peer_connection_factory.rs | 4 +- crates/livekit-webrtc/src/rtc_error.rs | 1 - 26 files changed, 865 insertions(+), 140 deletions(-) rename crates/{livekit-webrtc => }/.cargo/config (100%) create mode 100644 crates/livekit-core/src/lk_runtime.rs create mode 100644 crates/livekit-core/src/pc_transport.rs delete mode 100644 crates/livekit-webrtc/libwebrtc-sys/.cargo/config diff --git a/Cargo.lock b/Cargo.lock index 43aa7f5..a2c991e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -101,6 +101,22 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "core-foundation" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "194a7a9e6de53fa55116934067c844d9d749312f75c6f6d0980e8c252f8c2146" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5827cebf4670468b8772dd191856768aedcb1b0278a04f989f7766351917b9dc" + [[package]] name = "cpufeatures" version = "0.2.4" @@ -214,6 +230,21 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foreign-types" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1" +dependencies = [ + "foreign-types-shared", +] + +[[package]] +name = "foreign-types-shared" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" + [[package]] name = "form_urlencoded" version = "1.0.1" @@ -438,6 +469,8 @@ dependencies = [ "anyhow", "env_logger", "futures-util", + "lazy_static", + "livekit-webrtc", "log", "prost 0.11.0", "prost-build", @@ -509,6 +542,24 @@ version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a" +[[package]] +name = "native-tls" +version = "0.2.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd7e2f3618557f980e0b17e8856252eee3c97fa12c54dff0ca290fb6266ca4a9" +dependencies = [ + "lazy_static", + "libc", + "log", + "openssl", + "openssl-probe", + "openssl-sys", + "schannel", + "security-framework", + "security-framework-sys", + "tempfile", +] + [[package]] name = "num_cpus" version = "1.13.1" @@ -525,6 +576,51 @@ version = "1.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "074864da206b4973b84eb91683020dbefd6a8c3f0f38e054d93954e891935e4e" +[[package]] +name = "openssl" +version = "0.10.41" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "618febf65336490dfcf20b73f885f5651a0c89c64c2d4a8c3662585a70bf5bd0" +dependencies = [ + "bitflags", + "cfg-if", + "foreign-types", + "libc", + "once_cell", + "openssl-macros", + "openssl-sys", +] + +[[package]] +name = "openssl-macros" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b501e44f11665960c7e7fcf062c7d96a14ade4aa98116c004b2e37b5be7d736c" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "openssl-probe" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" + +[[package]] +name = "openssl-sys" +version = "0.9.75" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5f9bd0c2710541a3cda73d6f9ac4f1b240de4ae261065d309dbe73d9dceb42f" +dependencies = [ + "autocfg", + "cc", + "libc", + "pkg-config", + "vcpkg", +] + [[package]] name = "parking_lot" version = "0.12.1" @@ -576,6 +672,12 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "pkg-config" +version = "0.3.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1df8c4ec4b0627e53bdf214615ad287367e482558cf84b109250b37464dc03ae" + [[package]] name = "ppv-lite86" version = "0.2.16" @@ -753,6 +855,16 @@ dependencies = [ "winapi", ] +[[package]] +name = "schannel" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88d6731146462ea25d9244b2ed5fd1d716d25c52e4d54aa4fb0f3c4e9854dbe2" +dependencies = [ + "lazy_static", + "windows-sys", +] + [[package]] name = "scopeguard" version = "1.1.0" @@ -765,6 +877,29 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c8132065adcfd6e02db789d9285a0deb2f3fcb04002865ab67d5fb103533898" +[[package]] +name = "security-framework" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2bc1bb97804af6631813c55739f771071e0f2ed33ee20b68c86ec505d906356c" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0160a13a177a45bfb43ce71c01580998474f556ad854dcbca936dd2841a5c556" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "sha-1" version = "0.10.0" @@ -911,6 +1046,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-native-tls" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b" +dependencies = [ + "native-tls", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.17.2" @@ -919,7 +1064,9 @@ checksum = "f714dd15bead90401d77e04243611caec13726c2408afd5b31901dfcdcb3b181" dependencies = [ "futures-util", "log", + "native-tls", "tokio", + "tokio-native-tls", "tungstenite", ] @@ -935,6 +1082,7 @@ dependencies = [ "http", "httparse", "log", + "native-tls", "rand", "sha-1", "thiserror", @@ -993,6 +1141,12 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + [[package]] name = "version_check" version = "0.9.4" diff --git a/crates/livekit-webrtc/.cargo/config b/crates/.cargo/config similarity index 100% rename from crates/livekit-webrtc/.cargo/config rename to crates/.cargo/config diff --git a/crates/livekit-core/Cargo.toml b/crates/livekit-core/Cargo.toml index 486f2fd..ea4ee25 100644 --- a/crates/livekit-core/Cargo.toml +++ b/crates/livekit-core/Cargo.toml @@ -5,7 +5,7 @@ edition = "2021" [dependencies] log = "0.4" -tokio-tungstenite = "0.17.2" +tokio-tungstenite = { version = "0.17.2", features = ["native-tls"] } tokio = { version = "1.20.1", features = ["full"] } url = "2.2.2" futures-util = "0.3.23" @@ -13,6 +13,8 @@ thiserror = "1.0" prost = "0.11.0" prost-types = "0.11.1" anyhow = "1.0.63" +livekit-webrtc = { path = "../livekit-webrtc" } +lazy_static = "1.4.0" [build-dependencies] prost-build = { version = "0.10" } diff --git a/crates/livekit-core/src/lib.rs b/crates/livekit-core/src/lib.rs index 15669d3..186eef3 100644 --- a/crates/livekit-core/src/lib.rs +++ b/crates/livekit-core/src/lib.rs @@ -4,5 +4,7 @@ pub mod proto { mod rtc_engine; mod signal_client; +mod lk_runtime; +mod pc_transport; pub mod room; diff --git a/crates/livekit-core/src/lk_runtime.rs b/crates/livekit-core/src/lk_runtime.rs new file mode 100644 index 0000000..041158b --- /dev/null +++ b/crates/livekit-core/src/lk_runtime.rs @@ -0,0 +1,25 @@ +use log::trace; + +use livekit_webrtc::peer_connection_factory::PeerConnectionFactory; +use livekit_webrtc::webrtc::RTCRuntime; + +pub struct LKRuntime { + pub rtc_runtime: RTCRuntime, + pub pc_factory: PeerConnectionFactory, +} + +impl LKRuntime { + pub fn new() -> Self { + trace!("LKRuntime::new()"); + Self { + rtc_runtime: RTCRuntime::new(), + pc_factory: PeerConnectionFactory::new(), + } + } +} + +impl Drop for LKRuntime { + fn drop(&mut self) { + trace!("LKRuntime::drop()"); + } +} diff --git a/crates/livekit-core/src/pc_transport.rs b/crates/livekit-core/src/pc_transport.rs new file mode 100644 index 0000000..7c9e02c --- /dev/null +++ b/crates/livekit-core/src/pc_transport.rs @@ -0,0 +1,105 @@ +use std::sync::Arc; +use std::time::Duration; + +use log::{error, trace}; + +use livekit_webrtc::jsep::{IceCandidate, SessionDescription}; +use livekit_webrtc::peer_connection::{PeerConnection, RTCOfferAnswerOptions, SdpError, SignalingState}; +use livekit_webrtc::peer_connection_factory::RTCConfiguration; +use livekit_webrtc::rtc_error::RTCError; + +use crate::lk_runtime::LKRuntime; + +const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); // TODO(theomonnom) + +pub type OnOfferHandler = Box; + +pub struct PCTransport { + peer_connection: PeerConnection, + pending_candidates: Vec, + on_offer_handler: Option, + restarting_ice: bool, + renegotiate: bool, +} + +impl PCTransport { + pub fn new(lk_runtime: Arc, cfg: RTCConfiguration) -> Result { + let peer_connection = lk_runtime.pc_factory.create_peer_connection(cfg)?; + + Ok(Self { + peer_connection, + pending_candidates: Vec::default(), + on_offer_handler: None, + restarting_ice: false, + renegotiate: false, + }) + } + + pub fn peer_connection(&mut self) -> &mut PeerConnection { + &mut self.peer_connection + } + + pub fn on_offer(&mut self, handler: OnOfferHandler) { + self.on_offer_handler = Some(handler); + } + + pub fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) { + if self.peer_connection.remote_description().is_none() { + self.pending_candidates.push(ice_candidate); + return; + } + + self.peer_connection.add_ice_candidate(ice_candidate); + } + + pub async fn set_remote_description(&mut self, remote_description: SessionDescription) -> Result<(), SdpError> { + self.peer_connection.set_remote_description(remote_description).await?; + + for ic in self.pending_candidates.drain(..) { + self.peer_connection.add_ice_candidate(ic); + } + self.restarting_ice = false; + + if self.renegotiate { + self.renegotiate = false; + self.create_and_send_offer(RTCOfferAnswerOptions::default()).await?; + } + + Ok(()) + } + + pub async fn negotiate(&mut self) -> Result<(), SdpError> { + // TODO(theomonnom) Debounce here with NEGOTIATION_FREQUENCY + self.create_and_send_offer(RTCOfferAnswerOptions::default()).await + } + + async fn create_and_send_offer(&mut self, options: RTCOfferAnswerOptions) -> Result<(), SdpError> { + if self.on_offer_handler.is_none() { + return Ok(()); + } + + if options.ice_restart { + trace!("restarting ICE"); + self.restarting_ice = true; + } + + if self.peer_connection.signaling_state() == SignalingState::HaveLocalOffer { + if options.ice_restart { + if let Some(remote_description) = self.peer_connection.remote_description() { + self.peer_connection.set_remote_description(remote_description).await?; + } else { + error!("trying to ice restart when the pc doesn't have remote description"); + } + } else { + self.renegotiate = true; + return Ok(()); + } + } + + let offer = self.peer_connection.create_offer(options).await?; + trace!("created offer {:?}", offer); + self.peer_connection.set_local_description(offer.clone()).await?; + self.on_offer_handler.as_mut().unwrap()(offer); + Ok(()) + } +} \ No newline at end of file diff --git a/crates/livekit-core/src/rtc_engine.rs b/crates/livekit-core/src/rtc_engine.rs index fa61959..df943d9 100644 --- a/crates/livekit-core/src/rtc_engine.rs +++ b/crates/livekit-core/src/rtc_engine.rs @@ -1,49 +1,264 @@ -use crate::proto::signal_response; -use crate::signal_client::{SignalClient, SignalClientError}; -use log::error; -use tokio::sync::broadcast; +use std::sync::{Arc, Mutex, Weak}; -pub struct RTCEngine { - signal_client: SignalClient, +use lazy_static::lazy_static; +use log::{error, trace}; +use prost::Message as ProstMessage; +use thiserror::Error; +use tokio::sync::mpsc; + +use livekit_webrtc::data_channel::{DataChannel, DataChannelInit}; +use livekit_webrtc::jsep::{IceCandidate, SessionDescription}; +use livekit_webrtc::peer_connection::PeerConnectionState; +use livekit_webrtc::peer_connection_factory::{ + ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration, +}; +use livekit_webrtc::rtc_error::RTCError; + +use crate::lk_runtime::LKRuntime; +use crate::pc_transport::PCTransport; +use crate::proto::{DataPacket, JoinResponse, signal_request, SignalTarget, TrickleRequest}; +use crate::proto::signal_response::Message; +use crate::signal_client; +use crate::signal_client::{SignalClient, SignalError}; + +const LOSSY_DC_LABEL: &str = "_lossy"; +const RELIABLE_DC_LABEL: &str = "_reliable"; + +#[derive(Error, Debug)] +pub enum EngineError { + #[error("signal failure")] + SignalError(#[from] SignalError), + #[error("internal webrtc failure")] + RTCError(#[from] RTCError), } -impl RTCEngine { - pub fn new() -> RTCEngine { - Self { - signal_client: SignalClient::new(), - } - } +lazy_static! { + // Share one LKRuntime across all RTCEngine instances + static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); +} - pub async fn connect(&mut self, url: &str, token: &str) -> Result<(), SignalClientError> { - self.signal_client.connect(url, token).await?; +enum EngineMessage {} - tokio::spawn(Self::handle_rtc( - self.signal_client.response_rx.resubscribe(), - )); +struct PeerInternal { + publisher_pc: PCTransport, + subscriber_pc: PCTransport, - Ok(()) - } + lossy_dc: DataChannel, + reliable_dc: DataChannel, - pub fn update(&self) {} + pub_ice_rx: mpsc::Receiver, + sub_ice_rx: mpsc::Receiver, - async fn handle_rtc(mut signal_receiver: broadcast::Receiver) { - loop { - let msg = match signal_receiver.recv().await { - Ok(msg) => msg, - Err(error) => { - error!("Failed to receive SignalResponse: {:?}", error); - continue; - } - }; + pub_offer_rx: mpsc::Receiver, - match msg { - signal_response::Message::Join(join) => {} - signal_response::Message::Trickle(trickle) => {} - signal_response::Message::Answer(answer) => {} - signal_response::Message::Offer(offer) => {} - _ => {} + primary_connection_state_rx: mpsc::Receiver, + secondary_connection_state_rx: mpsc::Receiver, + + lossy_data_rx: mpsc::Receiver, + reliable_data_rx: mpsc::Receiver, +} + +struct RTCInternal { + lk_runtime: Arc, + signal_client: Arc, + pc_internal: PeerInternal, +} + +impl RTCInternal { + async fn connect( + url: &str, + token: &str, + ) -> Result { + let mut lk_runtime = None; + { + // Acquire an existing/a new LKRuntime + let mut lk_runtime_ref = LK_RUNTIME.lock().unwrap(); + lk_runtime = lk_runtime_ref.upgrade(); + + if lk_runtime.is_none() { + let new_runtime = Arc::new(LKRuntime::new()); + *lk_runtime_ref = Arc::downgrade(&new_runtime); + lk_runtime = Some(new_runtime); } } + let lk_runtime = lk_runtime.unwrap(); + let signal_client = Arc::new(signal_client::connect(url, token).await?); + + if let Message::Join(join) = signal_client.recv().await? { + let pc_internal = Self::configure(lk_runtime.clone(), join)?; + + Ok(Self { + lk_runtime, + signal_client, + pc_internal, + }) + } else { + panic!("the first received message isn't a JoinResponse"); + } + } + + async fn run(&mut self) { + loop { + tokio::select! { + Ok(signal) = self.signal_client.recv() => { + + }, + Some(ice_candidate) = self.pc_internal.pub_ice_rx.recv() => { + tokio::spawn({ + let sc = self.signal_client.clone(); + + async move { + let _ = sc.send(signal_request::Message::Trickle(TrickleRequest { + candidate_init: ice_candidate.to_string(), + target: SignalTarget::Publisher as i32 + })).await; + } + }); + }, + Some(ice_candidate) = self.pc_internal.sub_ice_rx.recv() => { + tokio::spawn({ + let sc = self.signal_client.clone(); + + async move { + let _ = sc.send(signal_request::Message::Trickle(TrickleRequest { + candidate_init: ice_candidate.to_string(), + target: SignalTarget::Subscriber as i32 + })).await; + } + }); + }, + Some(sdp) = self.pc_internal.pub_offer_rx.recv() => { + + }, + Some(state) = self.pc_internal.primary_connection_state_rx.recv() => { + + }, + Some(state) = self.pc_internal.secondary_connection_state_rx.recv() => { + + }, + Some(data) = self.pc_internal.lossy_data_rx.recv() => { + + }, + Some(data) = self.pc_internal.reliable_data_rx.recv() => { + + } + } + } + } + + fn configure(lk_runtime: Arc, join: JoinResponse) -> Result { + let cfg = RTCConfiguration { + ice_servers: { + let mut servers = vec![]; + for is in join.ice_servers { + servers.push(ICEServer { + urls: is.urls, + username: is.username, + password: is.credential, + }) + } + servers + }, + continual_gathering_policy: ContinualGatheringPolicy::GatherContinually, + ice_transport_type: IceTransportsType::All, + }; + + // Create the PeerConnections + let mut publisher_pc = PCTransport::new(lk_runtime.clone(), cfg.clone())?; + let mut subscriber_pc = PCTransport::new(lk_runtime, cfg)?; + + let (pub_ice_tx, pub_ice_rx) = mpsc::channel(8); + let (sub_ice_tx, sub_ice_rx) = mpsc::channel(8); + let (pub_offer_tx, pub_offer_rx) = mpsc::channel(8); + let (primary_connection_state_tx, primary_connection_state_rx) = mpsc::channel(8); + let (secondary_connection_state_tx, secondary_connection_state_rx) = mpsc::channel(8); + let (lossy_data_tx, lossy_data_rx) = mpsc::channel(8); + let (reliable_data_tx, reliable_data_rx) = mpsc::channel(8); + + publisher_pc.peer_connection().on_ice_candidate(Box::new(move |ice_candidate| { + trace!("publisher - on_ice_candidate: {:?}", ice_candidate); + let _ = pub_ice_tx.blocking_send(ice_candidate); + })); + + subscriber_pc.peer_connection().on_ice_candidate(Box::new(move |ice_candidate| { + trace!("subscriber - on_ice_candidate: {:?}", ice_candidate); + let _ = sub_ice_tx.blocking_send(ice_candidate); + })); + + publisher_pc.on_offer(Box::new(move |offer| { + trace!("publisher - on_offer: {:?}", offer); + let _ = pub_offer_tx.blocking_send(offer); // TODO(theomonnom) Don't use blocking_send here + })); + + let mut primary_pc = &publisher_pc; + let mut secondary_pc = &subscriber_pc; + if join.subscriber_primary { + primary_pc = &subscriber_pc; + secondary_pc = &publisher_pc; + } + + primary_pc.peer_connection().on_connection_change(Box::new(move |state| { + let _ = primary_connection_state_tx.blocking_send(state); + })); + + secondary_pc.peer_connection().on_connection_change(Box::new(move |state| { + let _ = secondary_connection_state_tx.blocking_send(state); + })); + + let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(LOSSY_DC_LABEL, { + let mut dc_init = DataChannelInit::default(); + dc_init.ordered = true; + dc_init.max_retransmits = Some(0); + dc_init + })?; + + let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(RELIABLE_DC_LABEL, { + let mut dc_init = DataChannelInit::default(); + dc_init.ordered = true; + dc_init + })?; + + lossy_dc.on_message(Box::new(|data, binary| { + if let Ok(data) = DataPacket::decode(data) { + let _ = lossy_data_tx.blocking_send(data); + } else { + trace!("lossy_dc - failed to decode DataPacket"); + } + })); + + reliable_dc.on_message(Box::new(|data, binary| { + if let Ok(data) = DataPacket::decode(data) { + let _ = reliable_data_tx.blocking_send(data); + } else { + trace!("reliable_dc - failed to decode DataPacket"); + } + })); + + Ok(PeerInternal { + publisher_pc, + subscriber_pc, + lossy_dc, + reliable_dc, + pub_ice_rx, + sub_ice_rx, + pub_offer_rx, + primary_connection_state_rx, + secondary_connection_state_rx, + lossy_data_rx, + reliable_data_rx, + }) + } +} + +pub struct RTCEngine {} + +/// Initialize the SignalClient & the PeerConnections +//pub async fn connect(url: &str, token: &str) -> Result { +//} + +impl RTCEngine { + async fn rtc_handle() { + loop {} } } @@ -51,9 +266,25 @@ impl RTCEngine { async fn test_test() { env_logger::init(); - let mut engine = RTCEngine::new(); - engine.connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjQ1OTY4MDYsImlzcyI6IkFQSUNrSG04M01oZ2hQeCIsIm5hbWUiOiJ1c2VyMSIsIm5iZiI6MTY2MDk5NjgwNiwic3ViIjoidXNlcjEiLCJ2aWRlbyI6eyJyb29tIjoibXktZmlyc3Qtcm9vbSIsInJvb21Kb2luIjp0cnVlfX0.SWU_LETMK6ZmFOf38pYjVhpur0o7jJc6u61h8BH7g20").await.unwrap(); - - // Wait before exiting the program - tokio::time::sleep(core::time::Duration::from_millis(1000 * 25)).await; + //engine.connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjQ1OTY4MDYsImlzcyI6IkFQSUNrSG04M01oZ2hQeCIsIm5hbWUiOiJ1c2VyMSIsIm5iZiI6MTY2MDk5NjgwNiwic3ViIjoidXNlcjEiLCJ2aWRlbyI6eyJyb29tIjoibXktZmlyc3Qtcm9vbSIsInJvb21Kb2luIjp0cnVlfX0.SWU_LETMK6ZmFOf38pYjVhpur0o7jJc6u61h8BH7g20").await.unwrap(); } + +/*sync fn handle_rtc(mut signal_receiver: broadcast::Receiver) { + loop { + let msg = match signal_receiver.recv().await { + Ok(msg) => msg, + Err(error) => { + error!("Failed to receive SignalResponse: {:?}", error); + continue; + } + }; + + match msg { + Message::Join(join) => {} + Message::Trickle(trickle) => {} + Message::Answer(answer) => {} + Message::Offer(offer) => {} + _ => {} + } + } +}*/ diff --git a/crates/livekit-core/src/signal_client.rs b/crates/livekit-core/src/signal_client.rs index 717f948..5fa8715 100644 --- a/crates/livekit-core/src/signal_client.rs +++ b/crates/livekit-core/src/signal_client.rs @@ -1,100 +1,186 @@ -use futures_util::SinkExt; -use futures_util::StreamExt; +use futures_util::{SinkExt, StreamExt}; +use futures_util::stream::{SplitSink, SplitStream}; use log::{error, info}; -use prost::Message as ProtoMessage; -use std::borrow::Borrow; +use prost::Message as ProstMessage; use thiserror::Error; use tokio::net::TcpStream; -use tokio::sync::broadcast; +use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinHandle; -use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; +use tokio_tungstenite::tungstenite::{ + Error as WsError, + Message, protocol::frame::{CloseFrame, coding::CloseCode}, +}; -use crate::{proto, proto::signal_response}; +use crate::proto::{signal_request, signal_response, SignalRequest, SignalResponse}; + +pub const PROTOCOL_VERSION: u32 = 8; #[derive(Error, Debug)] -pub enum SignalClientError { +pub enum SignalError { #[error("websocket failure")] - WebSocket(#[from] tokio_tungstenite::tungstenite::Error), + WsError(#[from] WsError), #[error("failed to parse the url")] UrlParse(#[from] url::ParseError), - #[error("failed to parse messages from server")] + #[error("failed to decode messages from server")] ProtoParse(#[from] prost::DecodeError), } +type SignalResult = Result; +type WebSocket = WebSocketStream>; + +#[derive(Debug)] +struct RecvMessage { + response_chn: oneshot::Sender>, +} + +#[derive(Debug)] +struct SendMessage { + signal: signal_request::Message, + response_chn: oneshot::Sender>, +} + pub struct SignalClient { - ws_handle: Option>>, - response_tx: broadcast::Sender, - pub response_rx: broadcast::Receiver, + read_sender: mpsc::Sender, + write_sender: mpsc::Sender, + write_shutdown_sender: oneshot::Sender<()>, + read_shutdown_sender: oneshot::Sender<()>, + read_handle: JoinHandle<()>, + write_handle: JoinHandle<()>, +} + +pub async fn connect(url: &str, token: &str) -> SignalResult { + let mut lk_url = url::Url::parse(url)?; + lk_url.set_path("/rtc"); + lk_url + .query_pairs_mut() + .append_pair("access_token", token) + .append_pair("protocol", PROTOCOL_VERSION.to_string().as_str()); + + let (ws_stream, _) = connect_async(lk_url).await?; + let (ws_writer, ws_reader) = ws_stream.split(); + + let (read_tx, read_rx) = mpsc::channel::(1); + let (write_tx, write_rx) = mpsc::channel::(1); + let (read_shutdown_tx, read_shutdown_rx) = oneshot::channel(); + let (write_shutdown_tx, write_shutdown_rx) = oneshot::channel(); + + let read_handle = tokio::spawn(SignalClient::ws_read(read_rx, ws_reader, read_shutdown_rx)); + let write_handle = tokio::spawn(SignalClient::ws_write( + write_rx, + ws_writer, + write_shutdown_rx, + )); + + Ok(SignalClient { + read_sender: read_tx, + write_sender: write_tx, + write_shutdown_sender: write_shutdown_tx, + read_shutdown_sender: read_shutdown_tx, + read_handle, + write_handle, + }) } impl SignalClient { - pub fn new() -> Self { - let (tx, rx) = broadcast::channel(16); - - Self { - response_tx: tx, - response_rx: rx, - ws_handle: None, - } + pub async fn close(self) { + let _ = self.write_shutdown_sender.send(()); + let _ = self.write_handle.await; + let _ = self.read_shutdown_sender.send(()); + let _ = self.read_handle.await; } - pub async fn connect(&mut self, url: &str, token: &str) -> Result<(), SignalClientError> { - let mut lk_url = url::Url::parse(url)?; - lk_url.set_path("/rtc"); - lk_url - .query_pairs_mut() - .append_pair("access_token", token) - .append_pair("protocol", "8"); - - info!("Connecting to {}", lk_url); - let (ws, _) = connect_async(&lk_url).await?; - - self.ws_handle = Some(tokio::spawn(Self::handle_ws(ws, self.response_tx.clone()))); - Ok(()) + pub async fn recv(&self) -> SignalResult { + let (send, recv) = oneshot::channel(); + let msg = RecvMessage { response_chn: send }; + let _ = self.read_sender.send(msg).await; + recv.await.expect("channel closed") } - pub async fn disconnect() { - unimplemented!() + pub async fn send(&self, signal: signal_request::Message) -> SignalResult<()> { + let (send, recv) = oneshot::channel(); + let msg = SendMessage { + signal, + response_chn: send, + }; + let _ = self.write_sender.send(msg).await; + recv.await.expect("channel closed") } - async fn handle_ws( - mut ws: WebSocketStream>, - response_tx: broadcast::Sender, - ) -> Result<(), SignalClientError> { + async fn ws_write( + mut write_receiver: mpsc::Receiver, + mut ws_writer: SplitSink, + mut shutdown_receiver: oneshot::Receiver<()>, + ) { loop { tokio::select! { - next_msg = ws.next() => { - let ws_msg = match next_msg { - Some(msg) => msg?, - None => break, + Some(msg) = write_receiver.recv() => { + let req = SignalRequest { + message: Some(msg.signal), }; - let data = match ws_msg { - Message::Binary(data) => data, - Message::Ping(data) => { - ws.send(Message::Pong(data)).await?; - continue - }, - Message::Close(_frame) => break, - _ => continue, - }; - - let proto_msg = proto::SignalResponse::decode(data.borrow())?; - let signal_response = proto_msg.message.unwrap(); - - match signal_response { - signal_response::Message::Pong(ts) => { - - }, - _ => { - response_tx.send(signal_response).unwrap(); - } + let write_res = ws_writer.send(Message::Binary(req.encode_to_vec())).await; + if let Err(err) = write_res { + error!("failed to send message to ws: {:?}", err); + let _ = msg.response_chn.send(Err(err.into())); + break; } + + let _ = msg.response_chn.send(Ok(())); + }, + _ = (&mut shutdown_receiver) => { + let _ = ws_writer.send(Message::Close(Some(CloseFrame { + code: CloseCode::Normal, + reason: "disconnected by client".into() + }))).await; + let _ = ws_writer.flush().await; + break; } } } + } - Ok(()) + async fn ws_read( + mut write_receiver: mpsc::Receiver, + mut ws_reader: SplitStream, + mut shutdown_receiver: oneshot::Receiver<()>, + ) { + loop { + tokio::select! { + Some(msg) = write_receiver.recv() => { + let read = ws_reader.next().await; + if read.is_none() { + let _ = msg.response_chn.send(Err(SignalError::WsError(WsError::ConnectionClosed))); + break; + } + let read = read.unwrap(); + match read { + Ok(Message::Binary(data)) => { + let res = SignalResponse::decode(data.as_slice()).expect("failed to decode incoming SignalResponse"); + + // TODO(theomonnon) Handle Message::Pong + let res = res.message.unwrap(); + let _ = msg.response_chn.send(Ok(res)); + } + _ => { + error!("unhandled websocket message: {:?}", read); + let _ = msg.response_chn.send(Err(SignalError::WsError(WsError::ConnectionClosed))); + break; + } + } + }, + _ = (&mut shutdown_receiver) => break + } + } } } + +#[tokio::test] +async fn test_test() { + env_logger::init(); + let client = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap(); + let msg = client.recv().await.unwrap(); + + client.close().await; + info!("Received message {:?}", msg); +} diff --git a/crates/livekit-webrtc/libwebrtc-sys/.cargo/config b/crates/livekit-webrtc/libwebrtc-sys/.cargo/config deleted file mode 100644 index 0c17df0..0000000 --- a/crates/livekit-webrtc/libwebrtc-sys/.cargo/config +++ /dev/null @@ -1,2 +0,0 @@ -[target.x86_64-pc-windows-msvc] -rustflags = ["-C", "target-feature=+crt-static"] \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h index cf97783..60f0220 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h @@ -19,6 +19,7 @@ class IceCandidate { explicit IceCandidate( std::unique_ptr ice_candidate); + rust::String stringify() const; std::unique_ptr release(); private: 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 7dd3afb..e0f7e67 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h @@ -34,6 +34,10 @@ class PeerConnection { std::unique_ptr init); void add_ice_candidate(std::unique_ptr candidate, NativeAddIceCandidateObserver& observer); + std::unique_ptr local_description() const; + std::unique_ptr remote_description() const; + SignalingState signaling_state() const; + IceGatheringState ice_gathering_state() const; void close(); private: 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 ed61c54..6bf5796 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h @@ -17,6 +17,10 @@ struct DataChannelObserverWrapper; struct AddIceCandidateObserverWrapper; // Shared types +enum class PeerConnectionState; +enum class SignalingState; +enum class IceConnectionState; +enum class IceGatheringState; struct RTCOfferAnswerOptions; struct RTCError; struct DataChannelInit; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs index 98b187a..26c5991 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs @@ -3,7 +3,7 @@ use std::slice; #[cxx::bridge(namespace = "livekit")] pub mod ffi { #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum Priority { VeryLow, Low, @@ -36,7 +36,7 @@ pub mod ffi { } #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum DataState { Connecting, Open, diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp index e214b42..418d308 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp @@ -16,6 +16,12 @@ IceCandidate::IceCandidate( std::unique_ptr ice_candidate) : ice_candidate_(std::move(ice_candidate)) {} +rust::String IceCandidate::stringify() const { + std::string str; + ice_candidate_->ToString(&str); + return rust::String{str}; +} + std::unique_ptr IceCandidate::release() { return std::move(ice_candidate_); } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs index f752f99..9be0d9f 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs @@ -32,6 +32,8 @@ pub mod ffi { type NativeSetLocalSdpObserverHandle; type NativeSetRemoteSdpObserverHandle; + fn stringify(self: &IceCandidate) -> String; + fn stringify(self: &SessionDescription) -> String; fn clone(self: &SessionDescription) -> UniquePtr; @@ -46,7 +48,7 @@ pub mod ffi { ) -> UniquePtr; fn _unique_ice_candidate() -> UniquePtr; // Ignore - fn _unique_session_description() -> UniquePtr; // Ignore + fn _unique_session_description() -> UniquePtr; // Ignore } } @@ -60,12 +62,14 @@ 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) + write!(f, "{}", self.stringify()) } } unsafe impl Send for ffi::IceCandidate {} +unsafe impl Sync for ffi::IceCandidate {} + // CreateSdpObserver pub trait CreateSdpObserver: Send { diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp index 3b18976..1d020c8 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp @@ -76,10 +76,35 @@ void PeerConnection::add_ice_candidate( [&](const webrtc::RTCError& err) { observer.OnComplete(to_error(err)); }); } +std::unique_ptr PeerConnection::local_description() const { + auto local_description = peer_connection_->local_description(); + if (local_description) + return std::make_unique(local_description->Clone()); + + return std::unique_ptr(); +} + +std::unique_ptr PeerConnection::remote_description() const { + auto remote_description = peer_connection_->remote_description(); + if (remote_description) + return std::make_unique(remote_description->Clone()); + + return std::unique_ptr(); +} + +SignalingState PeerConnection::signaling_state() const { + return static_cast(peer_connection_->signaling_state()); +} + +IceGatheringState PeerConnection::ice_gathering_state() const { + return static_cast(peer_connection_->ice_gathering_state()); +} + void PeerConnection::close() { peer_connection_->Close(); } + // AddIceCandidateObserver NativeAddIceCandidateObserver::NativeAddIceCandidateObserver( diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index 2bd8442..fd98852 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs @@ -23,7 +23,7 @@ pub mod ffi { } #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum PeerConnectionState { New, Connecting, @@ -34,7 +34,7 @@ pub mod ffi { } #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum SignalingState { Stable, HaveLocalOffer, @@ -45,7 +45,7 @@ pub mod ffi { } #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum IceConnectionState { IceConnectionNew, IceConnectionChecking, @@ -58,7 +58,7 @@ pub mod ffi { } #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum IceGatheringState { IceGatheringNew, IceGatheringGathering, @@ -158,6 +158,14 @@ pub mod ffi { observer: Pin<&mut NativeAddIceCandidateObserver>, ); + fn local_description(self: &PeerConnection) -> UniquePtr; + + fn remote_description(self: &PeerConnection) -> UniquePtr; + + fn signaling_state(self: &PeerConnection) -> SignalingState; + + fn ice_gathering_state(self: &PeerConnection) -> IceGatheringState; + fn close(self: Pin<&mut PeerConnection>); fn create_native_peer_connection_observer( 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 1de4ff4..07cb05c 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp @@ -86,8 +86,13 @@ std::unique_ptr create_rtc_configuration( for (auto& url : item.urls) { ice_server.urls.emplace_back(url.c_str()); } - rtc->servers.push_back(ice_server); + rtc->continual_gathering_policy = + static_cast( + conf.continual_gathering_policy); + + rtc->type = static_cast( + conf.ice_transport_type); } return rtc; 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 4116eca..5c45e31 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs @@ -12,9 +12,27 @@ pub mod ffi { pub password: String, } + #[derive(Debug)] + #[repr(i32)] + pub enum ContinualGatheringPolicy { + GatherOnce, + GatherContinually, + } + + #[derive(Debug)] + #[repr(i32)] + pub enum IceTransportsType { + None, + Relay, + NoHost, + All, + } + #[derive(Debug, Clone)] pub struct RTCConfiguration { pub ice_servers: Vec, + pub continual_gathering_policy: ContinualGatheringPolicy, + pub ice_transport_type: IceTransportsType, } unsafe extern "C++" { @@ -22,7 +40,7 @@ pub mod ffi { type PeerConnection = crate::peer_connection::ffi::PeerConnection; type NativePeerConnectionObserver = - crate::peer_connection::ffi::NativePeerConnectionObserver; + crate::peer_connection::ffi::NativePeerConnectionObserver; type PeerConnectionFactory; type NativeRTCConfiguration; @@ -38,3 +56,6 @@ pub mod ffi { ) -> Result>; } } + +unsafe impl Send for ffi::PeerConnectionFactory {} +unsafe impl Sync for ffi::PeerConnectionFactory {} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs b/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs index 0c2e1bd..c3839e3 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs @@ -7,7 +7,7 @@ use std::fmt::{Display, Formatter}; #[cxx::bridge(namespace = "livekit")] pub mod ffi { #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum RTCErrorType { None, UnsupportedOperation, @@ -24,7 +24,7 @@ pub mod ffi { } #[derive(Debug)] - #[repr(u32)] + #[repr(i32)] pub enum RTCErrorDetailType { None, DataChannelFailure, diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs index d890fc9..74df992 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs @@ -10,3 +10,6 @@ pub mod ffi { fn create_rtc_runtime() -> UniquePtr; } } + +unsafe impl Send for ffi::RTCRuntime {} +unsafe impl Sync for ffi::RTCRuntime {} diff --git a/crates/livekit-webrtc/src/data_channel.rs b/crates/livekit-webrtc/src/data_channel.rs index ad5145d..5a0d012 100644 --- a/crates/livekit-webrtc/src/data_channel.rs +++ b/crates/livekit-webrtc/src/data_channel.rs @@ -128,14 +128,14 @@ impl Default for InternalDataChannelObserver { #[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, + pub reliable: bool, + pub ordered: bool, + pub max_retransmit_time: Option, + pub max_retransmits: Option, + pub protocol: String, + pub negotiated: bool, + pub id: i32, + pub priority: Option, } impl Default for DataChannelInit { diff --git a/crates/livekit-webrtc/src/jsep.rs b/crates/livekit-webrtc/src/jsep.rs index d979b9a..4d4c25f 100644 --- a/crates/livekit-webrtc/src/jsep.rs +++ b/crates/livekit-webrtc/src/jsep.rs @@ -2,6 +2,7 @@ use cxx::UniquePtr; use libwebrtc_sys::jsep as sys_jsep; +// TODO Maybe we can replace that by a serialized IceCandidateInit #[derive(Debug)] pub struct IceCandidate { cxx_handle: UniquePtr, @@ -17,6 +18,12 @@ impl IceCandidate { } } +impl ToString for IceCandidate { + fn to_string(&self) -> String { + self.cxx_handle.stringify() + } +} + #[derive(Debug)] pub struct SessionDescription { cxx_handle: UniquePtr, @@ -32,6 +39,12 @@ impl SessionDescription { } } +impl ToString for SessionDescription { + fn to_string(&self) -> String { + self.cxx_handle.stringify() + } +} + impl Clone for SessionDescription { fn clone(&self) -> Self { SessionDescription::new(self.cxx_handle.clone()) diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index d310435..909a7ff 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -51,7 +51,7 @@ impl PeerConnection { } } - pub async fn create_offer(&mut self) -> Result { + pub async fn create_offer(&mut self, options: RTCOfferAnswerOptions) -> Result { let (tx, mut rx) = mpsc::channel(1); let wrapper = @@ -62,7 +62,7 @@ impl PeerConnection { unsafe { self.cxx_handle .pin_mut() - .create_offer(native_wrapper.pin_mut(), RTCOfferAnswerOptions::default()); + .create_offer(native_wrapper.pin_mut(), options); } match rx.recv().await { @@ -71,7 +71,7 @@ impl PeerConnection { } } - pub async fn create_answer(&mut self) -> Result { + pub async fn create_answer(&mut self, options: RTCOfferAnswerOptions) -> Result { let (tx, mut rx) = mpsc::channel(1); let wrapper = @@ -82,7 +82,7 @@ impl PeerConnection { unsafe { self.cxx_handle .pin_mut() - .create_answer(native_wrapper.pin_mut(), RTCOfferAnswerOptions::default()); + .create_answer(native_wrapper.pin_mut(), options); } match rx.recv().await { @@ -154,6 +154,7 @@ impl PeerConnection { } } + // TODO(theomonnom) Use IceCandidateInit instead of IceCandidate 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| { @@ -172,6 +173,32 @@ impl PeerConnection { } } + pub fn local_description(&self) -> Option { + let local_description = self.cxx_handle.local_description(); + if local_description.is_null() { + None + } else { + Some(SessionDescription::new(local_description)) + } + } + + pub fn remote_description(&self) -> Option { + let remote_description = self.cxx_handle.remote_description(); + if remote_description.is_null() { + None + } else { + Some(SessionDescription::new(remote_description)) + } + } + + pub fn signaling_state(&self) -> SignalingState { + self.cxx_handle.signaling_state() + } + + pub fn ice_gathering_state(&self) -> IceGatheringState { + self.cxx_handle.ice_gathering_state() + } + pub fn close(&mut self) { self.cxx_handle.pin_mut().close(); } @@ -344,16 +371,16 @@ pub type OnRenegotiationNeededHandler = Box; pub type OnNegotiationNeededEventHandler = Box; pub type OnIceConnectionChangeHandler = Box; pub type OnStandardizedIceConnectionChangeHandler = -Box; + Box; pub type OnConnectionChangeHandler = Box; pub type OnIceGatheringChangeHandler = Box; pub type OnIceCandidateHandler = Box; pub type OnIceCandidateErrorHandler = -Box; + Box; pub type OnIceCandidatesRemovedHandler = Box) + Send + Sync>; pub type OnIceConnectionReceivingChangeHandler = Box; pub type OnIceSelectedCandidatePairChangedHandler = -Box; + Box; pub type OnAddTrackHandler = Box) + Send + Sync>; pub type OnTrackHandler = Box; pub type OnRemoveTrackHandler = Box; @@ -368,16 +395,16 @@ pub(crate) struct InternalObserver { on_negotiation_needed_event_handler: Arc>>, on_ice_connection_change_handler: Arc>>, on_standardized_ice_connection_change_handler: - Arc>>, + 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>>, + Arc>>, on_ice_selected_candidate_pair_changed_handler: - Arc>>, + Arc>>, on_add_track_handler: Arc>>, on_track_handler: Arc>>, on_remove_track_handler: Arc>>, diff --git a/crates/livekit-webrtc/src/peer_connection_factory.rs b/crates/livekit-webrtc/src/peer_connection_factory.rs index 6ff7e75..5f37612 100644 --- a/crates/livekit-webrtc/src/peer_connection_factory.rs +++ b/crates/livekit-webrtc/src/peer_connection_factory.rs @@ -2,7 +2,9 @@ use cxx::UniquePtr; use libwebrtc_sys::peer_connection as sys_pc; use libwebrtc_sys::peer_connection_factory as sys_factory; -pub use sys_factory::ffi::{ICEServer, RTCConfiguration}; +pub use sys_factory::ffi::{ + ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration, +}; use crate::peer_connection::{InternalObserver, PeerConnection}; use crate::rtc_error::RTCError; diff --git a/crates/livekit-webrtc/src/rtc_error.rs b/crates/livekit-webrtc/src/rtc_error.rs index e054cf4..86c81e1 100644 --- a/crates/livekit-webrtc/src/rtc_error.rs +++ b/crates/livekit-webrtc/src/rtc_error.rs @@ -1,3 +1,2 @@ // TODO(theomonnom) Wrap the RTCError ffi so we can use Option(u16) pub use libwebrtc_sys::rtc_error::ffi::RTCError; -