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

refactored rtc_engine

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