From d1e0f39d59fc3ad19a1918c63887bdbd0cad50b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sat, 24 Dec 2022 00:23:05 +0100 Subject: [PATCH] room_session wip --- crates/livekit-core/src/room/internal.rs | 317 +++++++++++++++ crates/livekit-core/src/room/mod.rs | 367 ++---------------- .../room/participant/remote_participant.rs | 2 +- crates/livekit-core/src/rtc_engine/mod.rs | 4 +- .../src/rtc_engine/rtc_session.rs | 10 +- 5 files changed, 359 insertions(+), 341 deletions(-) create mode 100644 crates/livekit-core/src/room/internal.rs diff --git a/crates/livekit-core/src/room/internal.rs b/crates/livekit-core/src/room/internal.rs new file mode 100644 index 0000000..e901655 --- /dev/null +++ b/crates/livekit-core/src/room/internal.rs @@ -0,0 +1,317 @@ +use parking_lot::{Mutex, RwLock}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::Arc; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; + +use crate::events::{ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents}; +use crate::proto::{self, participant_info}; +use crate::room::ConnectionState; +use crate::rtc_engine::{EngineEvent, EngineEvents, RTCEngine}; +use crate::signal_client::SignalOptions; + +use super::id::{ParticipantIdentity, ParticipantSid}; +use super::participant::local_participant::LocalParticipant; +use super::participant::remote_participant::RemoteParticipant; +use super::participant::{ParticipantInternalTrait, ParticipantTrait}; +use super::{RoomError, RoomHandle, RoomResult}; +use tracing::{error, instrument, Level}; + +#[derive(Debug)] +pub struct SessionInner { + pub state: AtomicU8, // ConnectionState + pub sid: Mutex, + pub name: Mutex, + pub participants: RwLock>>, + pub rtc_engine: Arc, + pub local_participant: Arc, + pub room_events: Arc, +} + +#[derive(Debug)] +pub struct RoomSession { + inner: Arc, + session_task: JoinHandle<()>, + close_emitter: oneshot::Sender<()>, +} + +impl RoomSession { + pub async fn connect( + room_events: Arc, + url: &str, + token: &str, + ) -> RoomResult { + let (rtc_engine, engine_events) = RTCEngine::new(); + let rtc_engine = Arc::new(rtc_engine); + rtc_engine + .connect(url, token, SignalOptions::default()) + .await?; + + let join_response = rtc_engine.join_response().unwrap(); + let pi = join_response.participant.unwrap().clone(); + let local_participant = Arc::new(LocalParticipant::new( + rtc_engine.clone(), + pi.sid.into(), + pi.identity.into(), + pi.name, + pi.metadata, + )); + let room_info = join_response.room.unwrap(); + let inner = Arc::new(SessionInner { + state: AtomicU8::new(ConnectionState::Connecting as u8), + sid: Mutex::new(room_info.sid), + name: Mutex::new(room_info.name), + participants: Default::default(), + rtc_engine, + local_participant, + room_events, + }); + + for pi in join_response.other_participants { + let participant = { + let pi = pi.clone(); + inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) + }; + participant.update_info(pi.clone()); + participant + .update_tracks(RoomHandle::from(inner.clone()), pi.tracks) + .await; + } + + let (close_emitter, close_receiver) = oneshot::channel(); + let session_task = tokio::spawn(inner.room_task(engine_events, close_receiver)); + + let session = Self { + inner, + session_task, + close_emitter, + }; + Ok(session) + } + + pub async fn close(self) { + self.inner.close(); + let _ = self.close_emitter.send(()); + self.session_task.await; + } +} + +impl SessionInner { + async fn room_task( + self: Arc, + mut engine_events: EngineEvents, + mut close_receiver: oneshot::Receiver<()>, + ) { + loop { + tokio::select! { + res = engine_events.recv() => { + if let Some(event) = res { + if let Err(err) = self.on_engine_event(event).await { + error!("failed to handle engine event: {:?}", err); + } + } else { + panic!("engine_events has been closed unexpectedly"); + } + }, + _ = &mut close_receiver => { + break; + } + } + } + } + + #[instrument(level = Level::DEBUG)] + async fn on_engine_event(self: &Arc, event: EngineEvent) -> RoomResult<()> { + match event { + EngineEvent::ParticipantUpdate(update) => self.handle_participant_update(update).await, + EngineEvent::MediaTrack { + track, + stream, + receiver, + } => { + let stream_id = stream.id(); + let lk_stream_id = unpack_stream_id(&stream_id); + if lk_stream_id.is_none() { + Err(RoomError::Internal(format!( + "MediaTrack event with invalid track_id: {:?}", + &stream_id + )))?; + } + + let (participant_sid, track_sid) = lk_stream_id.unwrap(); + let remote_participant = self.get_participant(&participant_sid.to_string().into()); + + if let Some(remote_participant) = remote_participant { + tokio::spawn({ + let room_internal = self.clone(); + { + let track_sid = track_sid.to_owned().into(); + async move { + remote_participant + .add_subscribed_media_track( + RoomHandle::from(room_internal), + track_sid, + track, + ) + .await; + } + } + }); + } else { + // The server should send participant updates before sending a new offer + // So this should never happen. + Err(RoomError::Internal(format!( + "AddTrack event with invalid participant_sid: {:?}", + participant_sid + )))?; + } + } + EngineEvent::Resuming => {} + EngineEvent::Resumed => {} + EngineEvent::Restarting => {} + EngineEvent::Restarted => {} + EngineEvent::Disconnected => {} + } + + Ok(()) + } + + async fn close(&self) { + self.rtc_engine.close().await; + } + + fn get_participant(self: &Arc, sid: &ParticipantSid) -> Option> { + self.participants.read().get(sid).cloned() + } + + /// Update the participants inside a Room. + /// It'll create, update or remove a participant + /// It also update the participant tracks. + #[instrument(level = Level::DEBUG)] + async fn handle_participant_update(self: &Arc, update: proto::ParticipantUpdate) { + for pi in update.participants { + if pi.sid == self.local_participant.sid() + || pi.identity == self.local_participant.identity() + { + self.local_participant.clone().update_info(pi); + continue; + } + + let remote_participant = self.get_participant(&pi.sid.clone().into()); + + if let Some(remote_participant) = remote_participant { + if pi.state == participant_info::State::Disconnected as i32 { + // Participant disconnected + self.handle_participant_disconnect(remote_participant) + } else { + // Participant is already connected, update the it + remote_participant.update_info(pi.clone()); + remote_participant + .update_tracks(RoomHandle::from(self.clone()), pi.tracks) + .await; + } + } else { + // Create a new participant + let remote_participant = { + let pi = pi.clone(); + self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) + }; + let mut handler = self.room_events.on_participant_connected.lock(); + if let Some(cb) = handler.as_mut() { + cb(ParticipantConnectedEvent { + room_handle: RoomHandle::from(self.clone()), + participant: remote_participant.clone(), + }); + } + + remote_participant.update_info(pi.clone()); + remote_participant + .update_tracks(RoomHandle::from(self.clone()), pi.tracks) + .await; + } + } + } + + #[instrument(level = Level::DEBUG)] + fn handle_participant_disconnect(self: &Arc, remote_participant: Arc) { + self.participants.write().remove(&remote_participant.sid()); + + // TODO(theomonnom): Unpublish all tracks + + let mut handler = self.room_events.on_participant_disconnected.lock(); + if let Some(cb) = handler.as_mut() { + cb(ParticipantDisconnectedEvent { + room_handle: RoomHandle::from(self.clone()), + participant: remote_participant.clone(), + }); + } + } + + fn create_participant( + self: &Arc, + sid: ParticipantSid, + identity: ParticipantIdentity, + name: String, + metadata: String, + ) -> Arc { + let p = Arc::new(RemoteParticipant::new( + sid.clone(), + identity, + name, + metadata, + )); + + macro_rules! forward_event { + ($type:ident, when_connected) => { + p.internal_events().$type({ + let room_internal = self.clone(); + move |event| { + let room_internal = room_internal.clone(); + async move { + if room_internal.state.load(Ordering::SeqCst) + == ConnectionState::Connected as u8 + { + if let Some(cb) = room_internal.room_events.$type.lock().as_mut() { + cb(event).await; + } + } + } + } + }) + }; + ($type:ident) => { + p.internal_events().$type({ + let room_internal = self.clone(); + move |event| { + let room_internal = room_internal.clone(); + async move { + if let Some(cb) = room_internal.room_events.$type.lock().as_mut() { + cb(event).await; + } + } + } + }) + }; + } + + // Forward participantevents to room events + forward_event!(on_track_published, when_connected); + forward_event!(on_track_subscribed); + forward_event!(on_track_subscription_failed); + + self.participants.write().insert(sid, p.clone()); + p + } +} + +fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> { + let split: Vec<&str> = stream_id.split('|').collect(); + if split.len() == 2 { + let participant_sid = split.get(0).unwrap(); + let track_sid = split.get(1).unwrap(); + Some((participant_sid, track_sid)) + } else { + None + } +} diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index 1c59af9..97c7a26 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -4,6 +4,7 @@ use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::Arc; use self::id::{ParticipantIdentity, ParticipantSid}; +use self::internal::{RoomInternal, RoomSession}; use self::participant::local_participant::LocalParticipant; use self::participant::remote_participant::RemoteParticipant; use self::participant::ParticipantInternalTrait; @@ -17,14 +18,13 @@ use crate::proto::participant_info; use thiserror::Error; use tracing::{debug, error, instrument, trace_span, Level}; -use crate::rtc_engine::{ - EngineError, EngineEvent, EngineEvents, EngineResult, RTCEngine, -}; +use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, EngineResult, RTCEngine}; use crate::signal_client::SignalOptions; pub use crate::rtc_engine::SimulateScenario; pub mod id; +mod internal; pub mod participant; pub mod publication; pub mod track; @@ -47,351 +47,52 @@ pub enum ConnectionState { Reconnecting, } -#[derive(Debug)] -struct RoomInner { - state: AtomicU8, // ConnectionState - sid: Mutex, - name: Mutex, - participants: RwLock>>, - rtc_engine: Arc, - local_participant: Arc, +#[derive(Clone, Debug)] +pub struct RoomHandle { + session: Arc, } -#[derive(Debug)] +impl RoomHandle { + pub fn sid(&self) -> String { + self.session.sid.lock().clone() + } + + pub fn name(&self) -> String { + self.internal.name.lock().clone() + } + + pub fn local_participant(&self) -> Arc { + self.internal.local_participant.clone() + } + + pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { + self.internal.rtc_engine.simulate_scenario(scenario).await + } +} + +#[derive(Debug, Default)] pub struct Room { - inner: Option>, - events: Arc, + session: Option, + events: Arc, // Keep the same RoomEvents across sessions } impl Room { - pub fn new() -> Room { - Self { - inner: None, - events: Default::default(), - } - } - #[instrument(level = Level::DEBUG)] - pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> { - // Initialize the RTCEngine - let (rtc_engine, engine_events) = RTCEngine::new(); - let rtc_engine = Arc::new(rtc_engine); - rtc_engine - .connect(url, token, SignalOptions::default()) - .await?; - - let join_response = rtc_engine.join_response().unwrap(); - let pi = join_response.participant.unwrap().clone(); - let local_participant = Arc::new(LocalParticipant::new( - rtc_engine.clone(), - pi.sid.into(), - pi.identity.into(), - pi.name, - pi.metadata, - )); - let room_info = join_response.room.unwrap(); - let inner = Arc::new(RoomInner { - state: AtomicU8::new(ConnectionState::Connecting as u8), - sid: Mutex::new(room_info.sid), - name: Mutex::new(room_info.name), - participants: Default::default(), - rtc_engine, - local_participant, - }); - - for pi in join_response.other_participants { - let participant = { - let pi = pi.clone(); - Self::create_participant( - inner.clone(), - self.events.clone(), - pi.sid.into(), - pi.identity.into(), - pi.name, - pi.metadata, - ) - }; - participant.update_info(pi.clone()); - participant - .update_tracks(RoomHandle::from(inner.clone()), pi.tracks) - .await; - } - - self.inner = Some(inner.clone()); - tokio::spawn(Self::room_task(inner, self.events.clone(), engine_events)); - + pub async fn connect(&self, url: &str, token: &str) -> RoomResult<()> { + let room_session = Arc::new(RoomSession::connect(self.events.clone(), url, token).await?); + self.session = Some(room_session.clone()); Ok(()) } + pub async fn close(&self) {} + pub fn events(&self) -> Arc { self.events.clone() } pub fn get_handle(&self) -> Option { - self.inner.as_ref().map(|inner| RoomHandle { - inner: inner.clone(), + self.internal.as_ref().map(|internal| RoomHandle { + internal: internal.clone(), }) } - - async fn room_task( - room_inner: Arc, - room_events: Arc, - mut engine_events: EngineEvents, - ) { - while let Some(event) = engine_events.recv().await { - if let Err(err) = - Self::handle_event(room_inner.clone(), room_events.clone(), event).await - { - error!("failed to handle engine event: {:?}", err); - } - } - } - - #[instrument(level = Level::DEBUG, skip(room_inner, room_events))] - async fn handle_event( - room_inner: Arc, - room_events: Arc, - event: EngineEvent, - ) -> RoomResult<()> { - match event { - EngineEvent::ParticipantUpdate(update) => { - Self::handle_participant_update(room_inner.clone(), room_events.clone(), update) - .await - } - EngineEvent::MediaTrack { - track, - stream, - receiver, - } => { - let stream_id = stream.id(); - let lk_stream_id = unpack_stream_id(&stream_id); - if lk_stream_id.is_none() { - Err(RoomError::Internal(format!( - "MediaTrack event with invalid track_id: {:?}", - &stream_id - )))?; - } - - let (participant_sid, track_sid) = lk_stream_id.unwrap(); - let remote_participant = - Self::get_participant(room_inner.clone(), &participant_sid.to_string().into()); - - if let Some(remote_participant) = remote_participant { - tokio::spawn({ - let track_sid = track_sid.to_owned().into(); - async move { - remote_participant - .add_subscribed_media_track( - RoomHandle::from(room_inner), - track_sid, - track, - ) - .await; - } - }); - } else { - // The server should send participant updates before sending a new offer - // So this should not happen. - Err(RoomError::Internal(format!( - "AddTrack event with invalid participant_sid: {:?}", - participant_sid - )))?; - } - } - EngineEvent::Resuming => {} - EngineEvent::Resumed => {} - EngineEvent::Restarting => {} - EngineEvent::Restarted => {} - EngineEvent::Disconnected => {} - } - - Ok(()) - } - - #[instrument(level = Level::DEBUG, skip(room_inner, room_events))] - async fn handle_participant_update( - room_inner: Arc, - room_events: Arc, - update: proto::ParticipantUpdate, - ) { - for pi in update.participants { - if pi.sid == room_inner.local_participant.sid() - || pi.identity == room_inner.local_participant.identity() - { - room_inner.local_participant.clone().update_info(pi); - continue; - } - - let remote_participant = - Self::get_participant(room_inner.clone(), &pi.sid.clone().into()); - - if let Some(remote_participant) = remote_participant { - if pi.state == participant_info::State::Disconnected as i32 { - // Participant disconencted - Self::handle_participant_disconnect( - room_inner.clone(), - room_events.clone(), - remote_participant, - ) - } else { - // Participant is already connected, update the informations - remote_participant.update_info(pi.clone()); - remote_participant - .update_tracks(RoomHandle::from(room_inner.clone()), pi.tracks) - .await; - } - } else { - // Create a new participant and call OnConnect event - let remote_participant = { - let pi = pi.clone(); - Self::create_participant( - room_inner.clone(), - room_events.clone(), - pi.sid.into(), - pi.identity.into(), - pi.name, - pi.metadata, - ) - }; - let mut handler = room_events.on_participant_connected.lock(); - if let Some(cb) = handler.as_mut() { - cb(ParticipantConnectedEvent { - room_handle: RoomHandle::from(room_inner.clone()), - participant: remote_participant.clone(), - }); - } - - remote_participant.update_info(pi.clone()); - remote_participant - .update_tracks(RoomHandle::from(room_inner.clone()), pi.tracks) - .await; - } - } - } - - #[instrument(level = Level::DEBUG, skip(room_inner, room_events))] - fn handle_participant_disconnect( - room_inner: Arc, - room_events: Arc, - remote_participant: Arc, - ) { - room_inner - .participants - .write() - .remove(&remote_participant.sid()); - - // TODO(theomonnom): Unpublish all tracks - - let mut handler = room_events.on_participant_disconnected.lock(); - if let Some(cb) = handler.as_mut() { - cb(ParticipantDisconnectedEvent { - room_handle: RoomHandle::from(room_inner.clone()), - participant: remote_participant.clone(), - }); - } - } - - fn get_participant( - room_inner: Arc, - sid: &ParticipantSid, - ) -> Option> { - room_inner.participants.read().get(sid).cloned() - } - - fn create_participant( - room_inner: Arc, - room_events: Arc, - sid: ParticipantSid, - identity: ParticipantIdentity, - name: String, - metadata: String, - ) -> Arc { - let p = Arc::new(RemoteParticipant::new( - sid.clone(), - identity, - name, - metadata, - )); - - macro_rules! forward_event { - ($type:ident, when_connected) => { - p.internal_events().$type({ - let room_events = room_events.clone(); - let room_inner = room_inner.clone(); - move |event| { - let room_events = room_events.clone(); - let room_inner = room_inner.clone(); - async move { - if room_inner.state.load(Ordering::SeqCst) - == ConnectionState::Connected as u8 - { - if let Some(cb) = room_events.$type.lock().as_mut() { - cb(event).await; - } - } - } - } - }) - }; - ($type:ident) => { - p.internal_events().$type({ - let room_events = room_events.clone(); - move |event| { - let room_events = room_events.clone(); - async move { - if let Some(cb) = room_events.$type.lock().as_mut() { - cb(event).await; - } - } - } - }) - }; - } - - // Forward participantevents to room events - forward_event!(on_track_published, when_connected); - forward_event!(on_track_subscribed); - forward_event!(on_track_subscription_failed); - - room_inner.participants.write().insert(sid, p.clone()); - p - } -} - -#[derive(Clone, Debug)] -pub struct RoomHandle { - inner: Arc, -} - -impl RoomHandle { - fn from(room_inner: Arc) -> Self { - Self { inner: room_inner } - } - - pub fn sid(&self) -> String { - self.inner.sid.lock().clone() - } - - pub fn name(&self) -> String { - self.inner.name.lock().clone() - } - - pub fn local_participant(&self) -> Arc { - self.inner.local_participant.clone() - } - - pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { - self.inner.rtc_engine.simulate_scenario(scenario).await - } -} - -fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> { - let split: Vec<&str> = stream_id.split('|').collect(); - if split.len() == 2 { - let participant_sid = split.get(0).unwrap(); - let track_sid = split.get(1).unwrap(); - Some((participant_sid, track_sid)) - } else { - None - } } diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index 81653c4..d1c3d48 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -68,7 +68,7 @@ impl RemoteParticipant { return publication; } - sleep(Duration::from_millis(50)).await; + tokio::task::yield_now(); } } }; diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index f436915..9f9a7f5 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -352,7 +352,9 @@ impl EngineInner { let res = inner.reconnect_task().await; inner.reconnecting.store(false, Ordering::SeqCst); - if res.is_err() { + if res.is_ok() { + warn!("RTCEngine successfully reconnected") + } else { error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS); inner.close().await; } diff --git a/crates/livekit-core/src/rtc_engine/rtc_session.rs b/crates/livekit-core/src/rtc_engine/rtc_session.rs index a3410fc..a39eef4 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_session.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_session.rs @@ -200,9 +200,7 @@ impl RTCSession { join_response, }; - // Start tasks let (close_emitter, close_receiver) = watch::channel(false); - let inner = Arc::new(SessionInner { info: session_info, pc_state: AtomicU8::new(PCState::New as u8), @@ -224,6 +222,10 @@ impl RTCSession { ); let rtc_task = tokio::spawn(inner.clone().rtc_task(rtc_events, close_receiver.clone())); + if !inner.info.join_response.subscriber_primary { + inner.negotiate_publisher().await?; + } + let session = Self { lk_runtime, inner: inner.clone(), @@ -232,10 +234,6 @@ impl RTCSession { rtc_task, }; - if !inner.info.join_response.subscriber_primary { - inner.negotiate_publisher().await?; - } - Ok(session) }