events, observers & happy new year 🎆

Cargo.lock

@@ -512,6 +512,10 @@ dependencies = [
 [[package]]
 name = "livekit-utils"
 version = "0.1.0"
+dependencies = [
+ "parking_lot",
+ "tokio",
+]
 
 [[package]]
 name = "livekit-webrtc"

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<ParticipantHandle>,
+        speakers: Vec<Participant>,
     },
     ConnectionQualityChanged {
         quality: ConnectionQuality,
-        participant: ParticipantHandle,
+        participant: Participant,
     },
     DataReceived {
-        payload: Vec<u8>,
+        payload: Arc<Vec<u8>>,
         kind: data_packet::Kind,
         participant: Arc<RemoteParticipant>,
     },

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<Self>, info: ParticipantInfo) {
+    fn update_info(self: &Arc<Self>, info: ParticipantInfo, _emit_events: bool) {
         self.shared.update_info(info);
     }
 

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<Vec<u8>>,
+        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<Dispatcher<ParticipantEvent>>,
 }
 
 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<ParticipantEvent> {
+        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<Self>, info: ParticipantInfo);
+    fn update_info(self: &Arc<Self>, 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<HashMap<TrackSid, TrackPublication>>;
+    fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent>;
 }
 
 #[derive(Debug, Clone)]
-pub enum ParticipantHandle {
+pub enum Participant {
     Local(Arc<LocalParticipant>),
     Remote(Arc<RemoteParticipant>),
 }
 
-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<HashMap<TrackSid, TrackPublication>>);
+        fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver<ParticipantEvent>);
     );
 }
 
 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<HashMap<TrackSid, TrackPublication>> {
+                self.shared.tracks.read()
+            }
+
+            fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
+                self.shared.register_observer()
+            }
         }
     };
 }

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<Vec<u8>>, 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<Self>,
@@ -107,30 +122,57 @@ impl RemoteParticipant {
                 .add_track_publication(TrackPublication::Remote(remote_publication.clone()));
             track.start();
 
-            let _ = self
+            self.shared
-                .shared
+                .dispatcher
-                .internal_tx
+                .lock()
-                .send(SessionEvent::Room(RoomEvent::TrackSubscribed {
+                .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(
+            self.shared
-                RoomEvent::TrackSubscriptionFailed {
+                .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<Self>, 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<Self>, info: ParticipantInfo) {
+    fn update_info(self: &Arc<Self>, info: ParticipantInfo, emit_events: bool) {
         self.shared.update_info(info.clone());
 
         let mut valid_tracks = HashSet::<TrackSid>::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
+                // This is a new track, dispatch publish event
-                let _ =
+                if emit_events {
                     self.shared
-                        .internal_tx
+                        .dispatcher
-                        .send(SessionEvent::Room(RoomEvent::TrackPublished {
+                        .lock()
-                            publication: publication.clone(),
+                        .dispatch(&ParticipantEvent::TrackPublished { publication });
-                            participant: self.clone(),
+                }
-                        }));
             }
 
             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);
     }

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<TrackHandle>);
@@ -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<TrackDimension>,
     pub(super) mime_type: Mutex<String>,
-    pub(super) participant: ParticipantSid, // TODO(theomonnom) Use WeakParticipant instead
+    pub(super) muted: AtomicBool,
+    pub(super) participant: ParticipantSid,
+    pub(super) dispatcher: Mutex<Dispatcher<TrackEvent>>,
+    pub(super) close_sender: Mutex<Option<oneshot::Sender<()>>>,
 }
 
 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<Self>, track: Option<TrackHandle>) {
+        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<Self>,
+        mut close_receiver: oneshot::Receiver<()>,
+        mut track_receiver: mpsc::UnboundedReceiver<TrackEvent>,
+    ) {
+        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.muted.store(info.muted, Ordering::SeqCst);
-        *self.mime_type.lock() = info.mime_type;
+
+        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<TrackHandle>) {
-                *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<TrackHandle>) {
+        self.shared.update_track(track);
+    }
+
+    fn update_info(&self, info: TrackInfo) {
+        self.shared.update_info(info);
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct RemoteTrackPublication {
     shared: Arc<TrackPublicationShared>,
@@ -183,5 +248,15 @@ impl RemoteTrackPublication {
     }
 }
 
+impl TrackPublicationInternalTrait for RemoteTrackPublication {
+    fn update_track(&self, track: Option<TrackHandle>) {
+        self.shared.update_track(track);
+    }
+
+    fn update_info(&self, info: TrackInfo) {
+        self.shared.update_info(info);
+    }
+}
+
 impl_publication_trait!(LocalTrackPublication);
 impl_publication_trait!(RemoteTrackPublication);

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<SessionEvent>;
-pub(crate) type SessionEvents = mpsc::UnboundedReceiver<SessionEvent>;
-
-/// 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<String>,
     name: Mutex<String>,
     participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
-    active_speakers: RwLock<Vec<ParticipantHandle>>,
+    participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
+    active_speakers: RwLock<Vec<Participant>>,
     rtc_engine: Arc<RTCEngine>,
     local_participant: Arc<LocalParticipant>,
-    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(
+        let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver));
-            engine_events,
-            internal_rx,
-            close_receiver,
-            room_emitter,
-        ));
 
         inner.update_connection_state(ConnectionState::Connected);
 
