audio_level & active_speakers

crates/livekit-core/src/room/participant/local_participant.rs

@@ -52,6 +52,14 @@ impl ParticipantInternalTrait for LocalParticipant {
     fn update_info(self: &Arc<Self>, info: ParticipantInfo) {
         self.shared.update_info(info);
     }
+
+    fn set_speaking(&self, speaking: bool) {
+        self.shared.set_speaking(speaking);
+    }
+
+    fn set_audio_level(&self, level: f32) {
+        self.shared.set_audio_level(level);
+    }
 }
 
 impl_participant_trait!(LocalParticipant);

crates/livekit-core/src/room/participant/mod.rs

@@ -7,6 +7,7 @@ use crate::room::room_session::SessionEmitter;
 use livekit_utils::enum_dispatch;
 use parking_lot::{Mutex, RwLock};
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::Arc;
 
 pub mod local_participant;
@@ -19,6 +20,8 @@ pub(super) struct ParticipantShared {
     pub(super) name: Mutex<String>,
     pub(super) metadata: Mutex<String>,
     pub(super) tracks: RwLock<HashMap<TrackSid, TrackPublication>>,
+    pub(super) speaking: AtomicBool,
+    pub(super) audio_level: AtomicU32,
     pub(super) internal_tx: SessionEmitter,
 }
 
@@ -36,6 +39,8 @@ impl ParticipantShared {
             name: Mutex::new(name),
             metadata: Mutex::new(metadata),
             tracks: Default::default(),
+            speaking: Default::default(),
+            audio_level: Default::default(),
             internal_tx,
         }
     }
@@ -47,12 +52,23 @@ impl ParticipantShared {
         *self.metadata.lock() = info.metadata; // TODO(theomonnom): callback MetadataChanged
     }
 
+    pub(crate) fn set_speaking(&self, speaking: bool) {
+        self.speaking.store(speaking, Ordering::SeqCst);
+    }
+
+    pub(crate) fn set_audio_level(&self, audio_level: f32) {
+        self.audio_level
+            .store(audio_level.to_bits(), Ordering::SeqCst)
+    }
+
     pub(crate) fn add_track_publication(&self, publication: TrackPublication) {
         self.tracks.write().insert(publication.sid(), publication);
     }
 }
 
 pub(crate) trait ParticipantInternalTrait {
+    fn set_speaking(&self, speaking: bool);
+    fn set_audio_level(&self, level: f32);
     fn update_info(self: &Arc<Self>, info: ParticipantInfo);
 }
 
@@ -61,6 +77,8 @@ pub trait ParticipantTrait {
     fn identity(&self) -> ParticipantIdentity;
     fn name(&self) -> String;
     fn metadata(&self) -> String;
+    fn is_speaking(&self) -> bool;
+    fn audio_level(&self) -> f32;
 }
 
 #[derive(Debug, Clone)]
@@ -72,7 +90,9 @@ pub enum ParticipantHandle {
 impl ParticipantHandle {
     enum_dispatch!(
         [Local, Remote]
-        fnc!(update_info, &Self, [info: ParticipantInfo], ());
+        fnc!(pub(crate), update_info, &Self, [info: ParticipantInfo], ());
+        fnc!(pub(crate), set_speaking, &Self, [speaking: bool], ());
+        fnc!(pub(crate), set_audio_level, &Self, [audio_level: f32], ());
     );
 }
 
@@ -83,6 +103,8 @@ impl ParticipantTrait for ParticipantHandle {
         fnc!(identity, &Self, [], ParticipantIdentity);
         fnc!(name, &Self, [], String);
         fnc!(metadata, &Self, [], String);
+        fnc!(is_speaking, &Self, [], bool);
+        fnc!(audio_level, &Self, [], f32);
     );
 }
 
