cleanup: room & dependencies (#76)

livekit/src/prelude.rs

@@ -1,6 +1,6 @@
 pub use crate::participant::{LocalParticipant, Participant, ParticipantEvent, RemoteParticipant};
 
-pub use crate::{ConnectionState, Room, RoomError, RoomEvent, RoomResult, RoomSession};
+pub use crate::{ConnectionState, Room, RoomError, RoomEvent, RoomResult};
 
 pub use crate::publication::{LocalTrackPublication, RemoteTrackPublication, TrackPublication};
 

livekit/src/room/mod.rs

@@ -2,11 +2,19 @@ use self::track::RemoteTrack;
 use crate::participant::ConnectionQuality;
 use crate::prelude::*;
 use crate::rtc_engine::EngineError;
+use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RtcEngine};
+use crate::signal_client::SignalOptions;
 use livekit_protocol as proto;
+use livekit_protocol::observer::Dispatcher;
+use parking_lot::{Mutex, RwLock, RwLockReadGuard};
+use std::collections::HashMap;
 use std::fmt::Debug;
+use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use thiserror::Error;
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
+use tokio::task::JoinHandle;
+use tracing::{error, info, instrument, trace, Level};
 
 pub use crate::rtc_engine::SimulateScenario;
 
@@ -14,11 +22,8 @@ pub mod id;
 pub mod options;
 pub mod participant;
 pub mod publication;
-pub mod room_session;
 pub mod track;
 
-pub use room_session::*;
-
 pub type RoomResult<T> = Result<T, RoomError>;
 
 #[derive(Error, Debug)]
@@ -29,6 +34,8 @@ pub enum RoomError {
     Internal(String),
     #[error("this track or a track of the same source is already published")]
     TrackAlreadyPublished,
+    #[error("already closed")]
+    AlreadyClosed,
 }
 
 #[derive(Clone, Debug)]
@@ -85,9 +92,32 @@ pub enum RoomEvent {
     Reconnected,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum ConnectionState {
+    Disconnected,
+    Connected,
+    Reconnecting,
+    Unknown,
+}
+
+struct RoomHandle {
+    session_task: JoinHandle<()>,
+    close_emitter: oneshot::Sender<()>,
+}
+
 pub struct Room {
-    handle: SessionHandle,
+    inner: Arc<SessionInner>,
+    handle: Mutex<Option<RoomHandle>>,
+}
+
+impl Debug for Room {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Room")
+            .field("sid", &self.sid())
+            .field("name", &self.name())
+            .field("connection_state", &self.connection_state())
+            .finish()
+    }
 }
 
 impl Room {
@@ -95,21 +125,538 @@ impl Room {
         url: &str,
         token: &str,
     ) -> RoomResult<(Self, mpsc::UnboundedReceiver<RoomEvent>)> {
-        let handle = SessionHandle::connect(url, token).await?;
-        let events = handle.subscribe();
-        Ok((Self { handle }, events))
+        let (rtc_engine, engine_events) = RtcEngine::new();
+        let rtc_engine = Arc::new(rtc_engine);
+        rtc_engine
+            .connect(url, token, SignalOptions::default())
+            .await?;
+
+        let join_response = rtc_engine.join_response().unwrap();
+        let pi = join_response.participant.unwrap().clone();
+        let local_participant = LocalParticipant::new(
+            rtc_engine.clone(),
+            pi.sid.into(),
+            pi.identity.into(),
+            pi.name,
+            pi.metadata,
+        );
+
+        let room_info = join_response.room.unwrap();
+        let inner = Arc::new(SessionInner {
+            state: AtomicU8::new(ConnectionState::Disconnected as u8),
+            sid: Mutex::new(room_info.sid.into()),
+            name: Mutex::new(room_info.name),
+            metadata: Mutex::new(room_info.metadata),
+            participants: Default::default(),
+            participants_tasks: Default::default(),
+            active_speakers: Default::default(),
+            rtc_engine,
+            local_participant,
+            dispatcher: Default::default(),
+        });
+
+        for pi in join_response.other_participants {
+            let participant = {
+                let pi = pi.clone();
+                inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
+            };
+            participant.update_info(pi.clone());
+        }
+
+        let (close_emitter, close_receiver) = oneshot::channel();
+        let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver));
+
+        inner.update_connection_state(ConnectionState::Connected);
+
+        let session = Self {
+            inner,
+            handle: Mutex::new(Some(RoomHandle {
+                session_task,
+                close_emitter,
+            })),
+        };
+
+        let events = session.subscribe();
+        Ok((session, events))
     }
 
