diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index 6c6648c..43c24fc 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -35,7 +35,7 @@ pub enum RoomError { pub type RoomResult = Result; -#[derive(Debug, Clone, Copy, Eq, PartialEq)] +#[derive(Debug, Clone, Eq, PartialEq)] pub enum ConnectionState { Disconnected, Connecting, @@ -69,10 +69,14 @@ impl Room { #[instrument(level = Level::DEBUG)] pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> { - let (rtc_engine, engine_events) = - RTCEngine::connect(url, token, SignalOptions::default()).await?; + // Initialize the RTCEngine + let (rtc_engine, engine_events) = RTCEngine::new(); let rtc_engine = Arc::new(rtc_engine); - let join_response = rtc_engine.join_response(); + rtc_engine + .connect(url, token, SignalOptions::default()) + .await?; + + let join_response = rtc_engine.join_response().unwrap(); let pi = join_response.participant.unwrap().clone(); let local_participant = Arc::new(LocalParticipant::new( rtc_engine.clone(), @@ -150,26 +154,21 @@ impl Room { Self::handle_participant_update(room_inner.clone(), room_events.clone(), update) .await } - EngineEvent::AddTrack { - rtp_receiver, - streams, + EngineEvent::MediaTrack { + track, + stream, + receiver, } => { - if streams.is_empty() { - Err(RoomError::Internal( - "AddTrack event with empty streams".to_string(), - ))?; - } - - let first_stream_id = streams.first().unwrap().id(); - let stream_id = unpack_stream_id(&first_stream_id); - if stream_id.is_none() { + let stream_id = stream.id(); + let lk_stream_id = unpack_stream_id(&stream_id); + if lk_stream_id.is_none() { Err(RoomError::Internal(format!( - "AddTrack event with invalid track_id: {:?}", - first_stream_id + "MediaTrack event with invalid track_id: {:?}", + &stream_id )))?; } - let (participant_sid, track_sid) = stream_id.unwrap(); + let (participant_sid, track_sid) = lk_stream_id.unwrap(); let remote_participant = Self::get_participant(room_inner.clone(), &participant_sid.to_string().into()); @@ -181,7 +180,7 @@ impl Room { .add_subscribed_media_track( RoomHandle::from(room_inner), track_sid, - rtp_receiver.track(), + track, ) .await; } @@ -195,6 +194,11 @@ impl Room { )))?; } } + EngineEvent::Resuming => {} + EngineEvent::Resumed => {} + EngineEvent::Restarting => {} + EngineEvent::Restarted => {} + EngineEvent::Disconnected => {} } Ok(()) diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index 6819173..5fd9521 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -1,13 +1,15 @@ +use futures::FutureExt; use livekit_webrtc::data_channel::DataSendError; use livekit_webrtc::jsep::SdpParseError; -use livekit_webrtc::media_stream::MediaStream; +use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle}; use livekit_webrtc::rtc_error::RTCError; use livekit_webrtc::rtp_receiver::RtpReceiver; -use parking_lot::{Mutex, RwLock}; +use parking_lot::Mutex; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Weak}; use std::time::Duration; use thiserror::Error; +use tokio::sync::RwLock as AsyncRwLock; use tokio::task::JoinHandle; use lazy_static::lazy_static; @@ -18,7 +20,7 @@ use crate::proto::{data_packet, DataPacket, JoinResponse, ParticipantUpdate}; use crate::rtc_engine::lk_runtime::LKRuntime; use crate::signal_client::{SignalError, SignalOptions}; -use self::rtc_session::{RTCSession, SessionEvent, SessionEvents}; +use self::rtc_session::{RTCSession, SessionEvent, SessionEvents, SessionInfo}; mod lk_runtime; mod pc_transport; @@ -52,25 +54,39 @@ pub enum EngineError { #[derive(Debug)] pub enum EngineEvent { ParticipantUpdate(ParticipantUpdate), - AddTrack { - rtp_receiver: RtpReceiver, - streams: Vec, + MediaTrack { + track: MediaStreamTrackHandle, + stream: MediaStream, + receiver: RtpReceiver, }, Resuming, Resumed, Restarting, Restarted, + Disconnected, } // TODO(theomonnom): Smarter retry intervals -pub(crate) const RECONNECT_ATTEMPTS: u32 = 10; -pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300); +pub const RECONNECT_ATTEMPTS: u32 = 10; +pub const RECONNECT_INTERVAL: Duration = Duration::from_millis(300); lazy_static! { // Share one LKRuntime across all RTCEngine instances static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); } +#[derive(Debug, Clone, Eq, PartialEq)] +#[repr(u8)] +pub enum SimulateScenario { + SignalReconnect, + Speaker, + NodeFailure, + ServerLeave, + Migration, + ForceTcp, + ForceTls, +} + /// Represents a running RTCSession with the ability to close the session /// and the engine_task #[derive(Debug)] @@ -82,7 +98,9 @@ struct EngineHandle { #[derive(Debug)] struct EngineInner { - running_handle: RwLock>, + lk_runtime: Arc, + session_info: Mutex>, // Last/Current Sessioninfo + running_handle: AsyncRwLock>, reconnecting: AtomicBool, opened: AtomicBool, engine_emitter: EngineEmitter, @@ -90,7 +108,6 @@ struct EngineInner { #[derive(Debug)] pub struct RTCEngine { - lk_runtime: Arc, inner: Arc, } @@ -110,19 +127,15 @@ impl RTCEngine { let (engine_emitter, engine_events) = mpsc::channel(8); let inner = Arc::new(EngineInner { + lk_runtime: lk_runtime.unwrap(), + session_info: Default::default(), running_handle: Default::default(), reconnecting: Default::default(), opened: Default::default(), engine_emitter, }); - ( - Self { - lk_runtime: lk_runtime.unwrap(), - inner, - }, - engine_events, - ) + (Self { inner }, engine_events) } #[tracing::instrument] @@ -132,37 +145,18 @@ impl RTCEngine { token: &str, options: SignalOptions, ) -> EngineResult<()> { - let (session_emitter, session_events) = mpsc::unbounded_channel(); - let session = RTCSession::connect( - url, - token, - options, - self.lk_runtime.clone(), - session_emitter, - ) - .await?; - - let (close_sender, close_receiver) = oneshot::channel(); - let engine_task = tokio::spawn( - self.inner - .clone() - .engine_task(session_events, close_receiver), - ); - - self.inner.opened.store(true, Ordering::SeqCst); - *self.inner.running_handle.write() = Some(EngineHandle { - session, - engine_task, - close_sender, - }); - - Ok(()) + self.inner.connect(url, token, options).await } #[tracing::instrument] pub async fn close(&self) { self.inner.opened.store(false, Ordering::SeqCst); self.inner.close(); + let _ = self + .inner + .engine_emitter + .send(EngineEvent::Disconnected) + .await; } #[tracing::instrument(skip(data))] @@ -175,6 +169,7 @@ impl RTCEngine { self.inner .running_handle .read() + .await .as_ref() .unwrap() .session @@ -185,8 +180,8 @@ impl RTCEngine { } pub fn join_response(&self) -> Option { - if let Some(handle) = self.inner.running_handle.read().as_ref() { - Some(handle.session.info().join_response.clone()) + if let Some(info) = self.inner.session_info.lock().as_ref() { + Some(info.join_response.clone()) } else { None } @@ -217,15 +212,68 @@ impl EngineInner { } } - async fn on_session_event(&self, event: SessionEvent) -> EngineResult<()> { + async fn on_session_event(self: &Arc, event: SessionEvent) -> EngineResult<()> { + match event { + SessionEvent::Close { reason } => { + info!("received session close: {}", reason); + self.handle_disconnected().await; + } + SessionEvent::Data { data } => {} + SessionEvent::MediaTrack { + track, + stream, + receiver, + } => { + let _ = self + .engine_emitter + .send(EngineEvent::MediaTrack { + track, + stream, + receiver, + }) + .await; + } + SessionEvent::Connected => {} + } + Ok(()) + } + + async fn connect( + self: &Arc, + url: &str, + token: &str, + options: SignalOptions, + ) -> EngineResult<()> { + let (session_emitter, session_events) = mpsc::unbounded_channel(); + let session = RTCSession::connect( + url, + token, + options, + self.lk_runtime.clone(), + session_emitter, + ) + .await?; + + let (close_sender, close_receiver) = oneshot::channel(); + let engine_task = tokio::spawn(self.clone().engine_task(session_events, close_receiver)); + + *self.session_info.lock() = Some(session.info().clone()); + *self.running_handle.write().await = Some(EngineHandle { + session, + engine_task, + close_sender, + }); + + self.opened.store(true, Ordering::SeqCst); + Ok(()) } async fn close(&self) { - if let Some(handle) = self.running_handle.write().take() { + if let Some(handle) = self.running_handle.write().await.take() { handle.session.close().await; let _ = handle.close_sender.send(()); - handle.engine_task.await; + let _ = handle.engine_task.await; } } @@ -238,7 +286,7 @@ impl EngineInner { tokio::task::yield_now().await; } - if self.running_handle.read().is_none() { + if self.running_handle.read().await.is_none() { Err(EngineError::Connection("reconnection failed".to_owned()))? } @@ -247,7 +295,7 @@ impl EngineInner { /// Called every time the PeerConnection or the SignalClient is closed /// We first try to resume the connection, if it fails, we start a full reconnect. - async fn handle_disconnected(&self) { + async fn handle_disconnected(self: &Arc) { if !self.opened.load(Ordering::SeqCst) || self.reconnecting.load(Ordering::SeqCst) { return; } @@ -255,6 +303,7 @@ impl EngineInner { self.reconnecting.store(true, Ordering::SeqCst); warn!("RTCEngine disconnected unexpectedly, reconnecting..."); + let mut connected = false; let mut full_reconnect = false; for i in 0..RECONNECT_ATTEMPTS { if full_reconnect { @@ -267,7 +316,8 @@ impl EngineInner { error!("restarting connection failed: {}", err); } else { let _ = self.engine_emitter.send(EngineEvent::Restarted).await; - return; + connected = true; + break; } } else { if i == 0 { @@ -282,27 +332,45 @@ impl EngineInner { } } else { let _ = self.engine_emitter.send(EngineEvent::Resumed).await; - return; + connected = true; + break; } } tokio::time::sleep(RECONNECT_INTERVAL).await; } - error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS); + self.reconnecting.store(false, Ordering::SeqCst); - // TODO DISCONNECT + if !connected { + error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS); + let _ = self.engine_emitter.send(EngineEvent::Disconnected).await; + self.close().await; + } } /// Try to recover the connection by doing a full reconnect. /// It recreates a new RTCSession - async fn try_restart_connection(&self) -> EngineResult<()> { + async fn try_restart_connection(self: &Arc) -> EngineResult<()> { + let info = self.session_info.lock().clone().unwrap(); self.close().await; - Ok(()) + self.connect(&info.url, &info.token, info.options).await?; + self.running_handle + .read() + .await + .as_ref() + .unwrap() + .session + .wait_pc_connection() + .await + + // TODO(theomonnom): Resend SignalClient queue } /// Try to restart the current session async fn try_resume_connection(&self) -> EngineResult<()> { - Ok(()) + let handle = self.running_handle.read().await; + handle.as_ref().unwrap().session.restart().await?; + handle.as_ref().unwrap().session.wait_pc_connection().await } } diff --git a/crates/livekit-core/src/rtc_engine/pc_transport.rs b/crates/livekit-core/src/rtc_engine/pc_transport.rs index 74477cc..34b2fa2 100644 --- a/crates/livekit-core/src/rtc_engine/pc_transport.rs +++ b/crates/livekit-core/src/rtc_engine/pc_transport.rs @@ -21,7 +21,7 @@ pub type OnOfferHandler = Box< + Sync, >; -pub(crate) struct PCTransport { +pub struct PCTransport { signal_target: SignalTarget, peer_connection: PeerConnection, pending_candidates: Vec, diff --git a/crates/livekit-core/src/rtc_engine/rtc_events.rs b/crates/livekit-core/src/rtc_engine/rtc_events.rs index 6725202..6ccd0ed 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_events.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_events.rs @@ -28,14 +28,17 @@ pub enum RTCEvent { }, DataChannel { data_channel: DataChannel, + target: SignalTarget, }, // TODO (theomonnom): Move Offer to PCTransport Offer { offer: SessionDescription, + target: SignalTarget, }, AddTrack { rtp_receiver: RtpReceiver, streams: Vec, + target: SignalTarget, }, Data { data: Vec, @@ -91,28 +94,22 @@ fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler } pub fn forward_pc_events(transport: &mut PCTransport, rtc_emitter: RTCEmitter) { + let signal_target = transport.signal_target(); 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(), - )); + .on_ice_candidate(on_ice_candidate(signal_target, rtc_emitter.clone())); transport .peer_connection() - .on_add_track(on_add_track(transport.signal_target(), rtc_emitter.clone())); + .on_data_channel(on_data_channel(signal_target, rtc_emitter.clone())); transport .peer_connection() - .on_connection_change(on_connection_change( - transport.signal_target(), - rtc_emitter.clone(), - )); + .on_add_track(on_add_track(signal_target, rtc_emitter.clone())); + + transport + .peer_connection() + .on_connection_change(on_connection_change(signal_target, rtc_emitter.clone())); transport.on_offer(on_offer(transport.signal_target(), rtc_emitter.clone())); } diff --git a/crates/livekit-core/src/rtc_engine/rtc_session.rs b/crates/livekit-core/src/rtc_engine/rtc_session.rs index f802525..db3830f 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_session.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_session.rs @@ -1,6 +1,7 @@ use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle}; use livekit_webrtc::rtp_receiver::RtpReceiver; use parking_lot::Mutex; +use std::convert::TryInto; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -65,6 +66,21 @@ pub enum PCState { Closed, } +impl TryInto for u8 { + type Error = &'static str; + + fn try_into(self) -> Result { + match self { + 0 => Ok(PCState::New), + 1 => Ok(PCState::Connected), + 2 => Ok(PCState::Disconnected), + 3 => Ok(PCState::Reconnecting), + 4 => Ok(PCState::Closed), + _ => Err("invalid PCState"), + } + } +} + #[derive(Serialize, Deserialize)] #[allow(non_snake_case)] struct IceCandidateJSON { @@ -236,7 +252,11 @@ impl RTCSession { self.inner.publish_data(data, kind).await } - pub async fn wait_pc_connectiom(&self) -> EngineResult<()> { + pub async fn restart(&self) -> EngineResult<()> { + self.inner.restart_session().await + } + + pub async fn wait_pc_connection(&self) -> EngineResult<()> { self.inner.wait_pc_connection().await } } @@ -246,6 +266,14 @@ impl RTCSession { &self.inner.info } + pub fn state(&self) -> PCState { + self.inner + .pc_state + .load(Ordering::SeqCst) + .try_into() + .unwrap() + } + pub fn publisher(&self) -> &AsyncMutex { &self.inner.publisher_pc } @@ -412,10 +440,13 @@ impl SessionInner { self.on_session_disconnected("pc_state failed"); } } - RTCEvent::DataChannel { data_channel } => { + RTCEvent::DataChannel { + data_channel, + target: _, + } => { self.subscriber_dc.lock().push(data_channel); } - RTCEvent::Offer { offer } => { + RTCEvent::Offer { offer, target: _ } => { // Send the publisher offer to the server self.signal_client .send(signal_request::Message::Offer(proto::SessionDescription { @@ -427,6 +458,7 @@ impl SessionInner { RTCEvent::AddTrack { rtp_receiver, mut streams, + target: _, } => { if !streams.is_empty() { let _ = self.emitter.send(SessionEvent::MediaTrack { diff --git a/crates/livekit-core/src/signal_client/mod.rs b/crates/livekit-core/src/signal_client/mod.rs index beeec8c..e840ec5 100644 --- a/crates/livekit-core/src/signal_client/mod.rs +++ b/crates/livekit-core/src/signal_client/mod.rs @@ -36,14 +36,14 @@ pub enum SignalError { /// Events used by the RTCEngine who will handle the reconnection logic #[derive(Debug)] -pub(crate) enum SignalEvent { +pub enum SignalEvent { Open, Signal(signal_response::Message), Close, } #[derive(Debug, Clone)] -pub(crate) struct SignalOptions { +pub struct SignalOptions { pub(crate) reconnect: bool, pub(crate) sid: String, pub auto_subscribe: bool, diff --git a/examples/.cargo/config b/examples/.cargo/config new file mode 100644 index 0000000..ac2b23f --- /dev/null +++ b/examples/.cargo/config @@ -0,0 +1,2 @@ +[target.x86_64-pc-windows-msvc] +rustflags = ["-C", "target-feature=+crt-static"] diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 6696429..20abc68 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -1,4 +1,4 @@ [workspace] members = ["*"] -exclude = ["target"] +exclude = ["target", ".cargo"] resolver = "2" diff --git a/examples/simple_room/src/main.rs b/examples/simple_room/src/main.rs index 0d97987..351ba58 100644 --- a/examples/simple_room/src/main.rs +++ b/examples/simple_room/src/main.rs @@ -1,15 +1,10 @@ -use tracing_subscriber::prelude::*; mod app; mod events; mod video_grid; mod video_renderer; fn main() { - let fmt_layer = tracing_subscriber::fmt::Layer::default(); - - tracing_subscriber::registry() - .with(fmt_layer) - .init(); + tracing_subscriber::fmt::init(); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all()