talksik/client-sdk-rust
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

wip x2

Browse Source
This commit is contained in:
Théo Monnom
2022-12-16 00:04:53 +01:00
parent 4606c0a9ec
commit 86bec130bb
1 changed files with 106 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/livekit-core/src/rtc_engine/engine_internal.rs
+106 -92
View File
@@ -1,28 +1,23 @@
use parking_lot::{Mutex, RwLock}; use parking_lot::{Mutex, RwLock};
use std::error;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::{Arc, Weak}; use std::sync::{Arc, Weak};
use std::time::Duration; use std::time::Duration;
use tokio::time::sleep;
use tokio::sync::{mpsc, Mutex as AsyncMutex}; use tokio::sync::{mpsc, Mutex as AsyncMutex};
use lazy_static::lazy_static; use lazy_static::lazy_static;
use prost::Message; use prost::Message;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use thiserror::Error;
use tokio::time::sleep;
use tracing::{debug, error, info, trace, warn}; use tracing::{debug, error, info, trace, warn};
use crate::{proto, signal_client}; use crate::{proto, signal_client};
use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState}; use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState};
use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription}; use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription};
use livekit_webrtc::media_stream::MediaStream;
use livekit_webrtc::peer_connection::{ use livekit_webrtc::peer_connection::{
IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions, IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
}; };
use livekit_webrtc::peer_connection_factory::RTCConfiguration; 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::data_packet::Value;
use crate::proto::{ use crate::proto::{
@@ -33,10 +28,8 @@ use crate::rtc_engine::lk_runtime::LKRuntime;
use crate::rtc_engine::pc_transport::PCTransport; use crate::rtc_engine::pc_transport::PCTransport;
use crate::rtc_engine::rtc_events::{RTCEmitter, RTCEvent, RTCEvents}; use crate::rtc_engine::rtc_events::{RTCEmitter, RTCEvent, RTCEvents};
use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions}; use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions};
use std::cell::RefCell;
use super::{rtc_events, EngineEvents}; use super::{rtc_events, EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult};
use super::{EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult};
// //
// TODO(theomonnom): Smarter retry intervals // TODO(theomonnom): Smarter retry intervals
pub(crate) const RECONNECT_ATTEMPTS: u32 = 10; pub(crate) const RECONNECT_ATTEMPTS: u32 = 10;
@@ -60,7 +53,7 @@ pub enum PCState {
Closed, Closed,
} }
#[derive(Debug)] #[derive(Debug, Default)]
pub struct SessionInfo { pub struct SessionInfo {
url: String, url: String,
token: String, token: String,
@@ -70,6 +63,7 @@ pub struct SessionInfo {
#[derive(Debug)] #[derive(Debug)]
pub struct EngineInternal { pub struct EngineInternal {
lk_runtime: Arc<LKRuntime>, lk_runtime: Arc<LKRuntime>,
info: Mutex<SessionInfo>,
signal_client: Arc<SignalClient>, signal_client: Arc<SignalClient>,
session: Arc<RwLock<Option<RTCSession>>>, session: Arc<RwLock<Option<RTCSession>>>,
reconnecting: AtomicBool, reconnecting: AtomicBool,
@@ -283,7 +277,12 @@ impl EngineInternal {
tokio::spawn(self.clone().engine_task(rtc_events)); tokio::spawn(self.clone().engine_task(rtc_events));
if !join_response.subscriber_primary { if !join_response.subscriber_primary {
session.read().unwrap().negotiate_publisher().await?; session
.read()
.as_ref()
.unwrap()
.negotiate_publisher()
.await?;
} }
Ok(engine_events) Ok(engine_events)
@@ -327,19 +326,24 @@ impl EngineInternal {
trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate); trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate);
tokio::spawn(async move { tokio::spawn({
signal_client let signal_client = self.signal_client.clone();
.send(signal_request::Message::Trickle(TrickleRequest { async move {
candidate_init: json, signal_client
target: target as i32, .send(signal_request::Message::Trickle(TrickleRequest {
})) candidate_init: json,
.await; target: target as i32,
}))
.await;
}
}); });
} }
RTCEvent::ConnectionChange { state, target } => { RTCEvent::ConnectionChange { state, target } => {
trace!("connection change, {:?} {:?}", state, target); trace!("connection change, {:?} {:?}", state, target);
let subscriber_primary = session let subscriber_primary = self
.session
.read() .read()
.as_ref()
.unwrap() .unwrap()
.join_response .join_response
.lock() .lock()
@@ -348,22 +352,26 @@ impl EngineInternal {
let is_primary = subscriber_primary && target == SignalTarget::Subscriber; let is_primary = subscriber_primary && target == SignalTarget::Subscriber;
if is_primary && state == PeerConnectionState::Connected { if is_primary && state == PeerConnectionState::Connected {
let old_state = session let old_state = self
.session
.read() .read()
.as_ref()
.unwrap() .unwrap()
.pc_state .pc_state
.swap(PCState::Connected as u8, Ordering::SeqCst); .swap(PCState::Connected as u8, Ordering::SeqCst);
if old_state == PCState::New as u8 { if old_state == PCState::New as u8 {
let _ = emitter.send(EngineEvent::Connected).await; // First time connected let _ = self.engine_emitter.send(EngineEvent::Connected).await;
// First time connected
} }
} else if state == PeerConnectionState::Failed { } else if state == PeerConnectionState::Failed {
session self.session
.read() .read()
.as_ref()
.unwrap() .unwrap()
.pc_state .pc_state
.store(PCState::Disconnected as u8, Ordering::SeqCst); .store(PCState::Disconnected as u8, Ordering::SeqCst);
Self::handle_disconnected(signal_client, engine_inner, emitter); self.handle_disconnected();
} }
} }
RTCEvent::DataChannel { RTCEvent::DataChannel {
@@ -372,22 +380,27 @@ impl EngineInternal {
} => { } => {
if target == SignalTarget::Subscriber { if target == SignalTarget::Subscriber {
if data_channel.label() == RELIABLE_DC_LABEL { if data_channel.label() == RELIABLE_DC_LABEL {
*session.read().unwrap().sub_reliable_dc.lock() = Some(data_channel); *self.session.read().as_ref().unwrap().sub_reliable_dc.lock() =
Some(data_channel);
} else { } else {
*session.read().unwrap().sub_lossy_dc.lock() = Some(data_channel); *self.session.read().as_ref().unwrap().sub_lossy_dc.lock() =
Some(data_channel);
} }
} }
} }
RTCEvent::Offer { offer, target } => { RTCEvent::Offer { offer, target } => {
if target == SignalTarget::Publisher { if target == SignalTarget::Publisher {
// Send the publisher offer to the server // Send the publisher offer to the server
tokio::spawn(async move { tokio::spawn({
signal_client let signal_client = self.signal_client.clone();
.send(signal_request::Message::Offer(proto::SessionDescription { async move {
r#type: "offer".to_string(), signal_client
sdp: offer.to_string(), .send(signal_request::Message::Offer(proto::SessionDescription {
})) r#type: "offer".to_string(),
.await; sdp: offer.to_string(),
}))
.await;
}
}); });
} }
} }
@@ -397,7 +410,8 @@ impl EngineInternal {
target, target,
} => { } => {
if target == SignalTarget::Subscriber { if target == SignalTarget::Subscriber {
let _ = emitter let _ = self
.engine_emitter
.send(EngineEvent::AddTrack { .send(EngineEvent::AddTrack {
rtp_receiver, rtp_receiver,
streams, streams,
@@ -435,6 +449,7 @@ impl EngineInternal {
let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
self.session self.session
.read() .read()
.as_ref()
.unwrap() .unwrap()
.publisher_pc .publisher_pc
.lock() .lock()
@@ -448,14 +463,8 @@ impl EngineInternal {
trace!("received offer for the subscriber: {:?}", offer); trace!("received offer for the subscriber: {:?}", offer);
let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
let subscriber_pc = self let session = self.session.read();
.session let mut subscriber_pc = session.as_ref().unwrap().subscriber_pc.lock().await;
.read()
.as_ref()
.unwrap()
.subscriber_pc
.lock()
.await;
subscriber_pc.set_remote_description(sdp).await?; subscriber_pc.set_remote_description(sdp).await?;
let answer = subscriber_pc let answer = subscriber_pc
@@ -467,13 +476,16 @@ impl EngineInternal {
.set_local_description(answer.clone()) .set_local_description(answer.clone())
.await?; .await?;
tokio::spawn(async move { tokio::spawn({
self.signal_client let signal_client = self.signal_client.clone();
.send(signal_request::Message::Answer(proto::SessionDescription { async move {
r#type: "answer".to_string(), signal_client
sdp: answer.to_string(), .send(signal_request::Message::Answer(proto::SessionDescription {
})) r#type: "answer".to_string(),
.await; sdp: answer.to_string(),
}))
.await;
}
}); });
} }
signal_response::Message::Trickle(trickle) => { signal_response::Message::Trickle(trickle) => {
@@ -490,6 +502,7 @@ impl EngineInternal {
if trickle.target == SignalTarget::Publisher as i32 { if trickle.target == SignalTarget::Publisher as i32 {
self.session self.session
.read() .read()
.as_ref()
.unwrap() .unwrap()
.publisher_pc .publisher_pc
.lock() .lock()
@@ -499,6 +512,7 @@ impl EngineInternal {
} else { } else {
self.session self.session
.read() .read()
.as_ref()
.unwrap() .unwrap()
.subscriber_pc .subscriber_pc
.lock() .lock()
@@ -508,7 +522,10 @@ impl EngineInternal {
} }
} }
signal_response::Message::Update(update) => { signal_response::Message::Update(update) => {
let _ = emitter.send(EngineEvent::ParticipantUpdate(update)).await; let _ = self
.engine_emitter
.send(EngineEvent::ParticipantUpdate(update))
.await;
} }
_ => {} _ => {}
} }
@@ -531,34 +548,22 @@ impl EngineInternal {
for i in 0..RECONNECT_ATTEMPTS { for i in 0..RECONNECT_ATTEMPTS {
if full_reconnect { if full_reconnect {
if i == 0 { if i == 0 {
let _ = emitter.send(EngineEvent::Restarting).await; let _ = self.engine_emitter.send(EngineEvent::Restarting).await;
} }
info!("restarting connection... attempt: {}", i); info!("restarting connection... attempt: {}", i);
if let Err(err) = Self::try_restart_connection( if let Err(err) = self.try_restart_connection().await {
signal_client.clone(),
engine_inner.clone(),
emitter.clone(),
)
.await
{
error!("restarting connection failed: {}", err); error!("restarting connection failed: {}", err);
} else { } else {
return; return;
} }
} else { } else {
if i == 0 { if i == 0 {
let _ = emitter.send(EngineEvent::Resuming).await; let _ = self.engine_emitter.send(EngineEvent::Resuming).await;
} }
info!("resuming connection... attempt: {}", i); info!("resuming connection... attempt: {}", i);
if let Err(err) = Self::try_resume_connection( if let Err(err) = self.try_resume_connection().await {
signal_client.clone(),
engine_inner.clone(),
emitter.clone(),
)
.await
{
error!("resuming connection failed: {}", err); error!("resuming connection failed: {}", err);
if let EngineError::Signal(_) = err { if let EngineError::Signal(_) = err {
full_reconnect = true; full_reconnect = true;
@@ -576,21 +581,17 @@ impl EngineInternal {
// TODO DISCONNECT // TODO DISCONNECT
} }
async fn try_restart_connection( async fn try_restart_connection(self: &Arc<Self>) -> EngineResult<()> {
self: Arc<Self>,
signal_client: Arc<SignalClient>,
emitter: EngineEmitter,
) -> EngineResult<()> {
Ok(()) Ok(())
} }
async fn try_resume_connection( async fn try_resume_connection(self: &Arc<Self>) -> EngineResult<()> {
self: Arc<Self>, let mut info = self.info.lock();
signal_client: Arc<SignalClient>, info.options.sid = self
emitter: EngineEmitter, .session
) -> EngineResult<()> { .read()
let mut options = engine_inner.options.lock().clone(); .as_ref()
options.sid = engine_inner .unwrap()
.join_response .join_response
.lock() .lock()
.participant .participant
@@ -599,24 +600,34 @@ impl EngineInternal {
.sid .sid
.clone(); .clone();
signal_client self.signal_client.close().await;
.reconnect( self.signal_client
&engine_inner.url, .connect(&info.url, &info.token.clone(), info.options.clone())
&engine_inner.token.lock().clone(),
options,
)
.await?; .await?;
let _ = emitter.send(EngineEvent::SignalResumed).await; self.engine_emitter.send(EngineEvent::SignalResumed).await;
engine_inner self.session
.read()
.as_ref()
.unwrap()
.subscriber_pc .subscriber_pc
.lock() .lock()
.await .await
.prepare_ice_restart(); .prepare_ice_restart();
if engine_inner.has_published.load(Ordering::SeqCst) { if self
engine_inner .session
.read()
.as_ref()
.unwrap()
.has_published
.load(Ordering::SeqCst)
{
self.session
.read()
.as_ref()
.unwrap()
.publisher_pc .publisher_pc
.lock() .lock()
.await .await
@@ -626,12 +637,15 @@ impl EngineInternal {
}) })
.await?; .await?;
} }
self.session
.read()
.as_ref()
.unwrap()
.wait_pc_connection()
.await?;
Self::wait_pc_connection(engine_inner).await?; self.signal_client.flush_queue().await;
signal_client.flush_queue().await; self.engine_emitter.send(EngineEvent::Resumed);
let _ = emitter.send(EngineEvent::Resumed);
Ok(()) Ok(())
} }
} }