connection_quality

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

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);

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<u8> for ConnectionQuality {
+    fn from(value: u8) -> Self {
+        match value {
+            1 => Self::Excellent,
+            2 => Self::Good,
+            3 => Self::Poor,
+            _ => Self::Unknown,
+        }
+    }
+}
+
+impl From<proto::ConnectionQuality> 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<ParticipantSid>,
@@ -22,6 +53,7 @@ pub(super) struct ParticipantShared {
     pub(super) tracks: RwLock<HashMap<TrackSid, TrackPublication>>,
     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<Self>, 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()
+            }
         }
     };
 }

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);

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<u8> for ConnectionState {
+impl From<u8> for ConnectionState {
-    type Error = &'static str;
+    fn from(value: u8) -> Self {
-
-    fn try_from(value: u8) -> Result<Self, Self::Error> {
         match value {
-            0 => Ok(ConnectionState::Disconnected),
+            0 => ConnectionState::Disconnected,
-            1 => Ok(ConnectionState::Connected),
+            1 => ConnectionState::Connected,
-            2 => Ok(ConnectionState::Reconnecting),
+            2 => ConnectionState::Reconnecting,
-            _ => Err("invalid ConnectionState"),
+            _ => 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<SpeakerInfo>) {
         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<proto::ConnectionQualityInfo>) {
+        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

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::id::TrackSid;
-        use crate::room::track::{StreamState, TrackKind, TrackTrait};
+        use $crate::room::track::{StreamState, TrackKind, TrackTrait};
         use std::sync::atomic::Ordering;
 
         impl TrackTrait for $x {

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<SpeakerInfo>,
     },
+    ConnectionQuality {
+        updates: Vec<proto::ConnectionQualityInfo>,
+    },
     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(())

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

@@ -56,6 +56,9 @@ pub enum SessionEvent {
     SpeakersChanged {
         speakers: Vec<proto::SpeakerInfo>,
     },
+    ConnectionQuality{
+        updates: Vec<proto::ConnectionQualityInfo>,
+    },
     // 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
+                let _ = self.emitter.send(SessionEvent::SpeakersChanged {
-                    .emitter
+                    speakers: speaker.speakers,
-                    .send(SessionEvent::SpeakersChanged { 
+                });
-                        speakers: speaker.speakers,
+            }
-                    });
+            signal_response::Message::ConnectionQuality(quality) => {
+                let _ = self.emitter.send(SessionEvent::ConnectionQuality{
+                    updates: quality.updates,
+                });
             }
             _ => {}
         }