-    pub async fn close(self) {
-        self.handle.close().await;
+    pub async fn close(&self) -> RoomResult<()> {
+        if let Some(handle) = self.handle.lock().take() {
+            self.inner.close().await;
+            handle.close_emitter.send(()).ok();
+            handle.session_task.await.ok();
+            Ok(())
+        } else {
+            Err(RoomError::AlreadyClosed)
+        }
     }
 
-    /// Allow multiple subscribers/observers to receive events
     pub fn subscribe(&self) -> mpsc::UnboundedReceiver<RoomEvent> {
-        self.handle.subscribe()
+        self.inner.dispatcher.register()
     }
 
-    pub fn session(&self) -> RoomSession {
-        self.handle.session()
+    pub fn sid(&self) -> RoomSid {
+        self.inner.sid.lock().clone()
+    }
+
+    pub fn name(&self) -> String {
+        self.inner.name.lock().clone()
+    }
+
+    pub fn metadata(&self) -> String {
+        self.inner.metadata.lock().clone()
+    }
+
+    pub fn local_participant(&self) -> LocalParticipant {
+        self.inner.local_participant.clone()
+    }
+
+    pub fn connection_state(&self) -> ConnectionState {
+        self.inner.state.load(Ordering::Acquire).try_into().unwrap()
+    }
+
+    pub fn participants(&self) -> RwLockReadGuard<HashMap<ParticipantSid, RemoteParticipant>> {
+        self.inner.participants.read()
+    }
+
+    pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
+        self.inner.rtc_engine.simulate_scenario(scenario).await
+    }
+}
+
+struct SessionInner {
+    state: AtomicU8, // ConnectionState
+    sid: Mutex<RoomSid>,
+    name: Mutex<String>,
+    metadata: Mutex<String>,
+    participants: RwLock<HashMap<ParticipantSid, RemoteParticipant>>,
+    participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
+    active_speakers: RwLock<Vec<Participant>>,
+    rtc_engine: Arc<RtcEngine>,
+    local_participant: LocalParticipant,
+    dispatcher: Dispatcher<RoomEvent>,
+}
+
+impl Debug for SessionInner {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SessionInner")
+            .field("sid", &self.sid)
+            .field("name", &self.name)
+            .field("rtc_engine", &self.rtc_engine)
+            .finish()
+    }
+}
+
+impl SessionInner {
+    #[instrument(level = Level::DEBUG)]
+    async fn room_task(
+        self: Arc<Self>,
+        mut engine_events: EngineEvents,
+        mut close_receiver: oneshot::Receiver<()>,
+    ) {
+        loop {
+            tokio::select! {
+                res = engine_events.recv() => {
+                    if let Some(event) = res {
+                        if let Err(err) = self.on_engine_event(event).await {
+                            error!("failed to handle engine event: {:?}", err);
+                        }
+                    }
+                },
+                 _ = &mut close_receiver => {
+                    trace!("closing room_task");
+                    break;
+                }
+            }
+        }
+    }
+
+    /// Listen to the Participant events and forward them to the Room Dispatcher
+    #[instrument(level = Level::DEBUG)]
+    async fn participant_task(
+        self: Arc<Self>,
+        participant: Participant,
+        mut participant_events: mpsc::UnboundedReceiver<ParticipantEvent>,
+        mut close_rx: oneshot::Receiver<()>,
+    ) {
+        loop {
+            tokio::select! {
+                res = participant_events.recv() => {
+                    if let Some(event) = res {
+                        if let Err(err) = self.on_participant_event(&participant, event).await {
+                            error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
+                        }
+                    }
+                },
+                _ = &mut close_rx => {
+                    trace!("closing participant_task for {:?}", participant.sid());
+                    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 } => {
+                    self.dispatcher.dispatch(&RoomEvent::TrackPublished {
+                        participant: remote_participant.clone(),
+                        publication,
+                    });
+                }
+                ParticipantEvent::TrackUnpublished { publication } => {
+                    self.dispatcher.dispatch(&RoomEvent::TrackUnpublished {
+                        participant: remote_participant.clone(),
+                        publication,
+                    });
+                }
+                ParticipantEvent::TrackSubscribed { track, publication } => {
+                    self.dispatcher.dispatch(&RoomEvent::TrackSubscribed {
+                        participant: remote_participant.clone(),
+                        track,
+                        publication,
+                    });
+                }
+                ParticipantEvent::TrackUnsubscribed { track, publication } => {
+                    self.dispatcher.dispatch(&RoomEvent::TrackUnsubscribed {
+                        participant: remote_participant.clone(),
+                        track,
+                        publication,
+                    });
+                }
+                _ => {}
+            };
+        }
+
+        Ok(())
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
+        match event {
+            EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates),
+            EngineEvent::MediaTrack {
+                track,
+                stream,
+                receiver: _,
+            } => {
+                let stream_id = stream.id();
+                let lk_stream_id = unpack_stream_id(&stream_id);
+                if lk_stream_id.is_none() {
+                    Err(RoomError::Internal(format!(
+                        "MediaTrack event with invalid track_id: {:?}",
+                        &stream_id
+                    )))?;
+                }
+
+                let (participant_sid, track_sid) = lk_stream_id.unwrap();
+                let track_sid = track_sid.to_owned().into();
+                let remote_participant = self.get_participant(&participant_sid.to_string().into());
+
+                if let Some(remote_participant) = remote_participant {
+                    tokio::spawn(async move {
+                        remote_participant
+                            .add_subscribed_media_track(track_sid, track)
+                            .await;
+                    });
+                } else {
+                    // The server should send participant updates before sending a new offer
+                    // So this should never happen.
+                    Err(RoomError::Internal(format!(
+                        "AddTrack event with invalid participant_sid: {:?}",
+                        participant_sid
+                    )))?;
+                }
+            }
+            EngineEvent::Resuming => {
+                if self.update_connection_state(ConnectionState::Reconnecting) {
+                    self.dispatcher.dispatch(&RoomEvent::Reconnecting);
+                }
+            }
+            EngineEvent::Resumed => {
+                self.update_connection_state(ConnectionState::Connected);
+                self.dispatcher.dispatch(&RoomEvent::Reconnected);
+
+                // TODO(theomonnom): Update subscriptions settings
+                // TODO(theomonnom): Send sync state
+            }
+            EngineEvent::Restarting => self.handle_restarting(),
+            EngineEvent::Restarted => self.handle_restarted(),
+            EngineEvent::Disconnected => self.handle_disconnected(),
+            EngineEvent::Data {
+                payload,
+                kind,
+                participant_sid,
+            } => {
+                let payload = Arc::new(payload);
+                if let Some(participant) = self.get_participant(&participant_sid.into()) {
+                    self.dispatcher.dispatch(&RoomEvent::DataReceived {
+                        payload: payload.clone(),
+                        kind,
+                        participant: participant.clone(),
+                    });
+
+                    participant.on_data_received(payload, kind);
+                }
+            }
+            EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers),
+            EngineEvent::ConnectionQuality { updates } => {
+                self.handle_connection_quality_update(updates)
+            }
+        }
+
+        Ok(())
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    async fn close(&self) {
+        self.rtc_engine.close().await;
+    }
+
+    /// Change the connection state and emit an event
+    /// Does nothing if the state is already the same
+    #[instrument(level = Level::DEBUG)]
+    fn update_connection_state(&self, state: ConnectionState) -> bool {
+        let old_state = self.state.load(Ordering::Acquire);
+        if old_state == state as u8 {
+            return false;
+        }
+
+        self.state.store(state as u8, Ordering::Release);
+        self.dispatcher
+            .dispatch(&RoomEvent::ConnectionStateChanged(state));
+        return true;
+    }
+
+    /// Update the participants inside a Room.
+    /// It'll create, update or remove a participant
+    /// It also update the participant tracks.
+    #[instrument(level = Level::DEBUG)]
+    fn handle_participant_update(self: &Arc<Self>, updates: Vec<proto::ParticipantInfo>) {
+        for pi in updates {
+            if pi.sid == self.local_participant.sid()
+                || pi.identity == self.local_participant.identity()
+            {
+                self.local_participant.clone().update_info(pi);
+                continue;
+            }
+
+            let remote_participant = self.get_participant(&pi.sid.clone().into());
+
+            if let Some(remote_participant) = remote_participant {
+                if pi.state == proto::participant_info::State::Disconnected as i32 {
+                    // Participant disconnected
+                    info!("Participant disconnected: {}", pi.sid);
+                    self.clone()
+                        .handle_participant_disconnect(remote_participant)
+                } else {
+                    // Participant is already connected, update the it
+                    remote_participant.update_info(pi.clone());
+                }
+            } else {
+                // Create a new participant
+                info!("Participant connected: {}", pi.sid);
+                let remote_participant = {
+                    let pi = pi.clone();
+                    self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
+                };
+
+                let _ = self
+                    .dispatcher
+                    .dispatch(&RoomEvent::ParticipantConnected(remote_participant.clone()));
+
+                remote_participant.update_info(pi.clone());
+            }
+        }
+    }
+
+    /// 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<proto::SpeakerInfo>) {
+        let mut speakers = Vec::new();
+
+        for speaker in speakers_info {
+            let participant = {
+                if speaker.sid == self.local_participant.sid() {
+                    Participant::Local(self.local_participant.clone())
+                } else {
+                    if let Some(participant) = self.get_participant(&speaker.sid.into()) {
+                        Participant::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
+            .dispatcher
+            .dispatch(&RoomEvent::ActiveSpeakersChanged { speakers });
+    }
+
+    /// 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() {
+                    Participant::Local(self.local_participant.clone())
+                } else {
+                    if let Some(participant) = self.get_participant(&update.participant_sid.into())
+                    {
+                        Participant::Remote(participant)
+                    } else {
+                        continue;
+                    }
+                }
+            };
+
+            let quality: ConnectionQuality = proto::ConnectionQuality::from_i32(update.quality)
+                .unwrap()
+                .into();
+
+            participant.set_connection_quality(quality);
+            self.dispatcher
+                .dispatch(&RoomEvent::ConnectionQualityChanged {
+                    participant,
+                    quality,
+                });
+        }
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    fn handle_restarting(self: &Arc<Self>) {
+        // Remove existing participants/subscriptions on full reconnect
+        for (_, participant) in self.participants.read().iter() {
+            self.clone()
+                .handle_participant_disconnect(participant.clone());
+        }
+
+        if self.update_connection_state(ConnectionState::Reconnecting) {
+            self.dispatcher.dispatch(&RoomEvent::Reconnecting);
+        }
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    fn handle_restarted(self: &Arc<Self>) {
+        // Full reconnect succeeded!
+        let join_response = self.rtc_engine.join_response().unwrap();
+
+        self.update_connection_state(ConnectionState::Connected);
+        self.dispatcher.dispatch(&RoomEvent::Reconnected);
+
+        if let Some(pi) = join_response.participant {
+            self.local_participant.update_info(pi); // The sid may have changed
+        }
+
+        self.handle_participant_update(join_response.other_participants);
+
+        // TODO(theomonnom): unpublish & republish tracks
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    fn handle_disconnected(&self) {
+        if self.state.load(Ordering::Acquire) == ConnectionState::Disconnected as u8 {
+            return;
+        }
+
+        self.update_connection_state(ConnectionState::Disconnected);
+        self.dispatcher.dispatch(&RoomEvent::Disconnected);
+    }
+
+    /// Create a new participant
+    /// Also add it to the participants list
+    #[instrument(level = Level::DEBUG)]
+    fn create_participant(
+        self: &Arc<Self>,
+        sid: ParticipantSid,
+        identity: ParticipantIdentity,
+        name: String,
+        metadata: String,
+    ) -> RemoteParticipant {
+        let participant = RemoteParticipant::new(sid.clone(), identity, name, metadata);
+
+        // Create the participant task
+        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: RemoteParticipant) {
+        tokio::spawn(async move {
+            for (sid, _) in &*remote_participant.tracks() {
+                remote_participant.unpublish_track(&sid);
+            }
+
+            // 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());
+            self.dispatcher
+                .dispatch(&RoomEvent::ParticipantDisconnected(remote_participant));
+        });
+    }
+
+    fn get_participant(&self, sid: &ParticipantSid) -> Option<RemoteParticipant> {
+        self.participants.read().get(sid).cloned()
+    }
+}
+
+fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {
+    let split: Vec<&str> = stream_id.split('|').collect();
+    if split.len() == 2 {
+        let participant_sid = split.get(0).unwrap();
+        let track_sid = split.get(1).unwrap();
+        Some((participant_sid, track_sid))
+    } else {
+        None
+    }
+}
+
+impl From<u8> for ConnectionState {
+    fn from(value: u8) -> Self {
+        match value {
+            0 => ConnectionState::Disconnected,
+            1 => ConnectionState::Connected,
+            2 => ConnectionState::Reconnecting,
+            _ => ConnectionState::Unknown,
+        }
     }
 }

livekit/src/room/observer.rs

@@ -1 +0,0 @@
-

livekit/src/room/participant/local_participant.rs

@@ -9,16 +9,27 @@ use livekit_protocol as proto;
 use livekit_webrtc::rtp_parameters::RtpEncodingParameters;
 use parking_lot::RwLockReadGuard;
 use std::collections::HashMap;
+use std::fmt::Debug;
 use std::sync::Arc;
 use tokio::sync::mpsc;
-use tracing::{debug, trace};
+use tracing::debug;
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct LocalParticipant {
     inner: Arc<ParticipantInner>,
     rtc_engine: Arc<RtcEngine>,
 }
 
+impl Debug for LocalParticipant {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("LocalParticipant")
+            .field("sid", &self.sid())
+            .field("identity", &self.identity())
+            .field("name", &self.name())
+            .finish()
+    }
+}
+
 impl LocalParticipant {
     pub(crate) fn new(
         rtc_engine: Arc<RtcEngine>,
@@ -110,7 +121,7 @@ impl LocalParticipant {
     pub async fn unpublish_track(
         &self,
         track: TrackSid,
-        stop_on_unpublish: bool,
+        _stop_on_unpublish: bool,
     ) -> RoomResult<LocalTrackPublication> {
         let mut tracks = self.inner.tracks.write();
         if let Some(TrackPublication::Local(publication)) = tracks.remove(&track) {

livekit/src/room/participant/mod.rs

@@ -5,6 +5,7 @@ use livekit_protocol::enum_dispatch;
 use livekit_protocol::observer::Dispatcher;
 use parking_lot::{Mutex, RwLock, RwLockReadGuard};
 use std::collections::HashMap;
+use std::fmt::Debug;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering};
 use std::sync::Arc;
 use tokio::sync::mpsc;

livekit/src/room/participant/remote_participant.rs

@@ -6,6 +6,7 @@ use livekit_webrtc as rtc;
 use parking_lot::RwLockReadGuard;
 use rtc::prelude::MediaStreamTrack;
 use std::collections::{HashMap, HashSet};
+use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
@@ -14,11 +15,21 @@ use tracing::{debug, error, instrument, Level};
 
 const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5);
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct RemoteParticipant {
     inner: Arc<ParticipantInner>,
 }
 
+impl Debug for RemoteParticipant {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RemoteParticipant")
+            .field("sid", &self.sid())
+            .field("identity", &self.identity())
+            .field("name", &self.name())
+            .finish()
+    }
+}
+
 impl RemoteParticipant {
     pub(crate) fn new(
         sid: ParticipantSid,
@@ -99,7 +110,6 @@ impl RemoteParticipant {
                         unreachable!()
                     }
                 }
-                _ => unreachable!(),
             };
 
             debug!("starting track: {:?}", sid);

livekit/src/room/publication/local.rs

@@ -9,6 +9,8 @@ use std::sync::Arc;
 #[derive(Debug)]
 struct LocalTrackPublicationInner {
     publication_inner: TrackPublicationInner,
+
+    #[allow(unused)] // TODO(theomonnom)
     options: Mutex<TrackPublishOptions>,
 }
 
@@ -89,6 +91,7 @@ impl LocalTrackPublication {
         self.inner.publication_inner.update_track(track);
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub(crate) fn update_info(&self, info: proto::TrackInfo) {
         self.inner.publication_inner.update_info(info);

livekit/src/room/room_session.rs

@@ -1,563 +0,0 @@
-use crate::participant::ConnectionQuality;
-use crate::prelude::*;
-use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RtcEngine};
-use crate::signal_client::SignalOptions;
-use crate::{RoomError, RoomEvent, RoomResult, SimulateScenario};
-use livekit_protocol as proto;
-use livekit_protocol::observer::Dispatcher;
-use parking_lot::{Mutex, RwLock, RwLockReadGuard};
-use std::collections::HashMap;
-use std::sync::atomic::{AtomicU8, Ordering};
-use std::sync::Arc;
-use tokio::sync::{mpsc, oneshot};
-use tokio::task::JoinHandle;
-use tracing::{error, info, instrument, trace, Level};
-
-#[derive(Debug, Clone, Copy, Eq, PartialEq)]
-pub enum ConnectionState {
-    Disconnected,
-    Connected,
-    Reconnecting,
-    Unknown,
-}
-
-/// Internal representation of a RoomSession
-#[derive(Debug)]
-struct SessionInner {
-    state: AtomicU8, // ConnectionState
-    sid: Mutex<RoomSid>,
-    name: Mutex<String>,
-    metadata: Mutex<String>,
-    participants: RwLock<HashMap<ParticipantSid, RemoteParticipant>>,
-    participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
-    active_speakers: RwLock<Vec<Participant>>,
-    rtc_engine: Arc<RtcEngine>,
-    local_participant: LocalParticipant,
-    dispatcher: Dispatcher<RoomEvent>,
-}
-
-#[derive(Debug)]
-pub(crate) struct SessionHandle {
-    session: RoomSession,
-    session_task: JoinHandle<()>,
-    close_emitter: oneshot::Sender<()>,
-}
-
-/// RoomSession represents a connection to a room.
-/// It can be cloned and shared across threads.
-#[derive(Debug, Clone)]
-pub struct RoomSession {
-    inner: Arc<SessionInner>,
-}
-
-impl SessionHandle {
-    pub async fn connect(url: &str, token: &str) -> RoomResult<Self> {
-        let (rtc_engine, engine_events) = RtcEngine::new();
-        let rtc_engine = Arc::new(rtc_engine);
-        rtc_engine
-            .connect(url, token, SignalOptions::default())
-            .await?;
-
-        let join_response = rtc_engine.join_response().unwrap();
-        let pi = join_response.participant.unwrap().clone();
-        let local_participant = LocalParticipant::new(
-            rtc_engine.clone(),
-            pi.sid.into(),
-            pi.identity.into(),
-            pi.name,
-            pi.metadata,
-        );
-
-        let room_info = join_response.room.unwrap();
-        let inner = Arc::new(SessionInner {
-            state: AtomicU8::new(ConnectionState::Disconnected as u8),
-            sid: Mutex::new(room_info.sid.into()),
-            name: Mutex::new(room_info.name),
-            metadata: Mutex::new(room_info.metadata),
-            participants: Default::default(),
-            participants_tasks: Default::default(),
-            active_speakers: Default::default(),
-            rtc_engine,
-            local_participant,
-            dispatcher: Default::default(),
-        });
-
-        for pi in join_response.other_participants {
-            let participant = {
-                let pi = pi.clone();
-                inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
-            };
-            participant.update_info(pi.clone());
-        }
-
-        let (close_emitter, close_receiver) = oneshot::channel();
-        let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver));
-
-        inner.update_connection_state(ConnectionState::Connected);
-
-        let session = Self {
-            session: RoomSession { inner },
-            session_task,
-            close_emitter,
-        };
-        Ok(session)
-    }
-
-    pub async fn close(self) {
-        self.session.inner.close().await;
-        let _ = self.close_emitter.send(());
-        let _ = self.session_task.await;
-    }
-
-    pub fn subscribe(&self) -> mpsc::UnboundedReceiver<RoomEvent> {
-        self.session.inner.dispatcher.register()
-    }
-
-    pub fn session(&self) -> RoomSession {
-        self.session.clone()
-    }
-}
-
-impl RoomSession {
-    pub fn sid(&self) -> RoomSid {
-        self.inner.sid.lock().clone()
-    }
-
-    pub fn name(&self) -> String {
-        self.inner.name.lock().clone()
-    }
-
-    pub fn metadata(&self) -> String {
-        self.inner.metadata.lock().clone()
-    }
-
-    pub fn local_participant(&self) -> LocalParticipant {
-        self.inner.local_participant.clone()
-    }
-
-    pub fn connection_state(&self) -> ConnectionState {
-        self.inner.state.load(Ordering::Acquire).try_into().unwrap()
-    }
-
-    pub fn participants(&self) -> RwLockReadGuard<HashMap<ParticipantSid, RemoteParticipant>> {
-        self.inner.participants.read()
-    }
-
-    pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
-        self.inner.rtc_engine.simulate_scenario(scenario).await
-    }
-}
-
-impl SessionInner {
-    #[instrument(level = Level::DEBUG)]
-    async fn room_task(
-        self: Arc<Self>,
-        mut engine_events: EngineEvents,
-        mut close_receiver: oneshot::Receiver<()>,
-    ) {
-        loop {
-            tokio::select! {
-                res = engine_events.recv() => {
-                    if let Some(event) = res {
-                        if let Err(err) = self.on_engine_event(event).await {
-                            error!("failed to handle engine event: {:?}", err);
-                        }
-                    }
-                },
-                 _ = &mut close_receiver => {
-                    trace!("closing room_task");
-                    break;
-                }
-            }
-        }
-    }
-
-    /// Listen to the Participant events and forward them to the Room Dispatcher
-    #[instrument(level = Level::DEBUG)]
-    async fn participant_task(
-        self: Arc<Self>,
-        participant: Participant,
-        mut participant_events: mpsc::UnboundedReceiver<ParticipantEvent>,
-        mut close_rx: oneshot::Receiver<()>,
-    ) {
-        loop {
-            tokio::select! {
-                res = participant_events.recv() => {
-                    if let Some(event) = res {
-                        if let Err(err) = self.on_participant_event(&participant, event).await {
-                            error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
-                        }
-                    }
-                },
-                _ = &mut close_rx => {
-                    trace!("closing participant_task for {:?}", participant.sid());
-                    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 } => {
-                    self.dispatcher.dispatch(&RoomEvent::TrackPublished {
-                        participant: remote_participant.clone(),
-                        publication,
-                    });
-                }
-                ParticipantEvent::TrackUnpublished { publication } => {
-                    self.dispatcher.dispatch(&RoomEvent::TrackUnpublished {
-                        participant: remote_participant.clone(),
-                        publication,
-                    });
-                }
-                ParticipantEvent::TrackSubscribed { track, publication } => {
-                    self.dispatcher.dispatch(&RoomEvent::TrackSubscribed {
-                        participant: remote_participant.clone(),
-                        track,
-                        publication,
-                    });
-                }
-                ParticipantEvent::TrackUnsubscribed { track, publication } => {
-                    self.dispatcher.dispatch(&RoomEvent::TrackUnsubscribed {
-                        participant: remote_participant.clone(),
-                        track,
-                        publication,
-                    });
-                }
-                _ => {}
-            };
-        }
-
-        Ok(())
-    }
-
-    #[instrument(level = Level::DEBUG)]
-    async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
-        match event {
-            EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates),
-            EngineEvent::MediaTrack {
-                track,
-                stream,
-                receiver: _,
-            } => {
-                let stream_id = stream.id();
-                let lk_stream_id = unpack_stream_id(&stream_id);
-                if lk_stream_id.is_none() {
-                    Err(RoomError::Internal(format!(
-                        "MediaTrack event with invalid track_id: {:?}",
-                        &stream_id
-                    )))?;
-                }
-
-                let (participant_sid, track_sid) = lk_stream_id.unwrap();
-                let track_sid = track_sid.to_owned().into();
-                let remote_participant = self.get_participant(&participant_sid.to_string().into());
-
-                if let Some(remote_participant) = remote_participant {
-                    tokio::spawn(async move {
-                        remote_participant
-                            .add_subscribed_media_track(track_sid, track)
-                            .await;
-                    });
-                } else {
-                    // The server should send participant updates before sending a new offer
-                    // So this should never happen.
-                    Err(RoomError::Internal(format!(
-                        "AddTrack event with invalid participant_sid: {:?}",
-                        participant_sid
-                    )))?;
-                }
-            }
-            EngineEvent::Resuming => {
-                if self.update_connection_state(ConnectionState::Reconnecting) {
-                    self.dispatcher.dispatch(&RoomEvent::Reconnecting);
-                }
-            }
-            EngineEvent::Resumed => {
-                self.update_connection_state(ConnectionState::Connected);
-                self.dispatcher.dispatch(&RoomEvent::Reconnected);
-
-                // TODO(theomonnom): Update subscriptions settings
-                // TODO(theomonnom): Send sync state
-            }
-            EngineEvent::Restarting => self.handle_restarting(),
-            EngineEvent::Restarted => self.handle_restarted(),
-            EngineEvent::Disconnected => self.handle_disconnected(),
-            EngineEvent::Data {
-                payload,
-                kind,
-                participant_sid,
-            } => {
-                let payload = Arc::new(payload);
-                if let Some(participant) = self.get_participant(&participant_sid.into()) {
-                    self.dispatcher.dispatch(&RoomEvent::DataReceived {
-                        payload: payload.clone(),
-                        kind,
-                        participant: participant.clone(),
-                    });
-
-                    participant.on_data_received(payload, kind);
-                }
-            }
-            EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers),
-            EngineEvent::ConnectionQuality { updates } => {
-                self.handle_connection_quality_update(updates)
-            }
-        }
-
-        Ok(())
-    }
-
-    #[instrument(level = Level::DEBUG)]
-    async fn close(&self) {
-        self.rtc_engine.close().await;
-    }
-
-    /// Change the connection state and emit an event
-    /// Does nothing if the state is already the same
-    #[instrument(level = Level::DEBUG)]
-    fn update_connection_state(&self, state: ConnectionState) -> bool {
-        let old_state = self.state.load(Ordering::Acquire);
-        if old_state == state as u8 {
-            return false;
-        }
-
-        self.state.store(state as u8, Ordering::Release);
-        self.dispatcher
-            .dispatch(&RoomEvent::ConnectionStateChanged(state));
-        return true;
-    }
-
-    /// Update the participants inside a Room.
-    /// It'll create, update or remove a participant
-    /// It also update the participant tracks.
-    #[instrument(level = Level::DEBUG)]
-    fn handle_participant_update(self: &Arc<Self>, updates: Vec<proto::ParticipantInfo>) {
-        for pi in updates {
-            if pi.sid == self.local_participant.sid()
-                || pi.identity == self.local_participant.identity()
-            {
-                self.local_participant.clone().update_info(pi);
-                continue;
-            }
-
-            let remote_participant = self.get_participant(&pi.sid.clone().into());
-
-            if let Some(remote_participant) = remote_participant {
-                if pi.state == proto::participant_info::State::Disconnected as i32 {
-                    // Participant disconnected
-                    info!("Participant disconnected: {}", pi.sid);
-                    self.clone()
-                        .handle_participant_disconnect(remote_participant)
-                } else {
-                    // Participant is already connected, update the it
-                    remote_participant.update_info(pi.clone());
-                }
-            } else {
-                // Create a new participant
-                info!("Participant connected: {}", pi.sid);
-                let remote_participant = {
-                    let pi = pi.clone();
-                    self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
-                };
-
-                let _ = self
-                    .dispatcher
-                    .dispatch(&RoomEvent::ParticipantConnected(remote_participant.clone()));
-
-                remote_participant.update_info(pi.clone());
-            }
-        }
-    }
-
-    /// 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<proto::SpeakerInfo>) {
-        let mut speakers = Vec::new();
-
-        for speaker in speakers_info {
-            let participant = {
-                if speaker.sid == self.local_participant.sid() {
-                    Participant::Local(self.local_participant.clone())
-                } else {
-                    if let Some(participant) = self.get_participant(&speaker.sid.into()) {
-                        Participant::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
-            .dispatcher
-            .dispatch(&RoomEvent::ActiveSpeakersChanged { speakers });
-    }
-
-    /// 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() {
-                    Participant::Local(self.local_participant.clone())
-                } else {
-                    if let Some(participant) = self.get_participant(&update.participant_sid.into())
-                    {
-                        Participant::Remote(participant)
-                    } else {
-                        continue;
-                    }
-                }
-            };
-
-            let quality: ConnectionQuality = proto::ConnectionQuality::from_i32(update.quality)
-                .unwrap()
-                .into();
-
-            participant.set_connection_quality(quality);
-            self.dispatcher
-                .dispatch(&RoomEvent::ConnectionQualityChanged {
-                    participant,
-                    quality,
-                });
-        }
-    }
-
-    #[instrument(level = Level::DEBUG)]
-    fn handle_restarting(self: &Arc<Self>) {
-        // Remove existing participants/subscriptions on full reconnect
-        for (_, participant) in self.participants.read().iter() {
-            self.clone()
-                .handle_participant_disconnect(participant.clone());
-        }
-
-        if self.update_connection_state(ConnectionState::Reconnecting) {
-            self.dispatcher.dispatch(&RoomEvent::Reconnecting);
-        }
-    }
-
-    #[instrument(level = Level::DEBUG)]
-    fn handle_restarted(self: &Arc<Self>) {
-        // Full reconnect succeeded!
-        let join_response = self.rtc_engine.join_response().unwrap();
-
-        self.update_connection_state(ConnectionState::Connected);
-        self.dispatcher.dispatch(&RoomEvent::Reconnected);
-
-        if let Some(pi) = join_response.participant {
-            self.local_participant.update_info(pi); // The sid may have changed
-        }
-
-        self.handle_participant_update(join_response.other_participants);
-
-        // TODO(theomonnom): unpublish & republish tracks
-    }
-
-    #[instrument(level = Level::DEBUG)]
-    fn handle_disconnected(&self) {
-        if self.state.load(Ordering::Acquire) == ConnectionState::Disconnected as u8 {
-            return;
-        }
-
-        self.update_connection_state(ConnectionState::Disconnected);
-        self.dispatcher.dispatch(&RoomEvent::Disconnected);
-    }
-
-    /// Create a new participant
-    /// Also add it to the participants list
-    #[instrument(level = Level::DEBUG)]
-    fn create_participant(
-        self: &Arc<Self>,
-        sid: ParticipantSid,
-        identity: ParticipantIdentity,
-        name: String,
-        metadata: String,
-    ) -> RemoteParticipant {
-        let participant = RemoteParticipant::new(sid.clone(), identity, name, metadata);
-
-        // Create the participant task
-        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: RemoteParticipant) {
-        tokio::spawn(async move {
-            for (sid, _) in &*remote_participant.tracks() {
-                remote_participant.unpublish_track(&sid);
-            }
-
-            // 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());
-            self.dispatcher
-                .dispatch(&RoomEvent::ParticipantDisconnected(remote_participant));
-        });
-    }
-
-    fn get_participant(&self, sid: &ParticipantSid) -> Option<RemoteParticipant> {
-        self.participants.read().get(sid).cloned()
-    }
-}
-
-fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {
-    let split: Vec<&str> = stream_id.split('|').collect();
-    if split.len() == 2 {
-        let participant_sid = split.get(0).unwrap();
-        let track_sid = split.get(1).unwrap();
-        Some((participant_sid, track_sid))
-    } else {
-        None
-    }
-}
-
-impl From<u8> for ConnectionState {
-    fn from(value: u8) -> Self {
-        match value {
-            0 => ConnectionState::Disconnected,
-            1 => ConnectionState::Connected,
-            2 => ConnectionState::Reconnecting,
-            _ => ConnectionState::Unknown,
-        }
-    }
-}