@@ -90,6 +112,7 @@ macro_rules! impl_participant_trait {
     ($x:ty) => {
         use crate::proto::ParticipantInfo;
         use crate::room::id::{ParticipantIdentity, ParticipantSid};
+        use std::sync::atomic::Ordering;
         use std::sync::Arc;
 
         impl crate::room::participant::ParticipantTrait for $x {
@@ -108,10 +131,16 @@ macro_rules! impl_participant_trait {
             fn metadata(&self) -> String {
                 self.shared.metadata.lock().clone()
             }
+
+            fn is_speaking(&self) -> bool {
+                self.shared.speaking.load(Ordering::SeqCst)
+            }
+
+            fn audio_level(&self) -> f32 {
+                f32::from_bits(self.shared.audio_level.load(Ordering::SeqCst))
+            }
         }
     };
 }
 
 pub(super) use impl_participant_trait;
-
-

crates/livekit-core/src/room/participant/remote_participant.rs

@@ -1,4 +1,3 @@
-use crate::proto::TrackInfo;
 use crate::room::id::TrackSid;
 use crate::room::participant::{
     impl_participant_trait, ParticipantInternalTrait, ParticipantShared,
@@ -156,6 +155,14 @@ impl ParticipantInternalTrait for RemoteParticipant {
             valid_tracks.insert(track.sid.into());
         }
     }
+
+    fn set_speaking(&self, speaking: bool) {
+        self.shared.set_speaking(speaking);
+    }
+
+    fn set_audio_level(&self, level: f32) {
+        self.shared.set_audio_level(level);
+    }
 }
 
 impl_participant_trait!(RemoteParticipant);

crates/livekit-core/src/room/room_session.rs

@@ -1,9 +1,10 @@
 use super::id::{ParticipantIdentity, ParticipantSid};
 use super::participant::local_participant::LocalParticipant;
 use super::participant::remote_participant::RemoteParticipant;
+use super::participant::ParticipantHandle;
 use super::participant::{ParticipantInternalTrait, ParticipantTrait};
 use super::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario};
-use crate::proto::{self, participant_info};
+use crate::proto::{self, participant_info, SpeakerInfo};
 use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine};
 use crate::signal_client::SignalOptions;
 use parking_lot::{Mutex, RwLock};
@@ -50,6 +51,7 @@ struct SessionInner {
     sid: Mutex<String>,
     name: Mutex<String>,
     participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
+    active_speakers: RwLock<Vec<ParticipantHandle>>,
     rtc_engine: Arc<RTCEngine>,
     local_participant: Arc<LocalParticipant>,
     internal_tx: SessionEmitter,
@@ -95,6 +97,7 @@ impl SessionHandle {
             sid: Mutex::new(room_info.sid),
             name: Mutex::new(room_info.name),
             participants: Default::default(),
+            active_speakers: Default::default(),
             rtc_engine,
             local_participant,
             internal_tx,
@@ -214,7 +217,13 @@ impl SessionInner {
         match event {
             SessionEvent::Room(event) => {
                 if self.state.load(Ordering::Acquire) != ConnectionState::Connected as u8
-                    && matches!(event, RoomEvent::TrackPublished { .. })
+                    && matches!(
+                        event,
+                        RoomEvent::TrackPublished { .. }
+                            | RoomEvent::ParticipantConnected { .. }
+                            | RoomEvent::ParticipantDisconnected { .. }
+                            | RoomEvent::ActiveSpeakersChanged { .. }
+                    )
                 {
                     return Ok(()); // Ignore the event
                 }
@@ -298,6 +307,7 @@ impl SessionInner {
                         }));
                 }
             }
+            EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers),
         }
 
         Ok(())
@@ -308,10 +318,6 @@ impl SessionInner {
         self.rtc_engine.close().await;
     }
 
-    fn get_participant(&self, sid: &ParticipantSid) -> Option<Arc<RemoteParticipant>> {
-        self.participants.read().get(sid).cloned()
-    }
-
     /// Change the connection state and emit an event
     /// Does nothing if the state is already the same
     #[instrument(level = Level::DEBUG)]
@@ -384,6 +390,40 @@ impl SessionInner {
             )));
     }
 
