diff --git a/crates/livekit-core/src/rtc_engine/engine_internal.rs b/crates/livekit-core/src/rtc_engine/engine_internal.rs deleted file mode 100644 index 9409fed..0000000 --- a/crates/livekit-core/src/rtc_engine/engine_internal.rs +++ /dev/null @@ -1,372 +0,0 @@ -use parking_lot::{Mutex, RwLock}; -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; -use std::sync::{Arc, Weak}; -use std::time::Duration; - -use tokio::sync::{mpsc, Mutex as AsyncMutex, RwLock as AsyncRwLock}; -use tokio::time::sleep; - -use lazy_static::lazy_static; -use prost::Message; -use serde::{Deserialize, Serialize}; -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::peer_connection::{ - IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions, -}; -use livekit_webrtc::peer_connection_factory::RTCConfiguration; - -use super::rtc_session::{RTCSession, SessionInfo}; -use super::{rtc_events, EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult}; -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}; -// -// TODO(theomonnom): Smarter retry intervals -pub(crate) const RECONNECT_ATTEMPTS: u32 = 10; -pub(crate) 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)] -pub struct EngineInternal { - lk_runtime: Arc, - session: AsyncRwLock, - signal_client: Arc, - reconnecting: AtomicBool, - closed: AtomicBool, - engine_emitter: EngineEmitter, -} - -impl EngineInternal { - #[tracing::instrument] - pub async fn connect( - url: &str, - token: &str, - options: SignalOptions, - ) -> EngineResult<(Arc, EngineEvents)> { - 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); - } - } - let lk_runtime = lk_runtime.unwrap(); - // Configure the PeerConnections/RTCSession - let (engine_emitter, engine_events) = mpsc::channel(8); - let session_info = SessionInfo { - url: url.to_owned(), - token: token.to_owned(), - join_response: join_response.clone(), - options, - }; - let (rtc_session, rtc_events) = RTCSession::new(lk_runtime.clone(), session_info.clone())?; - let rtc_session = AsyncRwLock::new(rtc_session); - let rtc_internal = Arc::new(Self { - lk_runtime, - session: rtc_session, - signal_client, - reconnecting: AtomicBool::new(false), - closed: AtomicBool::new(false), - has_published: AtomicBool::new(false), - pc_state: AtomicU8::new(PCState::New as u8), - engine_emitter, - }); - - // Start tasks - tokio::spawn(rtc_internal.clone().signal_task(signal_events)); - tokio::spawn(rtc_internal.clone().engine_task(rtc_events)); - - if !join_response.subscriber_primary { - rtc_internal.negotiate_publisher().await?; - } - - Ok((rtc_internal, 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, - } => { - trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate); - - self.signal_client - .send(signal_request::Message::Trickle(TrickleRequest { - candidate_init: serde_json::to_string(&IceCandidateJSON { - sdpMid: ice_candidate.sdp_mid(), - sdpMLineIndex: ice_candidate.sdp_mline_index(), - candidate: ice_candidate.candidate(), - })?, - target: target as i32, - })) - .await; - } - RTCEvent::ConnectionChange { state, target } => { - trace!("connection change, {:?} {:?}", state, target); - let is_primary = self - .session - .read() - .await - .info() - .join_response - .subscriber_primary - && target == SignalTarget::Subscriber; - - if is_primary && state == PeerConnectionState::Connected { - let old_state = self - .pc_state - .swap(PCState::Connected as u8, Ordering::SeqCst); - if old_state == PCState::New as u8 { - let _ = self.engine_emitter.send(EngineEvent::Connected).await; - } - } else if state == PeerConnectionState::Failed { - self.pc_state - .store(PCState::Disconnected as u8, Ordering::SeqCst); - - self.handle_disconnected(); - } - } - RTCEvent::DataChannel { - data_channel, - target, - } => { - if target == SignalTarget::Subscriber { - self.session.read().await.use_data_channel(data_channel); - } - } - RTCEvent::Offer { offer, target } => { - if target == SignalTarget::Publisher { - // Send the publisher offer to the server - self.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 _ = self - .engine_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<()> { - self.session - .read() - .await - .on_signal_event(self.signal_client.clone(), event.clone()) - .await?; - - match event { - signal_response::Message::Update(update) => { - let _ = self - .engine_emitter - .send(EngineEvent::ParticipantUpdate(update)) - .await; - } - _ => {} - } - - Ok(()) - } - -} - -/// Reconnection logic impl, LiveKit handles reconnection in two ways: -/// - If the connection is recoverable, the client performs an ICE Restart [`try_resume_connection()`] -/// - Othwerwise, a full reconnect is performed. See [`try_restart_connection()`] -impl EngineInternal { - /// 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: &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 _ = self.engine_emitter.send(EngineEvent::Restarting).await; - } - - info!("restarting connection... attempt: {}", i); - if let Err(err) = self.try_restart_connection().await { - error!("restarting connection failed: {}", err); - } else { - return; - } - } else { - if i == 0 { - let _ = self.engine_emitter.send(EngineEvent::Resuming).await; - } - - info!("resuming connection... attempt: {}", i); - if let Err(err) = self.try_resume_connection().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 - } - - /// Try to recover the connection by doing a full reconnect. - /// It creates a new RTCSession - async fn try_restart_connection(self: &Arc) -> EngineResult<()> { - Ok(()) - } - - /// Try to recover the connection by doing an ICE restart. - async fn try_resume_connection(self: &Arc) -> EngineResult<()> { - let mut session_info = self.info.lock(); - info.options.sid = self - .session - .read() - .await - .join_response - .lock() - .participant - .as_ref() - .unwrap() - .sid - .clone(); - - self.signal_client.close().await; - self.signal_client - .connect(&info.url, &info.token.clone(), info.options.clone()) - .await?; - - self.engine_emitter.send(EngineEvent::SignalResumed).await; - - self.session - .read() - .as_ref() - .unwrap() - .subscriber_pc - .lock() - .await - .prepare_ice_restart(); - - if self - .session - .read() - .as_ref() - .unwrap() - .has_published - .load(Ordering::SeqCst) - { - self.session - .read() - .as_ref() - .unwrap() - .publisher_pc - .lock() - .await - .create_and_send_offer(RTCOfferAnswerOptions { - ice_restart: true, - ..Default::default() - }) - .await?; - } - self.session - .read() - .as_ref() - .unwrap() - .wait_pc_connection() - .await?; - - self.signal_client.flush_queue().await; - self.engine_emitter.send(EngineEvent::Resumed); - Ok(()) - } -} diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index b432d58..395ba1b 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -1,28 +1,22 @@ -use parking_lot::Mutex; -use std::error; +use livekit_webrtc::data_channel::DataSendError; +use livekit_webrtc::jsep::SdpParseError; +use livekit_webrtc::media_stream::MediaStream; +use livekit_webrtc::rtc_error::RTCError; +use livekit_webrtc::rtp_receiver::RtpReceiver; +use parking_lot::{Mutex, RwLock}; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::{Arc, Weak}; use std::time::Duration; - -use tokio::sync::{mpsc, Mutex as AsyncMutex}; +use thiserror::Error; +use tokio::task::JoinHandle; +use tokio_tungstenite::tungstenite::Error; use lazy_static::lazy_static; use prost::Message; -use serde::{Deserialize, Serialize}; -use thiserror::Error; -use tokio::time::sleep; +use tokio::sync::{mpsc, oneshot, Mutex as AsyncMutex, RwLock as AsyncRwLock}; 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::{ @@ -34,7 +28,8 @@ 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; +use self::rtc_session::{RTCSession, SessionEvent, SessionEvents}; + mod lk_runtime; mod pc_transport; mod rtc_events; @@ -79,94 +74,247 @@ pub enum EngineEvent { Restarted, } +// +// TODO(theomonnom): Smarter retry intervals +pub(crate) const RECONNECT_ATTEMPTS: u32 = 10; +pub(crate) 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()); +} + +/// Represents a running RTCSession with the ability to close the session +/// and the engine_task +#[derive(Debug)] +struct EngineHandle { + session: RTCSession, + engine_task: JoinHandle<()>, + close_sender: oneshot::Sender<()>, +} + +#[derive(Debug)] +struct EngineInner { + running_handle: RwLock>, + reconnecting: AtomicBool, + opened: AtomicBool, + engine_emitter: EngineEmitter, +} + #[derive(Debug)] pub struct RTCEngine { - engine_inner: Arc, + lk_runtime: Arc, + inner: Arc, } impl RTCEngine { - pub fn new() -> Self { + pub fn new() -> (Self, EngineEvents) { 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::new()); + let new_runtime = Arc::new(LKRuntime::default()); *lk_runtime_ref = Arc::downgrade(&new_runtime); lk_runtime = Some(new_runtime); } } - let (signal_client, mut signal_events) = SignalClient::new(); + let (engine_emitter, engine_events) = mpsc::channel(8); + let inner = Arc::new(EngineInner { + running_handle: Default::default(), + reconnecting: Default::default(), + opened: Default::default(), + engine_emitter, + }); - 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?; - debug!("received JoinResponse: {:?}", join_response); - - let (engine_inner, rtc_events) = - Self::configure_engine(lk_runtime.clone(), join_response.clone())?; - let engine_inner = Arc::new(engine_inner); - let signal_client = Arc::new(signal_client); - - let (emitter, events) = mpsc::channel(8); - - tokio::spawn(Self::signal_task( - signal_client.clone(), - engine_inner.clone(), - signal_events, - emitter.clone(), - )); - - tokio::spawn(Self::engine_task( - signal_client.clone(), - engine_inner.clone(), - rtc_events, - emitter.clone(), - )); - - if !join_response.subscriber_primary { - engine_inner.negotiate_publisher().await?; - } - - let rtc_engine = Self { - signal_client, - engine_inner, - lk_runtime, - }; - - Ok((rtc_engine, events)) + ( + Self { + lk_runtime: lk_runtime.unwrap(), + inner, + }, + engine_events, + ) } #[tracing::instrument] + pub async fn connect( + &self, + 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.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(()) + } + + #[tracing::instrument] + pub async fn close(&self) { + self.inner.opened.store(false, Ordering::SeqCst); + self.inner.close(); + } + + #[tracing::instrument(skip(data))] pub async fn publish_data( &self, data: &DataPacket, kind: data_packet::Kind, ) -> Result<(), EngineError> { - self.engine_inner.ensure_publisher_connected(kind).await?; - self.engine_inner - .data_channel(kind) - .lock() - .send(&data.encode_to_vec(), true) - .map_err(Into::into) + self.inner.wait_reconnection().await?; + self.inner + .running_handle + .read() + .as_ref() + .unwrap() + .session + .publish_data(data, kind) + .await?; + + Ok(()) } - pub fn join_response(&self) -> JoinResponse { - self.engine_inner.join_response.lock().clone() - } - - fn close(&self) { - // TODO + pub fn join_response(&self) -> Option { + if let Some(handle) = self.inner.running_handle.read().as_ref() { + Some(handle.session.info().join_response.clone()) + } else { + None + } + } +} + +impl EngineInner { + async fn engine_task( + self: Arc, + mut session_events: SessionEvents, + mut close_receiver: oneshot::Receiver<()>, + ) { + loop { + tokio::select! { + res = session_events.recv() => { + if let Some(event) = res { + if let Err(err) = self.on_session_event(event).await { + error!("failed to handle session event: {:?}", err); + } + } else { + panic!("rtc_sessions has been closed unexpectedly"); + } + }, + _ = &mut close_receiver => { + break; + } + } + } + } + + async fn on_session_event(&self, event: SessionEvent) -> EngineResult<()> { + Ok(()) + } + + async fn close(&self) { + if let Some(handle) = self.running_handle.write().take() { + handle.session.close().await; + let _ = handle.close_sender.send(()); + handle.engine_task.await; + } + } + + async fn wait_reconnection(&self) -> EngineResult<()> { + if !self.opened.load(Ordering::SeqCst) { + Err(EngineError::Connection("not opened".to_owned()))? + } + + while self.reconnecting.load(Ordering::Acquire) { + tokio::task::yield_now().await; + } + + if self.running_handle.read().is_none() { + Err(EngineError::Connection("reconnection failed".to_owned()))? + } + + Ok(()) + } + + /// 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) { + if !self.opened.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 _ = self.engine_emitter.send(EngineEvent::Restarting).await; + } + + info!("restarting connection... attempt: {}", i); + if let Err(err) = self.try_restart_connection().await { + error!("restarting connection failed: {}", err); + } else { + let _ = self.engine_emitter.send(EngineEvent::Restarted).await; + return; + } + } else { + if i == 0 { + let _ = self.engine_emitter.send(EngineEvent::Resuming).await; + } + + info!("resuming connection... attempt: {}", i); + if let Err(err) = self.try_resume_connection().await { + error!("resuming connection failed: {}", err); + if let EngineError::Signal(_) = err { + full_reconnect = true; + } + } else { + let _ = self.engine_emitter.send(EngineEvent::Resumed).await; + return; + } + } + + tokio::time::sleep(RECONNECT_INTERVAL).await; + } + error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS); + self.reconnecting.store(false, Ordering::SeqCst); + + // TODO DISCONNECT + } + + /// Try to recover the connection by doing a full reconnect. + /// It creates a new RTCSession + async fn try_restart_connection(&self) -> EngineResult<()> { + Ok(()) + } + + /// Try to restart the current session + async fn try_resume_connection(&self) -> EngineResult<()> { + Ok(()) } } diff --git a/crates/livekit-core/src/rtc_engine/pc_transport.rs b/crates/livekit-core/src/rtc_engine/pc_transport.rs index f0d56d8..74477cc 100644 --- a/crates/livekit-core/src/rtc_engine/pc_transport.rs +++ b/crates/livekit-core/src/rtc_engine/pc_transport.rs @@ -76,14 +76,15 @@ impl PCTransport { #[tracing::instrument(level = Level::DEBUG)] pub async fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) -> Result<(), RTCError> { - if self.peer_connection.remote_description().is_none() { - self.pending_candidates.push(ice_candidate); + if self.peer_connection.remote_description().is_some() && !self.restarting_ice { + self.peer_connection + .add_ice_candidate(ice_candidate) + .await?; + return Ok(()); } - self.peer_connection - .add_ice_candidate(ice_candidate) - .await?; + self.pending_candidates.push(ice_candidate); Ok(()) } diff --git a/crates/livekit-core/src/rtc_engine/rtc_session.rs b/crates/livekit-core/src/rtc_engine/rtc_session.rs index 8fa99ad..393e03b 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_session.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_session.rs @@ -54,6 +54,7 @@ pub enum SessionEvent { Close { reason: String, }, + Connected, } #[repr(u8)] @@ -81,7 +82,7 @@ pub struct SessionInfo { pub join_response: JoinResponse, } -/// Fields shared with engine_task and signal_task +/// Fields shared with rtc_task and signal_task #[derive(Debug)] struct SessionInner { info: SessionInfo, @@ -114,7 +115,7 @@ pub struct RTCSession { inner: Arc, close_emitter: watch::Sender, // false = is_running signal_task: JoinHandle<()>, - engine_task: JoinHandle<()>, + rtc_task: JoinHandle<()>, } impl RTCSession { @@ -123,7 +124,8 @@ impl RTCSession { token: &str, options: SignalOptions, lk_runtime: Arc, - ) -> EngineResult<(Self, SessionEvents)> { + session_emitter: SessionEmitter, + ) -> EngineResult { // Connect to the SignalClient let (signal_client, mut signal_events) = SignalClient::new(); let signal_client = Arc::new(signal_client); @@ -165,6 +167,7 @@ impl RTCSession { }, )?; + // Forward events received in the Signaling Thread to our rtc channel 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()); @@ -178,7 +181,6 @@ impl RTCSession { }; // Start tasks - let (session_emitter, session_events) = mpsc::unbounded_channel(); let (close_emitter, close_receiver) = watch::channel(false); let inner = Arc::new(SessionInner { @@ -200,50 +202,42 @@ impl RTCSession { .clone() .signal_task(signal_events, close_receiver.clone()), ); - let engine_task = tokio::spawn( - inner - .clone() - .engine_task(rtc_events, close_receiver.clone()), - ); + let rtc_task = tokio::spawn(inner.clone().rtc_task(rtc_events, close_receiver.clone())); let session = Self { lk_runtime, - inner, + inner: inner.clone(), close_emitter, signal_task, - engine_task, + rtc_task, }; - if !join_response.subscriber_primary { + if !inner.info.join_response.subscriber_primary { inner.negotiate_publisher().await?; } - Ok((session, session_events)) + Ok(session) } /// Close the PeerConnections and the SignalClient pub async fn close(self) { // Close the tasks self.close_emitter.send(true); - self.engine_task.await; + self.rtc_task.await; self.signal_task.await; self.inner.close().await; } - pub async fn wait_pc_connection(&self) -> EngineResult<()> { - let wait_connected = async move { - while self.inner.pc_state.load(Ordering::Acquire) != PCState::Connected as u8 { - tokio::task::yield_now().await; - } - }; + pub async fn publish_data( + &self, + data: &DataPacket, + kind: data_packet::Kind, + ) -> Result<(), EngineError> { + self.inner.publish_data(data, kind).await + } - tokio::select! { - _ = wait_connected => Ok(()), - _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { - let err = EngineError::Connection("wait_pc_connection timed out".to_string()); - Err(err) - } - } + pub async fn wait_pc_connectiom(&self) -> EngineResult<()> { + self.inner.wait_pc_connection().await } } @@ -260,13 +254,17 @@ impl RTCSession { &self.inner.subscriber_pc } + pub fn signal_client(&self) -> &Arc { + &self.inner.signal_client + } + pub fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel { &self.inner.data_channel(kind) } } impl SessionInner { - async fn engine_task( + async fn rtc_task( self: Arc, mut rtc_events: RTCEvents, mut close_receiver: watch::Receiver, @@ -405,7 +403,7 @@ impl SessionInner { .pc_state .swap(PCState::Connected as u8, Ordering::SeqCst); if old_state == PCState::New as u8 { - let _ = self.engine_emitter.send(EngineEvent::Connected).await; + let _ = self.emitter.send(SessionEvent::Connected); } } else if state == PeerConnectionState::Failed { self.pc_state @@ -428,15 +426,17 @@ impl SessionInner { } RTCEvent::AddTrack { rtp_receiver, - streams, + mut streams, } => { - let _ = self - .engine_emitter - .send(EngineEvent::AddTrack { - rtp_receiver, - streams, - }) - .await; + if !streams.is_empty() { + let _ = self.emitter.send(SessionEvent::MediaTrack { + track: rtp_receiver.track(), + stream: streams.remove(0), + receiver: rtp_receiver, + }); + } else { + warn!("AddTrack event with no streams"); + } } RTCEvent::Data { data, binary } => { if !binary { @@ -474,6 +474,68 @@ impl SessionInner { self.subscriber_pc.lock().await.close(); } + #[tracing::instrument] + async fn publish_data( + &self, + data: &DataPacket, + kind: data_packet::Kind, + ) -> Result<(), EngineError> { + self.ensure_publisher_connected(kind).await?; + self.data_channel(kind) + .send(&data.encode_to_vec(), true) + .map_err(Into::into) + } + + /// Try to restart the session by doing an ICE Restart (The SignalClient is also restarted) + /// This reconnection if more seemless than the full reconnection implemented in ['RTCEngine'] + async fn restart_session(&self) -> EngineResult<()> { + self.signal_client.close().await; + + let mut options = self.info.options.clone(); + options.sid = self.info.join_response.participant.clone().unwrap().sid; + options.reconnect = true; + + self.signal_client + .connect(&self.info.url, &self.info.token, options) + .await?; + + self.subscriber_pc.lock().await.prepare_ice_restart(); + + if self.has_published.load(Ordering::Acquire) { + self.publisher_pc + .lock() + .await + .create_and_send_offer(RTCOfferAnswerOptions { + ice_restart: true, + ..Default::default() + }) + .await?; + } + + self.wait_pc_connection().await?; + self.signal_client.flush_queue().await; + + Ok(()) + } + + // Wait for PCState to become PCState::Connected + // Timeout after ['MAX_ICE_CONNECT_TIMEOUT'] + async fn wait_pc_connection(&self) -> EngineResult<()> { + let wait_connected = async move { + while self.pc_state.load(Ordering::Acquire) != PCState::Connected as u8 { + tokio::task::yield_now().await; + } + }; + + tokio::select! { + _ = wait_connected => Ok(()), + _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { + let err = EngineError::Connection("wait_pc_connection timed out".to_string()); + Err(err) + } + } + } + /// Start publisher negotiation async fn negotiate_publisher(&self) -> EngineResult<()> { self.has_published.store(true, Ordering::Release); 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 38078ef..6fcc6df 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h @@ -24,10 +24,10 @@ class DataChannel { void register_observer(NativeDataChannelObserver& observer); void unregister_observer(); - bool send(const DataBuffer& buffer); + bool send(const DataBuffer& buffer) const; rust::String label() const; DataState state() const; - void close(); + void close() const; private: std::shared_ptr rtc_runtime_; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp index 41986f1..e5be4b7 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp @@ -24,7 +24,7 @@ void DataChannel::unregister_observer() { data_channel_->UnregisterObserver(); } -bool DataChannel::send(const DataBuffer& buffer) { +bool DataChannel::send(const DataBuffer& buffer) const { return data_channel_->Send(webrtc::DataBuffer{ rtc::CopyOnWriteBuffer(buffer.ptr, buffer.len), buffer.binary}); } @@ -37,7 +37,7 @@ DataState DataChannel::state() const { return static_cast(data_channel_->state()); } -void DataChannel::close() { +void DataChannel::close() const { 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 dd16836..004be48 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs @@ -68,10 +68,10 @@ pub mod ffi { ); fn unregister_observer(self: Pin<&mut DataChannel>); - fn send(self: Pin<&mut DataChannel>, data: &DataBuffer) -> bool; + fn send(self: &DataChannel, data: &DataBuffer) -> bool; fn label(self: &DataChannel) -> String; fn state(self: &DataChannel) -> DataState; - fn close(self: Pin<&mut DataChannel>); + fn close(self: &DataChannel); fn create_data_channel_init(init: DataChannelInit) -> UniquePtr; fn create_native_data_channel_observer( diff --git a/crates/livekit-webrtc/src/data_channel.rs b/crates/livekit-webrtc/src/data_channel.rs index 284340f..2a7591b 100644 --- a/crates/livekit-webrtc/src/data_channel.rs +++ b/crates/livekit-webrtc/src/data_channel.rs @@ -58,7 +58,7 @@ impl DataChannel { dc } - pub fn send(&mut self, data: &[u8], binary: bool) -> Result<(), DataSendError> { + pub fn send(&self, data: &[u8], binary: bool) -> Result<(), DataSendError> { let buffer = sys_dc::ffi::DataBuffer { ptr: data.as_ptr(), len: data.len(), @@ -66,7 +66,6 @@ impl DataChannel { }; self.cxx_handle - .pin_mut() .send(&buffer) .then_some(()) .ok_or(DataSendError {}) @@ -80,8 +79,8 @@ impl DataChannel { self.cxx_handle.state() } - pub fn close(&mut self) { - self.cxx_handle.pin_mut().close(); + pub fn close(&self) { + self.cxx_handle.close(); } pub fn on_state_change(&mut self, handler: OnStateChangeHandler) {