diff --git a/crates/livekit-core/src/room/participant/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs index 2cd42f4..45df481 100644 --- a/crates/livekit-core/src/room/participant/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -52,6 +52,14 @@ impl ParticipantInternalTrait for LocalParticipant { fn update_info(self: &Arc, 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); diff --git a/crates/livekit-core/src/room/participant/mod.rs b/crates/livekit-core/src/room/participant/mod.rs index 911f5d4..f92d006 100644 --- a/crates/livekit-core/src/room/participant/mod.rs +++ b/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, pub(super) metadata: Mutex, pub(super) tracks: RwLock>, + 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, 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; - - diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index 98c4fe4..8bc973c 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/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); diff --git a/crates/livekit-core/src/room/room_session.rs b/crates/livekit-core/src/room/room_session.rs index fecd1da..3b51fca 100644 --- a/crates/livekit-core/src/room/room_session.rs +++ b/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, name: Mutex, participants: RwLock>>, + active_speakers: RwLock>, rtc_engine: Arc, local_participant: Arc, 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> { - 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) { + 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> { + self.participants.read().get(sid).cloned() + } } fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> { diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index ab29a65..32dfb0e 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/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, kind: data_packet::Kind, }, + SpeakersChanged { + speakers: Vec, + }, 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(()) diff --git a/crates/livekit-core/src/rtc_engine/rtc_session.rs b/crates/livekit-core/src/rtc_engine/rtc_session.rs index b375761..f1bd690 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_session.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_session.rs @@ -53,6 +53,9 @@ pub enum SessionEvent { stream: MediaStream, receiver: RtpReceiver, }, + SpeakersChanged { + speakers: Vec, + }, // 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(_) => { - // TODO(theomonnonm) - } + Value::Speaker(_) => {} } } } diff --git a/crates/livekit-utils/src/enum_dispatch.rs b/crates/livekit-utils/src/enum_dispatch.rs index c7f59df..d24eb4d 100644 --- a/crates/livekit-utils/src/enum_dispatch.rs +++ b/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),+]) => { - fn $fnc(self: $self, $($arg: $t),*) -> $ret { + ($vis:vis$(,)? $fnc:ident, $self:ty, [$($arg:ident: $t:ty),*], $ret:ty, [$($variant:ident),+]) => { + $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); )* }; } -