+    #[instrument(level = Level::DEBUG)]
+    fn handle_speakers_changed(&self, speakers_info: Vec<SpeakerInfo>) {
+        let mut speakers = Vec::new();
+
+        for speaker in speakers_info {
+            let participant = {
+                if speaker.sid == self.local_participant.sid() {
+                    ParticipantHandle::Local(self.local_participant.clone())
+                } else {
+                    if let Some(participant) = self.get_participant(&speaker.sid.into()) {
+                        ParticipantHandle::Remote(participant)
+                    } else {
+                        continue;
+                    }
+                }
+            };
+
+            participant.set_speaking(speaker.active);
+            participant.set_audio_level(speaker.level);
+
+            if speaker.active {
+                speakers.push(participant);
+            }
+        }
+
+        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,
+            }));
+    }
+
     #[instrument(level = Level::DEBUG)]
     fn handle_restarting(&self) {
         // Remove existing participants/subscriptions on full reconnect
@@ -451,6 +491,10 @@ impl SessionInner {
         self.participants.write().insert(sid, p.clone());
         p
     }
+
+    fn get_participant(&self, sid: &ParticipantSid) -> Option<Arc<RemoteParticipant>> {
+        self.participants.read().get(sid).cloned()
+    }
 }
 
 fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {

crates/livekit-core/src/rtc_engine/mod.rs

@@ -18,7 +18,7 @@ use lazy_static::lazy_static;
 use tokio::sync::{mpsc, oneshot};
 use tracing::{error, info, warn};
 
-use crate::proto::{data_packet, DataPacket, JoinResponse, ParticipantUpdate};
+use crate::proto::{data_packet, DataPacket, JoinResponse, ParticipantUpdate, SpeakerInfo};
 use crate::rtc_engine::lk_runtime::LKRuntime;
 use crate::signal_client::{SignalError, SignalOptions};
 
@@ -78,6 +78,9 @@ pub enum EngineEvent {
         payload: Vec<u8>,
         kind: data_packet::Kind,
     },
+    SpeakersChanged {
+        speakers: Vec<SpeakerInfo>,
+    },
     Resuming,
     Resumed,
     Restarting,
@@ -274,6 +277,12 @@ impl EngineInner {
                     })
                     .await;
             }
+            SessionEvent::SpeakersChanged { speakers } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::SpeakersChanged { speakers })
+                    .await;
+            }
             SessionEvent::Connected => {}
         }
         Ok(())

crates/livekit-core/src/rtc_engine/rtc_session.rs

@@ -53,6 +53,9 @@ pub enum SessionEvent {
         stream: MediaStream,
         receiver: RtpReceiver,
     },
+    SpeakersChanged {
+        speakers: Vec<proto::SpeakerInfo>,
+    },
     // TODO(theomonnom): Move entirely the reconnection logic on mod.rs
     Close {
         source: String,
@@ -416,6 +419,13 @@ impl SessionInner {
                     true,
                 );
             }
+            signal_response::Message::SpeakersChanged(speaker) => {
+                let _ = self
+                    .emitter
+                    .send(SessionEvent::SpeakersChanged { 
+                        speakers: speaker.speakers,
+                    });
+            }
             _ => {}
         }
 
@@ -510,9 +520,7 @@ impl SessionInner {
                             kind: data_packet::Kind::from_i32(data.kind).unwrap(),
                         });
                     }
-                    Value::Speaker(_) => {
+                    Value::Speaker(_) => {}
-                        // TODO(theomonnonm)
-                    }
                 }
             }
         }

crates/livekit-utils/src/enum_dispatch.rs

@@ -1,3 +1,6 @@
+
+// TODO(theomonnom): Match the complete function signature like:
+//  - pub(crate) fn update_info(&self, info: ParticipantInfo) -> ();
 #[macro_export]
 macro_rules! enum_dispatch {
     // This arm is used to avoid nested loops with the arguments
@@ -10,16 +13,15 @@ macro_rules! enum_dispatch {
         }
     };
 
-    ($fnc:ident, $self:ty, [$($arg:ident: $t:ty),*], $ret:ty, [$($variant:ident),+]) => {
+    ($vis:vis$(,)? $fnc:ident, $self:ty, [$($arg:ident: $t:ty),*], $ret:ty, [$($variant:ident),+]) => {
-        fn $fnc(self: $self, $($arg: $t),*) -> $ret {
+        $vis fn $fnc(self: $self, $($arg: $t),*) -> $ret {
             enum_dispatch!(@match self $fnc ($($arg,)*) [$($variant),+])
         }
     };
 
-    ($variants:tt $(fnc!($fnc:ident, $self:ty, $args:tt, $ret:ty);)+) => {
+    ($variants:tt $(fnc!($vis:vis$(,)? $fnc:ident, $self:ty, $args:tt, $ret:ty);)+) => {
         $(
-            enum_dispatch!($fnc, $self, $args, $ret, $variants);
+            enum_dispatch!($vis, $fnc, $self, $args, $ret, $variants);
         )*
     };
 }
-