From f7a746f0f6f7535fa0b3f2da4928cbca131d8187 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Mon, 26 Dec 2022 00:15:29 +0100 Subject: [PATCH] =?UTF-8?q?finally=20using=20RoomSession=20&=20Merry=20Chr?= =?UTF-8?q?istmas=20=F0=9F=8E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/livekit-core/src/events.rs | 14 +- crates/livekit-core/src/room/mod.rs | 120 ++----------- .../src/room/participant/local_participant.rs | 2 +- .../room/participant/remote_participant.rs | 16 +- crates/livekit-core/src/room/room_session.rs | 161 ++++++++++++++---- 5 files changed, 159 insertions(+), 154 deletions(-) diff --git a/crates/livekit-core/src/events.rs b/crates/livekit-core/src/events.rs index 7d56e0c..cca6613 100644 --- a/crates/livekit-core/src/events.rs +++ b/crates/livekit-core/src/events.rs @@ -1,8 +1,8 @@ use crate::room::id::TrackSid; use crate::room::participant::remote_participant::RemoteParticipant; use crate::room::publication::RemoteTrackPublication; +use crate::room::room_session::{ConnectionState, RoomSession}; use crate::room::track::remote_track::RemoteTrackHandle; -use crate::room::{ConnectionState, RoomHandle}; use futures::future::Future; use futures_util::future::BoxFuture; use parking_lot::Mutex; @@ -32,19 +32,19 @@ pub enum TrackError { #[derive(Clone, Debug)] pub struct ParticipantConnectedEvent { - pub room_handle: RoomHandle, + pub room_session: RoomSession, pub participant: Arc, } #[derive(Clone, Debug)] pub struct ParticipantDisconnectedEvent { - pub room_handle: RoomHandle, + pub room_session: RoomSession, pub participant: Arc, } #[derive(Clone, Debug)] pub struct TrackSubscribedEvent { - pub room_handle: RoomHandle, + pub room_session: RoomSession, pub track: RemoteTrackHandle, pub publication: RemoteTrackPublication, pub participant: Arc, @@ -52,14 +52,14 @@ pub struct TrackSubscribedEvent { #[derive(Clone, Debug)] pub struct TrackPublishedEvent { - pub room_handle: RoomHandle, + pub room_session: RoomSession, pub publication: RemoteTrackPublication, pub participant: Arc, } #[derive(Clone, Debug)] pub struct TrackSubscriptionFailedEvent { - pub room_handle: RoomHandle, + pub room_session: RoomSession, pub error: TrackError, pub sid: TrackSid, pub participant: Arc, @@ -67,7 +67,7 @@ pub struct TrackSubscriptionFailedEvent { #[derive(Clone, Debug)] pub struct ConnectionStateChangedEvent { - pub room_handle: RoomHandle, + pub room_session: RoomSession, pub state: ConnectionState, } diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index 75a84f8..cf7da7e 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -4,11 +4,11 @@ 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; use self::participant::ParticipantTrait; +use self::room_session::{RoomInternal, RoomSession}; use crate::events::{ ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents, TrackPublishedEvent, TrackSubscribedEvent, @@ -23,10 +23,10 @@ use crate::signal_client::SignalOptions; pub use crate::rtc_engine::SimulateScenario; -mod room_session; pub mod id; pub mod participant; pub mod publication; +pub mod room_session; pub mod track; #[derive(Error, Debug)] @@ -39,124 +39,32 @@ pub enum RoomError { pub type RoomResult = Result; -#[derive(Debug, Clone, Eq, PartialEq)] -pub enum ConnectionState { - Disconnected, - Connecting, - Connected, - Reconnecting, -} - -#[derive(Clone, Debug)] -pub struct RoomSession { - internal: Arc, -} - -impl RoomSession { - 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 { - session: Option>, + internal: Option, events: Arc, // Keep the same RoomEvents across sessions } impl Room { #[instrument(level = Level::DEBUG)] - 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()); + pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> { + let internal = RoomInternal::connect(self.events.clone(), url, token).await?; + self.internal = Some(internal); Ok(()) } - pub async fn close(&self) {} + #[instrument(level = Level::DEBUG)] + pub async fn close(&mut self) { + if let Some(internal) = self.internal.take() { + internal.close().await; + } + } pub fn events(&self) -> Arc { self.events.clone() } - pub fn session(&self) -> Option<> { - self.internal.as_ref().map(|internal| RoomHandle { - internal: internal.clone(), - }) - } -} - -#[derive(Debug)] -pub struct RoomInternal { - inner: Arc, - session_task: JoinHandle<()>, - close_emitter: oneshot::Sender<()>, -} - -impl RoomInternal { - 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; + pub fn session(&self) -> Option { + self.internal.as_ref().map(RoomInternal::session) } } diff --git a/crates/livekit-core/src/room/participant/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs index 05ce33d..8d2227f 100644 --- a/crates/livekit-core/src/room/participant/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -4,7 +4,7 @@ use crate::proto::{data_packet, DataPacket, UserPacket}; use crate::room::participant::{ impl_participant_trait, ParticipantInternalTrait, ParticipantShared, }; -use crate::room::{RoomError, RoomInner}; +use crate::room::RoomError; use crate::rtc_engine::RTCEngine; #[derive(Debug)] diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index d1c3d48..baddddb 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -9,11 +9,11 @@ use crate::room::participant::{ use crate::room::publication::{ RemoteTrackPublication, TrackPublication, TrackPublicationInternalTrait, TrackPublicationTrait, }; +use crate::room::room_session::RoomSession; use crate::room::track::remote_audio_track::RemoteAudioTrack; use crate::room::track::remote_track::RemoteTrackHandle; use crate::room::track::remote_video_track::RemoteVideoTrack; use crate::room::track::{TrackKind, TrackTrait}; -use crate::room::{RoomHandle, RoomInner}; use livekit_webrtc::media_stream::MediaStreamTrackHandle; use std::collections::HashSet; use std::time::Duration; @@ -51,10 +51,10 @@ impl RemoteParticipant { }) } - #[instrument(level = Level::DEBUG, skip(room_handle))] + #[instrument(level = Level::DEBUG, skip(room_session))] pub(crate) async fn add_subscribed_media_track( self: Arc, - room_handle: RoomHandle, + room_session: RoomSession, sid: TrackSid, media_track: MediaStreamTrackHandle, ) { @@ -110,7 +110,7 @@ impl RemoteParticipant { track.start(); let event = TrackSubscribedEvent { - room_handle, + room_session, track, publication: remote_publication, participant: self.clone(), @@ -133,7 +133,7 @@ impl RemoteParticipant { error!("could not find published track with sid: {:?}", sid); let event = TrackSubscriptionFailedEvent { - room_handle, + room_session, sid: sid.clone(), error: TrackError::TrackNotFound(sid.clone().to_string()), participant: self.clone(), @@ -161,10 +161,10 @@ impl RemoteParticipant { } } - #[instrument(level = Level::DEBUG, skip(room_handle))] + #[instrument(level = Level::DEBUG, skip(room_session))] pub(crate) async fn update_tracks( self: Arc, - room_handle: RoomHandle, + room_session: RoomSession, tracks: Vec, ) { let mut valid_tracks = HashSet::::new(); @@ -179,7 +179,7 @@ impl RemoteParticipant { // This is a new track, fire publish events let event = TrackPublishedEvent { - room_handle: room_handle.clone(), + room_session: room_session.clone(), participant: self.clone(), publication: publication.clone(), }; diff --git a/crates/livekit-core/src/room/room_session.rs b/crates/livekit-core/src/room/room_session.rs index 5222f10..360f786 100644 --- a/crates/livekit-core/src/room/room_session.rs +++ b/crates/livekit-core/src/room/room_session.rs @@ -7,7 +7,6 @@ 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, EngineResult, RTCEngine}; use crate::signal_client::SignalOptions; @@ -18,51 +17,149 @@ use super::participant::{ParticipantInternalTrait, ParticipantTrait}; use super::{RoomError, RoomResult, SimulateScenario}; 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, Clone, Eq, PartialEq)] +pub enum ConnectionState { + Disconnected, + Connecting, + Connected, + Reconnecting, } -#[derive(Clone, Debug)] +impl TryFrom for ConnectionState { + type Error = &'static str; + + fn try_from(value: u8) -> Result { + match value { + 0 => Ok(ConnectionState::Disconnected), + 1 => Ok(ConnectionState::Connecting), + 2 => Ok(ConnectionState::Connected), + 3 => Ok(ConnectionState::Reconnecting), + _ => Err("invalid ConnectionState"), + } + } +} + +#[derive(Debug)] +struct SessionInner { + state: AtomicU8, // ConnectionState + sid: Mutex, + name: Mutex, + participants: RwLock>>, + rtc_engine: Arc, + local_participant: Arc, + room_events: Arc, +} + +/// RoomSession represents a connection to a room. +/// It can be cloned and shared across threads. +#[derive(Debug, Clone)] pub struct RoomSession { inner: Arc, } -impl RoomSession { - pub fn sid(&self) -> String { - self.session.sid.lock().clone() +/// Responsible for creating and closing the room session. +#[derive(Debug)] +pub struct RoomInternal { + inner: Arc, + session_task: JoinHandle<()>, + close_emitter: oneshot::Sender<()>, +} + +impl RoomInternal { + 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(RoomSession::from(inner.clone()), pi.tracks) + .await; + } + + let (close_emitter, close_receiver) = oneshot::channel(); + let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver)); + + let session = Self { + inner, + session_task, + close_emitter, + }; + Ok(session) } - pub fn name(&self) -> String { - self.internal.name.lock().clone() + pub async fn close(self) { + self.inner.close(); + let _ = self.close_emitter.send(()); + self.session_task.await; } - pub fn local_participant(&self) -> Arc { - self.internal.local_participant.clone() + pub fn session(&self) -> RoomSession { + RoomSession::from(self.inner.clone()) } pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { - self.internal.rtc_engine.simulate_scenario(scenario).await + self.inner.rtc_engine.simulate_scenario(scenario).await } } impl RoomSession { + fn from(inner: Arc) -> Self { + Self { inner } + } - pub(crate) async fn close(&self) -> RoomResult<()> { - self.internal.rtc_engine.close().await?; - self.internal.room_events.close(); - Ok(()) - } + 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 fn connection_state(&self) -> ConnectionState { + self.inner.state.load(Ordering::Acquire).try_into().unwrap() + } + + pub fn participants(&self) -> &RwLock>> { + &self.inner.participants + } + + pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { + self.inner.rtc_engine.simulate_scenario(scenario).await + } } -// Connect me to a database - impl SessionInner { async fn room_task( self: Arc, @@ -110,13 +207,13 @@ impl SessionInner { if let Some(remote_participant) = remote_participant { tokio::spawn({ - let room_internal = self.clone(); + let session_inner = self.clone(); { let track_sid = track_sid.to_owned().into(); async move { remote_participant .add_subscribed_media_track( - RoomHandle::from(room_internal), + RoomSession::from(session_inner), track_sid, track, ) @@ -174,7 +271,7 @@ impl SessionInner { // Participant is already connected, update the it remote_participant.update_info(pi.clone()); remote_participant - .update_tracks(RoomHandle::from(self.clone()), pi.tracks) + .update_tracks(RoomSession::from(self.clone()), pi.tracks) .await; } } else { @@ -186,14 +283,14 @@ impl SessionInner { 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()), + room_session: RoomSession::from(self.clone()), participant: remote_participant.clone(), }); } remote_participant.update_info(pi.clone()); remote_participant - .update_tracks(RoomHandle::from(self.clone()), pi.tracks) + .update_tracks(RoomSession::from(self.clone()), pi.tracks) .await; } } @@ -208,7 +305,7 @@ impl SessionInner { 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()), + room_session: RoomSession::from(self.clone()), participant: remote_participant.clone(), }); }