@@ -174,23 +161,10 @@ impl SessionInner {
     async fn room_task(
         self: Arc<Self>,
         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(
+    /// Listen to the Participant events and forward them to the Room Dispatcher
-        &self,
+    #[instrument(level = Level::DEBUG)]
-        event: SessionEvent,
+    async fn participant_task(
-        room_emitter: &RoomEmitter,
+        self: Arc<Self>,
-    ) -> EngineResult<()> {
+        participant: Participant,
-        match event {
+        mut participant_events: mpsc::UnboundedReceiver<ParticipantEvent>,
-            SessionEvent::Room(event) => {
+        mut close_rx: oneshot::Receiver<()>,
-                if self.state.load(Ordering::Acquire) != ConnectionState::Connected as u8
+    ) {
-                    && matches!(
+        loop {
-                        event,
+            tokio::select! {
-                        RoomEvent::TrackPublished { .. }
+                res = participant_events.recv() => {
-                            | RoomEvent::TrackUnpublished { .. }
+                    match res {
-                            | RoomEvent::ParticipantConnected { .. }
+                        Some(event) => {
-                            | RoomEvent::ParticipantDisconnected { .. }
+                            if let Err(err) = self.on_participant_event(&participant, event).await {
-                            | RoomEvent::ActiveSpeakersChanged { .. }
+                                error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
-                    )
+                            }
-                {
+                        },
-                    return Ok(()); // Ignore the event
+                        _ => panic!("engine_events has been closed unexpectedly")
-                }
+                    };
-
+                },
-                // Forward the event to the public channel
+                _ = &mut close_rx => {
-                let _ = room_emitter.send(event);
+                    break;
+                },
             }
         }
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    async fn on_participant_event(
+        self: &Arc<Self>,
+        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
+                    let _ = self.room_emitter.send(RoomEvent::Reconnecting);
-                        .internal_tx
-                        .send(SessionEvent::Room(RoomEvent::Reconnecting));
                 }
             }
             EngineEvent::Resumed => {
                 self.update_connection_state(ConnectionState::Connected);
-                let _ = self
+                let _ = self.room_emitter.send(RoomEvent::Reconnected);
-                    .internal_tx
-                    .send(SessionEvent::Room(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
+                    let _ = self.room_emitter.send(RoomEvent::DataReceived {
-                        .internal_tx
+                        payload: payload.clone(),
-                        .send(SessionEvent::Room(RoomEvent::DataReceived {
+                        kind,
-                            payload,
+                        participant: participant.clone(),
-                            kind,
+                    });
-                            participant,
+
-                        }));
+                    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
+            .room_emitter
-            .send(SessionEvent::Room(RoomEvent::ConnectionStateChanged(state)));
+            .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<Self>, 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
+                    .room_emitter
-                    .send(SessionEvent::Room(RoomEvent::ParticipantConnected(
+                    .send(RoomEvent::ParticipantConnected(remote_participant.clone()));
-                        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<RemoteParticipant>) {
-        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
+            .room_emitter
-            .send(SessionEvent::Room(RoomEvent::ActiveSpeakersChanged {
+            .send(RoomEvent::ActiveSpeakersChanged { speakers });
-                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 _ =
+            let _ = self.room_emitter.send(RoomEvent::ConnectionQualityChanged {
-                self.internal_tx
+                participant,
-                    .send(SessionEvent::Room(RoomEvent::ConnectionQualityChanged {
+                quality,
-                        participant,
+            });
-                        quality,
-                    }));
         }
     }
 
     #[instrument(level = Level::DEBUG)]
-    fn handle_restarting(&self) {
+    fn handle_restarting(self: &Arc<Self>) {
         // 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
+            let _ = self.room_emitter.send(RoomEvent::Reconnecting);
-                .internal_tx
-                .send(SessionEvent::Room(RoomEvent::Reconnecting));
         }
     }
 
     #[instrument(level = Level::DEBUG)]
-    fn handle_restarted(&self) {
+    fn handle_restarted(self: &Arc<Self>) {
         // Full reconnect succeeded!
         let join_response = self.rtc_engine.join_response().unwrap();
 
         self.update_connection_state(ConnectionState::Connected);
-        let _ = self
+        let _ = self.room_emitter.send(RoomEvent::Reconnected);
-            .internal_tx
-            .send(SessionEvent::Room(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
+        let _ = self.room_emitter.send(RoomEvent::Disconnected);
-            .internal_tx
-            .send(SessionEvent::Room(RoomEvent::Disconnected));
     }
 
     /// Create a new participant
     /// Also add it to the participants list
+    #[instrument(level = Level::DEBUG)]
     fn create_participant(
-        &self,
+        self: &Arc<Self>,
         sid: ParticipantSid,
         identity: ParticipantIdentity,
         name: String,
         metadata: String,
     ) -> Arc<RemoteParticipant> {
-        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());
+        // Create the participant task
-        p
+        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<Self>, remote_participant: Arc<RemoteParticipant>) {
+        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<Arc<RemoteParticipant>> {

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<TrackEvent>;
+    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<String>,
     pub(super) kind: AtomicU8,         // TrackKind
     pub(super) stream_state: AtomicU8, // StreamState
+    pub(super) muted: AtomicBool,
     pub(super) rtc_track: MediaStreamTrackHandle,
+    pub(super) dispatcher: Mutex<Dispatcher<TrackEvent>>,
 }
 
 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<TrackEvent> {
+        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<TrackEvent>);
+        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<TrackEvent> {
+                self.shared.register_observer()
+            }
+
+            fn set_muted(&self, muted: bool) {
+                self.shared.set_muted(muted);
+            }
         }
     };
 }

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<TrackEvent>);
+        fnc!(set_muted, &Self, [muted: bool], ());
     );
 }
 

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"] }

crates/livekit-utils/src/lib.rs

@@ -1 +1,2 @@
 pub mod enum_dispatch;
+pub mod observer;

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<T>
+where
+    T: Clone,
+{
+    senders: Vec<mpsc::UnboundedSender<T>>,
+}
+
+impl<T> Default for Dispatcher<T>
+where
+    T: Clone,
+{
+    fn default() -> Self {
+        Self {
+            senders: Default::default(),
+        }
+    }
+}
+
+impl<T> Dispatcher<T>
+where
+    T: Clone,
+{
+    pub fn register(&mut self) -> mpsc::UnboundedReceiver<T> {
+        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());
+    }
+}