From 4606c0a9ec8bda529e355e5b34f878322167a66e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Thu, 15 Dec 2022 23:27:43 +0100 Subject: [PATCH] wip --- .../src/rtc_engine/engine_internal.rs | 637 +++++++++++++++++ .../livekit-core/src/rtc_engine/lk_runtime.rs | 10 +- crates/livekit-core/src/rtc_engine/mod.rs | 655 +----------------- .../src/rtc_engine/pc_transport.rs | 10 +- .../livekit-core/src/rtc_engine/rtc_events.rs | 56 +- crates/livekit-core/src/signal_client/mod.rs | 26 +- .../src/signal_client/signal_stream.rs | 9 +- 7 files changed, 720 insertions(+), 683 deletions(-) create mode 100644 crates/livekit-core/src/rtc_engine/engine_internal.rs diff --git a/crates/livekit-core/src/rtc_engine/engine_internal.rs b/crates/livekit-core/src/rtc_engine/engine_internal.rs new file mode 100644 index 0000000..eb7ca8e --- /dev/null +++ b/crates/livekit-core/src/rtc_engine/engine_internal.rs @@ -0,0 +1,637 @@ +use parking_lot::{Mutex, RwLock}; +use std::error; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::{Arc, Weak}; +use std::time::Duration; + +use tokio::sync::{mpsc, Mutex as AsyncMutex}; + +use lazy_static::lazy_static; +use prost::Message; +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use tokio::time::sleep; +use tracing::{debug, error, info, trace, warn}; + +use crate::{proto, signal_client}; +use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState}; +use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription}; +use livekit_webrtc::media_stream::MediaStream; +use livekit_webrtc::peer_connection::{ + IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions, +}; +use livekit_webrtc::peer_connection_factory::RTCConfiguration; +use livekit_webrtc::rtc_error::RTCError; +use livekit_webrtc::rtp_receiver::RtpReceiver; + +use crate::proto::data_packet::Value; +use crate::proto::{ + data_packet, signal_request, signal_response, DataPacket, JoinResponse, ParticipantUpdate, + SignalTarget, TrickleRequest, +}; +use crate::rtc_engine::lk_runtime::LKRuntime; +use crate::rtc_engine::pc_transport::PCTransport; +use crate::rtc_engine::rtc_events::{RTCEmitter, RTCEvent, RTCEvents}; +use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions}; +use std::cell::RefCell; + +use super::{rtc_events, EngineEvents}; +use super::{EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult}; +// +// TODO(theomonnom): Smarter retry intervals +pub(crate) const RECONNECT_ATTEMPTS: u32 = 10; +pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300); + +pub(crate) const LOSSY_DC_LABEL: &str = "_lossy"; +pub(crate) const RELIABLE_DC_LABEL: &str = "_reliable"; +pub(crate) 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(Debug, Copy, Clone, PartialEq, Eq)] +pub enum PCState { + New, + Connected, + Disconnected, + Reconnecting, + Closed, +} + +#[derive(Debug)] +pub struct SessionInfo { + url: String, + token: String, + options: SignalOptions, +} + +#[derive(Debug)] +pub struct EngineInternal { + lk_runtime: Arc, + signal_client: Arc, + session: Arc>>, + reconnecting: AtomicBool, + closed: AtomicBool, + engine_emitter: EngineEmitter, +} + +/// This struct holds a WebRTC session +/// The session changes at every reconnection +#[derive(Debug)] +pub struct RTCSession { + join_response: Mutex, + has_published: AtomicBool, + pc_state: AtomicU8, // Casted to PCState enum + + publisher_pc: AsyncMutex, + subscriber_pc: AsyncMutex, + + // Publisher data channels + // Used to send data to other participants ( The SFU forwards the messages ) + lossy_dc: Mutex, + reliable_dc: Mutex, + + // Subscriber data channels + // These fields are never used, we just keep a strong reference to them, + // so we can receive data from other participants + sub_reliable_dc: Mutex>, + sub_lossy_dc: Mutex>, +} + +#[derive(Serialize, Deserialize)] +#[allow(non_snake_case)] +struct IceCandidateJSON { + sdpMid: String, + sdpMLineIndex: i32, + candidate: String, +} + +impl RTCSession { + pub fn configure( + lk_runtime: Arc, + join_response: JoinResponse, + ) -> EngineResult<(Self, RTCEvents)> { + let (rtc_emitter, events) = mpsc::unbounded_channel(); + let rtc_config = RTCConfiguration::from(join_response.clone()); + + let mut publisher_pc = PCTransport::new( + lk_runtime + .pc_factory + .create_peer_connection(rtc_config.clone())?, + SignalTarget::Publisher, + ); + + let mut subscriber_pc = PCTransport::new( + lk_runtime + .pc_factory + .create_peer_connection(rtc_config.clone())?, + SignalTarget::Subscriber, + ); + + 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, + DataChannelInit { + ordered: true, + ..DataChannelInit::default() + }, + )?; + + rtc_events::forward_pc_events(&mut publisher_pc, rtc_emitter.clone()); + rtc_events::forward_pc_events(&mut subscriber_pc, rtc_emitter.clone()); + rtc_events::forward_dc_events(&mut lossy_dc, rtc_emitter.clone()); + rtc_events::forward_dc_events(&mut reliable_dc, rtc_emitter.clone()); + + Ok(( + Self { + join_response: Mutex::new(join_response), + has_published: AtomicBool::new(false), + pc_state: AtomicU8::new(PCState::New as u8), + publisher_pc: AsyncMutex::new(publisher_pc), + subscriber_pc: AsyncMutex::new(subscriber_pc), + lossy_dc: Mutex::new(lossy_dc), + reliable_dc: Mutex::new(reliable_dc), + sub_lossy_dc: Mutex::new(None), + sub_reliable_dc: Mutex::new(None), + }, + events, + )) + } + + async fn negotiate_publisher(&self) -> EngineResult<()> { + self.has_published.store(true, Ordering::SeqCst); + let res = self.publisher_pc.lock().await.negotiate().await; + if let Err(err) = &res { + error!("failed to negotiate the publisher: {:?}", err); + } + res.map_err(Into::into) + } + + async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> { + if !self.join_response.lock().subscriber_primary { + return Ok(()); + } + + let publisher = &self.publisher_pc; + { + let mut publisher = publisher.lock().await; + if !publisher.is_connected() + && publisher.peer_connection().ice_connection_state() + != IceConnectionState::IceConnectionChecking + { + let _ = self.negotiate_publisher().await; + } + } + + let dc = self.data_channel(kind); + if dc.lock().state() == DataState::Open { + return Ok(()); + } + + // Wait until the PeerConnection is connected + let wait_connected = async move { + while publisher.lock().await.is_connected() && dc.lock().state() == DataState::Open { + sleep(Duration::from_millis(50)).await; + } + }; + + tokio::select! { + _ = wait_connected => Ok(()), + _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { + let err = EngineError::Connection("could not establish publisher connection: timeout".to_string()); + error!(error = ?err); + Err(err) + } + } + } + + async fn wait_pc_connection(&self) -> EngineResult<()> { + let wait_connected = async move { + while self.pc_state.load(Ordering::SeqCst) != PCState::Connected as u8 { + sleep(Duration::from_millis(50)).await; + } + }; + + tokio::select! { + _ = wait_connected => Ok(()), + _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { + let err = EngineError::Connection("wait_pc_connection timed out".to_string()); + Err(err) + } + } + } + + fn data_channel(&self, kind: data_packet::Kind) -> &Mutex { + if kind == data_packet::Kind::Reliable { + &self.reliable_dc + } else { + &self.lossy_dc + } + } +} + +impl Default for EngineInternal { + fn default() -> Self { + let mut lk_runtime = None; + { + let mut lk_runtime_ref = LK_RUNTIME.lock(); + lk_runtime = lk_runtime_ref.upgrade(); + + if lk_runtime.is_none() { + let new_runtime = Arc::new(LKRuntime::default()); + *lk_runtime_ref = Arc::downgrade(&new_runtime); + lk_runtime = Some(new_runtime); + } + } + + Self { + lk_runtime: lk_runtime.unwrap(), + ..Default::default() + } + } +} + +impl EngineInternal { + #[tracing::instrument] + pub async fn connect( + self: Arc, + url: &str, + token: &str, + options: SignalOptions, + ) -> EngineResult { + let mut signal_events = self.signal_client.connect(url, token, options).await?; + let join_response = signal_client::utils::next_join_response(&mut signal_events).await?; + debug!("received JoinResponse: {:?}", join_response); + + let (session, rtc_events) = + RTCSession::configure(self.lk_runtime.clone(), join_response.clone())?; + let session = Arc::new(RwLock::new(Some(session))); + + let (engine_emitter, engine_events) = mpsc::channel(8); + + tokio::spawn(self.clone().signal_task(signal_events)); + tokio::spawn(self.clone().engine_task(rtc_events)); + + if !join_response.subscriber_primary { + session.read().unwrap().negotiate_publisher().await?; + } + + Ok(engine_events) + } + + async fn engine_task(self: Arc, mut rtc_events: RTCEvents) { + while let Some(event) = rtc_events.recv().await { + if let Err(err) = self.handle_rtc(event).await { + error!("failed to handle rtc event: {:?}", err); + } + } + } + + async fn signal_task(self: Arc, mut signal_events: SignalEvents) { + while let Some(signal) = signal_events.recv().await { + match signal { + SignalEvent::Open => {} + SignalEvent::Signal(signal) => { + if let Err(err) = self.handle_signal(signal).await { + error!("failed to handle signal: {:?}", err); + } + } + SignalEvent::Close => { + self.handle_disconnected(); + } + } + } + } + + async fn handle_rtc(self: &Arc, event: RTCEvent) -> EngineResult<()> { + match event { + RTCEvent::IceCandidate { + ice_candidate, + target, + } => { + let json = serde_json::to_string(&IceCandidateJSON { + sdpMid: ice_candidate.sdp_mid(), + sdpMLineIndex: ice_candidate.sdp_mline_index(), + candidate: ice_candidate.candidate(), + })?; + + trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate); + + tokio::spawn(async move { + signal_client + .send(signal_request::Message::Trickle(TrickleRequest { + candidate_init: json, + target: target as i32, + })) + .await; + }); + } + RTCEvent::ConnectionChange { state, target } => { + trace!("connection change, {:?} {:?}", state, target); + let subscriber_primary = session + .read() + .unwrap() + .join_response + .lock() + .subscriber_primary; + + let is_primary = subscriber_primary && target == SignalTarget::Subscriber; + + if is_primary && state == PeerConnectionState::Connected { + let old_state = session + .read() + .unwrap() + .pc_state + .swap(PCState::Connected as u8, Ordering::SeqCst); + if old_state == PCState::New as u8 { + let _ = emitter.send(EngineEvent::Connected).await; // First time connected + } + } else if state == PeerConnectionState::Failed { + session + .read() + .unwrap() + .pc_state + .store(PCState::Disconnected as u8, Ordering::SeqCst); + + Self::handle_disconnected(signal_client, engine_inner, emitter); + } + } + RTCEvent::DataChannel { + data_channel, + target, + } => { + if target == SignalTarget::Subscriber { + if data_channel.label() == RELIABLE_DC_LABEL { + *session.read().unwrap().sub_reliable_dc.lock() = Some(data_channel); + } else { + *session.read().unwrap().sub_lossy_dc.lock() = Some(data_channel); + } + } + } + RTCEvent::Offer { offer, target } => { + if target == SignalTarget::Publisher { + // Send the publisher offer to the server + tokio::spawn(async move { + signal_client + .send(signal_request::Message::Offer(proto::SessionDescription { + r#type: "offer".to_string(), + sdp: offer.to_string(), + })) + .await; + }); + } + } + RTCEvent::AddTrack { + rtp_receiver, + streams, + target, + } => { + if target == SignalTarget::Subscriber { + let _ = emitter + .send(EngineEvent::AddTrack { + rtp_receiver, + streams, + }) + .await; + } + } + RTCEvent::Data { data, binary } => { + if !binary { + Err(EngineError::Internal( + "text messages aren't supported".to_string(), + ))?; + } + + let data = DataPacket::decode(&*data)?; + match data.value.unwrap() { + Value::User(user) => { + // TODO(theomonnom) Send event + } + Value::Speaker(_) => { + // TODO(theomonnonm) + } + } + } + } + + Ok(()) + } + + async fn handle_signal(self: &Arc, event: signal_response::Message) -> EngineResult<()> { + match event { + signal_response::Message::Answer(answer) => { + trace!("received answer from the publisher: {:?}", answer); + + let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; + self.session + .read() + .unwrap() + .publisher_pc + .lock() + .await + .set_remote_description(sdp) + .await?; + } + signal_response::Message::Offer(offer) => { + // Handle the subscriber offer & send an answer to livekit-server + // We always get an offer from the server when connecting + trace!("received offer for the subscriber: {:?}", offer); + let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; + + let subscriber_pc = self + .session + .read() + .as_ref() + .unwrap() + .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?; + + tokio::spawn(async move { + self.signal_client + .send(signal_request::Message::Answer(proto::SessionDescription { + r#type: "answer".to_string(), + sdp: answer.to_string(), + })) + .await; + }); + } + signal_response::Message::Trickle(trickle) => { + // Add the IceCandidate received from the livekit-server + let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?; + let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?; + + trace!( + "received ice_candidate {:?} - {:?}", + SignalTarget::from_i32(trickle.target).unwrap(), + ice + ); + + if trickle.target == SignalTarget::Publisher as i32 { + self.session + .read() + .unwrap() + .publisher_pc + .lock() + .await + .add_ice_candidate(ice) + .await?; + } else { + self.session + .read() + .unwrap() + .subscriber_pc + .lock() + .await + .add_ice_candidate(ice) + .await?; + } + } + signal_response::Message::Update(update) => { + let _ = emitter.send(EngineEvent::ParticipantUpdate(update)).await; + } + _ => {} + } + + Ok(()) + } +} + +/// Reconnection Logic for the RTCEngine, it is responsable for: TODO +impl EngineInternal { + async fn handle_disconnected(self: &Arc) { + if self.closed.load(Ordering::SeqCst) || self.reconnecting.load(Ordering::SeqCst) { + return; + } + + self.reconnecting.store(true, Ordering::SeqCst); + warn!("RTCEngine disconnected unexpectedly, reconnecting..."); + + let mut full_reconnect = false; + for i in 0..RECONNECT_ATTEMPTS { + if full_reconnect { + if i == 0 { + let _ = emitter.send(EngineEvent::Restarting).await; + } + + info!("restarting connection... attempt: {}", i); + if let Err(err) = Self::try_restart_connection( + signal_client.clone(), + engine_inner.clone(), + emitter.clone(), + ) + .await + { + error!("restarting connection failed: {}", err); + } else { + return; + } + } else { + if i == 0 { + let _ = emitter.send(EngineEvent::Resuming).await; + } + + info!("resuming connection... attempt: {}", i); + if let Err(err) = Self::try_resume_connection( + signal_client.clone(), + engine_inner.clone(), + emitter.clone(), + ) + .await + { + error!("resuming connection failed: {}", err); + if let EngineError::Signal(_) = err { + full_reconnect = true; + } + } else { + return; + } + } + + tokio::time::sleep(RECONNECT_INTERVAL).await; + } + error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS); + self.reconnecting.store(false, Ordering::SeqCst); + + // TODO DISCONNECT + } + + async fn try_restart_connection( + self: Arc, + signal_client: Arc, + emitter: EngineEmitter, + ) -> EngineResult<()> { + Ok(()) + } + + async fn try_resume_connection( + self: Arc, + signal_client: Arc, + emitter: EngineEmitter, + ) -> EngineResult<()> { + let mut options = engine_inner.options.lock().clone(); + options.sid = engine_inner + .join_response + .lock() + .participant + .as_ref() + .unwrap() + .sid + .clone(); + + signal_client + .reconnect( + &engine_inner.url, + &engine_inner.token.lock().clone(), + options, + ) + .await?; + + let _ = emitter.send(EngineEvent::SignalResumed).await; + + engine_inner + .subscriber_pc + .lock() + .await + .prepare_ice_restart(); + + if engine_inner.has_published.load(Ordering::SeqCst) { + engine_inner + .publisher_pc + .lock() + .await + .create_and_send_offer(RTCOfferAnswerOptions { + ice_restart: true, + ..Default::default() + }) + .await?; + } + + Self::wait_pc_connection(engine_inner).await?; + signal_client.flush_queue().await; + + let _ = emitter.send(EngineEvent::Resumed); + + Ok(()) + } +} diff --git a/crates/livekit-core/src/rtc_engine/lk_runtime.rs b/crates/livekit-core/src/rtc_engine/lk_runtime.rs index 061455d..5ff17c6 100644 --- a/crates/livekit-core/src/rtc_engine/lk_runtime.rs +++ b/crates/livekit-core/src/rtc_engine/lk_runtime.rs @@ -1,6 +1,6 @@ use std::fmt::{Debug, Formatter}; -use tracing::{event, Level}; +use tracing::trace; use livekit_webrtc::peer_connection_factory::PeerConnectionFactory; use livekit_webrtc::webrtc::RTCRuntime; @@ -19,9 +19,9 @@ impl Debug for LKRuntime { } } -impl LKRuntime { - pub fn new() -> Self { - event!(Level::TRACE, "LKRuntime::new()"); +impl Default for LKRuntime { + fn default() -> Self { + trace!("LKRuntime::default()"); let rtc_runtime = RTCRuntime::new(); Self { pc_factory: PeerConnectionFactory::new(rtc_runtime.clone()), @@ -32,6 +32,6 @@ impl LKRuntime { impl Drop for LKRuntime { fn drop(&mut self) { - event!(Level::TRACE, "LKRuntime::drop()"); + trace!("LKRuntime::drop()"); } } diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index 95dba5d..a061d5a 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -34,44 +34,15 @@ use crate::rtc_engine::pc_transport::PCTransport; use crate::rtc_engine::rtc_events::{RTCEmitter, RTCEvent, RTCEvents}; use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions}; +mod engine_internal; mod lk_runtime; mod pc_transport; mod rtc_events; -lazy_static! { - // Share one LKRuntime across all RTCEngine instances - static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); -} - pub(crate) type EngineEmitter = mpsc::Sender; pub(crate) type EngineEvents = mpsc::Receiver; pub(crate) type EngineResult = Result; -// TODO(theomonnom): Smarter retry intervals -pub(crate) const RECONNECT_ATTEMPTS: u32 = 10; -pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300); - -pub(crate) const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); -pub(crate) const LOSSY_DC_LABEL: &str = "_lossy"; -pub(crate) const RELIABLE_DC_LABEL: &str = "_reliable"; - -#[derive(Debug, Copy, Clone, PartialEq, Eq)] -pub(crate) enum PCState { - New, - Connected, - Disconnected, - Reconnecting, - Closed, -} - -#[derive(Serialize, Deserialize)] -#[allow(non_snake_case)] -struct IceCandidateJSON { - sdpMid: String, - sdpMLineIndex: i32, - candidate: String, -} - #[derive(Error, Debug)] pub enum EngineError { #[error("signal failure: {0}")] @@ -93,7 +64,7 @@ pub enum EngineError { } #[derive(Debug)] -pub(crate) enum EngineEvent { +pub enum EngineEvent { ParticipantUpdate(ParticipantUpdate), AddTrack { rtp_receiver: RtpReceiver, @@ -107,115 +78,13 @@ pub(crate) enum EngineEvent { Restarted, } -#[derive(Debug)] -struct EngineInner { - // Join infornation - url: String, - token: Mutex, // The token is refreshed periodically - options: Mutex, - join_response: Mutex, - - has_published: AtomicBool, - pc_state: AtomicU8, // Casted to PCState enum - reconnecting: AtomicBool, - - publisher_pc: AsyncMutex, - subscriber_pc: AsyncMutex, - - // Publisher data channels - // Used to send data to other participants ( The SFU forward the messages ) - lossy_dc: Mutex, - reliable_dc: Mutex, - // Subscriber data channels - // These fields are never used, we just keep a strong reference to them, - // so we can receive data from other participants - sub_reliable_dc: Mutex>, - sub_lossy_dc: Mutex>, - - closed: AtomicBool, -} - #[derive(Debug)] pub struct RTCEngine { - signal_client: Arc, - engine_inner: Arc, - - #[allow(unused)] - lk_runtime: Arc, // Keep a reference while we're using the RTCEngine -} - -impl EngineInner { - async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> { - if !self.join_response.lock().subscriber_primary { - return Ok(()); - } - - let publisher = &self.publisher_pc; - { - let mut publisher = publisher.lock().await; - if !publisher.is_connected() - && publisher.peer_connection().ice_connection_state() - != IceConnectionState::IceConnectionChecking - { - let _ = self.negotiate_publisher().await; - } - } - - let dc = self.data_channel(kind); - if dc.lock().state() == DataState::Open { - return Ok(()); - } - - // Wait until the PeerConnection is connected - let wait_connected = async move { - while publisher.lock().await.is_connected() && dc.lock().state() == DataState::Open { - sleep(Duration::from_millis(50)).await; - } - }; - - tokio::select! { - _ = wait_connected => Ok(()), - _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { - let err = EngineError::Connection("could not establish publisher connection: timeout".to_string()); - error!(error = ?err); - Err(err) - } - } - } - - async fn negotiate_publisher(&self) -> EngineResult<()> { - self.has_published.store(true, Ordering::SeqCst); - if let Err(err) = self.publisher_pc.lock().await.negotiate().await { - error!("failed to negotiate the publisher: {:?}", err); - Err(err)? - } else { - Ok(()) - } - } - - fn data_channel(&self, kind: data_packet::Kind) -> &Mutex { - if kind == data_packet::Kind::Reliable { - &self.reliable_dc - } else { - &self.lossy_dc - } - } + engine_inner: Arc, } impl RTCEngine { pub fn new() -> Self { - - Self { - - } - } - - #[tracing::instrument(skip(url, token))] - pub(crate) async fn connect( - url: &str, - token: &str, - options: SignalOptions, - ) -> EngineResult<(RTCEngine, EngineEvents)> { let mut lk_runtime = None; { let mut lk_runtime_ref = LK_RUNTIME.lock(); @@ -228,7 +97,17 @@ impl RTCEngine { } } - let lk_runtime = lk_runtime.unwrap(); + let (signal_client, mut signal_events) = SignalClient::new(); + + Self { lk_runtime } + } + + #[tracing::instrument(skip(url, token))] + pub(crate) async fn connect( + url: &str, + token: &str, + options: SignalOptions, + ) -> EngineResult<(RTCEngine, EngineEvents)> { let (signal_client, mut signal_events) = SignalClient::connect(url, token, options).await?; let join_response = signal_client::utils::next_join_response(&mut signal_events).await?; @@ -286,513 +165,7 @@ impl RTCEngine { self.engine_inner.join_response.lock().clone() } - async fn engine_task( - signal_client: Arc, - engine_inner: Arc, - mut rtc_events: RTCEvents, - emitter: EngineEmitter, - ) { - while let Some(event) = rtc_events.recv().await { - if let Err(err) = Self::handle_rtc( - event, - signal_client.clone(), - engine_inner.clone(), - emitter.clone(), - ) - .await - { - error!("failed to handle rtc event: {:?}", err); - } - } - } - - async fn signal_task( - signal_client: Arc, - engine_inner: Arc, - mut signal_events: SignalEvents, - emitter: EngineEmitter, - ) { - while let Some(signal) = signal_events.recv().await { - match signal { - SignalEvent::Open => {} - SignalEvent::Signal(signal) => { - if let Err(err) = Self::handle_signal( - signal, - signal_client.clone(), - engine_inner.clone(), - emitter.clone(), - ) - .await - { - error!("failed to handle signal: {:?}", err); - } - } - SignalEvent::Close => { - Self::handle_disconnected( - signal_client.clone(), - engine_inner.clone(), - emitter.clone(), - ); - } - } - } - } - - async fn handle_rtc( - event: RTCEvent, - signal_client: Arc, - engine_inner: Arc, - emitter: EngineEmitter, - ) -> EngineResult<()> { - match event { - RTCEvent::IceCandidate { - ice_candidate, - target, - } => { - let json = serde_json::to_string(&IceCandidateJSON { - sdpMid: ice_candidate.sdp_mid(), - sdpMLineIndex: ice_candidate.sdp_mline_index(), - candidate: ice_candidate.candidate(), - })?; - - trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate); - - tokio::spawn(async move { - signal_client - .send(signal_request::Message::Trickle(TrickleRequest { - candidate_init: json, - target: target as i32, - })) - .await; - }); - } - RTCEvent::ConnectionChange { state, target } => { - trace!("connection change, {:?} {:?}", state, target); - let subscriber_primary = engine_inner.join_response.lock().subscriber_primary; - let is_primary = subscriber_primary && target == SignalTarget::Subscriber; - - if is_primary && state == PeerConnectionState::Connected { - let old_state = engine_inner - .pc_state - .swap(PCState::Connected as u8, Ordering::SeqCst); - if old_state == PCState::New as u8 { - let _ = emitter.send(EngineEvent::Connected).await; // First time connected - } - } else if state == PeerConnectionState::Failed { - engine_inner - .pc_state - .store(PCState::Disconnected as u8, Ordering::SeqCst); - - Self::handle_disconnected(signal_client, engine_inner, emitter); - } - } - RTCEvent::DataChannel { - data_channel, - target, - } => { - if target == SignalTarget::Subscriber { - if data_channel.label() == RELIABLE_DC_LABEL { - *engine_inner.sub_reliable_dc.lock() = Some(data_channel); - } else { - *engine_inner.sub_lossy_dc.lock() = Some(data_channel); - } - } - } - RTCEvent::Offer { offer, target } => { - if target == SignalTarget::Publisher { - // Send the publisher offer to the server - tokio::spawn(async move { - signal_client - .send(signal_request::Message::Offer(proto::SessionDescription { - r#type: "offer".to_string(), - sdp: offer.to_string(), - })) - .await; - }); - } - } - RTCEvent::AddTrack { - rtp_receiver, - streams, - target, - } => { - if target == SignalTarget::Subscriber { - let _ = emitter - .send(EngineEvent::AddTrack { - rtp_receiver, - streams, - }) - .await; - } - } - RTCEvent::Data { data, binary } => { - if !binary { - Err(EngineError::Internal( - "text messages aren't supported".to_string(), - ))?; - } - - let data = DataPacket::decode(&*data)?; - match data.value.unwrap() { - Value::User(user) => { - // TODO(theomonnom) Send event - } - Value::Speaker(_) => { - // TODO(theomonnonm) - } - } - } - } - - Ok(()) - } - - async fn handle_signal( - event: signal_response::Message, - signal_client: Arc, - engine_inner: Arc, - emitter: EngineEmitter, - ) -> EngineResult<()> { - match event { - signal_response::Message::Answer(answer) => { - trace!("received answer from the publisher: {:?}", answer); - - let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; - engine_inner - .publisher_pc - .lock() - .await - .set_remote_description(sdp) - .await?; - } - signal_response::Message::Offer(offer) => { - // Handle the subscriber offer & send an answer to livekit-server - // We always get an offer from the server when connecting - trace!("received offer for the subscriber: {:?}", offer); - let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; - - engine_inner - .subscriber_pc - .lock() - .await - .set_remote_description(sdp) - .await?; - let answer = engine_inner - .subscriber_pc - .lock() - .await - .peer_connection() - .create_answer(RTCOfferAnswerOptions::default()) - .await?; - engine_inner - .subscriber_pc - .lock() - .await - .peer_connection() - .set_local_description(answer.clone()) - .await?; - - tokio::spawn(async move { - signal_client - .send(signal_request::Message::Answer(proto::SessionDescription { - r#type: "answer".to_string(), - sdp: answer.to_string(), - })) - .await; - }); - } - signal_response::Message::Trickle(trickle) => { - // Add the IceCandidate received from the livekit-server - let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?; - let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?; - - trace!( - "received ice_candidate {:?} - {:?}", - SignalTarget::from_i32(trickle.target).unwrap(), - ice - ); - - if trickle.target == SignalTarget::Publisher as i32 { - engine_inner - .publisher_pc - .lock() - .await - .add_ice_candidate(ice) - .await?; - } else { - engine_inner - .subscriber_pc - .lock() - .await - .add_ice_candidate(ice) - .await?; - } - } - signal_response::Message::Update(update) => { - let _ = emitter.send(EngineEvent::ParticipantUpdate(update)).await; - } - _ => {} - } - - Ok(()) - } - - async fn handle_disconnected( - signal_client: Arc, - engine_inner: Arc, - emitter: EngineEmitter, - ) { - if engine_inner.closed.load(Ordering::SeqCst) - || engine_inner.reconnecting.load(Ordering::SeqCst) - { - return; - } - - engine_inner.reconnecting.store(true, Ordering::SeqCst); - warn!("RTCEngine disconnected unexpectedly, reconnecting..."); - - let mut full_reconnect = false; - for i in 0..RECONNECT_ATTEMPTS { - if full_reconnect { - if i == 0 { - let _ = emitter.send(EngineEvent::Restarting).await; - } - - info!("restarting connection... attempt: {}", i); - if let Err(err) = Self::try_restart_connection( - signal_client.clone(), - engine_inner.clone(), - emitter.clone(), - ) - .await - { - error!("restarting connection failed: {}", err); - } else { - return; - } - } else { - if i == 0 { - let _ = emitter.send(EngineEvent::Resuming).await; - } - - info!("resuming connection... attempt: {}", i); - if let Err(err) = Self::try_resume_connection( - signal_client.clone(), - engine_inner.clone(), - emitter.clone(), - ) - .await - { - error!("resuming connection failed: {}", err); - if let EngineError::Signal(_) = err { - full_reconnect = true; - } - } else { - return; - } - } - - tokio::time::sleep(RECONNECT_INTERVAL).await; - } - error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS); - engine_inner.reconnecting.store(false, Ordering::SeqCst); - - // TODO DISCONNECT - } - - async fn try_restart_connection( - signal_client: Arc, - engine_inner: Arc, - emitter: EngineEmitter, - ) -> EngineResult<()> { - Ok(()) - } - - async fn try_resume_connection( - signal_client: Arc, - engine_inner: Arc, - emitter: EngineEmitter, - ) -> EngineResult<()> { - let mut options = engine_inner.options.lock().clone(); - options.sid = engine_inner - .join_response - .lock() - .participant - .as_ref() - .unwrap() - .sid - .clone(); - - signal_client - .reconnect( - &engine_inner.url, - &engine_inner.token.lock().clone(), - options, - ) - .await?; - - let _ = emitter.send(EngineEvent::SignalResumed).await; - - engine_inner - .subscriber_pc - .lock() - .await - .prepare_ice_restart(); - - if engine_inner.has_published.load(Ordering::SeqCst) { - engine_inner - .publisher_pc - .lock() - .await - .create_and_send_offer(RTCOfferAnswerOptions { - ice_restart: true, - ..Default::default() - }) - .await?; - } - - Self::wait_pc_connection(engine_inner).await?; - signal_client.flush_queue().await; - - let _ = emitter.send(EngineEvent::Resumed); - - Ok(()) - } - - async fn wait_pc_connection(engine_inner: Arc) -> EngineResult<()> { - let wait_connected = async move { - while engine_inner.pc_state.load(Ordering::SeqCst) != PCState::Connected as u8 { - sleep(Duration::from_millis(50)).await; - } - }; - - tokio::select! { - _ = wait_connected => Ok(()), - _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { - let err = EngineError::Connection("wait_pc_connection timed out".to_string()); - Err(err) - } - } - } - fn close(&self) { // TODO } - - fn configure_engine( - lk_runtime: Arc, - join_response: JoinResponse, - ) -> EngineResult<(EngineInner, RTCEvents)> { - let (rtc_emitter, events) = mpsc::unbounded_channel(); - let rtc_config = RTCConfiguration::from(join_response.clone()); - - 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.clone())?, - ); - - 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, - DataChannelInit { - ordered: true, - ..DataChannelInit::default() - }, - )?; - - publisher_pc - .peer_connection() - .on_ice_candidate(rtc_events::on_ice_candidate( - SignalTarget::Publisher, - rtc_emitter.clone(), - )); - subscriber_pc - .peer_connection() - .on_ice_candidate(rtc_events::on_ice_candidate( - SignalTarget::Subscriber, - rtc_emitter.clone(), - )); - - publisher_pc.on_offer(rtc_events::on_offer( - SignalTarget::Publisher, - rtc_emitter.clone(), - )); - subscriber_pc.on_offer(rtc_events::on_offer( - SignalTarget::Subscriber, - rtc_emitter.clone(), - )); - - publisher_pc - .peer_connection() - .on_data_channel(rtc_events::on_data_channel( - SignalTarget::Publisher, - rtc_emitter.clone(), - )); - subscriber_pc - .peer_connection() - .on_data_channel(rtc_events::on_data_channel( - SignalTarget::Subscriber, - rtc_emitter.clone(), - )); - - publisher_pc - .peer_connection() - .on_add_track(rtc_events::on_add_track( - SignalTarget::Publisher, - rtc_emitter.clone(), - )); - subscriber_pc - .peer_connection() - .on_add_track(rtc_events::on_add_track( - SignalTarget::Subscriber, - rtc_emitter.clone(), - )); - - publisher_pc - .peer_connection() - .on_connection_change(rtc_events::on_connection_change( - SignalTarget::Publisher, - rtc_emitter.clone(), - )); - subscriber_pc - .peer_connection() - .on_connection_change(rtc_events::on_connection_change( - SignalTarget::Subscriber, - rtc_emitter.clone(), - )); - - lossy_dc.on_message(rtc_events::on_message(rtc_emitter.clone())); - reliable_dc.on_message(rtc_events::on_message(rtc_emitter.clone())); - - Ok(( - EngineInner { - has_published: AtomicBool::new(false), - join_response: Mutex::new(join_response), - pc_state: AtomicU8::new(PCState::New as u8), - publisher_pc: AsyncMutex::new(publisher_pc), - subscriber_pc: AsyncMutex::new(subscriber_pc), - lossy_dc: Mutex::new(lossy_dc), - reliable_dc: Mutex::new(reliable_dc), - sub_lossy_dc: Mutex::new(None), - sub_reliable_dc: Mutex::new(None), - closed: AtomicBool::new(false), - }, - events, - )) - } } diff --git a/crates/livekit-core/src/rtc_engine/pc_transport.rs b/crates/livekit-core/src/rtc_engine/pc_transport.rs index 19c44a9..cae529b 100644 --- a/crates/livekit-core/src/rtc_engine/pc_transport.rs +++ b/crates/livekit-core/src/rtc_engine/pc_transport.rs @@ -11,6 +11,8 @@ use livekit_webrtc::peer_connection::{ }; use livekit_webrtc::rtc_error::RTCError; +use crate::proto::SignalTarget; + const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); pub type OnOfferHandler = Box< @@ -20,6 +22,7 @@ pub type OnOfferHandler = Box< >; pub(crate) struct PCTransport { + signal_target: SignalTarget, peer_connection: PeerConnection, pending_candidates: Vec, on_offer_handler: Option, @@ -34,8 +37,9 @@ impl Debug for PCTransport { } impl PCTransport { - pub fn new(peer_connection: PeerConnection) -> Self { + pub fn new(peer_connection: PeerConnection, signal_target: SignalTarget) -> Self { Self { + signal_target, peer_connection, pending_candidates: Vec::default(), on_offer_handler: None, @@ -54,6 +58,10 @@ impl PCTransport { &mut self.peer_connection } + pub fn signal_target(&self) -> SignalTarget { + self.signal_target.clone() + } + pub fn on_offer(&mut self, handler: OnOfferHandler) { self.on_offer_handler = Some(handler); } diff --git a/crates/livekit-core/src/rtc_engine/rtc_events.rs b/crates/livekit-core/src/rtc_engine/rtc_events.rs index 78712d7..2047582 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_events.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_events.rs @@ -11,11 +11,13 @@ use tokio::sync::mpsc; use crate::proto::SignalTarget; use crate::rtc_engine::pc_transport::OnOfferHandler; -pub(super) type RTCEmitter = mpsc::UnboundedSender; -pub(super) type RTCEvents = mpsc::UnboundedReceiver; +use super::pc_transport::PCTransport; + +pub type RTCEmitter = mpsc::UnboundedSender; +pub type RTCEvents = mpsc::UnboundedReceiver; #[derive(Debug)] -pub(super) enum RTCEvent { +pub enum RTCEvent { IceCandidate { ice_candidate: IceCandidate, target: SignalTarget, @@ -43,19 +45,16 @@ pub(super) enum RTCEvent { }, } -/// Handlers used to forward event to a channel +/// Handlers used to forward events to a channel /// Every callback here is called on the signaling thread -pub(super) fn on_connection_change( - target: SignalTarget, - emitter: RTCEmitter, -) -> OnConnectionChangeHandler { +fn on_connection_change(target: SignalTarget, emitter: RTCEmitter) -> OnConnectionChangeHandler { Box::new(move |state| { let _ = emitter.send(RTCEvent::ConnectionChange { state, target }); }) } -pub(super) fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler { +fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler { Box::new(move |ice_candidate| { let _ = emitter.send(RTCEvent::IceCandidate { ice_candidate, @@ -64,7 +63,7 @@ pub(super) fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnI }) } -pub(super) fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler { +fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler { Box::new(move |offer| { let _ = emitter.send(RTCEvent::Offer { offer, target }); @@ -72,7 +71,7 @@ pub(super) fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHand }) } -pub(super) fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler { +fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler { Box::new(move |mut data_channel| { data_channel.on_message(on_message(emitter.clone())); @@ -83,7 +82,7 @@ pub(super) fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDa }) } -pub(super) fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler { +fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler { Box::new(move |rtp_receiver, streams| { let _ = emitter.send(RTCEvent::AddTrack { rtp_receiver, @@ -93,7 +92,34 @@ pub(super) fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTr }) } -pub(super) fn on_message(emitter: RTCEmitter) -> OnMessageHandler { +pub fn forward_pc_events(transport: &mut PCTransport, rtc_emitter: RTCEmitter) { + transport + .peer_connection() + .on_ice_candidate(on_ice_candidate( + transport.signal_target(), + rtc_emitter.clone(), + )); + + transport.peer_connection().on_data_channel(on_data_channel( + transport.signal_target(), + rtc_emitter.clone(), + )); + + transport + .peer_connection() + .on_add_track(on_add_track(transport.signal_target(), rtc_emitter.clone())); + + transport + .peer_connection() + .on_connection_change(on_connection_change( + transport.signal_target(), + rtc_emitter.clone(), + )); + + transport.on_offer(on_offer(transport.signal_target(), rtc_emitter.clone())); +} + +fn on_message(emitter: RTCEmitter) -> OnMessageHandler { Box::new(move |data, binary| { let _ = emitter.send(RTCEvent::Data { data: data.to_vec(), @@ -101,3 +127,7 @@ pub(super) fn on_message(emitter: RTCEmitter) -> OnMessageHandler { }); }) } + +pub fn forward_dc_events(dc: &mut DataChannel, rtc_emitter: RTCEmitter) { + dc.on_message(on_message(rtc_emitter.clone())); +} diff --git a/crates/livekit-core/src/signal_client/mod.rs b/crates/livekit-core/src/signal_client/mod.rs index 4567a25..15eb2d8 100644 --- a/crates/livekit-core/src/signal_client/mod.rs +++ b/crates/livekit-core/src/signal_client/mod.rs @@ -61,34 +61,23 @@ impl Default for SignalOptions { } } -#[derive(Debug)] +#[derive(Debug, Default)] pub struct SignalClient { stream: RwLock>, - emitter: SignalEmitter, } impl SignalClient { - pub fn new() -> (Self, SignalEvents) { - let (emitter, events) = mpsc::channel(8); - ( - Self { - stream: Default::default(), - emitter, - }, - events, - ) - } - #[instrument(level = Level::DEBUG, skip(url, token, options))] pub(crate) async fn connect( &self, url: &str, token: &str, options: SignalOptions, - ) -> SignalResult<()> { - let stream = SignalStream::connect(url, token, options, self.emitter.clone()).await?; + ) -> SignalResult { + let (emitter, events) = mpsc::channel(8); + let stream = SignalStream::connect(url, token, options, emitter).await?; *self.stream.write() = Some(stream); - Ok(()) + Ok(events) } #[instrument(level = Level::DEBUG)] @@ -142,14 +131,15 @@ impl From for RTCConfiguration { pub mod utils { use crate::proto::{signal_response, JoinResponse}; use crate::signal_client::{SignalError, SignalEvent, SignalResult, JOIN_RESPONSE_TIMEOUT}; - use tokio::sync::mpsc; use tokio::time::timeout; use tokio_tungstenite::tungstenite::Error as WsError; use tracing::{event, instrument, Level}; + use super::SignalEvents; + #[instrument(level = Level::DEBUG, skip(receiver))] pub(crate) async fn next_join_response( - receiver: &mut mpsc::Receiver, + receiver: &mut SignalEvents, ) -> SignalResult { let join = async { while let Some(event) = receiver.recv().await { diff --git a/crates/livekit-core/src/signal_client/signal_stream.rs b/crates/livekit-core/src/signal_client/signal_stream.rs index eb0680d..901b23a 100644 --- a/crates/livekit-core/src/signal_client/signal_stream.rs +++ b/crates/livekit-core/src/signal_client/signal_stream.rs @@ -77,9 +77,8 @@ impl SignalStream { let (ws_writer, ws_reader) = ws_stream.split(); let (internal_tx, internal_rx) = mpsc::channel::(8); - let write_handle = - tokio::spawn(Self::handle_write(internal_rx, ws_writer, emitter.clone())); - let read_handle = tokio::spawn(Self::handle_read(internal_tx.clone(), ws_reader, emitter)); + let write_handle = tokio::spawn(Self::write_task(internal_rx, ws_writer, emitter.clone())); + let read_handle = tokio::spawn(Self::read_task(internal_tx.clone(), ws_reader, emitter)); Ok(Self { internal_tx, @@ -119,7 +118,7 @@ impl SignalStream { /// This task is used to send messages to the websocket /// It is also responsible for closing the connection - async fn handle_write( + async fn write_task( mut internal_rx: mpsc::Receiver, mut ws_writer: SplitSink, emitter: SignalEmitter, @@ -170,7 +169,7 @@ impl SignalStream { /// and dispatch them through the EventEmitter. /// /// It can also send messages to [handle_write] task ( Used e.g. answer to pings ) - async fn handle_read( + async fn read_task( internal_tx: mpsc::Sender, mut ws_reader: SplitStream, emitter: SignalEmitter,