diff --git a/Cargo.lock b/Cargo.lock index 0dd70ea..dc34d0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -512,6 +512,10 @@ dependencies = [ [[package]] name = "livekit-utils" version = "0.1.0" +dependencies = [ + "parking_lot", + "tokio", +] [[package]] name = "livekit-webrtc" diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index 73fdc2e..e4b4a8d 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -3,7 +3,7 @@ use self::room_session::{ConnectionState, RoomSession, SessionHandle}; use crate::proto::data_packet; use crate::room::id::TrackSid; use crate::room::participant::remote_participant::RemoteParticipant; -use crate::room::participant::ParticipantHandle; +use crate::room::participant::Participant; use crate::room::publication::RemoteTrackPublication; use crate::room::publication::TrackPublication; use crate::room::track::remote_track::RemoteTrackHandle; @@ -68,21 +68,21 @@ pub enum RoomEvent { }, TrackMuted { publication: TrackPublication, - participant: ParticipantHandle, + participant: Participant, }, TrackUnmuted { publication: TrackPublication, - participant: ParticipantHandle, + participant: Participant, }, ActiveSpeakersChanged { - speakers: Vec, + speakers: Vec, }, ConnectionQualityChanged { quality: ConnectionQuality, - participant: ParticipantHandle, + participant: Participant, }, DataReceived { - payload: Vec, + payload: Arc>, kind: data_packet::Kind, participant: Arc, }, diff --git a/crates/livekit-core/src/room/participant/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs index 5e9030f..e2f2d5d 100644 --- a/crates/livekit-core/src/room/participant/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -1,10 +1,18 @@ -use crate::proto::{data_packet, DataPacket, UserPacket}; +use super::ConnectionQuality; +use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket}; +use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; use crate::room::participant::{ - impl_participant_trait, ParticipantInternalTrait, ParticipantShared, ParticipantTrait, + impl_participant_trait, ParticipantEvent, ParticipantInternalTrait, ParticipantShared, + ParticipantTrait, }; -use crate::room::room_session::SessionEmitter; +use crate::room::publication::TrackPublication; use crate::room::RoomError; use crate::rtc_engine::RTCEngine; +use parking_lot::RwLockReadGuard; +use std::collections::HashMap; +use std::sync::atomic::Ordering; +use std::sync::Arc; +use tokio::sync::mpsc; #[derive(Debug)] pub struct LocalParticipant { @@ -19,10 +27,9 @@ impl LocalParticipant { identity: ParticipantIdentity, name: String, metadata: String, - internal_tx: SessionEmitter, ) -> Self { Self { - shared: ParticipantShared::new(sid, identity, name, metadata, internal_tx), + shared: ParticipantShared::new(sid, identity, name, metadata), rtc_engine, } } @@ -49,7 +56,7 @@ impl LocalParticipant { } impl ParticipantInternalTrait for LocalParticipant { - fn update_info(self: &Arc, info: ParticipantInfo) { + fn update_info(self: &Arc, info: ParticipantInfo, _emit_events: bool) { self.shared.update_info(info); } diff --git a/crates/livekit-core/src/room/participant/mod.rs b/crates/livekit-core/src/room/participant/mod.rs index 47dddfe..2c2b71f 100644 --- a/crates/livekit-core/src/room/participant/mod.rs +++ b/crates/livekit-core/src/room/participant/mod.rs @@ -1,19 +1,62 @@ +use super::publication::RemoteTrackPublication; +use super::TrackError; use crate::proto; use crate::proto::ParticipantInfo; use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; use crate::room::participant::local_participant::LocalParticipant; use crate::room::participant::remote_participant::RemoteParticipant; use crate::room::publication::{TrackPublication, TrackPublicationTrait}; -use crate::room::room_session::SessionEmitter; +use crate::room::track::remote_track::RemoteTrackHandle; use livekit_utils::enum_dispatch; -use parking_lot::{Mutex, RwLock}; +use livekit_utils::observer::Dispatcher; +use parking_lot::{Mutex, RwLock, RwLockReadGuard}; +use proto::data_packet; use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering}; use std::sync::Arc; +use tokio::sync::mpsc; pub mod local_participant; pub mod remote_participant; +#[derive(Debug, Clone)] +pub enum ParticipantEvent { + TrackPublished { + publication: RemoteTrackPublication, + }, + TrackUnpublished { + publication: RemoteTrackPublication, + }, + TrackSubscribed { + track: RemoteTrackHandle, + publication: RemoteTrackPublication, + }, + TrackUnsubscribed { + track: RemoteTrackHandle, + publication: RemoteTrackPublication, + }, + TrackSubscriptionFailed { + error: TrackError, + sid: TrackSid, + }, + DataReceived { + payload: Arc>, + kind: data_packet::Kind, + }, + SpeakingChanged { + speaking: bool, + }, + TrackMuted { + publication: TrackPublication, + }, + TrackUnmuted { + publication: TrackPublication, + }, + ConnectionQualityChanged { + quality: ConnectionQuality, + }, +} + #[derive(Debug, Clone, Copy, Eq, PartialEq)] #[repr(u8)] pub enum ConnectionQuality { @@ -54,7 +97,7 @@ pub(super) struct ParticipantShared { pub(super) speaking: AtomicBool, pub(super) audio_level: AtomicU32, pub(super) connection_quality: AtomicU8, - pub(super) internal_tx: SessionEmitter, + pub(super) dispatcher: Mutex>, } impl ParticipantShared { @@ -63,7 +106,6 @@ impl ParticipantShared { identity: ParticipantIdentity, name: String, metadata: String, - internal_tx: SessionEmitter, ) -> Self { Self { sid: Mutex::new(sid), @@ -74,7 +116,7 @@ impl ParticipantShared { speaking: Default::default(), audio_level: Default::default(), connection_quality: AtomicU8::new(ConnectionQuality::Unknown as u8), - internal_tx, + dispatcher: Default::default(), } } @@ -94,6 +136,10 @@ impl ParticipantShared { .store(audio_level.to_bits(), Ordering::SeqCst) } + pub(crate) fn register_observer(&self) -> mpsc::UnboundedReceiver { + self.dispatcher.lock().register() + } + pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) { self.connection_quality .store(quality as u8, Ordering::SeqCst); @@ -108,7 +154,7 @@ pub(crate) trait ParticipantInternalTrait { fn set_speaking(&self, speaking: bool); fn set_audio_level(&self, level: f32); fn set_connection_quality(&self, quality: ConnectionQuality); - fn update_info(self: &Arc, info: ParticipantInfo); + fn update_info(self: &Arc, info: ParticipantInfo, emit_events: bool); } pub trait ParticipantTrait { @@ -119,25 +165,29 @@ pub trait ParticipantTrait { fn is_speaking(&self) -> bool; fn audio_level(&self) -> f32; fn connection_quality(&self) -> ConnectionQuality; + fn tracks(&self) -> RwLockReadGuard>; + fn register_observer(&self) -> mpsc::UnboundedReceiver; } #[derive(Debug, Clone)] -pub enum ParticipantHandle { +pub enum Participant { Local(Arc), Remote(Arc), } -impl ParticipantHandle { +// TODO(theomonnom): Should I provide a WeakParticipant here ? + +impl Participant { enum_dispatch!( [Local, Remote] - fnc!(pub(crate), update_info, &Self, [info: ParticipantInfo], ()); + fnc!(pub(crate), update_info, &Self, [info: ParticipantInfo, emit_events: bool], ()); fnc!(pub(crate), set_speaking, &Self, [speaking: bool], ()); fnc!(pub(crate), set_audio_level, &Self, [audio_level: f32], ()); fnc!(pub(crate), set_connection_quality, &Self, [quality: ConnectionQuality], ()); ); } -impl ParticipantTrait for ParticipantHandle { +impl ParticipantTrait for Participant { enum_dispatch!( [Local, Remote] fnc!(sid, &Self, [], ParticipantSid); @@ -147,17 +197,13 @@ impl ParticipantTrait for ParticipantHandle { fnc!(is_speaking, &Self, [], bool); fnc!(audio_level, &Self, [], f32); fnc!(connection_quality, &Self, [], ConnectionQuality); + fnc!(tracks, &Self, [], RwLockReadGuard>); + fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver); ); } macro_rules! impl_participant_trait { ($x:ty) => { - use std::sync::atomic::Ordering; - use std::sync::Arc; - use $crate::proto::ParticipantInfo; - use $crate::room::id::{ParticipantIdentity, ParticipantSid}; - use $crate::room::participant::ConnectionQuality; - impl crate::room::participant::ParticipantTrait for $x { fn sid(&self) -> ParticipantSid { self.shared.sid.lock().clone() @@ -186,6 +232,14 @@ macro_rules! impl_participant_trait { fn connection_quality(&self) -> ConnectionQuality { self.shared.connection_quality.load(Ordering::SeqCst).into() } + + fn tracks(&self) -> RwLockReadGuard> { + self.shared.tracks.read() + } + + fn register_observer(&self) -> mpsc::UnboundedReceiver { + self.shared.register_observer() + } } }; } diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index ee492ec..b5a37c8 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -1,25 +1,29 @@ -use crate::room::id::TrackSid; +use super::ConnectionQuality; +use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket}; +use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; use crate::room::participant::{ - impl_participant_trait, ParticipantInternalTrait, ParticipantShared, + impl_participant_trait, ParticipantEvent, ParticipantInternalTrait, ParticipantShared, + ParticipantTrait, }; use crate::room::publication::{ RemoteTrackPublication, TrackPublication, TrackPublicationInternalTrait, TrackPublicationTrait, }; -use crate::room::room_session::SessionEmitter; -use crate::room::room_session::SessionEvent; 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::{RoomEvent, TrackError}; +use crate::room::TrackError; use livekit_webrtc::media_stream::MediaStreamTrackHandle; +use parking_lot::RwLockReadGuard; +use std::collections::HashMap; use std::collections::HashSet; +use std::sync::atomic::Ordering; +use std::sync::Arc; use std::time::Duration; +use tokio::sync::mpsc; use tokio::time::timeout; use tracing::{debug, error, instrument, Level}; -use super::ParticipantTrait; - const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5); #[derive(Debug)] @@ -33,10 +37,9 @@ impl RemoteParticipant { identity: ParticipantIdentity, name: String, metadata: String, - internal_tx: SessionEmitter, ) -> Self { Self { - shared: ParticipantShared::new(sid, identity, name, metadata, internal_tx), + shared: ParticipantShared::new(sid, identity, name, metadata), } } @@ -50,6 +53,18 @@ impl RemoteParticipant { }) } + /// Called by the RoomSession when receiving data by the RTCSession + /// It is just used to emit the Data event on the participant dispatcher. + pub(crate) fn on_data_received(&self, data: Arc>, kind: data_packet::Kind) { + self.shared + .dispatcher + .lock() + .dispatch(&ParticipantEvent::DataReceived { + payload: data, + kind, + }); + } + #[instrument(level = Level::DEBUG)] pub(crate) async fn add_subscribed_media_track( self: Arc, @@ -107,30 +122,57 @@ impl RemoteParticipant { .add_track_publication(TrackPublication::Remote(remote_publication.clone())); track.start(); - let _ = self - .shared - .internal_tx - .send(SessionEvent::Room(RoomEvent::TrackSubscribed { + self.shared + .dispatcher + .lock() + .dispatch(&ParticipantEvent::TrackSubscribed { track: track, publication: remote_publication, - participant: self.clone(), - })); + }); } else { error!("could not find published track with sid: {:?}", sid); - let _ = self.shared.internal_tx.send(SessionEvent::Room( - RoomEvent::TrackSubscriptionFailed { + self.shared + .dispatcher + .lock() + .dispatch(&ParticipantEvent::TrackSubscriptionFailed { sid: sid.clone(), error: TrackError::TrackNotFound(sid.clone().to_string()), - participant: self.clone(), - }, - )); + }); + } + } + + pub(crate) fn unpublish_track(self: &Arc, sid: &TrackSid, emit_events: bool) { + if let Some(publication) = self.get_track_publication(sid) { + // Unsubscribe to the track if needed + if let Some(track) = publication.track() { + track.stop(); + + self.shared + .dispatcher + .lock() + .dispatch(&ParticipantEvent::TrackUnsubscribed { + track: track.clone(), + publication: publication.clone(), + }); + } + + if emit_events { + self.shared + .dispatcher + .lock() + .dispatch(&ParticipantEvent::TrackUnpublished { + publication: publication.clone(), + }); + } + + publication.update_track(None); } } } impl ParticipantInternalTrait for RemoteParticipant { - fn update_info(self: &Arc, info: ParticipantInfo) { + fn update_info(self: &Arc, info: ParticipantInfo, emit_events: bool) { self.shared.update_info(info.clone()); let mut valid_tracks = HashSet::::new(); @@ -142,18 +184,26 @@ impl ParticipantInternalTrait for RemoteParticipant { self.shared .add_track_publication(TrackPublication::Remote(publication.clone())); - // This is a new track, fire publish event - let _ = + // This is a new track, dispatch publish event + if emit_events { self.shared - .internal_tx - .send(SessionEvent::Room(RoomEvent::TrackPublished { - publication: publication.clone(), - participant: self.clone(), - })); + .dispatcher + .lock() + .dispatch(&ParticipantEvent::TrackPublished { publication }); + } } valid_tracks.insert(track.sid.into()); } + + // remove tracks that are no longer valid + for (sid, _) in self.shared.tracks.read().iter() { + if valid_tracks.contains(sid) { + continue; + } + + self.unpublish_track(sid, emit_events); + } } fn set_speaking(&self, speaking: bool) { @@ -163,7 +213,7 @@ impl ParticipantInternalTrait for RemoteParticipant { fn set_audio_level(&self, level: f32) { self.shared.set_audio_level(level); } - + fn set_connection_quality(&self, quality: ConnectionQuality) { self.shared.set_connection_quality(quality); } diff --git a/crates/livekit-core/src/room/publication/mod.rs b/crates/livekit-core/src/room/publication/mod.rs index 7f7864d..0e8a60e 100644 --- a/crates/livekit-core/src/room/publication/mod.rs +++ b/crates/livekit-core/src/room/publication/mod.rs @@ -4,13 +4,15 @@ use crate::room::id::ParticipantSid; use crate::room::id::TrackSid; use crate::room::track::local_track::LocalTrackHandle; use crate::room::track::remote_track::RemoteTrackHandle; -use crate::room::track::{TrackHandle, TrackKind, TrackSource}; +use crate::room::track::{TrackHandle, TrackKind, TrackSource, TrackTrait}; use livekit_utils::enum_dispatch; +use livekit_utils::observer::Dispatcher; use parking_lot::Mutex; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; +use tokio::sync::{mpsc, oneshot}; -use super::track::TrackDimension; +use super::track::{TrackDimension, TrackEvent}; pub(crate) trait TrackPublicationInternalTrait { fn update_track(&self, track: Option); @@ -22,6 +24,7 @@ pub trait TrackPublicationTrait { fn sid(&self) -> TrackSid; fn kind(&self) -> TrackKind; fn source(&self) -> TrackSource; + fn muted(&self) -> bool; fn simulcasted(&self) -> bool; } @@ -35,7 +38,10 @@ pub(super) struct TrackPublicationShared { pub(super) simulcasted: AtomicBool, pub(super) dimension: Mutex, pub(super) mime_type: Mutex, - pub(super) participant: ParticipantSid, // TODO(theomonnom) Use WeakParticipant instead + pub(super) muted: AtomicBool, + pub(super) participant: ParticipantSid, + pub(super) dispatcher: Mutex>, + pub(super) close_sender: Mutex>>, } impl TrackPublicationShared { @@ -55,13 +61,56 @@ impl TrackPublicationShared { simulcasted: AtomicBool::new(info.simulcast), dimension: Mutex::new(TrackDimension(info.width, info.height)), mime_type: Mutex::new(info.mime_type), + muted: AtomicBool::new(info.muted), + dispatcher: Default::default(), + close_sender: Default::default(), participant, }) } + pub fn update_track(self: &Arc, track: Option) { + let mut old_track = self.track.lock(); + + if let Some(close_sender) = self.close_sender.lock().take() { + let _ = close_sender.send(()); + } + + *old_track = track.clone(); + if let Some(track) = track { + let (close_sender, close_receiver) = oneshot::channel(); + self.close_sender.lock().replace(close_sender); + + let track_receiver = track.register_observer(); + tokio::spawn( + self.clone() + .publication_task(close_receiver, track_receiver), + ); + } + } + + /// Task used to forward TrackHandle's events to the TrackPublications's dispatcher + async fn publication_task( + self: Arc, + mut close_receiver: oneshot::Receiver<()>, + mut track_receiver: mpsc::UnboundedReceiver, + ) { + loop { + tokio::select! { + Some(event) = track_receiver.recv() => { + self.dispatcher.lock().dispatch(&event); + } + _ = &mut close_receiver => { + break; + } + } + } + } + pub fn update_info(&self, info: TrackInfo) { *self.name.lock() = info.name; *self.sid.lock() = info.sid.into(); + *self.dimension.lock() = TrackDimension(info.width, info.height); + *self.mime_type.lock() = info.mime_type; self.kind.store( TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8, Ordering::SeqCst, @@ -71,8 +120,19 @@ impl TrackPublicationShared { Ordering::SeqCst, ); self.simulcasted.store(info.simulcast, Ordering::SeqCst); - *self.dimension.lock() = TrackDimension(info.width, info.height); - *self.mime_type.lock() = info.mime_type; + self.muted.store(info.muted, Ordering::SeqCst); + + if let Some(track) = self.track.lock().as_ref() { + track.set_muted(info.muted); + } + } +} + +impl Drop for TrackPublicationShared { + fn drop(&mut self) { + if let Some(close_sender) = self.close_sender.lock().take() { + let _ = close_sender.send(()); + } } } @@ -107,22 +167,13 @@ impl TrackPublicationTrait for TrackPublication { fnc!(name, &Self, [], String); fnc!(kind, &Self, [], TrackKind); fnc!(source, &Self, [], TrackSource); + fnc!(muted, &Self, [], bool); fnc!(simulcasted, &Self, [], bool); ); } macro_rules! impl_publication_trait { ($x:ident) => { - impl TrackPublicationInternalTrait for $x { - fn update_track(&self, track: Option) { - *self.shared.track.lock() = track; - } - - fn update_info(&self, info: TrackInfo) { - self.shared.update_info(info); - } - } - impl TrackPublicationTrait for $x { fn name(&self) -> String { self.shared.name.lock().clone() @@ -143,6 +194,10 @@ macro_rules! impl_publication_trait { fn simulcasted(&self) -> bool { self.shared.simulcasted.load(Ordering::SeqCst) } + + fn muted(&self) -> bool { + self.shared.muted.load(Ordering::SeqCst) + } } }; } @@ -162,6 +217,16 @@ impl LocalTrackPublication { } } +impl TrackPublicationInternalTrait for LocalTrackPublication { + fn update_track(&self, track: Option) { + self.shared.update_track(track); + } + + fn update_info(&self, info: TrackInfo) { + self.shared.update_info(info); + } +} + #[derive(Clone, Debug)] pub struct RemoteTrackPublication { shared: Arc, @@ -183,5 +248,15 @@ impl RemoteTrackPublication { } } +impl TrackPublicationInternalTrait for RemoteTrackPublication { + fn update_track(&self, track: Option) { + self.shared.update_track(track); + } + + fn update_info(&self, info: TrackInfo) { + self.shared.update_info(info); + } +} + impl_publication_trait!(LocalTrackPublication); impl_publication_trait!(RemoteTrackPublication); diff --git a/crates/livekit-core/src/room/room_session.rs b/crates/livekit-core/src/room/room_session.rs index 34b0866..4f284d1 100644 --- a/crates/livekit-core/src/room/room_session.rs +++ b/crates/livekit-core/src/room/room_session.rs @@ -1,7 +1,7 @@ use super::id::{ParticipantIdentity, ParticipantSid}; use super::participant::local_participant::LocalParticipant; use super::participant::remote_participant::RemoteParticipant; -use super::participant::{ConnectionQuality, ParticipantHandle}; +use super::participant::{ConnectionQuality, Participant, ParticipantEvent}; use super::participant::{ParticipantInternalTrait, ParticipantTrait}; use super::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario}; use crate::proto::{self, participant_info, SpeakerInfo}; @@ -16,14 +16,6 @@ use tokio::sync::oneshot; use tokio::task::JoinHandle; use tracing::{error, instrument, Level}; -pub(crate) type SessionEmitter = mpsc::UnboundedSender; -pub(crate) type SessionEvents = mpsc::UnboundedReceiver; - -/// Used internally for participants and tracks -pub(crate) enum SessionEvent { - Room(RoomEvent), // Send a public event -} - #[derive(Debug, Clone, Copy, Eq, PartialEq)] pub enum ConnectionState { Disconnected, @@ -50,10 +42,11 @@ struct SessionInner { sid: Mutex, name: Mutex, participants: RwLock>>, - active_speakers: RwLock>, + participants_tasks: RwLock, oneshot::Sender<()>)>>, + active_speakers: RwLock>, rtc_engine: Arc, local_participant: Arc, - internal_tx: SessionEmitter, + room_emitter: RoomEmitter, } #[derive(Debug)] @@ -78,7 +71,6 @@ impl SessionHandle { .connect(url, token, SignalOptions::default()) .await?; - let (internal_tx, internal_rx) = mpsc::unbounded_channel(); let join_response = rtc_engine.join_response().unwrap(); let pi = join_response.participant.unwrap().clone(); let local_participant = Arc::new(LocalParticipant::new( @@ -87,7 +79,6 @@ impl SessionHandle { pi.identity.into(), pi.name, pi.metadata, - internal_tx.clone(), )); let room_info = join_response.room.unwrap(); @@ -96,10 +87,11 @@ impl SessionHandle { sid: Mutex::new(room_info.sid), name: Mutex::new(room_info.name), participants: Default::default(), + participants_tasks: Default::default(), active_speakers: Default::default(), rtc_engine, local_participant, - internal_tx, + room_emitter, }); for pi in join_response.other_participants { @@ -107,16 +99,11 @@ impl SessionHandle { let pi = pi.clone(); inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) }; - participant.update_info(pi.clone()); + participant.update_info(pi.clone(), false); } let (close_emitter, close_receiver) = oneshot::channel(); - let session_task = tokio::spawn(inner.clone().room_task( - engine_events, - internal_rx, - close_receiver, - room_emitter, - )); + let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver)); inner.update_connection_state(ConnectionState::Connected); @@ -174,23 +161,10 @@ impl SessionInner { async fn room_task( self: Arc, mut engine_events: EngineEvents, - mut internal_rx: SessionEvents, mut close_receiver: oneshot::Receiver<()>, - room_emitter: RoomEmitter, ) { loop { tokio::select! { - biased; - res = internal_rx.recv() => { - match res { - Some(event) => { - if let Err(err) = self.on_internal_event(event, &room_emitter).await { - error!("failed to handle internal event: {:?}", err); - } - }, - _ => panic!("internal_rx has been closed unexpectedly") - }; - } res = engine_events.recv() => { match res { Some(event) => { @@ -208,30 +182,70 @@ impl SessionInner { } } - async fn on_internal_event( - &self, - event: SessionEvent, - room_emitter: &RoomEmitter, - ) -> EngineResult<()> { - match event { - SessionEvent::Room(event) => { - if self.state.load(Ordering::Acquire) != ConnectionState::Connected as u8 - && matches!( - event, - RoomEvent::TrackPublished { .. } - | RoomEvent::TrackUnpublished { .. } - | RoomEvent::ParticipantConnected { .. } - | RoomEvent::ParticipantDisconnected { .. } - | RoomEvent::ActiveSpeakersChanged { .. } - ) - { - return Ok(()); // Ignore the event - } - - // Forward the event to the public channel - let _ = room_emitter.send(event); + /// Listen to the Participant events and forward them to the Room Dispatcher + #[instrument(level = Level::DEBUG)] + async fn participant_task( + self: Arc, + participant: Participant, + mut participant_events: mpsc::UnboundedReceiver, + mut close_rx: oneshot::Receiver<()>, + ) { + loop { + tokio::select! { + res = participant_events.recv() => { + match res { + Some(event) => { + if let Err(err) = self.on_participant_event(&participant, event).await { + error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err); + } + }, + _ => panic!("engine_events has been closed unexpectedly") + }; + }, + _ = &mut close_rx => { + break; + }, } } + } + + #[instrument(level = Level::DEBUG)] + async fn on_participant_event( + self: &Arc, + participant: &Participant, + event: ParticipantEvent, + ) -> RoomResult<()> { + if let Participant::Remote(remote_participant) = participant { + match event { + ParticipantEvent::TrackPublished { publication } => { + let _ = self.room_emitter.send(RoomEvent::TrackPublished { + participant: remote_participant.clone(), + publication, + }); + } + ParticipantEvent::TrackUnpublished { publication } => { + let _ = self.room_emitter.send(RoomEvent::TrackUnpublished { + participant: remote_participant.clone(), + publication, + }); + } + ParticipantEvent::TrackSubscribed { track, publication } => { + let _ = self.room_emitter.send(RoomEvent::TrackSubscribed { + participant: remote_participant.clone(), + track, + publication, + }); + } + ParticipantEvent::TrackUnsubscribed { track, publication } => { + let _ = self.room_emitter.send(RoomEvent::TrackUnsubscribed { + participant: remote_participant.clone(), + track, + publication, + }); + } + _ => {} + }; + } Ok(()) } @@ -275,16 +289,12 @@ impl SessionInner { } EngineEvent::Resuming => { if self.update_connection_state(ConnectionState::Reconnecting) { - let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::Reconnecting)); + let _ = self.room_emitter.send(RoomEvent::Reconnecting); } } EngineEvent::Resumed => { self.update_connection_state(ConnectionState::Connected); - let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::Reconnected)); + let _ = self.room_emitter.send(RoomEvent::Reconnected); // TODO(theomonnom): Update subscriptions settings // TODO(theomonnom): Send sync state @@ -297,14 +307,15 @@ impl SessionInner { kind, participant_sid, } => { + let payload = Arc::new(payload); if let Some(participant) = self.get_participant(&participant_sid.into()) { - let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::DataReceived { - payload, - kind, - participant, - })); + let _ = self.room_emitter.send(RoomEvent::DataReceived { + payload: payload.clone(), + kind, + participant: participant.clone(), + }); + + participant.on_data_received(payload, kind); } } EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers), @@ -332,9 +343,8 @@ impl SessionInner { self.state.store(state as u8, Ordering::Release); let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::ConnectionStateChanged(state))); - + .room_emitter + .send(RoomEvent::ConnectionStateChanged(state)); return true; } @@ -342,12 +352,12 @@ impl SessionInner { /// It'll create, update or remove a participant /// It also update the participant tracks. #[instrument(level = Level::DEBUG)] - fn handle_participant_update(&self, update: proto::ParticipantUpdate) { + 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); + self.local_participant.clone().update_info(pi, true); continue; } @@ -356,10 +366,11 @@ impl SessionInner { if let Some(remote_participant) = remote_participant { if pi.state == participant_info::State::Disconnected as i32 { // Participant disconnected - self.handle_participant_disconnect(remote_participant) + self.clone() + .handle_participant_disconnect(remote_participant) } else { // Participant is already connected, update the it - remote_participant.update_info(pi.clone()); + remote_participant.update_info(pi.clone(), true); } } else { // Create a new participant @@ -369,30 +380,14 @@ impl SessionInner { }; let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::ParticipantConnected( - remote_participant.clone(), - ))); + .room_emitter + .send(RoomEvent::ParticipantConnected(remote_participant.clone())); - remote_participant.update_info(pi.clone()); + remote_participant.update_info(pi.clone(), true); } } } - /// A participant has disconnected - /// Cleanup the participant and emit an event - #[instrument(level = Level::DEBUG)] - fn handle_participant_disconnect(&self, remote_participant: Arc) { - self.participants.write().remove(&remote_participant.sid()); - - // TODO(theomonnom): Unpublish all tracks - let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::ParticipantDisconnected( - remote_participant.clone(), - ))); - } - /// Active speakers changed /// Update the participants & sort the active_speakers by audio_level #[instrument(level = Level::DEBUG)] @@ -402,10 +397,10 @@ impl SessionInner { for speaker in speakers_info { let participant = { if speaker.sid == self.local_participant.sid() { - ParticipantHandle::Local(self.local_participant.clone()) + Participant::Local(self.local_participant.clone()) } else { if let Some(participant) = self.get_participant(&speaker.sid.into()) { - ParticipantHandle::Remote(participant) + Participant::Remote(participant) } else { continue; } @@ -422,11 +417,10 @@ impl SessionInner { speakers.sort_by(|a, b| a.audio_level().partial_cmp(&b.audio_level()).unwrap()); *self.active_speakers.write() = speakers.clone(); + let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::ActiveSpeakersChanged { - speakers, - })); + .room_emitter + .send(RoomEvent::ActiveSpeakersChanged { speakers }); } /// Handle a connection quality update @@ -436,11 +430,11 @@ impl SessionInner { for update in updates { let participant = { if update.participant_sid == self.local_participant.sid() { - ParticipantHandle::Local(self.local_participant.clone()) + Participant::Local(self.local_participant.clone()) } else { if let Some(participant) = self.get_participant(&update.participant_sid.into()) { - ParticipantHandle::Remote(participant) + Participant::Remote(participant) } else { continue; } @@ -452,41 +446,36 @@ impl SessionInner { .into(); participant.set_connection_quality(quality); - let _ = - self.internal_tx - .send(SessionEvent::Room(RoomEvent::ConnectionQualityChanged { - participant, - quality, - })); + let _ = self.room_emitter.send(RoomEvent::ConnectionQualityChanged { + participant, + quality, + }); } } #[instrument(level = Level::DEBUG)] - fn handle_restarting(&self) { + fn handle_restarting(self: &Arc) { // Remove existing participants/subscriptions on full reconnect for (_, participant) in self.participants.read().iter() { - self.handle_participant_disconnect(participant.clone()); + self.clone() + .handle_participant_disconnect(participant.clone()); } if self.update_connection_state(ConnectionState::Reconnecting) { - let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::Reconnecting)); + let _ = self.room_emitter.send(RoomEvent::Reconnecting); } } #[instrument(level = Level::DEBUG)] - fn handle_restarted(&self) { + fn handle_restarted(self: &Arc) { // Full reconnect succeeded! let join_response = self.rtc_engine.join_response().unwrap(); self.update_connection_state(ConnectionState::Connected); - let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::Reconnected)); + let _ = self.room_emitter.send(RoomEvent::Reconnected); if let Some(pi) = join_response.participant { - self.local_participant.update_info(pi); // The sid may have changed + self.local_participant.update_info(pi, true); // The sid may have changed } self.handle_participant_update(proto::ParticipantUpdate { @@ -503,30 +492,66 @@ impl SessionInner { } self.update_connection_state(ConnectionState::Disconnected); - let _ = self - .internal_tx - .send(SessionEvent::Room(RoomEvent::Disconnected)); + let _ = self.room_emitter.send(RoomEvent::Disconnected); } /// Create a new participant /// Also add it to the participants list + #[instrument(level = Level::DEBUG)] fn create_participant( - &self, + self: &Arc, sid: ParticipantSid, identity: ParticipantIdentity, name: String, metadata: String, ) -> Arc { - let p = Arc::new(RemoteParticipant::new( + let participant = Arc::new(RemoteParticipant::new( sid.clone(), identity, name, metadata, - self.internal_tx.clone(), )); - self.participants.write().insert(sid, p.clone()); - p + // Create the participant task + let (close_tx, close_rx) = oneshot::channel(); + let participant_task = tokio::spawn(self.clone().participant_task( + Participant::Remote(participant.clone()), + participant.register_observer(), + close_rx, + )); + self.participants_tasks + .write() + .insert(sid.clone(), (participant_task, close_tx)); + + self.participants.write().insert(sid, participant.clone()); + participant + } + + /// A participant has disconnected + /// Cleanup the participant and emit an event + #[instrument(level = Level::DEBUG)] + fn handle_participant_disconnect(self: Arc, remote_participant: Arc) { + tokio::spawn(async move { + for (sid, _) in &*remote_participant.tracks() { + remote_participant.unpublish_track(&sid, true); + } + + // Close the participant task + if let Some((task, close_tx)) = self + .participants_tasks + .write() + .remove(&remote_participant.sid()) + { + let _ = close_tx.send(()); + let _ = task.await; + } + + self.participants.write().remove(&remote_participant.sid()); + + let _ = self.room_emitter.send(RoomEvent::ParticipantDisconnected( + remote_participant.clone(), + )); + }); } fn get_participant(&self, sid: &ParticipantSid) -> Option> { diff --git a/crates/livekit-core/src/room/track/mod.rs b/crates/livekit-core/src/room/track/mod.rs index 31ceb70..b9eb468 100644 --- a/crates/livekit-core/src/room/track/mod.rs +++ b/crates/livekit-core/src/room/track/mod.rs @@ -5,10 +5,12 @@ use crate::room::track::local_video_track::LocalVideoTrack; use crate::room::track::remote_audio_track::RemoteAudioTrack; use crate::room::track::remote_video_track::RemoteVideoTrack; use livekit_utils::enum_dispatch; +use livekit_utils::observer::Dispatcher; use livekit_webrtc::media_stream::{MediaStreamTrackHandle, MediaStreamTrackTrait}; use parking_lot::Mutex; -use std::sync::atomic::AtomicU8; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; +use tokio::sync::mpsc; pub mod audio_track; pub mod local_audio_track; @@ -106,6 +108,14 @@ pub trait TrackTrait { fn stream_state(&self) -> StreamState; fn start(&self); fn stop(&self); + fn register_observer(&self) -> mpsc::UnboundedReceiver; + fn set_muted(&self, muted: bool); +} + +#[derive(Debug, Clone)] +pub enum TrackEvent { + Mute, + Unmute, } #[derive(Debug)] @@ -114,7 +124,9 @@ pub(super) struct TrackShared { pub(super) name: Mutex, pub(super) kind: AtomicU8, // TrackKind pub(super) stream_state: AtomicU8, // StreamState + pub(super) muted: AtomicBool, pub(super) rtc_track: MediaStreamTrackHandle, + pub(super) dispatcher: Mutex>, } impl TrackShared { @@ -129,7 +141,9 @@ impl TrackShared { name: Mutex::new(name), kind: AtomicU8::new(kind as u8), stream_state: AtomicU8::new(StreamState::Active as u8), + muted: AtomicBool::new(false), rtc_track, + dispatcher: Default::default(), } } @@ -140,6 +154,25 @@ impl TrackShared { pub(crate) fn stop(&self) { self.rtc_track.set_enabled(false); } + + pub(crate) fn set_muted(&self, muted: bool) { + if self.muted.load(Ordering::SeqCst) == muted { + return; + } + + self.muted.store(muted, Ordering::SeqCst); + self.rtc_track.set_enabled(!muted); + + self.dispatcher.lock().dispatch(if muted { + &TrackEvent::Mute + } else { + &TrackEvent::Unmute + }); + } + + pub(crate) fn register_observer(&self) -> mpsc::UnboundedReceiver { + self.dispatcher.lock().register() + } } #[derive(Clone, Debug)] @@ -159,6 +192,8 @@ impl TrackTrait for TrackHandle { fnc!(stream_state, &Self, [], StreamState); fnc!(start, &Self, [], ()); fnc!(stop, &Self, [], ()); + fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver); + fnc!(set_muted, &Self, [muted: bool], ()); ); } @@ -178,9 +213,10 @@ impl TrackHandle { macro_rules! impl_track_trait { ($x:ident) => { - use $crate::room::id::TrackSid; - use $crate::room::track::{StreamState, TrackKind, TrackTrait}; use std::sync::atomic::Ordering; + use tokio::sync::mpsc; + use $crate::room::id::TrackSid; + use $crate::room::track::{StreamState, TrackEvent, TrackKind, TrackTrait}; impl TrackTrait for $x { fn sid(&self) -> TrackSid { @@ -206,6 +242,14 @@ macro_rules! impl_track_trait { fn stop(&self) { self.shared.stop(); } + + fn register_observer(&self) -> mpsc::UnboundedReceiver { + self.shared.register_observer() + } + + fn set_muted(&self, muted: bool) { + self.shared.set_muted(muted); + } } }; } diff --git a/crates/livekit-core/src/room/track/remote_track.rs b/crates/livekit-core/src/room/track/remote_track.rs index c4badca..0ed2c09 100644 --- a/crates/livekit-core/src/room/track/remote_track.rs +++ b/crates/livekit-core/src/room/track/remote_track.rs @@ -1,10 +1,10 @@ use std::sync::Arc; - use super::{StreamState, TrackKind}; use crate::room::id::TrackSid; use crate::room::track::remote_audio_track::RemoteAudioTrack; use crate::room::track::remote_video_track::RemoteVideoTrack; -use crate::room::track::TrackHandle; +use crate::room::track::{TrackHandle, TrackEvent}; +use tokio::sync::mpsc; use livekit_utils::enum_dispatch; use super::TrackTrait; @@ -24,6 +24,8 @@ impl TrackTrait for RemoteTrackHandle { fnc!(stream_state, &Self, [], StreamState); fnc!(start, &Self, [], ()); fnc!(stop, &Self, [], ()); + fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver); + fnc!(set_muted, &Self, [muted: bool], ()); ); } diff --git a/crates/livekit-utils/Cargo.toml b/crates/livekit-utils/Cargo.toml index 27b9ca7..b2f868c 100644 --- a/crates/livekit-utils/Cargo.toml +++ b/crates/livekit-utils/Cargo.toml @@ -6,3 +6,5 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +parking_lot = "0.12.1" +tokio = { version = "1", features = ["full"] } diff --git a/crates/livekit-utils/src/lib.rs b/crates/livekit-utils/src/lib.rs index f328b57..a58478e 100644 --- a/crates/livekit-utils/src/lib.rs +++ b/crates/livekit-utils/src/lib.rs @@ -1 +1,2 @@ pub mod enum_dispatch; +pub mod observer; diff --git a/crates/livekit-utils/src/observer.rs b/crates/livekit-utils/src/observer.rs new file mode 100644 index 0000000..46cb800 --- /dev/null +++ b/crates/livekit-utils/src/observer.rs @@ -0,0 +1,39 @@ +// Really basic implementation of the observer pattern using mpsc channels. +// Currently unbounded channels + +use tokio::sync::mpsc; + +#[derive(Debug)] +pub struct Dispatcher +where + T: Clone, +{ + senders: Vec>, +} + +impl Default for Dispatcher +where + T: Clone, +{ + fn default() -> Self { + Self { + senders: Default::default(), + } + } +} + +impl Dispatcher +where + T: Clone, +{ + pub fn register(&mut self) -> mpsc::UnboundedReceiver { + let (tx, rx) = mpsc::unbounded_channel(); + self.senders.push(tx); + rx + } + + pub fn dispatch(&mut self, msg: &T) { + self.senders + .retain(|sender| sender.send(msg.clone()).is_err()); + } +}