diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index 8a6b17a..73fdc2e 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -1,3 +1,4 @@ +use self::participant::ConnectionQuality; use self::room_session::{ConnectionState, RoomSession, SessionHandle}; use crate::proto::data_packet; use crate::room::id::TrackSid; @@ -76,6 +77,10 @@ pub enum RoomEvent { ActiveSpeakersChanged { speakers: Vec, }, + ConnectionQualityChanged { + quality: ConnectionQuality, + participant: ParticipantHandle, + }, DataReceived { payload: Vec, kind: data_packet::Kind, diff --git a/crates/livekit-core/src/room/participant/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs index 45df481..5e9030f 100644 --- a/crates/livekit-core/src/room/participant/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -60,6 +60,10 @@ impl ParticipantInternalTrait for LocalParticipant { 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); + } } 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 f92d006..47dddfe 100644 --- a/crates/livekit-core/src/room/participant/mod.rs +++ b/crates/livekit-core/src/room/participant/mod.rs @@ -1,3 +1,4 @@ +use crate::proto; use crate::proto::ParticipantInfo; use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; use crate::room::participant::local_participant::LocalParticipant; @@ -7,12 +8,42 @@ 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::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering}; use std::sync::Arc; pub mod local_participant; pub mod remote_participant; +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +#[repr(u8)] +pub enum ConnectionQuality { + Unknown, + Excellent, + Good, + Poor, +} + +impl From for ConnectionQuality { + fn from(value: u8) -> Self { + match value { + 1 => Self::Excellent, + 2 => Self::Good, + 3 => Self::Poor, + _ => Self::Unknown, + } + } +} + +impl From for ConnectionQuality { + fn from(value: proto::ConnectionQuality) -> Self { + match value { + proto::ConnectionQuality::Excellent => Self::Excellent, + proto::ConnectionQuality::Good => Self::Good, + proto::ConnectionQuality::Poor => Self::Poor, + } + } +} + #[derive(Debug)] pub(super) struct ParticipantShared { pub(super) sid: Mutex, @@ -22,6 +53,7 @@ pub(super) struct ParticipantShared { pub(super) tracks: RwLock>, pub(super) speaking: AtomicBool, pub(super) audio_level: AtomicU32, + pub(super) connection_quality: AtomicU8, pub(super) internal_tx: SessionEmitter, } @@ -41,6 +73,7 @@ impl ParticipantShared { tracks: Default::default(), speaking: Default::default(), audio_level: Default::default(), + connection_quality: AtomicU8::new(ConnectionQuality::Unknown as u8), internal_tx, } } @@ -61,6 +94,11 @@ impl ParticipantShared { .store(audio_level.to_bits(), Ordering::SeqCst) } + pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) { + self.connection_quality + .store(quality as u8, Ordering::SeqCst); + } + pub(crate) fn add_track_publication(&self, publication: TrackPublication) { self.tracks.write().insert(publication.sid(), publication); } @@ -69,6 +107,7 @@ impl ParticipantShared { 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); } @@ -79,6 +118,7 @@ pub trait ParticipantTrait { fn metadata(&self) -> String; fn is_speaking(&self) -> bool; fn audio_level(&self) -> f32; + fn connection_quality(&self) -> ConnectionQuality; } #[derive(Debug, Clone)] @@ -93,6 +133,7 @@ impl ParticipantHandle { 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], ()); + fnc!(pub(crate), set_connection_quality, &Self, [quality: ConnectionQuality], ()); ); } @@ -105,15 +146,17 @@ impl ParticipantTrait for ParticipantHandle { fnc!(metadata, &Self, [], String); fnc!(is_speaking, &Self, [], bool); fnc!(audio_level, &Self, [], f32); + fnc!(connection_quality, &Self, [], ConnectionQuality); ); } 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; + 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 { @@ -139,6 +182,10 @@ macro_rules! impl_participant_trait { fn audio_level(&self) -> f32 { f32::from_bits(self.shared.audio_level.load(Ordering::SeqCst)) } + + fn connection_quality(&self) -> ConnectionQuality { + self.shared.connection_quality.load(Ordering::SeqCst).into() + } } }; } diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index 8bc973c..ee492ec 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -163,6 +163,10 @@ 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); + } } 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 3b51fca..34b0866 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::ParticipantHandle; +use super::participant::{ConnectionQuality, ParticipantHandle}; use super::participant::{ParticipantInternalTrait, ParticipantTrait}; use super::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario}; use crate::proto::{self, participant_info, SpeakerInfo}; @@ -29,17 +29,16 @@ pub enum ConnectionState { Disconnected, Connected, Reconnecting, + Unknown, } -impl TryFrom for ConnectionState { - type Error = &'static str; - - fn try_from(value: u8) -> Result { +impl From for ConnectionState { + fn from(value: u8) -> Self { match value { - 0 => Ok(ConnectionState::Disconnected), - 1 => Ok(ConnectionState::Connected), - 2 => Ok(ConnectionState::Reconnecting), - _ => Err("invalid ConnectionState"), + 0 => ConnectionState::Disconnected, + 1 => ConnectionState::Connected, + 2 => ConnectionState::Reconnecting, + _ => ConnectionState::Unknown, } } } @@ -220,6 +219,7 @@ impl SessionInner { && matches!( event, RoomEvent::TrackPublished { .. } + | RoomEvent::TrackUnpublished { .. } | RoomEvent::ParticipantConnected { .. } | RoomEvent::ParticipantDisconnected { .. } | RoomEvent::ActiveSpeakersChanged { .. } @@ -308,6 +308,9 @@ impl SessionInner { } } EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers), + EngineEvent::ConnectionQuality { updates } => { + self.handle_connection_quality_update(updates) + } } Ok(()) @@ -390,6 +393,8 @@ impl SessionInner { ))); } + /// Active speakers changed + /// Update the participants & sort the active_speakers by audio_level #[instrument(level = Level::DEBUG)] fn handle_speakers_changed(&self, speakers_info: Vec) { let mut speakers = Vec::new(); @@ -424,6 +429,38 @@ impl SessionInner { })); } + /// Handle a connection quality update + /// Emit ConnectionQualityChanged event for the concerned participants + #[instrument(level = Level::DEBUG)] + fn handle_connection_quality_update(&self, updates: Vec) { + for update in updates { + let participant = { + if update.participant_sid == self.local_participant.sid() { + ParticipantHandle::Local(self.local_participant.clone()) + } else { + if let Some(participant) = self.get_participant(&update.participant_sid.into()) + { + ParticipantHandle::Remote(participant) + } else { + continue; + } + } + }; + + let quality: ConnectionQuality = proto::ConnectionQuality::from_i32(update.quality) + .unwrap() + .into(); + + participant.set_connection_quality(quality); + let _ = + self.internal_tx + .send(SessionEvent::Room(RoomEvent::ConnectionQualityChanged { + participant, + quality, + })); + } + } + #[instrument(level = Level::DEBUG)] fn handle_restarting(&self) { // Remove existing participants/subscriptions on full reconnect diff --git a/crates/livekit-core/src/room/track/mod.rs b/crates/livekit-core/src/room/track/mod.rs index 28d4a96..31ceb70 100644 --- a/crates/livekit-core/src/room/track/mod.rs +++ b/crates/livekit-core/src/room/track/mod.rs @@ -178,8 +178,8 @@ impl TrackHandle { macro_rules! impl_track_trait { ($x:ident) => { - use crate::room::id::TrackSid; - use crate::room::track::{StreamState, TrackKind, TrackTrait}; + use $crate::room::id::TrackSid; + use $crate::room::track::{StreamState, TrackKind, TrackTrait}; use std::sync::atomic::Ordering; impl TrackTrait for $x { diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index 32dfb0e..22b0b3d 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -18,7 +18,9 @@ use lazy_static::lazy_static; use tokio::sync::{mpsc, oneshot}; use tracing::{error, info, warn}; -use crate::proto::{data_packet, DataPacket, JoinResponse, ParticipantUpdate, SpeakerInfo}; +use crate::proto::{ + self as proto, data_packet, DataPacket, JoinResponse, ParticipantUpdate, SpeakerInfo, +}; use crate::rtc_engine::lk_runtime::LKRuntime; use crate::signal_client::{SignalError, SignalOptions}; @@ -81,6 +83,9 @@ pub enum EngineEvent { SpeakersChanged { speakers: Vec, }, + ConnectionQuality { + updates: Vec, + }, Resuming, Resumed, Restarting, @@ -283,6 +288,12 @@ impl EngineInner { .send(EngineEvent::SpeakersChanged { speakers }) .await; } + SessionEvent::ConnectionQuality { updates } => { + let _ = self + .engine_emitter + .send(EngineEvent::ConnectionQuality { updates }) + .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 f1bd690..475de55 100644 --- a/crates/livekit-core/src/rtc_engine/rtc_session.rs +++ b/crates/livekit-core/src/rtc_engine/rtc_session.rs @@ -56,6 +56,9 @@ pub enum SessionEvent { SpeakersChanged { speakers: Vec, }, + ConnectionQuality{ + updates: Vec, + }, // TODO(theomonnom): Move entirely the reconnection logic on mod.rs Close { source: String, @@ -420,11 +423,14 @@ impl SessionInner { ); } signal_response::Message::SpeakersChanged(speaker) => { - let _ = self - .emitter - .send(SessionEvent::SpeakersChanged { - speakers: speaker.speakers, - }); + let _ = self.emitter.send(SessionEvent::SpeakersChanged { + speakers: speaker.speakers, + }); + } + signal_response::Message::ConnectionQuality(quality) => { + let _ = self.emitter.send(SessionEvent::ConnectionQuality{ + updates: quality.updates, + }); } _ => {} }