diff --git a/crates/livekit-core/src/rtc_engine.rs b/crates/livekit-core/src/rtc_engine.rs index 56b08f1..1d0bd66 100644 --- a/crates/livekit-core/src/rtc_engine.rs +++ b/crates/livekit-core/src/rtc_engine.rs @@ -1,13 +1,12 @@ -use std::sync::{Arc, Mutex, Weak}; +use std::sync::{Arc, Weak}; +use std::sync::atomic::{AtomicU8, Ordering}; use std::time::Duration; use lazy_static::lazy_static; use log::{error, trace}; -use prost::Message as ProstMessage; use thiserror::Error; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, Mutex}; use tokio::time::sleep; -use tokio_tungstenite::tungstenite::protocol::frame::coding::Data; use livekit_webrtc::data_channel::{DataChannel, DataChannelInit}; use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription}; @@ -20,14 +19,17 @@ use livekit_webrtc::rtc_error::RTCError; use crate::{proto, signal_client}; use crate::lk_runtime::LKRuntime; use crate::pc_transport::PCTransport; -use crate::proto::{ - DataPacket, JoinResponse, signal_request, signal_response, SignalTarget, TrickleRequest, -}; +use crate::proto::{DataPacket, JoinResponse, signal_request, signal_response, SignalResponse, SignalTarget, TrickleRequest}; use crate::signal_client::{SignalClient, SignalError}; const LOSSY_DC_LABEL: &str = "_lossy"; const RELIABLE_DC_LABEL: &str = "_reliable"; +lazy_static! { + // Share one LKRuntime across all RTCEngine instances + static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); +} + #[derive(Error, Debug)] pub enum EngineError { #[error("signal failure")] @@ -40,7 +42,7 @@ pub enum EngineError { Serde(#[from] serde_json::Error), } -#[derive(PartialEq, Debug, Copy, Clone)] +#[derive(Debug, Copy, Clone, PartialEq, Eq)] enum PCState { New, Connected, @@ -49,107 +51,121 @@ enum PCState { Closed, } -lazy_static! { - // Share one LKRuntime across all RTCEngine instances - static ref LK_RUNTIME: Mutex> = Mutex::new(Weak::new()); -} - -enum EngineMessage {} - -struct PeerInternal { - publisher_pc: PCTransport, - subscriber_pc: PCTransport, - - lossy_dc: DataChannel, - reliable_dc: DataChannel, - - pub_ice_rx: mpsc::Receiver, - sub_ice_rx: mpsc::Receiver, - - pub_offer_rx: mpsc::Receiver, - - primary_connection_state_rx: mpsc::Receiver, - secondary_connection_state_rx: mpsc::Receiver, - - lossy_data_rx: mpsc::Receiver, - reliable_data_rx: mpsc::Receiver, - - sub_dc_rx: mpsc::Receiver, - - pc_state: PCState, -} - -struct RTCInternal { - #[allow(unused)] - lk_runtime: Arc, - signal_client: Arc, - pc_internal: PeerInternal, -} - -impl RTCInternal { - async fn connect(url: &str, token: &str) -> Result { - let mut lk_runtime = None; - { - // Acquire an existing/a new LKRuntime - let mut lk_runtime_ref = LK_RUNTIME.lock().unwrap(); - 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 = Arc::new(signal_client::connect(url, token).await?); - - trace!("waiting join_response.."); - if let signal_response::Message::Join(join) = signal_client.recv().await? { - trace!("configuring peer_connections: {:?}", join); - let mut pc_internal = Self::configure(lk_runtime.clone(), join.clone())?; - - if !join.subscriber_primary { - pc_internal.publisher_pc.negotiate().await?; - } - - Ok(Self { - lk_runtime, - signal_client, - pc_internal, - }) - } else { - panic!("the first received message isn't a JoinResponse"); - } +#[derive(Debug)] +pub enum EngineMessage { + IceCandidate { + ice_candidate: IceCandidate, + publisher: bool + }, + ConnectionChange { + state: PeerConnectionState, + primary: bool + }, + PrimaryDataChannel { + data_channel: DataChannel, + }, + PublisherOffer { + offer: SessionDescription, + }, + Data { + data: Vec, + binary: bool, + reliable: bool } +} + +struct EngineInternal { + publisher_pc: Arc>, + subscriber_pc: Arc>, + + lossy_dc: Arc>, + reliable_dc: Arc>, + + msg_sender: mpsc::Sender, + + pc_state: AtomicU8, // PCState +} + +pub struct RTCEngine { + #[allow(unused)] + lk_runtime: Arc, // Keep a reference while we're using the RTCEngine + signal_client: Arc, + internal: Arc +} + +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); + } + let lk_runtime = lk_runtime.unwrap(); + let signal_client = Arc::new(signal_client::connect(url, token).await?); + + if let signal_response::Message::Join(join_response) = signal_client.recv().await? { + trace!("received join_response: {:?}", join_response); + let (sender, receiver) = mpsc::channel(8); + let internal = Arc::new(RTCEngine::configure(lk_runtime.clone(), sender, join_response.clone())?); + + if !join_response.subscriber_primary { + internal.publisher_pc.lock().await.negotiate().await?; + } - fn request_signal(&mut self, msg: signal_request::Message) { tokio::spawn({ - let sc = self.signal_client.clone(); + let signal_client = signal_client.clone(); + let internal = internal.clone(); async move { - if let Err(err) = sc.send(msg).await { - error!("failed to send signal: {:?}", err); - } + RTCEngine::handle_loop(receiver, signal_client, internal).await; + } + }); + + Ok(RTCEngine { + lk_runtime, + signal_client, + internal, + }) + } else { + panic!("the first received message isn't a JoinResponse"); + } +} + +impl RTCEngine { + + async fn send_data() { + + } + + fn send_request(msg: signal_request::Message, signal_client: Arc) { + tokio::spawn(async move { + if let Err(err) = signal_client.send(msg).await { + error!("failed to send signal: {:?}", err); } }); } - async fn handle_signal(&mut self, signal: signal_response::Message) -> Result<(), EngineError> { + async fn handle_signal(signal: signal_response::Message, signal_client: &Arc, rtc_internal: &Arc) -> Result<(), EngineError> { match signal { signal_response::Message::Answer(answer) => { let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; - self.pc_internal.publisher_pc.set_remote_description(sdp).await?; + rtc_internal.publisher_pc.lock().await.set_remote_description(sdp).await?; }, signal_response::Message::Offer(offer) => { let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; - self.pc_internal.subscriber_pc.set_remote_description(sdp).await?; - let answer = self.pc_internal.subscriber_pc.peer_connection().create_answer(RTCOfferAnswerOptions::default()).await?; - self.pc_internal.subscriber_pc.peer_connection().set_local_description(answer.clone()).await?; + let mut subscriber_pc = rtc_internal.subscriber_pc.lock().await; - self.request_signal(signal_request::Message::Answer(proto::SessionDescription { + subscriber_pc.set_remote_description(sdp).await?; + let answer = subscriber_pc.peer_connection().create_answer(RTCOfferAnswerOptions::default()).await?; + subscriber_pc.peer_connection().set_local_description(answer.clone()).await?; + + Self::send_request(signal_request::Message::Answer(proto::SessionDescription { r#type: "answer".to_string(), sdp: answer.to_string(), - })); + }), signal_client.clone()); }, signal_response::Message::Trickle(trickle) => { let json: serde_json::Value = serde_json::from_str(&trickle.candidate_init)?; @@ -160,9 +176,9 @@ impl RTCInternal { )?; if trickle.target == SignalTarget::Publisher as i32 { - self.pc_internal.publisher_pc.add_ice_candidate(ice).await?; + rtc_internal.publisher_pc.lock().await.add_ice_candidate(ice).await?; } else { - self.pc_internal.subscriber_pc.add_ice_candidate(ice).await?; + rtc_internal.subscriber_pc.lock().await.add_ice_candidate(ice).await?; } } _ => {}, @@ -171,94 +187,67 @@ impl RTCInternal { Ok(()) } - async fn run(&mut self) { + async fn handle_loop(mut receiver: mpsc::Receiver, signal_client: Arc, rtc_internal: Arc) { loop { tokio::select! { - Ok(signal) = self.signal_client.recv() => { - if let Err(err) = self.handle_signal(signal).await { + Ok(signal) = signal_client.recv() => { + if let Err(err) = Self::handle_signal(signal, &signal_client, &rtc_internal).await { error!("failed to handle signal: {:?}", err); } }, - Some(ice_candidate) = self.pc_internal.pub_ice_rx.recv() => { - self.request_signal(signal_request::Message::Trickle(TrickleRequest { - candidate_init: ice_candidate.to_string(), - target: SignalTarget::Publisher as i32 - })); - }, - Some(ice_candidate) = self.pc_internal.sub_ice_rx.recv() => { - self.request_signal(signal_request::Message::Trickle(TrickleRequest { - candidate_init: ice_candidate.to_string(), - target: SignalTarget::Subscriber as i32 - })); - }, - Some(sdp) = self.pc_internal.pub_offer_rx.recv() => { - trace!("received publisher offer: {:?}", sdp); - self.request_signal(signal_request::Message::Offer(proto::SessionDescription { - r#type: "offer".to_string(), - sdp: sdp.to_string(), - })); - }, - Some(state) = self.pc_internal.primary_connection_state_rx.recv() => { - if state == PeerConnectionState::Connected { - let old_state = self.pc_internal.pc_state; - self.pc_internal.pc_state = PCState::Connected; - - if old_state == PCState::New { - // TODO(theomonnom) OnConnected + Some(msg) = receiver.recv() => { + match msg { + EngineMessage::IceCandidate { ice_candidate, publisher } => { + trace!("received ice_candidate: {:?} (publisher: {:?})", ice_candidate, publisher); + // Send the ice_candidate to the server + Self::send_request(signal_request::Message::Trickle(TrickleRequest { + candidate_init: ice_candidate.to_string(), + target: if publisher {SignalTarget::Publisher} else {SignalTarget::Subscriber} as i32 + }), signal_client.clone()); } - } else if state == PeerConnectionState::Failed { - self.pc_internal.pc_state = PCState::Disconnected; - // TODO(theomonnom) Handle Disconnect - } - }, - Some(state) = self.pc_internal.secondary_connection_state_rx.recv() => { - if state == PeerConnectionState::Failed { - self.pc_internal.pc_state = PCState::Disconnected; - // TODO(theomonnom) Handle Disconnect - } - }, - Some(data) = self.pc_internal.lossy_data_rx.recv() => { + EngineMessage::ConnectionChange { state, primary } => { + if primary && state == PeerConnectionState::Connected { + let old_state = rtc_internal.pc_state.load(Ordering::SeqCst); + rtc_internal.pc_state.store(PCState::Connected as u8, Ordering::SeqCst); - }, - Some(data) = self.pc_internal.reliable_data_rx.recv() => { + if old_state == PCState::New as u8 { + // TODO(theomonnom) OnConnected + } + } else if state == PeerConnectionState::Failed { + rtc_internal.pc_state.store(PCState::Disconnected as u8, Ordering::SeqCst); + // TODO(theomonnom) handle Disconnect + } + } + EngineMessage::PrimaryDataChannel { mut data_channel } => { + let reliable = data_channel.label() == RELIABLE_DC_LABEL; + Self::configure_dc(&mut data_channel, reliable, rtc_internal.msg_sender.clone()); - }, - Some(mut dc) = self.pc_internal.sub_dc_rx.recv() => { - // Subscriber DataChannels - // Only received when the subscriber_primary is enabled - trace!("using subscriber data channels"); - - let (data_tx, data_rx) = mpsc::channel(8); - Self::configure_dc(&mut dc, data_tx); - - if dc.label() == RELIABLE_DC_LABEL { - self.pc_internal.reliable_dc = dc; - self.pc_internal.reliable_data_rx = data_rx; - } else { - self.pc_internal.lossy_dc = dc; - self.pc_internal.lossy_data_rx = data_rx; + trace!("received and using subscriber datachannel (reliable: {:?})", reliable); + if reliable { + *rtc_internal.reliable_dc.lock().await = data_channel; + } else { + *rtc_internal.lossy_dc.lock().await = data_channel; + } + } + EngineMessage::PublisherOffer { offer } => { + trace!("received publisher offer: {:?}", offer); + // Send the offer to the server + Self::send_request(signal_request::Message::Offer(proto::SessionDescription { + r#type: "offer".to_string(), + sdp: offer.to_string(), + }), signal_client.clone()); + } + EngineMessage::Data { data, binary, reliable } => {} } } } } } - fn configure_dc(data_channel: &mut DataChannel, data_tx: mpsc::Sender) { - let label = data_channel.label(); - data_channel.on_message(Box::new(move |data, _| { - if let Ok(data) = DataPacket::decode(data) { - let _ = data_tx.blocking_send(data); - } else { - trace!("{} - failed to decode DataPacket", label); - } - })); - } - - fn configure( - lk_runtime: Arc, - join: JoinResponse, - ) -> Result { - let cfg = RTCConfiguration { + /// This function is called on connect & on reconnect + /// It creates the PeerConnections, the DataChannels & the libwebrtc listeners + fn configure(lk_runtime: Arc, sender: mpsc::Sender, join: JoinResponse) -> Result { + let rtc_config = RTCConfiguration { ice_servers: { let mut servers = vec![]; for is in join.ice_servers { @@ -274,36 +263,34 @@ impl RTCInternal { ice_transport_type: IceTransportsType::All, }; - // Create the PeerConnections - let mut publisher_pc = PCTransport::new(lk_runtime.clone(), cfg.clone())?; - let mut subscriber_pc = PCTransport::new(lk_runtime, cfg)?; + let mut publisher_pc = PCTransport::new(lk_runtime.pc_factory.create_peer_connection(rtc_config.clone())?); + let mut subscriber_pc = PCTransport::new(lk_runtime.pc_factory.create_peer_connection(rtc_config)?); - let (pub_ice_tx, pub_ice_rx) = mpsc::channel(8); - let (sub_ice_tx, sub_ice_rx) = mpsc::channel(8); - let (pub_offer_tx, pub_offer_rx) = mpsc::channel(8); - let (primary_connection_state_tx, primary_connection_state_rx) = mpsc::channel(8); - let (secondary_connection_state_tx, secondary_connection_state_rx) = mpsc::channel(8); - let (lossy_data_tx, lossy_data_rx) = mpsc::channel(8); - let (reliable_data_tx, reliable_data_rx) = mpsc::channel(8); - let (sub_dc_tx, sub_dc_rx) = mpsc::channel(8); + publisher_pc.peer_connection().on_ice_candidate(Box::new({ + let sender = sender.clone(); + move |ice_candidate| { + let _ = sender.blocking_send(EngineMessage::IceCandidate { + ice_candidate, + publisher: true + }); + } + })); - publisher_pc - .peer_connection() - .on_ice_candidate(Box::new(move |ice_candidate| { - trace!("publisher - on_ice_candidate: {:?}", ice_candidate); - let _ = pub_ice_tx.blocking_send(ice_candidate); - })); + subscriber_pc.peer_connection().on_ice_candidate(Box::new({ + let sender = sender.clone(); + move |ice_candidate| { + let _ = sender.blocking_send(EngineMessage::IceCandidate { + ice_candidate, + publisher: false + }); + } + })); - subscriber_pc - .peer_connection() - .on_ice_candidate(Box::new(move |ice_candidate| { - trace!("subscriber - on_ice_candidate: {:?}", ice_candidate); - let _ = sub_ice_tx.blocking_send(ice_candidate); - })); - - publisher_pc.on_offer(Box::new(move |offer| { - trace!("publisher - on_offer: {:?}", offer); - let _ = pub_offer_tx.blocking_send(offer); // TODO(theomonnom) Don't use blocking_send here + publisher_pc.on_offer(Box::new({ + let sender = sender.clone(); + move |offer| { + let _ = sender.blocking_send(EngineMessage::PublisherOffer {offer}); + } })); let mut primary_pc = &mut publisher_pc; @@ -312,21 +299,36 @@ impl RTCInternal { primary_pc = &mut subscriber_pc; secondary_pc = &mut publisher_pc; - primary_pc.peer_connection().on_data_channel(Box::new(move |dc| { - let _ = sub_dc_tx.blocking_send(dc); - })); + primary_pc.peer_connection().on_data_channel(Box::new({{ + let sender = sender.clone(); + move |data_channel| { + let _ = sender.blocking_send(EngineMessage::PrimaryDataChannel {data_channel}); + } + }})); } primary_pc .peer_connection() - .on_connection_change(Box::new(move |state| { - let _ = primary_connection_state_tx.blocking_send(state); + .on_connection_change(Box::new({ + let sender = sender.clone(); + move |state| { + let _ = sender.blocking_send(EngineMessage::ConnectionChange { + state, + primary: true + }); + } })); secondary_pc .peer_connection() - .on_connection_change(Box::new(move |state| { - let _ = secondary_connection_state_tx.blocking_send(state); + .on_connection_change(Box::new({ + let sender = sender.clone(); + move |state| { + let _ = sender.blocking_send(EngineMessage::ConnectionChange { + state, + primary: false + }); + } })); // Note that when subscriber_primary feature is enabled, @@ -348,72 +350,35 @@ impl RTCInternal { }, )?; - Self::configure_dc(&mut lossy_dc, lossy_data_tx); - Self::configure_dc(&mut reliable_dc, reliable_data_tx); + Self::configure_dc(&mut lossy_dc, true, sender.clone()); + Self::configure_dc(&mut reliable_dc, false, sender.clone()); - Ok(PeerInternal { - publisher_pc, - subscriber_pc, - lossy_dc, - reliable_dc, - pub_ice_rx, - sub_ice_rx, - pub_offer_rx, - primary_connection_state_rx, - secondary_connection_state_rx, - lossy_data_rx, - reliable_data_rx, - sub_dc_rx, - pc_state: PCState::New, + Ok(EngineInternal { + publisher_pc: Arc::new(Mutex::new(publisher_pc)), + subscriber_pc: Arc::new(Mutex::new(subscriber_pc)), + lossy_dc: Arc::new(Mutex::new(lossy_dc)), + reliable_dc: Arc::new(Mutex::new(reliable_dc)), + pc_state: AtomicU8::new(PCState::New as u8), + msg_sender: sender }) } -} -pub struct RTCEngine {} - -/// Initialize the SignalClient & the PeerConnections -pub async fn connect(url: &str, token: &str) -> Result { - let mut rtc_internal = RTCInternal::connect(url, token).await?; - tokio::spawn(async move { - rtc_internal.run().await - }); - - Ok(RTCEngine{}) -} - -impl RTCEngine { - async fn rtc_handle() { - loop {} + /// Map the libwebrtc listeners to a mpsc channel + fn configure_dc(data_channel: &mut DataChannel, reliable: bool, sender: mpsc::Sender) { + data_channel.on_message(Box::new(move |data, binary| { + let _ = sender.blocking_send(EngineMessage::Data { + data: data.to_vec(), + reliable, + binary + }); + })); } } #[tokio::test] async fn test_test() { env_logger::init(); - let engine = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap(); - sleep(Duration::from_secs(60)).await; - } - -/*sync fn handle_rtc(mut signal_receiver: broadcast::Receiver) { - loop { - let msg = match signal_receiver.recv().await { - Ok(msg) => msg, - Err(error) => { - error!("Failed to receive SignalResponse: {:?}", error); - continue; - } - }; - - match msg { - Message::Join(join) => {} - Message::Trickle(trickle) => {} - Message::Answer(answer) => {} - Message::Offer(offer) => {} - _ => {} - } - } -}*/