diff --git a/Cargo.lock b/Cargo.lock index a2c991e..d20a7e5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -475,6 +475,7 @@ dependencies = [ "prost 0.11.0", "prost-build", "prost-types 0.11.1", + "serde_json", "thiserror", "tokio", "tokio-tungstenite", @@ -855,6 +856,12 @@ dependencies = [ "winapi", ] +[[package]] +name = "ryu" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4501abdff3ae82a1c1b477a17252eb69cee9e66eb915c1abaa4f44d873df9f09" + [[package]] name = "schannel" version = "0.1.20" @@ -900,6 +907,23 @@ dependencies = [ "libc", ] +[[package]] +name = "serde" +version = "1.0.145" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b" + +[[package]] +name = "serde_json" +version = "1.0.85" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e55a28e3aaef9d5ce0506d0a14dbba8054ddc7e499ef522dd8b26859ec9d4a44" +dependencies = [ + "itoa", + "ryu", + "serde", +] + [[package]] name = "sha-1" version = "0.10.0" diff --git a/crates/livekit-core/Cargo.toml b/crates/livekit-core/Cargo.toml index ea4ee25..8ddd856 100644 --- a/crates/livekit-core/Cargo.toml +++ b/crates/livekit-core/Cargo.toml @@ -4,6 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] +serde_json = "1.0" log = "0.4" tokio-tungstenite = { version = "0.17.2", features = ["native-tls"] } tokio = { version = "1.20.1", features = ["full"] } diff --git a/crates/livekit-core/src/lib.rs b/crates/livekit-core/src/lib.rs index 186eef3..de8d910 100644 --- a/crates/livekit-core/src/lib.rs +++ b/crates/livekit-core/src/lib.rs @@ -2,9 +2,9 @@ pub mod proto { include!(concat!(env!("OUT_DIR"), "/livekit.rs")); } -mod rtc_engine; -mod signal_client; mod lk_runtime; mod pc_transport; +mod rtc_engine; +mod signal_client; pub mod room; diff --git a/crates/livekit-core/src/pc_transport.rs b/crates/livekit-core/src/pc_transport.rs index 7c9e02c..338cc4d 100644 --- a/crates/livekit-core/src/pc_transport.rs +++ b/crates/livekit-core/src/pc_transport.rs @@ -4,7 +4,9 @@ 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::{ + PeerConnection, RTCOfferAnswerOptions, SignalingState, +}; use livekit_webrtc::peer_connection_factory::RTCConfiguration; use livekit_webrtc::rtc_error::RTCError; @@ -12,7 +14,7 @@ use crate::lk_runtime::LKRuntime; const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); // TODO(theomonnom) -pub type OnOfferHandler = Box; +pub type OnOfferHandler = Box; pub struct PCTransport { peer_connection: PeerConnection, @@ -43,37 +45,48 @@ impl PCTransport { self.on_offer_handler = Some(handler); } - pub fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) { + pub async fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) -> Result<(), RTCError> { if self.peer_connection.remote_description().is_none() { self.pending_candidates.push(ice_candidate); - return; + return Ok(()); } - self.peer_connection.add_ice_candidate(ice_candidate); + self.peer_connection.add_ice_candidate(ice_candidate).await?; + Ok(()) } - pub async fn set_remote_description(&mut self, remote_description: SessionDescription) -> Result<(), SdpError> { - self.peer_connection.set_remote_description(remote_description).await?; + pub async fn set_remote_description( + &mut self, + remote_description: SessionDescription, + ) -> Result<(), RTCError> { + self.peer_connection + .set_remote_description(remote_description) + .await?; for ic in self.pending_candidates.drain(..) { - self.peer_connection.add_ice_candidate(ic); + self.peer_connection.add_ice_candidate(ic).await?; } self.restarting_ice = false; if self.renegotiate { self.renegotiate = false; - self.create_and_send_offer(RTCOfferAnswerOptions::default()).await?; + self.create_and_send_offer(RTCOfferAnswerOptions::default()) + .await?; } Ok(()) } - pub async fn negotiate(&mut self) -> Result<(), SdpError> { + pub async fn negotiate(&mut self) -> Result<(), RTCError> { // TODO(theomonnom) Debounce here with NEGOTIATION_FREQUENCY - self.create_and_send_offer(RTCOfferAnswerOptions::default()).await + self.create_and_send_offer(RTCOfferAnswerOptions::default()) + .await } - async fn create_and_send_offer(&mut self, options: RTCOfferAnswerOptions) -> Result<(), SdpError> { + async fn create_and_send_offer( + &mut self, + options: RTCOfferAnswerOptions, + ) -> Result<(), RTCError> { if self.on_offer_handler.is_none() { return Ok(()); } @@ -86,7 +99,9 @@ impl PCTransport { 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?; + self.peer_connection + .set_remote_description(remote_description) + .await?; } else { error!("trying to ice restart when the pc doesn't have remote description"); } @@ -98,8 +113,10 @@ impl PCTransport { let offer = self.peer_connection.create_offer(options).await?; trace!("created offer {:?}", offer); - self.peer_connection.set_local_description(offer.clone()).await?; + 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 df943d9..f66a7d9 100644 --- a/crates/livekit-core/src/rtc_engine.rs +++ b/crates/livekit-core/src/rtc_engine.rs @@ -1,24 +1,27 @@ use std::sync::{Arc, Mutex, Weak}; +use std::time::Duration; use lazy_static::lazy_static; use log::{error, trace}; use prost::Message as ProstMessage; use thiserror::Error; use tokio::sync::mpsc; +use tokio::time::sleep; use livekit_webrtc::data_channel::{DataChannel, DataChannelInit}; -use livekit_webrtc::jsep::{IceCandidate, SessionDescription}; -use livekit_webrtc::peer_connection::PeerConnectionState; +use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription}; +use livekit_webrtc::peer_connection::{PeerConnectionState, RTCOfferAnswerOptions}; use livekit_webrtc::peer_connection_factory::{ ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration, }; use livekit_webrtc::rtc_error::RTCError; +use crate::{proto, signal_client}; 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::proto::{ + DataPacket, JoinResponse, signal_request, signal_response, SignalTarget, TrickleRequest, +}; use crate::signal_client::{SignalClient, SignalError}; const LOSSY_DC_LABEL: &str = "_lossy"; @@ -27,9 +30,22 @@ const RELIABLE_DC_LABEL: &str = "_reliable"; #[derive(Error, Debug)] pub enum EngineError { #[error("signal failure")] - SignalError(#[from] SignalError), + Signal(#[from] SignalError), #[error("internal webrtc failure")] - RTCError(#[from] RTCError), + Rtc(#[from] RTCError), + #[error("failed to parse sdp")] + Parse(#[from] SdpParseError), + #[error("serde error")] + Serde(#[from] serde_json::Error), +} + +#[derive(PartialEq, Debug, Copy, Clone)] +enum PCState { + New, + Connected, + Disconnected, + Reconnecting, + Closed, } lazy_static! { @@ -56,19 +72,19 @@ struct PeerInternal { lossy_data_rx: mpsc::Receiver, reliable_data_rx: mpsc::Receiver, + + pc_state: PCState, } struct RTCInternal { + #[allow(unused)] lk_runtime: Arc, signal_client: Arc, pc_internal: PeerInternal, } impl RTCInternal { - async fn connect( - url: &str, - token: &str, - ) -> Result { + async fn connect(url: &str, token: &str) -> Result { let mut lk_runtime = None; { // Acquire an existing/a new LKRuntime @@ -84,8 +100,14 @@ impl RTCInternal { 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)?; + trace!("waiting JoinReponse.."); + if let signal_response::Message::Join(join) = signal_client.recv().await? { + trace!("configuring peerconnections: {:?}", join); + let mut pc_internal = Self::configure(lk_runtime.clone(), join.clone())?; + + if !join.subscriber_primary { + pc_internal.publisher_pc.negotiate().await?; + } Ok(Self { lk_runtime, @@ -97,44 +119,100 @@ impl RTCInternal { } } + fn request_signal(&mut self, msg: signal_request::Message) { + tokio::spawn({ + let sc = self.signal_client.clone(); + + async move { + if let Err(err) = sc.send(msg).await { + error!("failed to send signal: {:?}", err); + } + } + }); + } + + async fn handle_signal(&mut self, signal: signal_response::Message) -> Result<(), EngineError> { + match signal { + signal_response::Message::Answer(answer) => { + let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; + self.pc_internal.publisher_pc.set_remote_description(sdp).await?; + }, + signal_response::Message::Offer(offer) => { + let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; + self.pc_internal.subscriber_pc.set_remote_description(sdp).await?; + let answer = self.pc_internal.subscriber_pc.peer_connection().create_answer(RTCOfferAnswerOptions::default()).await?; + self.pc_internal.subscriber_pc.peer_connection().set_local_description(answer.clone()).await?; + + self.request_signal(signal_request::Message::Answer(proto::SessionDescription { + r#type: "answer".to_string(), + sdp: answer.to_string(), + })); + }, + signal_response::Message::Trickle(trickle) => { + let json: serde_json::Value = serde_json::from_str(&trickle.candidate_init)?; + let ice = IceCandidate::from( + json["sdpMid"].as_str().unwrap(), + json["sdpMLineIndex"].as_i64().unwrap().try_into().unwrap(), + json["candidate"].as_str().unwrap() + )?; + + if trickle.target == SignalTarget::Publisher as i32 { + self.pc_internal.publisher_pc.add_ice_candidate(ice).await?; + } else { + self.pc_internal.subscriber_pc.add_ice_candidate(ice).await?; + } + } + _ => {}, + } + + Ok(()) + } + async fn run(&mut self) { loop { tokio::select! { Ok(signal) = self.signal_client.recv() => { - + if let Err(err) = self.handle_signal(signal).await { + error!("failed to handle signal: {:?}", err); + } }, 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; - } - }); + self.request_signal(signal_request::Message::Trickle(TrickleRequest { + candidate_init: ice_candidate.to_string(), + target: SignalTarget::Publisher as i32 + })); }, 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; - } - }); + self.request_signal(signal_request::Message::Trickle(TrickleRequest { + candidate_init: ice_candidate.to_string(), + target: SignalTarget::Subscriber as i32 + })); }, Some(sdp) = self.pc_internal.pub_offer_rx.recv() => { - + trace!("received publisher offer: {:?}", sdp); + self.request_signal(signal_request::Message::Offer(proto::SessionDescription { + r#type: "offer".to_string(), + sdp: sdp.to_string(), + })); }, Some(state) = self.pc_internal.primary_connection_state_rx.recv() => { + if state == PeerConnectionState::Connected { + let old_state = self.pc_internal.pc_state; + self.pc_internal.pc_state = PCState::Connected; + if old_state == PCState::New { + // TODO(theomonnom) OnConnected + } + } else if state == PeerConnectionState::Failed { + self.pc_internal.pc_state = PCState::Disconnected; + // TODO(theomonnom) Handle Disconnect + } }, Some(state) = self.pc_internal.secondary_connection_state_rx.recv() => { - + if state == PeerConnectionState::Failed { + self.pc_internal.pc_state = PCState::Disconnected; + // TODO(theomonnom) Handle Disconnect + } }, Some(data) = self.pc_internal.lossy_data_rx.recv() => { @@ -146,7 +224,10 @@ impl RTCInternal { } } - fn configure(lk_runtime: Arc, join: JoinResponse) -> Result { + fn configure( + lk_runtime: Arc, + join: JoinResponse, + ) -> Result { let cfg = RTCConfiguration { ice_servers: { let mut servers = vec![]; @@ -175,50 +256,62 @@ impl RTCInternal { 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); - })); + 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); - })); + 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; + let mut primary_pc = &mut publisher_pc; + let mut secondary_pc = &mut subscriber_pc; if join.subscriber_primary { - primary_pc = &subscriber_pc; - secondary_pc = &publisher_pc; + primary_pc = &mut subscriber_pc; + secondary_pc = &mut publisher_pc; } - primary_pc.peer_connection().on_connection_change(Box::new(move |state| { - let _ = primary_connection_state_tx.blocking_send(state); - })); + 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); - })); + 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 lossy_dc = publisher_pc.peer_connection().create_data_channel( + LOSSY_DC_LABEL, + DataChannelInit { + ordered: true, + max_retransmits: Some(0), + ..DataChannelInit::default() + }, + )?; - 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 - })?; + let mut reliable_dc = publisher_pc.peer_connection().create_data_channel( + RELIABLE_DC_LABEL, + DataChannelInit { + ordered: true, + ..DataChannelInit::default() + }, + )?; - lossy_dc.on_message(Box::new(|data, binary| { + lossy_dc.on_message(Box::new(move |data, _| { if let Ok(data) = DataPacket::decode(data) { let _ = lossy_data_tx.blocking_send(data); } else { @@ -226,7 +319,7 @@ impl RTCInternal { } })); - reliable_dc.on_message(Box::new(|data, binary| { + reliable_dc.on_message(Box::new(move |data, _| { if let Ok(data) = DataPacket::decode(data) { let _ = reliable_data_tx.blocking_send(data); } else { @@ -246,6 +339,7 @@ impl RTCInternal { secondary_connection_state_rx, lossy_data_rx, reliable_data_rx, + pc_state: PCState::New, }) } } @@ -253,8 +347,14 @@ impl RTCInternal { pub struct RTCEngine {} /// Initialize the SignalClient & the PeerConnections -//pub async fn connect(url: &str, token: &str) -> Result { -//} +pub async fn connect(url: &str, token: &str) -> Result { + let mut rtc_internal = RTCInternal::connect(url, token).await?; + tokio::spawn(async move { + rtc_internal.run().await + }); + + Ok(RTCEngine{}) +} impl RTCEngine { async fn rtc_handle() { @@ -266,7 +366,11 @@ impl RTCEngine { async fn test_test() { env_logger::init(); - //engine.connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjQ1OTY4MDYsImlzcyI6IkFQSUNrSG04M01oZ2hQeCIsIm5hbWUiOiJ1c2VyMSIsIm5iZiI6MTY2MDk5NjgwNiwic3ViIjoidXNlcjEiLCJ2aWRlbyI6eyJyb29tIjoibXktZmlyc3Qtcm9vbSIsInJvb21Kb2luIjp0cnVlfX0.SWU_LETMK6ZmFOf38pYjVhpur0o7jJc6u61h8BH7g20").await.unwrap(); + let engine = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap(); + + + sleep(Duration::from_secs(60)).await; + } /*sync fn handle_rtc(mut signal_receiver: broadcast::Receiver) { diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h index 60f0220..7c1d044 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h @@ -26,6 +26,8 @@ class IceCandidate { std::unique_ptr ice_candidate_; }; +std::unique_ptr create_ice_candidate(rust::String sdp_mid, int sdp_mline_index, rust::String sdp); + static std::unique_ptr _unique_ice_candidate() { return nullptr; // Ignore } @@ -43,6 +45,8 @@ class SessionDescription { std::unique_ptr session_description_; }; +std::unique_ptr create_session_description(SdpType type, rust::String sdp); + static std::unique_ptr _unique_session_description() { return nullptr; // Ignore } 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 6bf5796..73a696b 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h @@ -21,6 +21,8 @@ enum class PeerConnectionState; enum class SignalingState; enum class IceConnectionState; enum class IceGatheringState; +enum class SdpType; +struct SdpParseError; struct RTCOfferAnswerOptions; struct RTCError; struct DataChannelInit; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp index 418d308..5ed542a 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp @@ -4,14 +4,23 @@ #include "livekit/jsep.h" +#include #include #include "libwebrtc-sys/src/jsep.rs.h" #include "livekit/rtc_error.h" -#include "rtc_base/ref_counted_object.h" namespace livekit { +const std::string& serialize_sdp_error(webrtc::SdpParseError error) { + std::stringstream ss; + ss << std::hex << std::setfill('0'); + ss << std::setw(8) << (uint32_t)error.line.length(); + ss << std::dec << std::setw(1) << error.line; + ss << std::dec << std::setw(1) << error.description; + return ss.str(); +} + IceCandidate::IceCandidate( std::unique_ptr ice_candidate) : ice_candidate_(std::move(ice_candidate)) {} @@ -26,6 +35,20 @@ std::unique_ptr IceCandidate::release() { return std::move(ice_candidate_); } +std::unique_ptr create_ice_candidate(rust::String sdp_mid, + int sdp_mline_index, + rust::String sdp) { + webrtc::SdpParseError error; + auto ice_rtc = webrtc::CreateIceCandidate(sdp_mid.c_str(), sdp_mline_index, + sdp.c_str(), &error); + if (!ice_rtc) { + throw std::runtime_error(serialize_sdp_error(error)); + } + + return std::make_unique( + std::unique_ptr(ice_rtc)); +} + SessionDescription::SessionDescription( std::unique_ptr session_description) : session_description_(std::move(session_description)) {} @@ -45,6 +68,19 @@ SessionDescription::release() { return std::move(session_description_); } +std::unique_ptr create_session_description( + SdpType type, + rust::String sdp) { + webrtc::SdpParseError error; + auto rtc_sdp = webrtc::CreateSessionDescription( + static_cast(type), sdp.c_str(), &error); + if (!rtc_sdp) { + throw std::runtime_error(serialize_sdp_error(error)); + } + + return std::make_unique(std::move(rtc_sdp)); +} + // CreateSdpObserver NativeCreateSdpObserver::NativeCreateSdpObserver( diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs index 9be0d9f..0a1cc4f 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs @@ -1,4 +1,6 @@ -use std::fmt::{Debug, Formatter}; +use std::error::Error; +use std::fmt::{Debug, Display, Formatter}; +use std::str::FromStr; use cxx::UniquePtr; @@ -6,6 +8,21 @@ use crate::rtc_error::ffi::RTCError; #[cxx::bridge(namespace = "livekit")] pub mod ffi { + #[derive(Debug)] + #[repr(i32)] + pub enum SdpType { + Offer, + PrAnswer, + Answer, + Rollback, + } + + #[derive(Debug)] + pub struct SdpParseError { + pub line: String, + pub description: String, + } + extern "Rust" { type CreateSdpObserverWrapper; fn on_success( @@ -47,11 +64,22 @@ pub mod ffi { observer: Box, ) -> UniquePtr; + fn create_ice_candidate(sdp_mid: String, sdp_mline_index: i32, sdp: String) -> Result>; + fn create_session_description(sdp_type: SdpType, sdp: String) -> Result>; + fn _unique_ice_candidate() -> UniquePtr; // Ignore fn _unique_session_description() -> UniquePtr; // Ignore } } +impl Error for ffi::SdpParseError {} + +impl Display for ffi::SdpParseError { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + write!(f, "SdpParseError occurred {}: {}", self.line, self.description) + } +} + impl Debug for ffi::SessionDescription { fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { write!(f, "{}", self.stringify()) @@ -59,6 +87,7 @@ impl Debug for ffi::SessionDescription { } unsafe impl Send for ffi::SessionDescription {} +unsafe impl Sync for ffi::SessionDescription {} impl Debug for ffi::IceCandidate { fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { @@ -67,9 +96,38 @@ impl Debug for ffi::IceCandidate { } unsafe impl Send for ffi::IceCandidate {} - unsafe impl Sync for ffi::IceCandidate {} +impl ffi::SdpParseError { + /// # Safety + /// The value must be correctly encoded + pub unsafe fn from(value: &str) -> Self { + // Parse the hex encoded error from c++ + let line_length = u32::from_str_radix(&value[0..8], 16).unwrap() as usize + 8; + let line = String::from(&value[8..line_length]); + let description = String::from(&value[line_length..]); + + Self { + line, + description, + } + } +} + +impl FromStr for ffi::SdpType { + type Err = (); + + fn from_str(s: &str) -> Result { + match s { + "offer" => Ok(ffi::SdpType::Offer), + "pranswer" => Ok(ffi::SdpType::PrAnswer), + "answer" => Ok(ffi::SdpType::Answer), + "rollback" => Ok(ffi::SdpType::Rollback), + _ => Err(()), + } + } +} + // CreateSdpObserver pub trait CreateSdpObserver: Send { @@ -134,3 +192,37 @@ impl SetRemoteSdpObserverWrapper { self.observer.on_set_remote_description_complete(error); } } + +#[cfg(test)] +mod tests { + use log::info; + + use crate::jsep::ffi; + + #[test] + fn throw_error() { + let sdp_string = "v=0 +o=- 6549709950142776241 2 IN IP4 127.0.0.1 +s=- +t=0 0 +======================== ERROR HERE +a=group:BUNDLE 0 +a=extmap-allow-mixed +a=msid-semantic: WMS +m=application 9 UDP/DTLS/SCTP webrtc-datachannel +c=IN IP4 0.0.0.0 +a=ice-ufrag:Tw7h +a=ice-pwd:6XOVUD6HpcB4c1M8EB8jXJE9 +a=ice-options:trickle +a=fingerprint:sha-256 4F:EC:23:59:5D:A5:E6:3E:3E:5D:8A:09:B6:FA:04:AA:19:99:49:67:BD:65:93:06:BB:EE:AC:D5:21:0F:57:D6 +a=setup:actpass +a=mid:0 +a=sctp-port:5000 +a=max-message-size:262144 +"; + + let sdp = ffi::create_session_description(ffi::SdpType::Offer, sdp_string.to_string()); + let err = unsafe { ffi::SdpParseError::from(sdp.err().unwrap().what()) }; + info!("parse err: {:?}", err) + } +} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index fd98852..4e5f10c 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs @@ -253,11 +253,23 @@ pub mod ffi { } // https://webrtc.github.io/webrtc-org/native-code/native-apis/ +unsafe impl Send for ffi::PeerConnection {} unsafe impl Sync for ffi::PeerConnection {} -unsafe impl Send for ffi::PeerConnection {} - unsafe impl Send for ffi::NativePeerConnectionObserver {} +unsafe impl Sync for ffi::NativePeerConnectionObserver {} + +unsafe impl Sync for ffi::NativeAddIceCandidateObserver {} +unsafe impl Send for ffi::NativeAddIceCandidateObserver {} + +unsafe impl Sync for ffi::NativeSetRemoteSdpObserverHandle {} +unsafe impl Send for ffi::NativeSetRemoteSdpObserverHandle {} + +unsafe impl Sync for ffi::NativeSetLocalSdpObserverHandle {} +unsafe impl Send for ffi::NativeSetLocalSdpObserverHandle {} + +unsafe impl Sync for ffi::NativeCreateSdpObserverHandle {} +unsafe impl Send for ffi::NativeCreateSdpObserverHandle {} impl Default for ffi::RTCOfferAnswerOptions { /* @@ -280,17 +292,21 @@ impl Default for ffi::RTCOfferAnswerOptions { } } +pub trait AddIceCandidateObserver: Send { + fn on_complete(&self, error: RTCError); +} + pub struct AddIceCandidateObserverWrapper { - observer: Box, + observer: Box, } impl AddIceCandidateObserverWrapper { - pub fn new(observer: Box) -> Self { + pub fn new(observer: Box) -> Self { Self { observer } } fn on_complete(&self, error: RTCError) { - (self.observer)(error); + self.observer.on_complete(error); } } diff --git a/crates/livekit-webrtc/src/jsep.rs b/crates/livekit-webrtc/src/jsep.rs index 4d4c25f..75e80ed 100644 --- a/crates/livekit-webrtc/src/jsep.rs +++ b/crates/livekit-webrtc/src/jsep.rs @@ -1,7 +1,8 @@ use cxx::UniquePtr; - use libwebrtc_sys::jsep as sys_jsep; +pub use sys_jsep::ffi::{SdpType, SdpParseError}; + // TODO Maybe we can replace that by a serialized IceCandidateInit #[derive(Debug)] pub struct IceCandidate { @@ -9,6 +10,15 @@ pub struct IceCandidate { } impl IceCandidate { + pub fn from(sdp_mid: &str, sdp_mline_index: i32, sdp: &str) -> Result { + let res = sys_jsep::ffi::create_ice_candidate(sdp_mid.to_string(), sdp_mline_index, sdp.to_string()); + + match res { + Ok(cxx_handle) => Ok(IceCandidate::new(cxx_handle)), + Err(e) => Err(unsafe { SdpParseError::from(e.what()) }), + } + } + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { Self { cxx_handle } } @@ -30,6 +40,15 @@ pub struct SessionDescription { } impl SessionDescription { + pub fn from(sdp_type: SdpType, description: &str) -> Result { + let res = sys_jsep::ffi::create_session_description(sdp_type, description.to_string()); + + match res { + Ok(cxx_handle) => Ok(SessionDescription::new(cxx_handle)), + Err(e) => Err(unsafe { SdpParseError::from(e.what()) }), + } + } + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { Self { cxx_handle } } diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index 909a7ff..a4860aa 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -1,9 +1,7 @@ -use std::fmt::Debug; use std::sync::{Arc, Mutex}; use cxx::UniquePtr; use log::trace; -use thiserror::Error; use tokio::sync::mpsc; use libwebrtc_sys::data_channel as sys_dc; @@ -22,14 +20,6 @@ use crate::rtc_error::RTCError; use crate::rtp_receiver::RtpReceiver; use crate::rtp_transceiver::RtpTransceiver; -#[derive(Error, Debug)] -pub enum SdpError { - #[error("recv failure: {0}")] - RecvError(String), - #[error("internal libwebrtc error")] - RTCError(#[from] RTCError), -} - pub struct PeerConnection { cxx_handle: UniquePtr, observer: Box, @@ -51,7 +41,7 @@ impl PeerConnection { } } - pub async fn create_offer(&mut self, options: RTCOfferAnswerOptions) -> Result { + pub async fn create_offer(&mut self, options: RTCOfferAnswerOptions) -> Result { let (tx, mut rx) = mpsc::channel(1); let wrapper = @@ -65,13 +55,10 @@ impl PeerConnection { .create_offer(native_wrapper.pin_mut(), options); } - match rx.recv().await { - Some(value) => value.map_err(Into::into), - None => Err(SdpError::RecvError("channel closed".to_string())), - } + rx.recv().await.unwrap() } - pub async fn create_answer(&mut self, options: RTCOfferAnswerOptions) -> Result { + pub async fn create_answer(&mut self, options: RTCOfferAnswerOptions) -> Result { let (tx, mut rx) = mpsc::channel(1); let wrapper = @@ -85,16 +72,13 @@ impl PeerConnection { .create_answer(native_wrapper.pin_mut(), options); } - match rx.recv().await { - Some(value) => value.map_err(Into::into), - None => Err(SdpError::RecvError("channel closed".to_string())), - } + rx.recv().await.unwrap() } pub async fn set_local_description( &mut self, desc: SessionDescription, - ) -> Result<(), SdpError> { + ) -> Result<(), RTCError> { let (tx, mut rx) = mpsc::channel(1); let wrapper = sys_jsep::SetLocalSdpObserverWrapper::new(Box::new(InternalSetLocalSdpObserver { tx })); @@ -107,16 +91,13 @@ impl PeerConnection { .set_local_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())), - } + rx.recv().await.unwrap() } pub async fn set_remote_description( &mut self, desc: SessionDescription, - ) -> Result<(), SdpError> { + ) -> Result<(), RTCError> { let (tx, mut rx) = mpsc::channel(1); let wrapper = sys_jsep::SetRemoteSdpObserverWrapper::new(Box::new(InternalSetRemoteSdpObserver { @@ -131,10 +112,7 @@ impl PeerConnection { .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())), - } + rx.recv().await.unwrap() } pub fn create_data_channel( @@ -155,10 +133,10 @@ impl PeerConnection { } // TODO(theomonnom) Use IceCandidateInit instead of IceCandidate - pub async fn add_ice_candidate(&mut self, candidate: IceCandidate) -> Result<(), SdpError> { + pub async fn add_ice_candidate(&mut self, candidate: IceCandidate) -> Result<(), RTCError> { let (tx, mut rx) = mpsc::channel(1); - let observer = sys_pc::AddIceCandidateObserverWrapper::new(Box::new(move |error| { - tx.blocking_send(error).unwrap(); + let observer = sys_pc::AddIceCandidateObserverWrapper::new(Box::new(InternalAddIceCandidateObserver { + tx, })); let mut native_observer = @@ -167,10 +145,7 @@ impl PeerConnection { .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())), - } + rx.recv().await.unwrap() } pub fn local_description(&self) -> Option { @@ -313,6 +288,20 @@ impl PeerConnection { } } + +// SetLocalSdpObserver + +struct InternalAddIceCandidateObserver { + tx: mpsc::Sender>, +} + +impl sys_pc::AddIceCandidateObserver for InternalAddIceCandidateObserver { + fn on_complete(&self, error: RTCError) { + let res = if error.ok() { Ok(()) } else { Err(error) }; + let _ = self.tx.blocking_send(res); + } +} + // CreateSdpObserver struct InternalCreateSdpObserver { @@ -324,13 +313,11 @@ impl sys_jsep::CreateSdpObserver for InternalCreateSdpObserver { &self, session_description: UniquePtr, ) { - self.tx - .blocking_send(Ok(SessionDescription::new(session_description))) - .unwrap(); + let _ = self.tx.blocking_send(Ok(SessionDescription::new(session_description))); } fn on_failure(&self, error: RTCError) { - self.tx.blocking_send(Err(error)).unwrap(); + let _ = self.tx.blocking_send(Err(error)); } } @@ -343,7 +330,7 @@ struct InternalSetLocalSdpObserver { impl sys_jsep::SetLocalSdpObserver for InternalSetLocalSdpObserver { fn on_set_local_description_complete(&self, error: RTCError) { let res = if error.ok() { Ok(()) } else { Err(error) }; - self.tx.blocking_send(res).unwrap(); + let _ = self.tx.blocking_send(res); } } @@ -356,7 +343,7 @@ struct InternalSetRemoteSdpObserver { impl sys_jsep::SetRemoteSdpObserver for InternalSetRemoteSdpObserver { fn on_set_remote_description_complete(&self, error: RTCError) { let res = if error.ok() { Ok(()) } else { Err(error) }; - self.tx.blocking_send(res).unwrap(); + let _ = self.tx.blocking_send(res); } } @@ -634,6 +621,8 @@ impl sys_pc::PeerConnectionObserver for InternalObserver { mod tests { use log::trace; use tokio::sync::mpsc; + use libwebrtc_sys::peer_connection::ffi::RTCOfferAnswerOptions; + use libwebrtc_sys::peer_connection_factory::ffi::{ContinualGatheringPolicy, IceTransportsType}; use crate::data_channel::{DataChannel, DataChannelInit}; use crate::jsep::IceCandidate; @@ -657,6 +646,8 @@ mod tests { username: "".into(), password: "".into(), }], + continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, + ice_transport_type: IceTransportsType::All }; let mut bob = factory.create_peer_connection(config.clone()).unwrap(); @@ -682,12 +673,12 @@ mod tests { .create_data_channel("test_dc", DataChannelInit::default()) .unwrap(); - let offer = bob.create_offer().await.unwrap(); + let offer = bob.create_offer(RTCOfferAnswerOptions::default()).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(); + let answer = alice.create_answer(RTCOfferAnswerOptions::default()).await.unwrap(); trace!("Alice answer: {:?}", answer); alice.set_local_description(answer.clone()).await.unwrap(); bob.set_remote_description(answer).await.unwrap();