livekit/src/room/track/local_audio_track.rs

@@ -7,6 +7,7 @@ use livekit_protocol as proto;
 use livekit_webrtc as rtc;
 use parking_lot::Mutex;
 use rtc::audio_source::native::NativeAudioSource;
+use std::fmt::Debug;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 
@@ -16,11 +17,21 @@ pub struct LocalAudioTrackInner {
     capture_options: Mutex<AudioCaptureOptions>,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct LocalAudioTrack {
     inner: Arc<LocalAudioTrackInner>,
 }
 
+impl Debug for LocalAudioTrack {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("LocalAudioTrack")
+            .field("sid", &self.sid())
+            .field("name", &self.name())
+            .field("source", &self.source())
+            .finish()
+    }
+}
+
 impl LocalAudioTrack {
     pub(crate) fn new(
         name: String,

livekit/src/room/track/local_video_track.rs

@@ -6,6 +6,7 @@ use livekit_webrtc as rtc;
 use livekit_webrtc::peer_connection_factory::native::PeerConnectionFactoryExt;
 use parking_lot::Mutex;
 use rtc::video_source::native::NativeVideoSource;
+use std::fmt::Debug;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 
@@ -15,11 +16,21 @@ struct LocalVideoTrackInner {
     capture_options: Mutex<VideoCaptureOptions>,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct LocalVideoTrack {
     inner: Arc<LocalVideoTrackInner>,
 }
 
+impl Debug for LocalVideoTrack {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("LocalVideoTrack")
+            .field("sid", &self.sid())
+            .field("name", &self.name())
+            .field("source", &self.source())
+            .finish()
+    }
+}
+
 impl LocalVideoTrack {
     pub fn new(
         name: String,

livekit/src/room/track/mod.rs

@@ -9,10 +9,10 @@ use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use thiserror::Error;
 use tokio::sync::mpsc;
 
-pub mod local_audio_track;
-pub mod local_video_track;
-pub mod remote_audio_track;
-pub mod remote_video_track;
+mod local_audio_track;
+mod local_video_track;
+mod remote_audio_track;
+mod remote_video_track;
 
 pub use local_audio_track::*;
 pub use local_video_track::*;

livekit/src/room/track/remote_audio_track.rs

@@ -2,14 +2,25 @@ use super::TrackInner;
 use crate::prelude::*;
 use livekit_protocol as proto;
 use livekit_webrtc as rtc;
+use std::fmt::Debug;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct RemoteAudioTrack {
     pub(crate) inner: Arc<TrackInner>,
 }
 
+impl Debug for RemoteAudioTrack {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RemoteAudioTrack")
+            .field("sid", &self.sid())
+            .field("name", &self.name())
+            .field("source", &self.source())
+            .finish()
+    }
+}
+
 impl RemoteAudioTrack {
     pub(crate) fn new(
         sid: TrackSid,
@@ -90,12 +101,14 @@ impl RemoteAudioTrack {
         true
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub(crate) fn transceiver(&self) -> Option<rtc::rtp_transceiver::RtpTransceiver> {
         self.inner.transceiver()
     }
 
     #[inline]
+    #[allow(dead_code)]
     pub(crate) fn update_transceiver(
         &self,
         transceiver: Option<rtc::rtp_transceiver::RtpTransceiver>,

livekit/src/room/track/remote_video_track.rs

@@ -2,14 +2,25 @@ use super::TrackInner;
 use crate::prelude::*;
 use livekit_protocol as proto;
 use livekit_webrtc as rtc;
+use std::fmt::Debug;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct RemoteVideoTrack {
     pub(crate) inner: Arc<TrackInner>,
 }
 
+impl Debug for RemoteVideoTrack {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RemoteVideoTrack")
+            .field("sid", &self.sid())
+            .field("name", &self.name())
+            .field("source", &self.source())
+            .finish()
+    }
+}
+
 impl RemoteVideoTrack {
     pub(crate) fn new(
         sid: TrackSid,
@@ -90,11 +101,13 @@ impl RemoteVideoTrack {
         true
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub(crate) fn transceiver(&self) -> Option<rtc::rtp_transceiver::RtpTransceiver> {
         self.inner.transceiver()
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub(crate) fn update_transceiver(
         &self,

livekit/src/rtc_engine/mod.rs

@@ -7,6 +7,7 @@ use livekit_protocol as proto;
 use livekit_webrtc::prelude::*;
 use livekit_webrtc::session_description::SdpParseError;
 use parking_lot::Mutex;
+use std::fmt::Debug;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -99,7 +100,6 @@ struct EngineHandle {
     close_sender: oneshot::Sender<()>,
 }
 
-#[derive(Debug)]
 struct EngineInner {
     lk_runtime: Arc<LkRuntime>,
     session_info: Mutex<Option<SessionInfo>>, // Last/Current Sessioninfo
@@ -113,6 +113,17 @@ struct EngineInner {
     reconnect_interval: Mutex<Interval>,
 }
 
+impl Debug for EngineInner {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("EngineInner")
+            .field("session_info", &self.session_info)
+            .field("opened", &self.opened)
+            .field("reconnecting", &self.reconnecting)
+            .field("full_reconnect", &self.full_reconnect)
+            .finish()
+    }
+}
+
 #[derive(Debug)]
 pub struct RtcEngine {
     inner: Arc<EngineInner>,

livekit/src/rtc_engine/peer_transport.rs

@@ -4,7 +4,7 @@ use std::fmt::{Debug, Formatter};
 use std::time::Duration;
 use tracing::{event, Level};
 
-const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150);
+const _NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150);
 
 pub type OnOfferCreated = Box<dyn FnMut(SessionDescription) + Send + Sync>;
 

livekit/src/rtc_engine/rtc_events.rs

@@ -3,7 +3,7 @@ use crate::rtc_engine::peer_transport::OnOfferCreated;
 use livekit_protocol as proto;
 use livekit_webrtc::{self as rtc, prelude::*};
 use tokio::sync::mpsc;
-use tracing::{debug, error};
+use tracing::{error};
 
 pub type RtcEmitter = mpsc::UnboundedSender<RtcEvent>;
 pub type RtcEvents = mpsc::UnboundedReceiver<RtcEvent>;

livekit/src/rtc_engine/rtc_session.rs

@@ -14,6 +14,7 @@ use prost::Message;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::convert::TryInto;
+use std::fmt::Debug;
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -104,7 +105,6 @@ pub struct SessionInfo {
 }
 
 /// Fields shared with rtc_task and signal_task
-#[derive(Debug)]
 struct SessionInner {
     info: SessionInfo,
     signal_client: Arc<SignalClient>,
@@ -128,6 +128,18 @@ struct SessionInner {
     closed: AtomicBool,
     emitter: SessionEmitter,
 }
+
+impl Debug for SessionInner {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SessionInner")
+            .field("info", &self.info)
+            .field("pc_state", &self.pc_state)
+            .field("has_published", &self.has_published)
+            .field("closed", &self.closed)
+            .finish()
+    }
+}
+
 /// This struct holds a WebRTC session
 /// The session changes at every reconnection
 ///
@@ -317,6 +329,7 @@ impl RtcSession {
         &self.inner.info
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub fn state(&self) -> PeerState {
         self.inner
@@ -326,21 +339,25 @@ impl RtcSession {
             .unwrap()
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub fn publisher(&self) -> &AsyncMutex<PeerTransport> {
         &self.inner.publisher_pc
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub fn subscriber(&self) -> &AsyncMutex<PeerTransport> {
         &self.inner.subscriber_pc
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub fn signal_client(&self) -> &Arc<SignalClient> {
         &self.inner.signal_client
     }
 
+    #[allow(dead_code)]
     #[inline]
     pub fn data_channel(&self, kind: proto::data_packet::Kind) -> &DataChannel {
         &self.inner.data_channel(kind)

livekit/src/signal_client/mod.rs

@@ -1,6 +1,6 @@
 use crate::signal_client::signal_stream::SignalStream;
 use livekit_protocol as proto;
-use livekit_webrtc::prelude::*;
+
 use parking_lot::RwLock;
 use std::fmt::Debug;
 use std::time::Duration;
@@ -104,9 +104,10 @@ impl SignalClient {
         // TODO(theomonnom): enqueue message
     }
 
+    /*#[allow(dead_code)]
     pub async fn clear_queue(&self) {
         // TODO(theomonnom): impl
-    }
+    }*/
 
     #[instrument(level = Level::DEBUG)]
     pub async fn flush_queue(&self) {