From 0d34e5a48e086014fc8ac2931fbc6e6751dbb07f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Thu, 29 Sep 2022 21:34:30 +0200 Subject: [PATCH] Started Room, publisher negotiation, add RTCRuntime dependencies across webrtc instances --- Cargo.lock | 15 + crates/livekit-core/Cargo.toml | 1 + crates/livekit-core/src/lib.rs | 3 +- crates/livekit-core/src/lk_runtime.rs | 10 +- crates/livekit-core/src/local_participant.rs | 47 ++ crates/livekit-core/src/pc_transport.rs | 31 +- crates/livekit-core/src/room.rs | 74 ++- crates/livekit-core/src/rtc_engine.rs | 484 +++++++++++++----- .../include/livekit/data_channel.h | 4 + .../libwebrtc-sys/include/livekit/jsep.h | 6 +- .../include/livekit/peer_connection.h | 8 +- .../include/livekit/peer_connection_factory.h | 11 +- .../include/livekit/rust_types.h | 1 + .../libwebrtc-sys/include/livekit/webrtc.h | 9 +- .../libwebrtc-sys/src/data_channel.cpp | 7 +- .../libwebrtc-sys/src/data_channel.rs | 1 + .../livekit-webrtc/libwebrtc-sys/src/jsep.cpp | 14 +- .../livekit-webrtc/libwebrtc-sys/src/jsep.rs | 3 + .../libwebrtc-sys/src/peer_connection.cpp | 24 +- .../libwebrtc-sys/src/peer_connection.rs | 4 + .../src/peer_connection_factory.cpp | 33 +- .../src/peer_connection_factory.rs | 3 +- .../libwebrtc-sys/src/webrtc.cpp | 27 +- .../libwebrtc-sys/src/webrtc.rs | 2 +- crates/livekit-webrtc/src/data_channel.rs | 33 +- crates/livekit-webrtc/src/jsep.rs | 12 + crates/livekit-webrtc/src/peer_connection.rs | 5 + .../src/peer_connection_factory.rs | 11 +- crates/livekit-webrtc/src/webrtc.rs | 9 +- 29 files changed, 707 insertions(+), 185 deletions(-) create mode 100644 crates/livekit-core/src/local_participant.rs diff --git a/Cargo.lock b/Cargo.lock index d20a7e5..8a3191f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -475,6 +475,7 @@ dependencies = [ "prost 0.11.0", "prost-build", "prost-types 0.11.1", + "serde", "serde_json", "thiserror", "tokio", @@ -912,6 +913,20 @@ name = "serde" version = "1.0.145" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.145" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81fa1584d3d1bcacd84c277a0dfe21f5b0f6accf4a23d04d4c6d61f1af522b4c" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] [[package]] name = "serde_json" diff --git a/crates/livekit-core/Cargo.toml b/crates/livekit-core/Cargo.toml index 8ddd856..893c954 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 = { version = "1", features = ["derive"] } serde_json = "1.0" log = "0.4" tokio-tungstenite = { version = "0.17.2", features = ["native-tls"] } diff --git a/crates/livekit-core/src/lib.rs b/crates/livekit-core/src/lib.rs index de8d910..c9bc7e1 100644 --- a/crates/livekit-core/src/lib.rs +++ b/crates/livekit-core/src/lib.rs @@ -3,8 +3,9 @@ pub mod proto { } mod lk_runtime; +mod signal_client; mod pc_transport; mod rtc_engine; -mod signal_client; +mod local_participant; pub mod room; diff --git a/crates/livekit-core/src/lk_runtime.rs b/crates/livekit-core/src/lk_runtime.rs index 041158b..269dce0 100644 --- a/crates/livekit-core/src/lk_runtime.rs +++ b/crates/livekit-core/src/lk_runtime.rs @@ -3,17 +3,21 @@ use log::trace; use livekit_webrtc::peer_connection_factory::PeerConnectionFactory; use livekit_webrtc::webrtc::RTCRuntime; +/// SAFETY: The order of initialization and deletion is important for LKRuntime. +/// See the C++ constructors & destructor of these fields + pub struct LKRuntime { - pub rtc_runtime: RTCRuntime, pub pc_factory: PeerConnectionFactory, + pub rtc_runtime: RTCRuntime, } impl LKRuntime { pub fn new() -> Self { trace!("LKRuntime::new()"); + let rtc_runtime = RTCRuntime::new(); Self { - rtc_runtime: RTCRuntime::new(), - pc_factory: PeerConnectionFactory::new(), + pc_factory: PeerConnectionFactory::new(rtc_runtime.clone()), + rtc_runtime, } } } diff --git a/crates/livekit-core/src/local_participant.rs b/crates/livekit-core/src/local_participant.rs new file mode 100644 index 0000000..248b2d0 --- /dev/null +++ b/crates/livekit-core/src/local_participant.rs @@ -0,0 +1,47 @@ +use std::sync::Arc; + +use futures_util::TryFutureExt; +use tokio::sync::Mutex; + +use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket}; +use crate::room::RoomError; +use crate::rtc_engine::RTCEngine; + +pub struct LocalParticipant { + sid: String, + identity: String, + name: String, + + engine: Arc>, +} + +impl LocalParticipant { + pub(crate) fn from(info: ParticipantInfo, engine: Arc>) -> Self { + Self { + sid: info.sid, + identity: info.identity, + name: info.name, + engine, + } + } + + pub(crate) fn update(info: ParticipantInfo) { + // TODO(theomonnom) + } + + // TODO(theomonnom) Add the destinations parameter + pub async fn publish_data(&mut self, data: &[u8], kind: data_packet::Kind) -> Result<(), RoomError> { + let data = DataPacket { + kind: kind as i32, + value: Some(data_packet::Value::User(UserPacket { + participant_sid: self.sid.clone(), + payload: data.to_vec(), + destination_sids: vec![], // TODO(theomonnom) + })), + }; + + self.engine.lock().await.publish_data(&data, kind).await.map_err(Into::into) + } +} + + diff --git a/crates/livekit-core/src/pc_transport.rs b/crates/livekit-core/src/pc_transport.rs index 338cc4d..2431e78 100644 --- a/crates/livekit-core/src/pc_transport.rs +++ b/crates/livekit-core/src/pc_transport.rs @@ -1,20 +1,18 @@ -use std::sync::Arc; +use std::future::Future; +use std::pin::Pin; use std::time::Duration; use log::{error, trace}; use livekit_webrtc::jsep::{IceCandidate, SessionDescription}; use livekit_webrtc::peer_connection::{ - PeerConnection, RTCOfferAnswerOptions, SignalingState, + IceConnectionState, PeerConnection, RTCOfferAnswerOptions, 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 type OnOfferHandler = Box Pin + Send + 'static>>) + Send + Sync>; pub struct PCTransport { peer_connection: PeerConnection, @@ -25,16 +23,20 @@ pub struct PCTransport { } impl PCTransport { - pub fn new(lk_runtime: Arc, cfg: RTCConfiguration) -> Result { - let peer_connection = lk_runtime.pc_factory.create_peer_connection(cfg)?; - - Ok(Self { + pub fn new(peer_connection: PeerConnection) -> Self { + Self { peer_connection, pending_candidates: Vec::default(), on_offer_handler: None, restarting_ice: false, renegotiate: false, - }) + } + } + + pub fn is_connected(&self) -> bool { + self.peer_connection.ice_connection_state() == IceConnectionState::IceConnectionConnected + || self.peer_connection.ice_connection_state() + == IceConnectionState::IceConnectionCompleted } pub fn peer_connection(&mut self) -> &mut PeerConnection { @@ -51,7 +53,9 @@ impl PCTransport { return Ok(()); } - self.peer_connection.add_ice_candidate(ice_candidate).await?; + self.peer_connection + .add_ice_candidate(ice_candidate) + .await?; Ok(()) } @@ -116,7 +120,8 @@ impl PCTransport { self.peer_connection .set_local_description(offer.clone()) .await?; - self.on_offer_handler.as_mut().unwrap()(offer); + self.on_offer_handler.as_mut().unwrap()(offer).await; Ok(()) } } + diff --git a/crates/livekit-core/src/room.rs b/crates/livekit-core/src/room.rs index 8be9529..b044574 100644 --- a/crates/livekit-core/src/room.rs +++ b/crates/livekit-core/src/room.rs @@ -1 +1,73 @@ -pub struct Room {} +use std::sync::Arc; +use std::time::Duration; +use log::trace; + +use thiserror::Error; +use tokio::sync::Mutex; +use tokio::time::sleep; + +use crate::local_participant::LocalParticipant; +use crate::proto::data_packet; +use crate::proto::signal_request::Message::Mute; +use crate::rtc_engine; +use crate::rtc_engine::{EngineError, RTCEngine}; + +#[derive(Error, Debug)] +pub enum RoomError { + #[error("internal RTCEngine failure")] + Engine(#[from] EngineError), +} + +pub struct Room { + sid: String, + name: String, + local_participant: LocalParticipant, + engine: Arc>, +} + +pub async fn connect(url: &str, token: &str) -> Result { + let engine = rtc_engine::connect(url, token).await?; + + engine.on_data(Box::new(|packet| { + trace!("This is a test"); + + Box::pin(async move { + trace!("Another test"); + }) + })).await; + + let join = engine.join_response().await; + let engine = Arc::new(Mutex::new(engine)); + let lp = LocalParticipant::from(join.participant.unwrap(), engine.clone()); + + let room_info = join.room.unwrap(); + Ok(Room { + sid: room_info.sid, + name: room_info.name, + local_participant: lp, + engine, + }) +} + +impl Room { + pub fn local_participant(&mut self) -> &mut LocalParticipant { + &mut self.local_participant + } + + pub fn sid(&self) -> &str { + &self.sid + } + + pub fn name(&self) -> &str { + &self.name + } +} + + +#[tokio::test] +async fn test_test() { + env_logger::init(); + let mut room = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap(); + room.local_participant().publish_data(b"This is a test", data_packet::Kind::Reliable).await.unwrap(); + //sleep(Duration::from_secs(60)).await; +} diff --git a/crates/livekit-core/src/rtc_engine.rs b/crates/livekit-core/src/rtc_engine.rs index 1d0bd66..1156e30 100644 --- a/crates/livekit-core/src/rtc_engine.rs +++ b/crates/livekit-core/src/rtc_engine.rs @@ -1,16 +1,21 @@ +use std::future::Future; +use std::pin::Pin; use std::sync::{Arc, Weak}; -use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::time::Duration; use lazy_static::lazy_static; use log::{error, trace}; +use prost::Message; use thiserror::Error; use tokio::sync::{mpsc, Mutex}; -use tokio::time::sleep; +use tokio::time; -use livekit_webrtc::data_channel::{DataChannel, DataChannelInit}; +use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState}; use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription}; -use livekit_webrtc::peer_connection::{PeerConnectionState, RTCOfferAnswerOptions}; +use livekit_webrtc::peer_connection::{ + IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions, +}; use livekit_webrtc::peer_connection_factory::{ ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration, }; @@ -19,17 +24,35 @@ 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, signal_response, SignalResponse, SignalTarget, TrickleRequest}; +use crate::proto::{ + data_packet, DataPacket, JoinResponse, signal_request, signal_response, SignalTarget, + TrickleRequest, UserPacket, +}; +use crate::proto::data_packet::Value; use crate::signal_client::{SignalClient, SignalError}; +use serde::{Deserialize, Serialize}; const LOSSY_DC_LABEL: &str = "_lossy"; const RELIABLE_DC_LABEL: &str = "_reliable"; +const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); lazy_static! { // Share one LKRuntime across all RTCEngine instances static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); } +#[derive(Serialize, Deserialize)] +struct IceCandidateJSON { + sdpMid: String, + sdpMLineIndex: i32, + candidate: String, +} + +pub struct Packet { + pub data: UserPacket, + pub kind: data_packet::Kind, +} + #[derive(Error, Debug)] pub enum EngineError { #[error("signal failure")] @@ -40,6 +63,14 @@ pub enum EngineError { Parse(#[from] SdpParseError), #[error("serde error")] Serde(#[from] serde_json::Error), + #[error("failed to send data to the datachannel")] + Data(#[from] DataSendError), + #[error("connection error: {0}")] + Connection(String), + #[error("decode error")] + Decode(#[from] prost::DecodeError), + #[error("internal error: {0}")] + Internal(String), // Unexpected error } #[derive(Debug, Copy, Clone, PartialEq, Eq)] @@ -55,11 +86,11 @@ enum PCState { pub enum EngineMessage { IceCandidate { ice_candidate: IceCandidate, - publisher: bool + publisher: bool, }, ConnectionChange { state: PeerConnectionState, - primary: bool + primary: bool, }, PrimaryDataChannel { data_channel: DataChannel, @@ -70,27 +101,35 @@ pub enum EngineMessage { Data { data: Vec, binary: bool, - reliable: bool - } + reliable: bool, + }, } +pub type OnDataHandler = +Box Pin + Send + 'static>>) + Send + Sync>; + struct EngineInternal { publisher_pc: Arc>, subscriber_pc: Arc>, - lossy_dc: Arc>, reliable_dc: Arc>, msg_sender: mpsc::Sender, + join_response: Mutex, + pc_state: AtomicU8, + // PCState + has_published: AtomicBool, - pc_state: AtomicU8, // PCState + // Listeners + on_data_handler: Arc>>, } pub struct RTCEngine { + signal_client: Arc, + internal: Arc, + #[allow(unused)] lk_runtime: Arc, // Keep a reference while we're using the RTCEngine - signal_client: Arc, - internal: Arc } pub async fn connect(url: &str, token: &str) -> Result { @@ -109,7 +148,11 @@ pub async fn connect(url: &str, token: &str) -> Result { if let signal_response::Message::Join(join_response) = signal_client.recv().await? { trace!("received join_response: {:?}", join_response); let (sender, receiver) = mpsc::channel(8); - let internal = Arc::new(RTCEngine::configure(lk_runtime.clone(), sender, join_response.clone())?); + let internal = Arc::new(RTCEngine::configure( + lk_runtime.clone(), + sender, + join_response.clone(), + )?); if !join_response.subscriber_primary { internal.publisher_pc.lock().await.negotiate().await?; @@ -135,9 +178,93 @@ pub async fn connect(url: &str, token: &str) -> Result { } impl RTCEngine { + /// Send data to other participants in the Room + pub async fn publish_data( + &mut self, + data: &DataPacket, + kind: data_packet::Kind, + ) -> Result<(), EngineError> { + self.ensure_publisher_connected(kind).await?; - async fn send_data() { + self.data_channel(kind) + .lock() + .await + .send(&data.encode_to_vec(), true) + .map_err(Into::into) + } + /// Return the last JoinResponse from the server + pub async fn join_response(&self) -> JoinResponse { + self.internal.join_response.lock().await.clone() + } + + pub async fn on_data(&self, f: OnDataHandler) { + *self.internal.on_data_handler.lock().await = Some(f); + } + + fn data_channel(&self, kind: data_packet::Kind) -> &Arc> { + if kind == data_packet::Kind::Reliable { + &self.internal.reliable_dc + } else { + &self.internal.lossy_dc + } + } + + async fn ensure_publisher_connected( + &mut self, + kind: data_packet::Kind, + ) -> Result<(), EngineError> { + if !self.join_response().await.subscriber_primary { + return Ok(()); + } + + { + let mut publisher = self.internal.publisher_pc.lock().await; + if !publisher.is_connected() + && publisher.peer_connection().ice_connection_state() + != IceConnectionState::IceConnectionChecking + { + tokio::spawn({ + let rtc_internal = self.internal.clone(); + async move { + let _ = Self::negotiate_publisher(rtc_internal).await; + } + }); + } + } + + let dc = self.data_channel(kind); + { + let dc = self.data_channel(kind).lock().await; + if dc.state() == DataState::Open { + return Ok(()); + } + } + + let res = time::timeout(MAX_ICE_CONNECT_TIMEOUT, { + let internal = self.internal.clone(); + + async move { + let mut interval = time::interval(Duration::from_millis(50)); + + loop { + if internal.publisher_pc.lock().await.is_connected() && dc.lock().await.state() == DataState::Open { + break; + } + + interval.tick().await; + } + } + }) + .await; + + if res.is_err() { + Err(EngineError::Connection( + "could not establish publisher connection".to_string(), + )) + } else { + Ok(()) + } } fn send_request(msg: signal_request::Message, signal_client: Arc) { @@ -148,109 +275,229 @@ impl RTCEngine { }); } - async fn handle_signal(signal: signal_response::Message, signal_client: &Arc, rtc_internal: &Arc) -> Result<(), EngineError> { + async fn handle_signal( + signal: signal_response::Message, + signal_client: &Arc, + rtc_internal: &Arc, + ) -> Result<(), EngineError> { match signal { signal_response::Message::Answer(answer) => { + trace!("received answer for publisher: {:?}", answer); let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; - rtc_internal.publisher_pc.lock().await.set_remote_description(sdp).await?; - }, + rtc_internal + .publisher_pc + .lock() + .await + .set_remote_description(sdp) + .await?; + } signal_response::Message::Offer(offer) => { let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; let mut subscriber_pc = rtc_internal.subscriber_pc.lock().await; subscriber_pc.set_remote_description(sdp).await?; - let answer = subscriber_pc.peer_connection().create_answer(RTCOfferAnswerOptions::default()).await?; - subscriber_pc.peer_connection().set_local_description(answer.clone()).await?; + let answer = subscriber_pc + .peer_connection() + .create_answer(RTCOfferAnswerOptions::default()) + .await?; + subscriber_pc + .peer_connection() + .set_local_description(answer.clone()) + .await?; - Self::send_request(signal_request::Message::Answer(proto::SessionDescription { - r#type: "answer".to_string(), - sdp: answer.to_string(), - }), signal_client.clone()); - }, + Self::send_request( + signal_request::Message::Answer(proto::SessionDescription { + r#type: "answer".to_string(), + sdp: answer.to_string(), + }), + signal_client.clone(), + ); + } 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() - )?; + let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?; + let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?; + + trace!( + "received ice_candidate: {:?} (publisher: {:?})", + ice, + trickle.target + ); if trickle.target == SignalTarget::Publisher as i32 { - rtc_internal.publisher_pc.lock().await.add_ice_candidate(ice).await?; + rtc_internal + .publisher_pc + .lock() + .await + .add_ice_candidate(ice) + .await?; } else { - rtc_internal.subscriber_pc.lock().await.add_ice_candidate(ice).await?; + rtc_internal + .subscriber_pc + .lock() + .await + .add_ice_candidate(ice) + .await?; } } - _ => {}, + _ => {} } Ok(()) } - async fn handle_loop(mut receiver: mpsc::Receiver, signal_client: Arc, rtc_internal: Arc) { + async fn handle_message( + msg: EngineMessage, + signal_client: &Arc, + rtc_internal: &Arc, + ) -> Result<(), EngineError> { + match msg { + EngineMessage::IceCandidate { + ice_candidate, + publisher, + } => { + trace!( + "sending ice_candidate: {:?} (publisher: {:?})", + ice_candidate, + publisher + ); + + let json = serde_json::to_string(&IceCandidateJSON { + sdpMid: ice_candidate.sdp_mid(), + sdpMLineIndex: ice_candidate.sdp_mline_index(), + candidate: ice_candidate.candidate() + })?; + + // Send the ice_candidate to the server + Self::send_request( + signal_request::Message::Trickle(TrickleRequest { + candidate_init: json, + target: if publisher { + SignalTarget::Publisher + } else { + SignalTarget::Subscriber + } as i32, + }), + signal_client.clone(), + ); + } + EngineMessage::ConnectionChange { state, primary } => { + if primary && state == PeerConnectionState::Connected { + let old_state = rtc_internal.pc_state.load(Ordering::SeqCst); + rtc_internal + .pc_state + .store(PCState::Connected as u8, Ordering::SeqCst); + + if old_state == PCState::New as u8 { + // TODO(theomonnom) OnConnected + } + } else if state == PeerConnectionState::Failed { + rtc_internal + .pc_state + .store(PCState::Disconnected as u8, Ordering::SeqCst); + + // TODO(theomonnom) handle Disconnect + } + } + EngineMessage::PrimaryDataChannel { mut data_channel } => { + let reliable = data_channel.label() == RELIABLE_DC_LABEL; + Self::configure_dc(&mut data_channel, reliable, rtc_internal.msg_sender.clone()); + + trace!( + "received and using subscriber datachannel (reliable: {:?})", + reliable + ); + if reliable { + *rtc_internal.reliable_dc.lock().await = data_channel; + } else { + *rtc_internal.lossy_dc.lock().await = data_channel; + } + } + EngineMessage::PublisherOffer { offer } => { + trace!("received publisher offer: {:?}", offer); + // Send the offer to the server + Self::send_request( + signal_request::Message::Offer(proto::SessionDescription { + r#type: "offer".to_string(), + sdp: offer.to_string(), + }), + signal_client.clone(), + ); + } + EngineMessage::Data { + data, + binary, + reliable: _, + } => { + if !binary { + return Err(EngineError::Internal( + "text message aren't supported by LiveKit".to_string(), + )); + } + + let data = DataPacket::decode(&*data)?; + match data.value.unwrap() { + Value::User(user) => { + let mut handler = rtc_internal.on_data_handler.lock().await; + if let Some(f) = &mut *handler { + f(Packet { + data: user, + kind: data_packet::Kind::from_i32(data.kind).unwrap(), + }).await; + } + } + Value::Speaker(_) => { + // TODO(theomonnonm) + } + } + } + } + + Ok(()) + } + + async fn handle_loop( + mut receiver: mpsc::Receiver, + signal_client: Arc, + rtc_internal: Arc, + ) { loop { tokio::select! { Ok(signal) = signal_client.recv() => { + trace!("received signal: {:?}", signal); if let Err(err) = Self::handle_signal(signal, &signal_client, &rtc_internal).await { error!("failed to handle signal: {:?}", err); } }, Some(msg) = receiver.recv() => { - match msg { - EngineMessage::IceCandidate { ice_candidate, publisher } => { - trace!("received ice_candidate: {:?} (publisher: {:?})", ice_candidate, publisher); - // Send the ice_candidate to the server - Self::send_request(signal_request::Message::Trickle(TrickleRequest { - candidate_init: ice_candidate.to_string(), - target: if publisher {SignalTarget::Publisher} else {SignalTarget::Subscriber} as i32 - }), signal_client.clone()); - } - EngineMessage::ConnectionChange { state, primary } => { - if primary && state == PeerConnectionState::Connected { - let old_state = rtc_internal.pc_state.load(Ordering::SeqCst); - rtc_internal.pc_state.store(PCState::Connected as u8, Ordering::SeqCst); - - if old_state == PCState::New as u8 { - // TODO(theomonnom) OnConnected - } - } else if state == PeerConnectionState::Failed { - rtc_internal.pc_state.store(PCState::Disconnected as u8, Ordering::SeqCst); - // TODO(theomonnom) handle Disconnect - } - } - EngineMessage::PrimaryDataChannel { mut data_channel } => { - let reliable = data_channel.label() == RELIABLE_DC_LABEL; - Self::configure_dc(&mut data_channel, reliable, rtc_internal.msg_sender.clone()); - - trace!("received and using subscriber datachannel (reliable: {:?})", reliable); - if reliable { - *rtc_internal.reliable_dc.lock().await = data_channel; - } else { - *rtc_internal.lossy_dc.lock().await = data_channel; - } - } - EngineMessage::PublisherOffer { offer } => { - trace!("received publisher offer: {:?}", offer); - // Send the offer to the server - Self::send_request(signal_request::Message::Offer(proto::SessionDescription { - r#type: "offer".to_string(), - sdp: offer.to_string(), - }), signal_client.clone()); - } - EngineMessage::Data { data, binary, reliable } => {} + if let Err(err) = Self::handle_message(msg, &signal_client, &rtc_internal).await { + error!("failed to handle engine message: {:?}", err); } } } } } + async fn negotiate_publisher(rtc_internal: Arc) -> Result<(), EngineError> { + rtc_internal.has_published.store(true, Ordering::SeqCst); + if let Err(err) = rtc_internal.publisher_pc.lock().await.negotiate().await { + error!("failed to negotiate the publisher: {:?}", err); + Err(EngineError::Rtc(err)) + } else { + Ok(()) + } + } + /// This function is called on connect & on reconnect /// It creates the PeerConnections, the DataChannels & the libwebrtc listeners - fn configure(lk_runtime: Arc, sender: mpsc::Sender, join: JoinResponse) -> Result { + fn configure( + lk_runtime: Arc, + sender: mpsc::Sender, + join: JoinResponse, + ) -> Result { let rtc_config = RTCConfiguration { ice_servers: { let mut servers = vec![]; - for is in join.ice_servers { + for is in join.ice_servers.clone() { servers.push(ICEServer { urls: is.urls, username: is.username, @@ -263,15 +510,20 @@ impl RTCEngine { ice_transport_type: IceTransportsType::All, }; - let mut publisher_pc = PCTransport::new(lk_runtime.pc_factory.create_peer_connection(rtc_config.clone())?); - let mut subscriber_pc = PCTransport::new(lk_runtime.pc_factory.create_peer_connection(rtc_config)?); + let mut publisher_pc = PCTransport::new( + lk_runtime + .pc_factory + .create_peer_connection(rtc_config.clone())?, + ); + let mut subscriber_pc = + PCTransport::new(lk_runtime.pc_factory.create_peer_connection(rtc_config)?); publisher_pc.peer_connection().on_ice_candidate(Box::new({ let sender = sender.clone(); move |ice_candidate| { let _ = sender.blocking_send(EngineMessage::IceCandidate { ice_candidate, - publisher: true + publisher: true, }); } })); @@ -281,17 +533,21 @@ impl RTCEngine { move |ice_candidate| { let _ = sender.blocking_send(EngineMessage::IceCandidate { ice_candidate, - publisher: false + publisher: false, }); } })); - publisher_pc.on_offer(Box::new({ + publisher_pc.on_offer({ let sender = sender.clone(); - move |offer| { - let _ = sender.blocking_send(EngineMessage::PublisherOffer {offer}); - } - })); + Box::new(move |offer| { + let sender = sender.clone(); + + Box::pin(async move { + let _ = sender.send(EngineMessage::PublisherOffer { offer }).await; + }) + }) + }); let mut primary_pc = &mut publisher_pc; let mut secondary_pc = &mut subscriber_pc; @@ -299,25 +555,24 @@ impl RTCEngine { primary_pc = &mut subscriber_pc; secondary_pc = &mut publisher_pc; - primary_pc.peer_connection().on_data_channel(Box::new({{ + primary_pc.peer_connection().on_data_channel(Box::new({ let sender = sender.clone(); move |data_channel| { - let _ = sender.blocking_send(EngineMessage::PrimaryDataChannel {data_channel}); - } - }})); - } - - primary_pc - .peer_connection() - .on_connection_change(Box::new({ - let sender = sender.clone(); - move |state| { - let _ = sender.blocking_send(EngineMessage::ConnectionChange { - state, - primary: true - }); + let _ = + sender.blocking_send(EngineMessage::PrimaryDataChannel { data_channel }); } })); + } + + primary_pc.peer_connection().on_connection_change(Box::new({ + let sender = sender.clone(); + move |state| { + let _ = sender.blocking_send(EngineMessage::ConnectionChange { + state, + primary: true, + }); + } + })); secondary_pc .peer_connection() @@ -326,7 +581,7 @@ impl RTCEngine { move |state| { let _ = sender.blocking_send(EngineMessage::ConnectionChange { state, - primary: false + primary: false, }); } })); @@ -358,27 +613,26 @@ impl RTCEngine { subscriber_pc: Arc::new(Mutex::new(subscriber_pc)), lossy_dc: Arc::new(Mutex::new(lossy_dc)), reliable_dc: Arc::new(Mutex::new(reliable_dc)), + msg_sender: sender, + join_response: Mutex::new(join), pc_state: AtomicU8::new(PCState::New as u8), - msg_sender: sender + has_published: AtomicBool::new(false), + on_data_handler: Default::default(), }) } /// Map the libwebrtc listeners to a mpsc channel - fn configure_dc(data_channel: &mut DataChannel, reliable: bool, sender: mpsc::Sender) { + fn configure_dc( + data_channel: &mut DataChannel, + reliable: bool, + sender: mpsc::Sender, + ) { data_channel.on_message(Box::new(move |data, binary| { let _ = sender.blocking_send(EngineMessage::Data { data: data.to_vec(), reliable, - binary + binary, }); })); } } - -#[tokio::test] -async fn test_test() { - env_logger::init(); - let engine = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap(); - - sleep(Duration::from_secs(60)).await; -} 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 e5be7b6..38078ef 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h @@ -10,6 +10,7 @@ #include "api/data_channel_interface.h" #include "rust/cxx.h" #include "rust_types.h" +#include "webrtc.h" namespace livekit { using NativeDataChannelInit = webrtc::DataChannelInit; @@ -18,15 +19,18 @@ class NativeDataChannelObserver; class DataChannel { public: explicit DataChannel( + std::shared_ptr rtc_runtime, rtc::scoped_refptr data_channel); void register_observer(NativeDataChannelObserver& observer); void unregister_observer(); bool send(const DataBuffer& buffer); rust::String label() const; + DataState state() const; void close(); private: + std::shared_ptr rtc_runtime_; rtc::scoped_refptr data_channel_; }; diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h index 7c1d044..2436ec1 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h @@ -18,7 +18,11 @@ class IceCandidate { public: explicit IceCandidate( std::unique_ptr ice_candidate); - + + rust::String sdp_mid() const; + int sdp_mline_index() const; + rust::String candidate() const; // TODO(theomonnom) Return livekit::Candidate instead of rust::String + rust::String stringify() 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 e0f7e67..f929a0a 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h @@ -12,6 +12,7 @@ #include "jsep.h" #include "rust/cxx.h" #include "rust_types.h" +#include "webrtc.h" namespace livekit { class NativeAddIceCandidateObserver; @@ -19,6 +20,7 @@ class NativeAddIceCandidateObserver; class PeerConnection { public: explicit PeerConnection( + std::shared_ptr rtc_runtime, rtc::scoped_refptr peer_connection); void create_offer(NativeCreateSdpObserverHandle& observer, @@ -38,9 +40,11 @@ class PeerConnection { std::unique_ptr remote_description() const; SignalingState signaling_state() const; IceGatheringState ice_gathering_state() const; + IceConnectionState ice_connection_state() const; void close(); private: + std::shared_ptr rtc_runtime_; rtc::scoped_refptr peer_connection_; }; @@ -65,7 +69,7 @@ create_native_add_ice_candidate_observer( class NativePeerConnectionObserver : public webrtc::PeerConnectionObserver { public: - explicit NativePeerConnectionObserver( + explicit NativePeerConnectionObserver(std::shared_ptr rtc_runtime, rust::Box observer); void OnSignalingChange( @@ -126,11 +130,13 @@ class NativePeerConnectionObserver : public webrtc::PeerConnectionObserver { void OnInterestingUsage(int usage_pattern) override; private: + std::shared_ptr rtc_runtime_; rust::Box observer_; }; std::unique_ptr create_native_peer_connection_observer( + std::shared_ptr rtc_runtime, rust::Box observer); } // namespace livekit 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 4749b87..e296b73 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 @@ -8,6 +8,7 @@ #include "api/peer_connection_interface.h" #include "peer_connection.h" #include "rust_types.h" +#include "webrtc.h" namespace livekit { using NativeRTCConfiguration = @@ -15,21 +16,19 @@ using NativeRTCConfiguration = class PeerConnectionFactory { public: - PeerConnectionFactory(); + explicit PeerConnectionFactory(std::shared_ptr rtc_runtime); + ~PeerConnectionFactory(); std::unique_ptr create_peer_connection( std::unique_ptr config, NativePeerConnectionObserver& observer) const; private: - std::unique_ptr network_thread_; - std::unique_ptr worker_thread_; - std::unique_ptr signaling_thread_; - + std::shared_ptr rtc_runtime_; rtc::scoped_refptr peer_factory_; }; -std::unique_ptr create_peer_connection_factory(); +std::unique_ptr create_peer_connection_factory(std::shared_ptr rtc_runtime); std::unique_ptr create_rtc_configuration( RTCConfiguration conf); } // namespace livekit 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 73a696b..4f766a1 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h @@ -22,6 +22,7 @@ enum class SignalingState; enum class IceConnectionState; enum class IceGatheringState; enum class SdpType; +enum class DataState; struct SdpParseError; struct RTCOfferAnswerOptions; struct RTCError; diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/webrtc.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/webrtc.h index 0896939..11b4a6f 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/webrtc.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/webrtc.h @@ -22,13 +22,20 @@ class RTCRuntime { RTCRuntime(const RTCRuntime&) = delete; RTCRuntime& operator=(const RTCRuntime&) = delete; + rtc::Thread* network_thread() const; + rtc::Thread* worker_thread() const; + rtc::Thread* signaling_thread() const; + private: + std::unique_ptr network_thread_; + std::unique_ptr worker_thread_; + std::unique_ptr signaling_thread_; #ifdef WEBRTC_WIN rtc::WinsockInitializer winsock_; #endif }; -std::unique_ptr create_rtc_runtime(); +std::shared_ptr create_rtc_runtime(); } // namespace livekit diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp index 88df064..03bbe43 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp @@ -11,8 +11,9 @@ namespace livekit { DataChannel::DataChannel( + std::shared_ptr rtc_runtime, rtc::scoped_refptr data_channel) - : data_channel_(std::move(data_channel)) {} + : rtc_runtime_(std::move(rtc_runtime)), data_channel_(std::move(data_channel)) {} void DataChannel::register_observer(NativeDataChannelObserver& observer) { data_channel_->RegisterObserver(&observer); @@ -31,6 +32,10 @@ rust::String DataChannel::label() const { return data_channel_->label(); } +DataState DataChannel::state() const { + return static_cast(data_channel_->state()); +} + void DataChannel::close() { return data_channel_->Close(); } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs index 26c5991..b41c1e7 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs @@ -69,6 +69,7 @@ pub mod ffi { fn unregister_observer(self: Pin<&mut DataChannel>); fn send(self: Pin<&mut DataChannel>, data: &DataBuffer) -> bool; fn label(self: &DataChannel) -> String; + fn state(self: &DataChannel) -> DataState; fn close(self: Pin<&mut DataChannel>); fn create_data_channel_init(init: DataChannelInit) -> UniquePtr; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp index 5ed542a..20ba43d 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp @@ -12,7 +12,7 @@ namespace livekit { -const std::string& serialize_sdp_error(webrtc::SdpParseError error) { +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(); @@ -25,6 +25,18 @@ IceCandidate::IceCandidate( std::unique_ptr ice_candidate) : ice_candidate_(std::move(ice_candidate)) {} +rust::String IceCandidate::sdp_mid() const { + return ice_candidate_->sdp_mid(); +} + +int IceCandidate::sdp_mline_index() const { + return ice_candidate_->sdp_mline_index(); +} + +rust::String IceCandidate::candidate() const { + return stringify(); +} + rust::String IceCandidate::stringify() const { std::string str; ice_candidate_->ToString(&str); diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs index 0a1cc4f..4c6c0be 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs @@ -49,6 +49,9 @@ pub mod ffi { type NativeSetLocalSdpObserverHandle; type NativeSetRemoteSdpObserverHandle; + fn sdp_mid(self: &IceCandidate) -> String; + fn sdp_mline_index(self: &IceCandidate) -> i32; + fn candidate(self: &IceCandidate) -> String; fn stringify(self: &IceCandidate) -> String; fn stringify(self: &SessionDescription) -> String; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp index 1d020c8..5ebde8b 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp @@ -24,8 +24,10 @@ toNativeOfferAnswerOptions(const RTCOfferAnswerOptions& options) { } PeerConnection::PeerConnection( + std::shared_ptr rtc_runtime, rtc::scoped_refptr peer_connection) - : peer_connection_(std::move(peer_connection)) {} + : rtc_runtime_(std::move(rtc_runtime)), + peer_connection_(std::move(peer_connection)) {} void PeerConnection::create_offer( NativeCreateSdpObserverHandle& observer_handle, @@ -65,7 +67,7 @@ std::unique_ptr PeerConnection::create_data_channel( throw std::runtime_error(serialize_error(to_error(result.error()))); } - return std::make_unique(result.value()); + return std::make_unique(rtc_runtime_, result.value()); } void PeerConnection::add_ice_candidate( @@ -97,14 +99,19 @@ SignalingState PeerConnection::signaling_state() const { } IceGatheringState PeerConnection::ice_gathering_state() const { - return static_cast(peer_connection_->ice_gathering_state()); + return static_cast( + peer_connection_->ice_gathering_state()); +} + +IceConnectionState PeerConnection::ice_connection_state() const { + return static_cast( + peer_connection_->ice_connection_state()); } void PeerConnection::close() { peer_connection_->Close(); } - // AddIceCandidateObserver NativeAddIceCandidateObserver::NativeAddIceCandidateObserver( @@ -124,8 +131,9 @@ create_native_add_ice_candidate_observer( // PeerConnectionObserver NativePeerConnectionObserver::NativePeerConnectionObserver( + std::shared_ptr rtc_runtime, rust::Box observer) - : observer_(std::move(observer)) {} + : rtc_runtime_(std::move(rtc_runtime)), observer_(std::move(observer)) {} void NativePeerConnectionObserver::OnSignalingChange( webrtc::PeerConnectionInterface::SignalingState new_state) { @@ -144,7 +152,7 @@ void NativePeerConnectionObserver::OnRemoveStream( void NativePeerConnectionObserver::OnDataChannel( rtc::scoped_refptr data_channel) { - observer_->on_data_channel(std::make_unique(data_channel)); + observer_->on_data_channel(std::make_unique(rtc_runtime_, data_channel)); } void NativePeerConnectionObserver::OnRenegotiationNeeded() { @@ -255,7 +263,9 @@ void NativePeerConnectionObserver::OnInterestingUsage(int usage_pattern) { std::unique_ptr create_native_peer_connection_observer( + std::shared_ptr rtc_runtime, rust::Box observer) { - return std::make_unique(std::move(observer)); + return std::make_unique(rtc_runtime, + std::move(observer)); } } // namespace livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index 4e5f10c..93860df 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs @@ -109,6 +109,7 @@ pub mod ffi { type NativeSetRemoteSdpObserverHandle = crate::jsep::ffi::NativeSetRemoteSdpObserverHandle; type NativeDataChannelInit = crate::data_channel::ffi::NativeDataChannelInit; type SessionDescription = crate::jsep::ffi::SessionDescription; + type RTCRuntime = crate::webrtc::ffi::RTCRuntime; type NativeAddIceCandidateObserver; type NativePeerConnectionObserver; @@ -166,9 +167,12 @@ pub mod ffi { fn ice_gathering_state(self: &PeerConnection) -> IceGatheringState; + fn ice_connection_state(self: &PeerConnection) -> IceConnectionState; + fn close(self: Pin<&mut PeerConnection>); fn create_native_peer_connection_observer( + rtc_runtime: SharedPtr, observer: Box, ) -> UniquePtr; 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 07cb05c..040f5a3 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp @@ -4,6 +4,8 @@ #include "livekit/peer_connection_factory.h" +#include + #include "api/audio_codecs/builtin_audio_decoder_factory.h" #include "api/audio_codecs/builtin_audio_encoder_factory.h" #include "api/rtc_event_log/rtc_event_log_factory.h" @@ -16,24 +18,15 @@ namespace livekit { -PeerConnectionFactory::PeerConnectionFactory() { - rtc::LogMessage::LogToDebug(rtc::LS_INFO); +PeerConnectionFactory::PeerConnectionFactory( + std::shared_ptr rtc_runtime) + : rtc_runtime_(std::move(rtc_runtime)) { RTC_LOG(LS_INFO) << "PeerConnectionFactory::PeerConnectionFactory()"; - network_thread_ = rtc::Thread::CreateWithSocketServer(); - network_thread_->SetName("network_thread", &network_thread_); - network_thread_->Start(); - worker_thread_ = rtc::Thread::Create(); - worker_thread_->SetName("worker_thread", &worker_thread_); - worker_thread_->Start(); - signaling_thread_ = rtc::Thread::Create(); - signaling_thread_->SetName("signaling_thread", &signaling_thread_); - signaling_thread_->Start(); - webrtc::PeerConnectionFactoryDependencies dependencies; - dependencies.network_thread = network_thread_.get(); - dependencies.worker_thread = worker_thread_.get(); - dependencies.signaling_thread = signaling_thread_.get(); + dependencies.network_thread = rtc_runtime_->network_thread(); + dependencies.worker_thread = rtc_runtime_->worker_thread(); + dependencies.signaling_thread = rtc_runtime_->signaling_thread(); dependencies.task_queue_factory = webrtc::CreateDefaultTaskQueueFactory(); dependencies.event_log_factory = std::make_unique( dependencies.task_queue_factory.get()); @@ -56,6 +49,10 @@ PeerConnectionFactory::PeerConnectionFactory() { } } +PeerConnectionFactory::~PeerConnectionFactory() { + RTC_LOG(LS_INFO) << "PeerConnectionFactory::~PeerConnectionFactory()"; +} + std::unique_ptr PeerConnectionFactory::create_peer_connection( std::unique_ptr config, NativePeerConnectionObserver& observer) const { @@ -67,11 +64,11 @@ std::unique_ptr PeerConnectionFactory::create_peer_connection( throw std::runtime_error(serialize_error(to_error(result.error()))); } - return std::make_unique(result.value()); + return std::make_unique(rtc_runtime_, result.value()); } -std::unique_ptr create_peer_connection_factory() { - return std::make_unique(); +std::unique_ptr create_peer_connection_factory(std::shared_ptr rtc_runtime) { + return std::make_unique(std::move(rtc_runtime)); } std::unique_ptr create_rtc_configuration( 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 5c45e31..82bbebc 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs @@ -43,8 +43,9 @@ pub mod ffi { crate::peer_connection::ffi::NativePeerConnectionObserver; type PeerConnectionFactory; type NativeRTCConfiguration; + type RTCRuntime = crate::webrtc::ffi::RTCRuntime; - fn create_peer_connection_factory() -> UniquePtr; + fn create_peer_connection_factory(runtime: SharedPtr) -> UniquePtr; fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr; /// SAFETY diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp index 37665f1..0afc9d9 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp @@ -8,8 +8,19 @@ namespace livekit { RTCRuntime::RTCRuntime() { + rtc::LogMessage::LogToDebug(rtc::LS_INFO); RTC_LOG(LS_INFO) << "RTCRuntime()"; RTC_CHECK(rtc::InitializeSSL()) << "Failed to InitializeSSL()"; + + network_thread_ = rtc::Thread::CreateWithSocketServer(); + network_thread_->SetName("network_thread", &network_thread_); + network_thread_->Start(); + worker_thread_ = rtc::Thread::Create(); + worker_thread_->SetName("worker_thread", &worker_thread_); + worker_thread_->Start(); + signaling_thread_ = rtc::Thread::Create(); + signaling_thread_->SetName("signaling_thread", &signaling_thread_); + signaling_thread_->Start(); } RTCRuntime::~RTCRuntime() { @@ -17,7 +28,19 @@ RTCRuntime::~RTCRuntime() { RTC_CHECK(rtc::CleanupSSL()) << "Failed to CleanupSSL()"; } -std::unique_ptr create_rtc_runtime() { - return std::make_unique(); +rtc::Thread* RTCRuntime::network_thread() const { + return network_thread_.get(); +} + +rtc::Thread* RTCRuntime::worker_thread() const { + return worker_thread_.get(); +} + +rtc::Thread* RTCRuntime::signaling_thread() const { + return signaling_thread_.get(); +} + +std::shared_ptr create_rtc_runtime() { + return std::make_shared(); } } // namespace 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 index 74df992..71120e3 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs @@ -7,7 +7,7 @@ pub mod ffi { type RTCRuntime; - fn create_rtc_runtime() -> UniquePtr; + fn create_rtc_runtime() -> SharedPtr; } } diff --git a/crates/livekit-webrtc/src/data_channel.rs b/crates/livekit-webrtc/src/data_channel.rs index 5a0d012..95efa7c 100644 --- a/crates/livekit-webrtc/src/data_channel.rs +++ b/crates/livekit-webrtc/src/data_channel.rs @@ -1,11 +1,12 @@ -use std::fmt::{Debug, Formatter}; +use std::error::Error; +use std::fmt::{Debug, Display, Formatter}; use std::sync::{Arc, Mutex}; - use cxx::UniquePtr; use log::trace; use libwebrtc_sys::data_channel as sys_dc; -pub use sys_dc::ffi::Priority; + +pub use sys_dc::ffi::{Priority, DataState}; pub struct DataChannel { cxx_handle: UniquePtr, @@ -21,6 +22,17 @@ impl Debug for DataChannel { } } +#[derive(Debug)] +pub struct DataSendError; + +impl Display for DataSendError { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + write!(f, "failed to send data to the DataChannel") + } +} + +impl Error for DataSendError { } + impl DataChannel { pub(crate) fn new(cxx_handle: UniquePtr) -> Self { let mut observer = Box::new(InternalDataChannelObserver::default()); @@ -44,19 +56,28 @@ impl DataChannel { dc } - pub fn send(&mut self, data: &[u8], binary: bool) -> bool { + pub fn send(&mut self, data: &[u8], binary: bool) -> Result<(), DataSendError> { let buffer = sys_dc::ffi::DataBuffer { ptr: data.as_ptr(), len: data.len(), binary, }; - self.cxx_handle.pin_mut().send(&buffer) + + self.cxx_handle + .pin_mut() + .send(&buffer) + .then_some(()) + .ok_or(DataSendError {}) } pub fn label(&self) -> String { self.cxx_handle.label() } + pub fn state(&self) -> DataState { + self.cxx_handle.state() + } + pub fn close(&mut self) { self.cxx_handle.pin_mut().close(); } @@ -69,7 +90,7 @@ impl DataChannel { *self.observer.on_message_handler.lock().unwrap() = Some(handler); } - pub fn on_buffer(&mut self, handler: OnBufferedAmountChangeHandler) { + pub fn on_buffered_amount_change(&mut self, handler: OnBufferedAmountChangeHandler) { *self .observer .on_buffered_amount_change_handler diff --git a/crates/livekit-webrtc/src/jsep.rs b/crates/livekit-webrtc/src/jsep.rs index 75e80ed..bf2d40a 100644 --- a/crates/livekit-webrtc/src/jsep.rs +++ b/crates/livekit-webrtc/src/jsep.rs @@ -26,6 +26,18 @@ impl IceCandidate { pub(crate) fn release(self) -> UniquePtr { self.cxx_handle } + + pub fn sdp_mid(&self) -> String { + self.cxx_handle.sdp_mid() + } + + pub fn sdp_mline_index(&self) -> i32 { + self.cxx_handle.sdp_mline_index() + } + + pub fn candidate(&self) -> String { + self.cxx_handle.candidate() + } } impl ToString for IceCandidate { diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index a4860aa..9e248b3 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -25,6 +25,7 @@ pub struct PeerConnection { observer: Box, // Keep alive for C++ + #[allow(unused)] native_observer: UniquePtr, } @@ -174,6 +175,10 @@ impl PeerConnection { self.cxx_handle.ice_gathering_state() } + pub fn ice_connection_state(&self) -> IceConnectionState { + self.cxx_handle.ice_connection_state() + } + pub fn close(&mut self) { self.cxx_handle.pin_mut().close(); } diff --git a/crates/livekit-webrtc/src/peer_connection_factory.rs b/crates/livekit-webrtc/src/peer_connection_factory.rs index 5f37612..09f06ff 100644 --- a/crates/livekit-webrtc/src/peer_connection_factory.rs +++ b/crates/livekit-webrtc/src/peer_connection_factory.rs @@ -8,15 +8,18 @@ pub use sys_factory::ffi::{ use crate::peer_connection::{InternalObserver, PeerConnection}; use crate::rtc_error::RTCError; +use crate::webrtc::RTCRuntime; pub struct PeerConnectionFactory { cxx_handle: UniquePtr, + rtc_runtime: RTCRuntime, } impl PeerConnectionFactory { - pub fn new() -> Self { + pub fn new(rtc_runtime: RTCRuntime) -> Self { Self { - cxx_handle: sys_factory::ffi::create_peer_connection_factory(), + cxx_handle: sys_factory::ffi::create_peer_connection_factory(rtc_runtime.clone().release()), + rtc_runtime, } } @@ -28,8 +31,8 @@ impl PeerConnectionFactory { 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)), + let mut native_observer = sys_pc::ffi::create_native_peer_connection_observer(self.rtc_runtime.clone().release(), + Box::new(sys_pc::PeerConnectionObserverWrapper::new(&mut *observer)), ); let res = self diff --git a/crates/livekit-webrtc/src/webrtc.rs b/crates/livekit-webrtc/src/webrtc.rs index 5065a79..684bd4e 100644 --- a/crates/livekit-webrtc/src/webrtc.rs +++ b/crates/livekit-webrtc/src/webrtc.rs @@ -1,9 +1,10 @@ -use cxx::UniquePtr; +use cxx::{SharedPtr}; use libwebrtc_sys::webrtc as sys_rtc; +#[derive(Clone)] pub struct RTCRuntime { - cxx_handle: UniquePtr, + cxx_handle: SharedPtr, } impl RTCRuntime { @@ -12,4 +13,8 @@ impl RTCRuntime { cxx_handle: sys_rtc::ffi::create_rtc_runtime(), } } + + pub(crate) fn release(self) -> SharedPtr { + self.cxx_handle + } }