From 2ecae706a90411c3de3b22dafe018657a7a522d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sat, 8 Oct 2022 23:04:50 +0200 Subject: [PATCH] refactored RTCEngine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Added comments & separated code into multiple parts. Maybe the next step is to put PeerConnections into another file 🤔 --- crates/livekit-core/src/lib.rs | 3 +- crates/livekit-core/src/lk_runtime.rs | 2 +- crates/livekit-core/src/local_participant.rs | 2 - crates/livekit-core/src/pc_transport.rs | 1 - crates/livekit-core/src/room.rs | 24 +- .../engine_internal.rs} | 478 +++++++----------- crates/livekit-core/src/rtc_engine/mod.rs | 153 ++++++ 7 files changed, 356 insertions(+), 307 deletions(-) rename crates/livekit-core/src/{rtc_engine.rs => rtc_engine/engine_internal.rs} (65%) create mode 100644 crates/livekit-core/src/rtc_engine/mod.rs diff --git a/crates/livekit-core/src/lib.rs b/crates/livekit-core/src/lib.rs index c9bc7e1..8b972d1 100644 --- a/crates/livekit-core/src/lib.rs +++ b/crates/livekit-core/src/lib.rs @@ -7,5 +7,6 @@ mod signal_client; mod pc_transport; mod rtc_engine; mod local_participant; +mod event; -pub mod room; +pub mod room; \ No newline at end of file diff --git a/crates/livekit-core/src/lk_runtime.rs b/crates/livekit-core/src/lk_runtime.rs index ea349ef..061455d 100644 --- a/crates/livekit-core/src/lk_runtime.rs +++ b/crates/livekit-core/src/lk_runtime.rs @@ -6,7 +6,7 @@ use livekit_webrtc::peer_connection_factory::PeerConnectionFactory; use livekit_webrtc::webrtc::RTCRuntime; /// SAFETY: The order of initialization and deletion is important for LKRuntime. -/// See the C++ constructors & destructor of these fields +/// See the C++ constructors & destructors of these fields pub struct LKRuntime { pub pc_factory: PeerConnectionFactory, diff --git a/crates/livekit-core/src/local_participant.rs b/crates/livekit-core/src/local_participant.rs index 8654368..4ffc925 100644 --- a/crates/livekit-core/src/local_participant.rs +++ b/crates/livekit-core/src/local_participant.rs @@ -1,6 +1,4 @@ use std::sync::Arc; - -use futures_util::TryFutureExt; use tokio::sync::Mutex; use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket}; diff --git a/crates/livekit-core/src/pc_transport.rs b/crates/livekit-core/src/pc_transport.rs index 8d1724d..ca4f2b2 100644 --- a/crates/livekit-core/src/pc_transport.rs +++ b/crates/livekit-core/src/pc_transport.rs @@ -134,4 +134,3 @@ impl PCTransport { Ok(()) } } - diff --git a/crates/livekit-core/src/room.rs b/crates/livekit-core/src/room.rs index 583044f..d2070a2 100644 --- a/crates/livekit-core/src/room.rs +++ b/crates/livekit-core/src/room.rs @@ -2,7 +2,6 @@ use std::sync::Arc; use thiserror::Error; use tokio::sync::Mutex; -use tracing::{event, Level, trace}; use crate::local_participant::LocalParticipant; use crate::rtc_engine; @@ -14,6 +13,11 @@ pub enum RoomError { Engine(#[from] EngineError), } +#[derive(Debug)] +pub enum RoomEvent { + +} + pub struct Room { sid: String, name: String, @@ -24,18 +28,16 @@ pub struct Room { #[tracing::instrument(skip(url, token))] pub async fn connect(url: &str, token: &str) -> Result { let engine = rtc_engine::connect(url, token).await?; - - engine.on_data(Box::new(|packet| { - event!(Level::DEBUG, "received data"); - Box::pin(async move {}) - })).await; - let join = engine.join_response().await; let engine = Arc::new(Mutex::new(engine)); let local_participant = LocalParticipant::from(join.participant.unwrap(), engine.clone()); - let internal = Arc::new(RoomInternal::new(engine)); let room_info = join.room.unwrap(); + + tokio::spawn(async move { + + }); + Ok(Room { sid: room_info.sid, name: room_info.name, @@ -45,7 +47,11 @@ pub async fn connect(url: &str, token: &str) -> Result { } impl Room { - pub fn local_participant(&mut self) -> &mut LocalParticipant { + pub fn local_participant(&self) -> &LocalParticipant { + &self.local_participant + } + + pub fn local_participant_mut(&mut self) -> &mut LocalParticipant { &mut self.local_participant } diff --git a/crates/livekit-core/src/rtc_engine.rs b/crates/livekit-core/src/rtc_engine/engine_internal.rs similarity index 65% rename from crates/livekit-core/src/rtc_engine.rs rename to crates/livekit-core/src/rtc_engine/engine_internal.rs index c8cb166..7c98870 100644 --- a/crates/livekit-core/src/rtc_engine.rs +++ b/crates/livekit-core/src/rtc_engine/engine_internal.rs @@ -1,47 +1,38 @@ use std::fmt::{Debug, Formatter}; -use std::future::Future; -use std::pin::Pin; -use std::sync::{Arc, Weak}; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::Arc; use std::time::Duration; -use lazy_static::lazy_static; use prost::Message; use serde::{Deserialize, Serialize}; -use thiserror::Error; use tokio::sync::{mpsc, Mutex}; use tokio::time; use tracing::{event, Level}; -use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState}; -use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription}; +use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataState}; +use livekit_webrtc::jsep::{IceCandidate, SessionDescription}; use livekit_webrtc::peer_connection::{ IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions, }; use livekit_webrtc::peer_connection_factory::{ ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration, }; -use livekit_webrtc::rtc_error::RTCError; -use crate::{proto, signal_client}; use crate::lk_runtime::LKRuntime; use crate::pc_transport::PCTransport; -use crate::proto::{ - data_packet, DataPacket, JoinResponse, signal_request, signal_response, SignalTarget, - TrickleRequest, UserPacket, -}; +use crate::proto; use crate::proto::data_packet::Value; -use crate::signal_client::{SignalClient, SignalError}; +use crate::proto::{ + data_packet, signal_request, signal_response, DataPacket, JoinResponse, SignalTarget, + TrickleRequest, +}; +use crate::rtc_engine::{EngineError, MAX_ICE_CONNECT_TIMEOUT}; +use crate::signal_client::SignalClient; const LOSSY_DC_LABEL: &str = "_lossy"; const RELIABLE_DC_LABEL: &str = "_reliable"; -const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); - -lazy_static! { - // Share one LKRuntime across all RTCEngine instances - static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); -} +// Used to communicate IceCandidate with the server #[derive(Serialize, Deserialize)] #[allow(non_snake_case)] struct IceCandidateJSON { @@ -50,36 +41,8 @@ struct IceCandidateJSON { candidate: String, } -pub struct Packet { - pub data: UserPacket, - pub kind: data_packet::Kind, -} - -pub type OnDataHandler = -Box Pin + Send + 'static>>) + Send + Sync>; - -#[derive(Error, Debug)] -pub enum EngineError { - #[error("signal failure")] - Signal(#[from] SignalError), - #[error("internal webrtc failure")] - Rtc(#[from] RTCError), - #[error("failed to parse sdp")] - Parse(#[from] SdpParseError), - #[error("serde error")] - Serde(#[from] serde_json::Error), - #[error("failed to send data to the datachannel")] - Data(#[from] DataSendError), - #[error("connection error: {0}")] - Connection(String), - #[error("decode error")] - Decode(#[from] prost::DecodeError), - #[error("internal error: {0}")] - Internal(String), // Unexpected error -} - #[derive(Debug, Copy, Clone, PartialEq, Eq)] -enum PCState { +pub(crate) enum PCState { New, Connected, Disconnected, @@ -88,7 +51,7 @@ enum PCState { } #[derive(Debug)] -pub enum EngineMessage { +pub(crate) enum InternalMessage { IceCandidate { ice_candidate: IceCandidate, publisher: bool, @@ -109,113 +72,37 @@ pub enum EngineMessage { }, } +pub(crate) struct EngineInternal { + pub(super) publisher_pc: Arc>, + pub(super) subscriber_pc: Arc>, + pub(super) lossy_dc: Arc>, + pub(super) reliable_dc: Arc>, + pub(super) lossy_dc_sub: Arc>>, + pub(super) reliable_dc_sub: Arc>>, -#[derive(Debug)] -pub struct RTCEngine { - signal_client: Arc, - internal: Arc, - - #[allow(unused)] - lk_runtime: Arc, // Keep a reference while we're using the RTCEngine + pub(super) msg_sender: mpsc::Sender, + pub(super) join_response: Mutex, + pub(super) pc_state: AtomicU8, // casted to PCState + pub(super) has_published: AtomicBool, } -#[tracing::instrument(skip(url, token))] -pub async fn connect(url: &str, token: &str) -> Result { - // Acquire an existing/a new LKRuntime - let mut lk_runtime_ref = LK_RUNTIME.lock().await; - let mut lk_runtime = lk_runtime_ref.upgrade(); - - if lk_runtime.is_none() { - let new_runtime = Arc::new(LKRuntime::new()); - *lk_runtime_ref = Arc::downgrade(&new_runtime); - lk_runtime = Some(new_runtime); +impl Debug for EngineInternal { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + write!(f, "EngineInternal") } - let lk_runtime = lk_runtime.unwrap(); - let signal_client = Arc::new(signal_client::connect(url, token).await?); - - if let Some(signal_response::Message::Join(join_response)) = signal_client.recv().await { - event!(Level::DEBUG, "received JoinResponse: {:?}", join_response); - let (sender, receiver) = mpsc::channel(8); - let internal = Arc::new(EngineInternal::new( - lk_runtime.clone(), - sender, - join_response.clone(), - )?); - - if !join_response.subscriber_primary { - internal.publisher_pc.lock().await.negotiate().await?; - } - - tokio::spawn({ - let signal_client = signal_client.clone(); - let internal = internal.clone(); - - async move { - internal.run(receiver, signal_client).await; - } - }); - - Ok(RTCEngine { - lk_runtime, - signal_client, - internal, - }) - } else { - panic!("the first received message isn't a JoinResponse"); - } -} - -impl RTCEngine { - /// Send data to other participants in the Room - #[tracing::instrument] - pub async fn publish_data( - &mut self, - data: &DataPacket, - kind: data_packet::Kind, - ) -> Result<(), EngineError> { - self.internal.ensure_publisher_connected(kind).await?; - self.internal.data_channel(kind) - .lock() - .await - .send(&data.encode_to_vec(), true) - .map_err(Into::into) - } - - /// Return the last received JoinResponse - pub async fn join_response(&self) -> JoinResponse { - self.internal.join_response.lock().await.clone() - } - - pub async fn on_data(&self, f: OnDataHandler) { - *self.internal.on_data_handler.lock().await = Some(f); - } -} - -struct EngineInternal { - publisher_pc: Arc>, - subscriber_pc: Arc>, - lossy_dc: Arc>, - reliable_dc: Arc>, - lossy_dc_sub: Arc>>, - reliable_dc_sub: Arc>>, - - msg_sender: mpsc::Sender, - join_response: Mutex, - pc_state: AtomicU8, - // PCState - has_published: AtomicBool, - - // Listeners - on_data_handler: Arc>>, } impl EngineInternal { - /// New internal is created on connect & on reconnect - /// It creates the PeerConnections, the DataChannels and the libwebrtc listeners + /// Configure the PeerConnections + /// + /// This is called on connect & on full reconnect. + /// Create the PeerConnections & the DataChannels. + /// Register listeners and send the internal messages + /// to the event_loop. #[tracing::instrument] - fn new( + pub(super) fn configure( lk_runtime: Arc, - sender: mpsc::Sender, + sender: mpsc::Sender, join: JoinResponse, ) -> Result { let rtc_config = RTCConfiguration { @@ -245,7 +132,7 @@ impl EngineInternal { publisher_pc.peer_connection().on_ice_candidate(Box::new({ let sender = sender.clone(); move |ice_candidate| { - let _ = sender.blocking_send(EngineMessage::IceCandidate { + let _ = sender.blocking_send(InternalMessage::IceCandidate { ice_candidate, publisher: true, }); @@ -255,7 +142,7 @@ impl EngineInternal { subscriber_pc.peer_connection().on_ice_candidate(Box::new({ let sender = sender.clone(); move |ice_candidate| { - let _ = sender.blocking_send(EngineMessage::IceCandidate { + let _ = sender.blocking_send(InternalMessage::IceCandidate { ice_candidate, publisher: false, }); @@ -268,7 +155,7 @@ impl EngineInternal { let sender = sender.clone(); tokio::spawn(async move { - let _ = sender.send(EngineMessage::PublisherOffer { offer }).await; + let _ = sender.send(InternalMessage::PublisherOffer { offer }).await; }); Box::pin(async move {}) @@ -285,7 +172,7 @@ impl EngineInternal { let sender = sender.clone(); move |data_channel| { let _ = - sender.blocking_send(EngineMessage::PrimaryDataChannel { data_channel }); + sender.blocking_send(InternalMessage::PrimaryDataChannel { data_channel }); } })); } @@ -293,7 +180,7 @@ impl EngineInternal { primary_pc.peer_connection().on_connection_change(Box::new({ let sender = sender.clone(); move |state| { - let _ = sender.blocking_send(EngineMessage::ConnectionChange { + let _ = sender.blocking_send(InternalMessage::ConnectionChange { state, primary: true, }); @@ -305,15 +192,13 @@ impl EngineInternal { .on_connection_change(Box::new({ let sender = sender.clone(); move |state| { - let _ = sender.blocking_send(EngineMessage::ConnectionChange { + let _ = sender.blocking_send(InternalMessage::ConnectionChange { state, primary: false, }); } })); - // Note that when subscriber_primary feature is enabled, - // the subscriber uses his own data channels created by the server. let mut lossy_dc = publisher_pc.peer_connection().create_data_channel( LOSSY_DC_LABEL, DataChannelInit { @@ -345,26 +230,26 @@ impl EngineInternal { join_response: Mutex::new(join), pc_state: AtomicU8::new(PCState::New as u8), has_published: AtomicBool::new(false), - on_data_handler: Default::default(), }) } - /// Map the libwebrtc listeners to a mpsc channel + /// Send InternalMessage when a datachannel receives data #[tracing::instrument] - fn configure_dc( - data_channel: &mut DataChannel, - sender: mpsc::Sender, - ) { + fn configure_dc(data_channel: &mut DataChannel, sender: mpsc::Sender) { data_channel.on_message(Box::new(move |data, binary| { - let _ = sender.blocking_send(EngineMessage::Data { + let _ = sender.blocking_send(InternalMessage::Data { data: data.to_vec(), binary, }); })); } + /// Ensure the publisher PeerConnection is connected + /// + /// When subscriber_primary is enabled, only the subscriber PeerConnection is negotiated. + /// This allows for faster connection when we don't need the publisher #[tracing::instrument] - async fn ensure_publisher_connected( + pub(super) async fn ensure_publisher_connected( self: &Arc, kind: data_packet::Kind, ) -> Result<(), EngineError> { @@ -377,7 +262,7 @@ impl EngineInternal { let mut publisher = publisher.lock().await; if !publisher.is_connected() && publisher.peer_connection().ice_connection_state() - != IceConnectionState::IceConnectionChecking + != IceConnectionState::IceConnectionChecking { tokio::spawn({ let internal = self.clone(); @@ -406,7 +291,7 @@ impl EngineInternal { interval.tick().await; } }) - .await; + .await; if res.is_err() { let err = @@ -418,88 +303,11 @@ impl EngineInternal { } } + /// Run the event_loop of the RTCEngine #[tracing::instrument] - async fn handle_signal( + pub(super) async fn run( self: &Arc, - signal: signal_response::Message, - signal_client: Arc, - ) -> Result<(), EngineError> { - match signal { - signal_response::Message::Answer(answer) => { - event!(Level::TRACE, "received answer for publisher: {:?}", answer); - let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; - self.publisher_pc - .lock() - .await - .set_remote_description(sdp) - .await?; - } - signal_response::Message::Offer(offer) => { - event!(Level::TRACE, "received offer for subscriber: {:?}", offer); - let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; - - self.subscriber_pc - .lock() - .await - .set_remote_description(sdp) - .await?; - let answer = self.subscriber_pc - .lock() - .await - .peer_connection() - .create_answer(RTCOfferAnswerOptions::default()) - .await?; - self.subscriber_pc - .lock() - .await - .peer_connection() - .set_local_description(answer.clone()) - .await?; - - tokio::spawn(async move { - let _ = 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 json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?; - let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?; - - event!( - Level::TRACE, - "received ice_candidate ({:?}) - {:?}", - SignalTarget::from_i32(trickle.target).unwrap(), - ice - ); - - if trickle.target == SignalTarget::Publisher as i32 { - self.publisher_pc - .lock() - .await - .add_ice_candidate(ice) - .await?; - } else { - self.subscriber_pc - .lock() - .await - .add_ice_candidate(ice) - .await?; - } - } - _ => {} - } - - Ok(()) - } - - #[tracing::instrument] - pub async fn run( - self: &Arc, - mut receiver: mpsc::Receiver, + mut receiver: mpsc::Receiver, signal_client: Arc, ) { loop { @@ -533,17 +341,116 @@ impl EngineInternal { } } + /// Handle SignalResponse messages coming from the server + /// + /// Run the needed livekit-protocol + #[tracing::instrument] + async fn handle_signal( + self: &Arc, + signal: signal_response::Message, + signal_client: Arc, + ) -> Result<(), EngineError> { + match signal { + signal_response::Message::Answer(answer) => { + event!(Level::TRACE, "received answer for publisher: {:?}", answer); + let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; + self.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 + event!(Level::TRACE, "received offer for subscriber: {:?}", offer); + let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; + + self.subscriber_pc + .lock() + .await + .set_remote_description(sdp) + .await?; + let answer = self + .subscriber_pc + .lock() + .await + .peer_connection() + .create_answer(RTCOfferAnswerOptions::default()) + .await?; + self.subscriber_pc + .lock() + .await + .peer_connection() + .set_local_description(answer.clone()) + .await?; + + tokio::spawn(async move { + let _ = 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)?; + + event!( + Level::TRACE, + "received ice_candidate ({:?}) - {:?}", + SignalTarget::from_i32(trickle.target).unwrap(), + ice + ); + + if trickle.target == SignalTarget::Publisher as i32 { + self.publisher_pc + .lock() + .await + .add_ice_candidate(ice) + .await?; + } else { + self.subscriber_pc + .lock() + .await + .add_ice_candidate(ice) + .await?; + } + } + _ => {} + } + + Ok(()) + } + + /// Handle libwebrtc messages + /// + /// Every message used inside this function comes from libwebrtc. + /// The messages are received in [EngineInternal](#run) + /// We're not handling the messages inside the signaling_thread, to return + /// as quickly as possible. #[tracing::instrument] async fn handle_message( self: &Arc, - msg: EngineMessage, + msg: InternalMessage, signal_client: Arc, ) -> Result<(), EngineError> { match msg { - EngineMessage::IceCandidate { + InternalMessage::IceCandidate { ice_candidate, publisher, } => { + // Send the IceCandidate to livekit-server + // Note that ContinualGatheringPolicy is set to GatherContinually + let json = serde_json::to_string(&IceCandidateJSON { + sdpMid: ice_candidate.sdp_mid(), + sdpMLineIndex: ice_candidate.sdp_mline_index(), + candidate: ice_candidate.candidate(), + })?; + let target = if publisher { SignalTarget::Publisher } else { @@ -557,23 +464,19 @@ impl EngineInternal { ice_candidate ); - let json = serde_json::to_string(&IceCandidateJSON { - sdpMid: ice_candidate.sdp_mid(), - sdpMLineIndex: ice_candidate.sdp_mline_index(), - candidate: ice_candidate.candidate(), - })?; - - // Send the ice_candidate to the server tokio::spawn(async move { - let _ = signal_client.send(signal_request::Message::Trickle( - TrickleRequest { + let _ = signal_client + .send(signal_request::Message::Trickle(TrickleRequest { candidate_init: json, target: target as i32, - }, - )).await; + })) + .await; }); } - EngineMessage::ConnectionChange { state, primary } => { + InternalMessage::ConnectionChange { state, primary } => { + // PeerConnectionState changed + // Reconnect if we've been disconnected unexpectedly + // If connected for the first time, send OnConnect event if primary && state == PeerConnectionState::Connected { let old_state = self.pc_state.load(Ordering::SeqCst); self.pc_state @@ -583,18 +486,21 @@ impl EngineInternal { // TODO(theomonnom) OnConnected } } else if state == PeerConnectionState::Failed { - self.pc_state.store(PCState::Disconnected as u8, Ordering::SeqCst); + self.pc_state + .store(PCState::Disconnected as u8, Ordering::SeqCst); // TODO(theomonnom) handle Disconnect } } - EngineMessage::PrimaryDataChannel { mut data_channel } => { + InternalMessage::PrimaryDataChannel { mut data_channel } => { + // Received datachannel from the primary PeerConnection. + // If subscriber_primary is enabled, the datachannel is used for downstream data let reliable = data_channel.label() == RELIABLE_DC_LABEL; Self::configure_dc(&mut data_channel, self.msg_sender.clone()); event!( Level::TRACE, - "received subscriber data_channel - {:?}", + "received primary data_channel - {:?}", data_channel ); @@ -604,44 +510,40 @@ impl EngineInternal { *self.lossy_dc_sub.lock().await = Some(data_channel); } } - EngineMessage::PublisherOffer { offer } => { - event!( - Level::TRACE, - "sending publisher offer - {:?}", - offer - ); + InternalMessage::PublisherOffer { offer } => { + // Send the publisher offer to livekit-server + event!(Level::TRACE, "sending publisher offer - {:?}", offer); - // Send the offer to the server tokio::spawn(async move { - let _ = signal_client.send(signal_request::Message::Offer( - proto::SessionDescription { + let _ = signal_client + .send(signal_request::Message::Offer(proto::SessionDescription { r#type: "offer".to_string(), sdp: offer.to_string(), - }, - )).await; + })) + .await; }); } - EngineMessage::Data { - data, - binary, - } => { + InternalMessage::Data { data, binary } => { + // Received data from a datachannel + // If this is a Speaker DataPacket, update the active speakers + // Send SpeakersChanged/OnData event if !binary { return Err(EngineError::Internal( - "text messages aren't supported by LiveKit".to_string(), + "text messages aren't supported".to_string(), )); } let data = DataPacket::decode(&*data)?; match data.value.unwrap() { Value::User(user) => { - let mut handler = self.on_data_handler.lock().await; + /*let mut handler = self.on_data_handler.lock().await; if let Some(f) = &mut *handler { f(Packet { data: user, kind: data_packet::Kind::from_i32(data.kind).unwrap(), }) .await; - } + }*/ } Value::Speaker(_) => { // TODO(theomonnonm) @@ -657,18 +559,14 @@ impl EngineInternal { async fn negotiate_publisher(self: &Arc) -> Result<(), EngineError> { self.has_published.store(true, Ordering::SeqCst); if let Err(err) = self.publisher_pc.lock().await.negotiate().await { - event!( - Level::ERROR, - "failed to negotiate the publisher: {:?}", - err, - ); + event!(Level::ERROR, "failed to negotiate the publisher: {:?}", err,); Err(EngineError::Rtc(err)) } else { Ok(()) } } - fn data_channel(&self, kind: data_packet::Kind) -> Arc> { + pub(super) fn data_channel(&self, kind: data_packet::Kind) -> Arc> { if kind == data_packet::Kind::Reliable { self.reliable_dc.clone() } else { @@ -676,9 +574,3 @@ impl EngineInternal { } } } - -impl Debug for EngineInternal { - fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { - write!(f, "EngineInternal") - } -} diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs new file mode 100644 index 0000000..180097f --- /dev/null +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -0,0 +1,153 @@ +use crate::lk_runtime::LKRuntime; +use crate::proto::{data_packet, signal_response, DataPacket, JoinResponse, UserPacket}; +use crate::rtc_engine::engine_internal::EngineInternal; +use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalOptions}; +use futures_util::{FutureExt, StreamExt}; +use lazy_static::lazy_static; +use livekit_webrtc::data_channel::DataSendError; +use livekit_webrtc::jsep::SdpParseError; +use livekit_webrtc::rtc_error::RTCError; +use prost::Message; +use std::sync::{Arc, Weak}; +use std::time::Duration; +use thiserror::Error; +use tokio::sync::{mpsc, Mutex}; +use tokio::time; +use tracing::{event, Level}; + +mod engine_internal; + +lazy_static! { + // Share one LKRuntime across all RTCEngine instances + static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); +} + +pub(crate) const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); +pub(crate) const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Error, Debug)] +pub enum EngineError { + #[error("signal failure")] + Signal(#[from] SignalError), + #[error("internal webrtc failure")] + Rtc(#[from] RTCError), + #[error("failed to parse sdp")] + Parse(#[from] SdpParseError), + #[error("serde error")] + Serde(#[from] serde_json::Error), + #[error("failed to send data to the datachannel")] + Data(#[from] DataSendError), + #[error("connection error: {0}")] + Connection(String), + #[error("decode error")] + Decode(#[from] prost::DecodeError), + #[error("internal error: {0}")] + Internal(String), // Unexpected error +} + +#[derive(Debug)] +pub struct Packet { + pub data: UserPacket, + pub kind: data_packet::Kind, +} + +#[derive(Debug)] +pub enum EngineEvent { + DataReceived(Packet), +} + +#[derive(Debug)] +pub struct RTCEngine { + signal_client: Arc, + internal: Arc, + + #[allow(unused)] + lk_runtime: Arc, // Keep a reference while we're using the RTCEngine +} + +#[tracing::instrument(skip(url, token))] +pub async fn connect( + url: &str, + token: &str, + options: SignalOptions, +) -> Result { + // Acquire an existing/a new LKRuntime + let mut lk_runtime_ref = LK_RUNTIME.lock().await; + let mut lk_runtime = lk_runtime_ref.upgrade(); + + if lk_runtime.is_none() { + let new_runtime = Arc::new(LKRuntime::new()); + *lk_runtime_ref = Arc::downgrade(&new_runtime); + lk_runtime = Some(new_runtime); + } + let lk_runtime = lk_runtime.unwrap(); + let (signal_client, mut signal_events) = SignalClient::connect(url, token, options).await?; + let signal_client = Arc::new(signal_client); + + let join_response = time::timeout(JOIN_RESPONSE_TIMEOUT, async move { + while let Some(event) = signal_events.next().await { + match event { + SignalEvent::Signal(signal_response::Message::Join(join)) => return join, + _ => { + // Should we try a reconnect on close here? + continue; + } + } + } + + unreachable!(); + }) + .await + .map_err(|_| EngineError::Internal("failed to receive JoinResponse".to_string()))?; + + event!(Level::DEBUG, "received JoinResponse: {:?}", join_response); + + let (sender, receiver) = mpsc::channel(8); + let internal = Arc::new(EngineInternal::configure( + lk_runtime.clone(), + sender, + join_response.clone(), + )?); + + if !join_response.subscriber_primary { + internal.publisher_pc.lock().await.negotiate().await?; + } + + tokio::spawn({ + let signal_client = signal_client.clone(); + let internal = internal.clone(); + + async move { + internal.run(receiver, signal_client).await; + } + }); + + Ok(RTCEngine { + lk_runtime, + signal_client, + internal, + }) +} + +impl RTCEngine { + /// Send data to other participants in the Room + #[tracing::instrument] + pub async fn publish_data( + &mut self, + data: &DataPacket, + kind: data_packet::Kind, + ) -> Result<(), EngineError> { + self.internal.ensure_publisher_connected(kind).await?; + self.internal + .data_channel(kind) + .lock() + .await + .send(&data.encode_to_vec(), true) + .map_err(Into::into) + } + + /// Return the last received JoinResponse + pub async fn join_response(&self) -> JoinResponse { + self.internal.join_response.lock().await.clone() + } +}