From 2e3c87f232c57b87266a408a63896b7caeff21c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Mon, 19 Dec 2022 17:44:10 +0100 Subject: [PATCH] rtc_session WIP --- .../src/rtc_engine/engine_internal.rs | 212 +------- .../src/rtc_engine/pc_transport.rs | 4 + .../livekit-core/src/rtc_engine/rtc_events.rs | 4 +- .../src/rtc_engine/rtc_session.rs | 511 ++++++++++++++++-- crates/livekit-core/src/signal_client/mod.rs | 23 +- 5 files changed, 524 insertions(+), 230 deletions(-) diff --git a/crates/livekit-core/src/rtc_engine/engine_internal.rs b/crates/livekit-core/src/rtc_engine/engine_internal.rs index a67a3ae..9409fed 100644 --- a/crates/livekit-core/src/rtc_engine/engine_internal.rs +++ b/crates/livekit-core/src/rtc_engine/engine_internal.rs @@ -19,6 +19,8 @@ use livekit_webrtc::peer_connection::{ }; 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, @@ -28,52 +30,28 @@ 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 super::{rtc_events, 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 EngineInternal { lk_runtime: Arc, session: AsyncRwLock, - signal_client: SignalClient, + signal_client: Arc, reconnecting: AtomicBool, closed: AtomicBool, - has_published: AtomicBool, - pc_state: AtomicU8, // Casted to PCState enum engine_emitter: EngineEmitter, } -#[derive(Serialize, Deserialize)] -#[allow(non_snake_case)] -struct IceCandidateJSON { - sdpMid: String, - sdpMLineIndex: i32, - candidate: String, -} - impl EngineInternal { #[tracing::instrument] pub async fn connect( @@ -93,13 +71,6 @@ impl EngineInternal { } } let lk_runtime = lk_runtime.unwrap(); - - // Connect to the SignalClient - let signal_client = SignalClient::default(); - let mut signal_events = signal_client.connect(url, token, options.clone()).await?; - let join_response = signal_client::utils::next_join_response(&mut signal_events).await?; - debug!("received JoinResponse: {:?}", join_response); - // Configure the PeerConnections/RTCSession let (engine_emitter, engine_events) = mpsc::channel(8); let session_info = SessionInfo { @@ -112,7 +83,6 @@ impl EngineInternal { let rtc_session = AsyncRwLock::new(rtc_session); let rtc_internal = Arc::new(Self { lk_runtime, - info: Mutex::new(session_info), session: rtc_session, signal_client, reconnecting: AtomicBool::new(false), @@ -165,23 +135,27 @@ impl EngineInternal { } => { trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate); - let json = serde_json::to_string(&IceCandidateJSON { - sdpMid: ice_candidate.sdp_mid(), - sdpMLineIndex: ice_candidate.sdp_mline_index(), - candidate: ice_candidate.candidate(), - })?; - self.signal_client .send(signal_request::Message::Trickle(TrickleRequest { - candidate_init: json, + 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 subscriber_primary = self.info.lock().join_response.subscriber_primary; - let is_primary = subscriber_primary && target == SignalTarget::Subscriber; + 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 @@ -189,7 +163,6 @@ impl EngineInternal { .swap(PCState::Connected as u8, Ordering::SeqCst); if old_state == PCState::New as u8 { let _ = self.engine_emitter.send(EngineEvent::Connected).await; - // First time connected } } else if state == PeerConnectionState::Failed { self.pc_state @@ -203,11 +176,7 @@ impl EngineInternal { target, } => { if target == SignalTarget::Subscriber { - if data_channel.label() == RELIABLE_DC_LABEL { - *self.session.read().await.sub_reliable_dc.lock() = Some(data_channel); - } else { - *self.session.read().await.sub_lossy_dc.lock() = Some(data_channel); - } + self.session.read().await.use_data_channel(data_channel); } } RTCEvent::Offer { offer, target } => { @@ -259,75 +228,13 @@ impl EngineInternal { } 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::Answer(answer) => { - trace!("received answer from the publisher: {:?}", answer); - - let answer = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; - self.session - .read() - .await - .publisher_pc - .lock() - .await - .set_remote_description(answer) - .await?; - } - signal_response::Message::Offer(offer) => { - trace!("received offer for the subscriber: {:?}", offer); - // Handle the subscriber offer & send an answer to livekit-server - // We always get an offer from the server when connecting - - let offer = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; - let answer = self - .session - .read() - .await - .subscriber_pc - .lock() - .await - .create_anwser(offer, RTCOfferAnswerOptions::default()) - .await?; - - 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_candidate = - IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?; - let signal_target = SignalTarget::from_i32(trickle.target).unwrap(); - trace!( - "received ice_candidate {:?} {:?}", - signal_target, - ice_candidate - ); - - if trickle.target == SignalTarget::Publisher as i32 { - self.session - .read() - .await - .publisher_pc - .lock() - .await - .add_ice_candidate(ice_candidate) - .await?; - } else { - self.session - .read() - .await - .subscriber_pc - .lock() - .await - .add_ice_candidate(ice_candidate) - .await?; - } - } signal_response::Message::Update(update) => { let _ = self .engine_emitter @@ -340,79 +247,6 @@ impl EngineInternal { Ok(()) } - /// Start publisher negotiation - 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) - } - - /// Ensure the Publisher PC is connected, if not, start the negotiation - /// This is required when sending data to the server - 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 - } - } } /// Reconnection logic impl, LiveKit handles reconnection in two ways: diff --git a/crates/livekit-core/src/rtc_engine/pc_transport.rs b/crates/livekit-core/src/rtc_engine/pc_transport.rs index 55e20be..f0d56d8 100644 --- a/crates/livekit-core/src/rtc_engine/pc_transport.rs +++ b/crates/livekit-core/src/rtc_engine/pc_transport.rs @@ -70,6 +70,10 @@ impl PCTransport { self.restarting_ice = true; } + pub fn close(&mut self) { + self.peer_connection.close(); + } + #[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() { diff --git a/crates/livekit-core/src/rtc_engine/rtc_events.rs b/crates/livekit-core/src/rtc_engine/rtc_events.rs index 2047582..6725202 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_events.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_events.rs @@ -28,16 +28,14 @@ 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, diff --git a/crates/livekit-core/src/rtc_engine/rtc_session.rs b/crates/livekit-core/src/rtc_engine/rtc_session.rs index 845a1b3..8fa99ad 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_session.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_session.rs @@ -1,38 +1,138 @@ -#[derive(Debug, Clone, Default)] -pub struct SessionInfo { - url: String, - token: String, - options: SignalOptions, - join_response: JoinResponse, +use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle}; +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::task::JoinHandle; + +use tokio::sync::{broadcast, mpsc, watch, 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 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 super::{rtc_events, EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult}; + +pub const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); +pub const LOSSY_DC_LABEL: &str = "_lossy"; +pub const RELIABLE_DC_LABEL: &str = "_reliable"; + +pub type SessionEmitter = mpsc::UnboundedSender; +pub type SessionEvents = mpsc::UnboundedReceiver; + +#[derive(Debug)] +pub enum SessionEvent { + Data { + data: Vec, + }, + MediaTrack { + track: MediaStreamTrackHandle, + stream: MediaStream, + receiver: RtpReceiver, + }, + Close { + reason: String, + }, } -/// This struct holds a WebRTC session -/// The session changes at every reconnection +#[repr(u8)] +pub enum PCState { + New, + Connected, + Disconnected, + Reconnecting, + Closed, +} + +#[derive(Serialize, Deserialize)] +#[allow(non_snake_case)] +struct IceCandidateJSON { + sdpMid: String, + sdpMLineIndex: i32, + candidate: String, +} + +#[derive(Debug, Clone, Default)] +pub struct SessionInfo { + pub url: String, + pub token: String, + pub options: SignalOptions, + pub join_response: JoinResponse, +} + +/// Fields shared with engine_task and signal_task #[derive(Debug)] -pub struct RTCSession { +struct SessionInner { info: SessionInfo, + signal_client: Arc, + pc_state: AtomicU8, // PCState + has_published: AtomicBool, + publisher_pc: AsyncMutex, subscriber_pc: AsyncMutex, // Publisher data channels - // Used to send data to other participants ( The SFU forwards the messages ) + // used to send data to other participants ( The SFU forwards the messages ) lossy_dc: DataChannel, reliable_dc: DataChannel, - // Subscriber data channels - // These fields are never used, we just keep a strong reference to them, + // Keep a strong reference to the subscriber datachannels, // so we can receive data from other participants - sub_reliable_dc: Mutex>, - sub_lossy_dc: Mutex>, + subscriber_dc: Mutex>, + + emitter: SessionEmitter, +} + +/// This struct holds a WebRTC session +/// The session changes at every reconnection +/// +/// RTCSession is also responsable for the signaling and the negotation +#[derive(Debug)] +pub struct RTCSession { + lk_runtime: Arc, + inner: Arc, + close_emitter: watch::Sender, // false = is_running + signal_task: JoinHandle<()>, + engine_task: JoinHandle<()>, } impl RTCSession { - pub fn new( + pub async fn connect( + url: &str, + token: &str, + options: SignalOptions, lk_runtime: Arc, - session_info: SessionInfo, - ) -> EngineResult<(Self, RTCEvents)> { - let (rtc_emitter, events) = mpsc::unbounded_channel(); - let rtc_config = RTCConfiguration::from(session_info.join_response); + ) -> EngineResult<(Self, SessionEvents)> { + // Connect to the SignalClient + let (signal_client, mut signal_events) = SignalClient::new(); + let signal_client = Arc::new(signal_client); + signal_client.connect(url, token, options.clone()).await?; + let join_response = signal_client::utils::next_join_response(&mut signal_events).await?; + debug!("received JoinResponse: {:?}", join_response); + + let (rtc_emitter, rtc_events) = mpsc::unbounded_channel(); + let rtc_config = RTCConfiguration::from(join_response.clone()); let mut publisher_pc = PCTransport::new( lk_runtime @@ -70,19 +170,366 @@ impl RTCSession { rtc_events::forward_dc_events(&mut lossy_dc, rtc_emitter.clone()); rtc_events::forward_dc_events(&mut reliable_dc, rtc_emitter.clone()); - Ok(( - Self { - info: session_info, - publisher_pc: AsyncMutex::new(publisher_pc), - subscriber_pc: AsyncMutex::new(subscriber_pc), - sub_lossy_dc: Default::default(), - sub_reliable_dc: Default::default(), - lossy_dc, - reliable_dc, - }, - events, - )) + let session_info = SessionInfo { + url: url.to_owned(), + token: token.to_owned(), + options, + join_response, + }; + + // Start tasks + let (session_emitter, session_events) = mpsc::unbounded_channel(); + let (close_emitter, close_receiver) = watch::channel(false); + + let inner = Arc::new(SessionInner { + info: session_info, + pc_state: AtomicU8::new(PCState::New as u8), + has_published: Default::default(), + signal_client, + publisher_pc: AsyncMutex::new(publisher_pc), + subscriber_pc: AsyncMutex::new(subscriber_pc), + lossy_dc, + reliable_dc, + subscriber_dc: Default::default(), + emitter: session_emitter, + }); + + // Start session tasks + let signal_task = tokio::spawn( + inner + .clone() + .signal_task(signal_events, close_receiver.clone()), + ); + let engine_task = tokio::spawn( + inner + .clone() + .engine_task(rtc_events, close_receiver.clone()), + ); + + let session = Self { + lk_runtime, + inner, + close_emitter, + signal_task, + engine_task, + }; + + if !join_response.subscriber_primary { + inner.negotiate_publisher().await?; + } + + Ok((session, session_events)) } - + /// Close the PeerConnections and the SignalClient + pub async fn close(self) { + // Close the tasks + self.close_emitter.send(true); + self.engine_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; + } + }; + + tokio::select! { + _ = wait_connected => Ok(()), + _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { + let err = EngineError::Connection("wait_pc_connection timed out".to_string()); + Err(err) + } + } + } +} + +impl RTCSession { + pub fn info(&self) -> &SessionInfo { + &self.inner.info + } + + pub fn publisher(&self) -> &AsyncMutex { + &self.inner.publisher_pc + } + + pub fn subscriber(&self) -> &AsyncMutex { + &self.inner.subscriber_pc + } + + pub fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel { + &self.inner.data_channel(kind) + } +} + +impl SessionInner { + async fn engine_task( + self: Arc, + mut rtc_events: RTCEvents, + mut close_receiver: watch::Receiver, + ) { + loop { + tokio::select! { + res = rtc_events.recv() => { + if let Some(event) = res { + if let Err(err) = self.on_rtc_event(event).await { + error!("failed to handle rtc event: {:?}", err); + } + } else { + panic!("rtc_events has been closed unexpectedly"); + } + }, + _ = close_receiver.changed() => { + break; + } + } + } + } + + async fn signal_task( + self: Arc, + mut signal_events: SignalEvents, + mut close_receiver: watch::Receiver, + ) { + loop { + tokio::select! { + res = signal_events.recv() => { + if let Some(signal) = res { + match signal { + SignalEvent::Open => {} + SignalEvent::Signal(signal) => { + if let Err(err) = self.on_signal_event(signal).await { + error!("failed to handle signal: {:?}", err); + } + } + SignalEvent::Close => { + self.on_session_disconnected("SignalClient closed"); + } + } + } else { + panic!("signal_events has been closed unexpectedly"); + } + + }, + _ = close_receiver.changed() => { + break; + } + } + } + } + + async fn on_signal_event(&self, event: signal_response::Message) -> EngineResult<()> { + match event { + signal_response::Message::Answer(answer) => { + trace!("received publisher answer: {:?}", answer); + let answer = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; + self.publisher_pc + .lock() + .await + .set_remote_description(answer) + .await?; + } + signal_response::Message::Offer(offer) => { + trace!("received subscriber offer: {:?}", offer); + let offer = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; + let answer = self + .subscriber_pc + .lock() + .await + .create_anwser(offer, RTCOfferAnswerOptions::default()) + .await?; + + 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) => { + let target = SignalTarget::from_i32(trickle.target).unwrap(); + let ice_candidate = { + let json = serde_json::from_str::(&trickle.candidate_init)?; + IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)? + }; + + trace!("received ice_candidate {:?} {:?}", target, ice_candidate); + + if target == SignalTarget::Publisher { + self.publisher_pc + .lock() + .await + .add_ice_candidate(ice_candidate) + .await?; + } else { + self.subscriber_pc + .lock() + .await + .add_ice_candidate(ice_candidate) + .await?; + } + } + _ => {} + } + + Ok(()) + } + + async fn on_rtc_event(&self, event: RTCEvent) -> EngineResult<()> { + match event { + RTCEvent::IceCandidate { + ice_candidate, + target, + } => { + 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.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.on_session_disconnected("pc_state failed"); + } + } + RTCEvent::DataChannel { data_channel } => { + self.subscriber_dc.lock().push(data_channel); + } + RTCEvent::Offer { offer } => { + // 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, + } => { + 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(()) + } + + /// Called when the SignalClient or one of the PeerConnection has lost the connection + /// The RTCEngine may try a reconnect. + fn on_session_disconnected(&self, reason: &str) { + let _ = self.emitter.send(SessionEvent::Close { + reason: reason.to_owned(), + }); + } + + async fn close(&self) { + self.signal_client.close().await; + self.publisher_pc.lock().await.close(); + self.subscriber_pc.lock().await.close(); + } + + /// Start publisher negotiation + async fn negotiate_publisher(&self) -> EngineResult<()> { + self.has_published.store(true, Ordering::Release); + 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) + } + + /// Ensure the Publisher PC is connected, if not, start the negotiation + /// This is required when sending data to the server + async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> { + if !self.info.join_response.subscriber_primary { + return Ok(()); + } + + if !self.publisher_pc.lock().await.is_connected() + && self + .publisher_pc + .lock() + .await + .peer_connection() + .ice_connection_state() + != IceConnectionState::IceConnectionChecking + { + let _ = self.negotiate_publisher().await; + } + + let dc = self.data_channel(kind); + if dc.state() == DataState::Open { + return Ok(()); + } + + // Wait until the PeerConnection is connected + let wait_connected = async { + while self.publisher_pc.lock().await.is_connected() && dc.state() == DataState::Open { + tokio::task::yield_now().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) + } + } + } + + fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel { + if kind == data_packet::Kind::Reliable { + &self.reliable_dc + } else { + &self.lossy_dc + } + } } diff --git a/crates/livekit-core/src/signal_client/mod.rs b/crates/livekit-core/src/signal_client/mod.rs index 15eb2d8..beeec8c 100644 --- a/crates/livekit-core/src/signal_client/mod.rs +++ b/crates/livekit-core/src/signal_client/mod.rs @@ -61,23 +61,34 @@ impl Default for SignalOptions { } } -#[derive(Debug, Default)] +#[derive(Debug)] 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( + pub async fn connect( &self, url: &str, token: &str, options: SignalOptions, - ) -> SignalResult { - let (emitter, events) = mpsc::channel(8); - let stream = SignalStream::connect(url, token, options, emitter).await?; + ) -> SignalResult<()> { + let stream = SignalStream::connect(url, token, options, self.emitter.clone()).await?; *self.stream.write() = Some(stream); - Ok(events) + Ok(()) } #[instrument(level = Level::DEBUG)]