wip

I still need to solve type checking cycle to be able to compile

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

@@ -35,7 +35,7 @@ pub enum RoomError {
 
 pub type RoomResult<T> = Result<T, RoomError>;
 
-#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+#[derive(Debug, Clone, Eq, PartialEq)]
 pub enum ConnectionState {
     Disconnected,
     Connecting,
@@ -69,10 +69,14 @@ impl Room {
 
     #[instrument(level = Level::DEBUG)]
     pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> {
-        let (rtc_engine, engine_events) =
+        // Initialize the RTCEngine
-            RTCEngine::connect(url, token, SignalOptions::default()).await?;
+        let (rtc_engine, engine_events) = RTCEngine::new();
         let rtc_engine = Arc::new(rtc_engine);
-        let join_response = rtc_engine.join_response();
+        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 = Arc::new(LocalParticipant::new(
             rtc_engine.clone(),
@@ -150,26 +154,21 @@ impl Room {
                 Self::handle_participant_update(room_inner.clone(), room_events.clone(), update)
                     .await
             }
-            EngineEvent::AddTrack {
+            EngineEvent::MediaTrack {
-                rtp_receiver,
+                track,
-                streams,
+                stream,
+                receiver,
             } => {
-                if streams.is_empty() {
+                let stream_id = stream.id();
-                    Err(RoomError::Internal(
+                let lk_stream_id = unpack_stream_id(&stream_id);
-                        "AddTrack event with empty streams".to_string(),
+                if lk_stream_id.is_none() {
-                    ))?;
-                }
-
-                let first_stream_id = streams.first().unwrap().id();
-                let stream_id = unpack_stream_id(&first_stream_id);
-                if stream_id.is_none() {
                     Err(RoomError::Internal(format!(
-                        "AddTrack event with invalid track_id: {:?}",
+                        "MediaTrack event with invalid track_id: {:?}",
-                        first_stream_id
+                        &stream_id
                     )))?;
                 }
 
-                let (participant_sid, track_sid) = stream_id.unwrap();
+                let (participant_sid, track_sid) = lk_stream_id.unwrap();
                 let remote_participant =
                     Self::get_participant(room_inner.clone(), &participant_sid.to_string().into());
 
@@ -181,7 +180,7 @@ impl Room {
                                 .add_subscribed_media_track(
                                     RoomHandle::from(room_inner),
                                     track_sid,
-                                    rtp_receiver.track(),
+                                    track,
                                 )
                                 .await;
                         }
@@ -195,6 +194,11 @@ impl Room {
                     )))?;
                 }
             }
+            EngineEvent::Resuming => {}
+            EngineEvent::Resumed => {}
+            EngineEvent::Restarting => {}
+            EngineEvent::Restarted => {}
+            EngineEvent::Disconnected => {}
         }
 
         Ok(())

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

@@ -1,13 +1,15 @@
+use futures::FutureExt;
 use livekit_webrtc::data_channel::DataSendError;
 use livekit_webrtc::jsep::SdpParseError;
-use livekit_webrtc::media_stream::MediaStream;
+use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle};
 use livekit_webrtc::rtc_error::RTCError;
 use livekit_webrtc::rtp_receiver::RtpReceiver;
-use parking_lot::{Mutex, RwLock};
+use parking_lot::Mutex;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Weak};
 use std::time::Duration;
 use thiserror::Error;
+use tokio::sync::RwLock as AsyncRwLock;
 use tokio::task::JoinHandle;
 
 use lazy_static::lazy_static;
@@ -18,7 +20,7 @@ use crate::proto::{data_packet, DataPacket, JoinResponse, ParticipantUpdate};
 use crate::rtc_engine::lk_runtime::LKRuntime;
 use crate::signal_client::{SignalError, SignalOptions};
 
-use self::rtc_session::{RTCSession, SessionEvent, SessionEvents};
+use self::rtc_session::{RTCSession, SessionEvent, SessionEvents, SessionInfo};
 
 mod lk_runtime;
 mod pc_transport;
@@ -52,25 +54,39 @@ pub enum EngineError {
 #[derive(Debug)]
 pub enum EngineEvent {
     ParticipantUpdate(ParticipantUpdate),
-    AddTrack {
+    MediaTrack {
-        rtp_receiver: RtpReceiver,
+        track: MediaStreamTrackHandle,
-        streams: Vec<MediaStream>,
+        stream: MediaStream,
+        receiver: RtpReceiver,
     },
     Resuming,
     Resumed,
     Restarting,
     Restarted,
+    Disconnected,
 }
 
 // TODO(theomonnom): Smarter retry intervals
-pub(crate) const RECONNECT_ATTEMPTS: u32 = 10;
+pub const RECONNECT_ATTEMPTS: u32 = 10;
-pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300);
+pub const RECONNECT_INTERVAL: Duration = Duration::from_millis(300);
 
 lazy_static! {
     // Share one LKRuntime across all RTCEngine instances
     static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
 }
 
+#[derive(Debug, Clone, Eq, PartialEq)]
+#[repr(u8)]
+pub enum SimulateScenario {
+    SignalReconnect,
+    Speaker,
+    NodeFailure,
+    ServerLeave,
+    Migration,
+    ForceTcp,
+    ForceTls,
+}
+
 /// Represents a running RTCSession with the ability to close the session
 /// and the engine_task
 #[derive(Debug)]
@@ -82,7 +98,9 @@ struct EngineHandle {
 
 #[derive(Debug)]
 struct EngineInner {
-    running_handle: RwLock<Option<EngineHandle>>,
+    lk_runtime: Arc<LKRuntime>,
+    session_info: Mutex<Option<SessionInfo>>, // Last/Current Sessioninfo
+    running_handle: AsyncRwLock<Option<EngineHandle>>,
     reconnecting: AtomicBool,
     opened: AtomicBool,
     engine_emitter: EngineEmitter,
@@ -90,7 +108,6 @@ struct EngineInner {
 
 #[derive(Debug)]
 pub struct RTCEngine {
-    lk_runtime: Arc<LKRuntime>,
     inner: Arc<EngineInner>,
 }
 
@@ -110,19 +127,15 @@ impl RTCEngine {
 
         let (engine_emitter, engine_events) = mpsc::channel(8);
         let inner = Arc::new(EngineInner {
+            lk_runtime: lk_runtime.unwrap(),
+            session_info: Default::default(),
             running_handle: Default::default(),
             reconnecting: Default::default(),
             opened: Default::default(),
             engine_emitter,
         });
 
-        (
+        (Self { inner }, engine_events)
-            Self {
-                lk_runtime: lk_runtime.unwrap(),
-                inner,
-            },
-            engine_events,
-        )
     }
 
     #[tracing::instrument]
@@ -132,37 +145,18 @@ impl RTCEngine {
         token: &str,
         options: SignalOptions,
     ) -> EngineResult<()> {
-        let (session_emitter, session_events) = mpsc::unbounded_channel();
+        self.inner.connect(url, token, options).await
-        let session = RTCSession::connect(
-            url,
-            token,
-            options,
-            self.lk_runtime.clone(),
-            session_emitter,
-        )
-        .await?;
-
-        let (close_sender, close_receiver) = oneshot::channel();
-        let engine_task = tokio::spawn(
-            self.inner
-                .clone()
-                .engine_task(session_events, close_receiver),
-        );
-
-        self.inner.opened.store(true, Ordering::SeqCst);
-        *self.inner.running_handle.write() = Some(EngineHandle {
-            session,
-            engine_task,
-            close_sender,
-        });
-
-        Ok(())
     }
 
     #[tracing::instrument]
     pub async fn close(&self) {
         self.inner.opened.store(false, Ordering::SeqCst);
         self.inner.close();
+        let _ = self
+            .inner
+            .engine_emitter
+            .send(EngineEvent::Disconnected)
+            .await;
     }
 
     #[tracing::instrument(skip(data))]
@@ -175,6 +169,7 @@ impl RTCEngine {
         self.inner
             .running_handle
             .read()
+            .await
             .as_ref()
             .unwrap()
             .session
@@ -185,8 +180,8 @@ impl RTCEngine {
     }
 
     pub fn join_response(&self) -> Option<JoinResponse> {
-        if let Some(handle) = self.inner.running_handle.read().as_ref() {
+        if let Some(info) = self.inner.session_info.lock().as_ref() {
-            Some(handle.session.info().join_response.clone())
+            Some(info.join_response.clone())
         } else {
             None
         }
@@ -217,15 +212,68 @@ impl EngineInner {
         }
     }
 
-    async fn on_session_event(&self, event: SessionEvent) -> EngineResult<()> {
+    async fn on_session_event(self: &Arc<Self>, event: SessionEvent) -> EngineResult<()> {
+        match event {
+            SessionEvent::Close { reason } => {
+                info!("received session close: {}", reason);
+                self.handle_disconnected().await;
+            }
+            SessionEvent::Data { data } => {}
+            SessionEvent::MediaTrack {
+                track,
+                stream,
+                receiver,
+            } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::MediaTrack {
+                        track,
+                        stream,
+                        receiver,
+                    })
+                    .await;
+            }
+            SessionEvent::Connected => {}
+        }
+        Ok(())
+    }
+
+    async fn connect(
+        self: &Arc<Self>,
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+    ) -> EngineResult<()> {
+        let (session_emitter, session_events) = mpsc::unbounded_channel();
+        let session = RTCSession::connect(
+            url,
+            token,
+            options,
+            self.lk_runtime.clone(),
+            session_emitter,
+        )
+        .await?;
+
+        let (close_sender, close_receiver) = oneshot::channel();
+        let engine_task = tokio::spawn(self.clone().engine_task(session_events, close_receiver));
+
+        *self.session_info.lock() = Some(session.info().clone());
+        *self.running_handle.write().await = Some(EngineHandle {
+            session,
+            engine_task,
+            close_sender,
+        });
+
+        self.opened.store(true, Ordering::SeqCst);
+
         Ok(())
     }
 
     async fn close(&self) {
-        if let Some(handle) = self.running_handle.write().take() {
+        if let Some(handle) = self.running_handle.write().await.take() {
             handle.session.close().await;
             let _ = handle.close_sender.send(());
-            handle.engine_task.await;
+            let _ = handle.engine_task.await;
         }
     }
 
@@ -238,7 +286,7 @@ impl EngineInner {
             tokio::task::yield_now().await;
         }
 
-        if self.running_handle.read().is_none() {
+        if self.running_handle.read().await.is_none() {
             Err(EngineError::Connection("reconnection failed".to_owned()))?
         }
 
@@ -247,7 +295,7 @@ impl EngineInner {
 
     /// Called every time the PeerConnection or the SignalClient is closed
     /// We first try to resume the connection, if it fails, we start a full reconnect.
-    async fn handle_disconnected(&self) {
+    async fn handle_disconnected(self: &Arc<Self>) {
         if !self.opened.load(Ordering::SeqCst) || self.reconnecting.load(Ordering::SeqCst) {
             return;
         }
@@ -255,6 +303,7 @@ impl EngineInner {
         self.reconnecting.store(true, Ordering::SeqCst);
         warn!("RTCEngine disconnected unexpectedly, reconnecting...");
 
+        let mut connected = false;
         let mut full_reconnect = false;
         for i in 0..RECONNECT_ATTEMPTS {
             if full_reconnect {
@@ -267,7 +316,8 @@ impl EngineInner {
                     error!("restarting connection failed: {}", err);
                 } else {
                     let _ = self.engine_emitter.send(EngineEvent::Restarted).await;
-                    return;
+                    connected = true;
+                    break;
                 }
             } else {
                 if i == 0 {
@@ -282,27 +332,45 @@ impl EngineInner {
                     }
                 } else {
                     let _ = self.engine_emitter.send(EngineEvent::Resumed).await;
-                    return;
+                    connected = true;
+                    break;
                 }
             }
 
             tokio::time::sleep(RECONNECT_INTERVAL).await;
         }
-        error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS);
+
         self.reconnecting.store(false, Ordering::SeqCst);
 
-        // TODO DISCONNECT
+        if !connected {
+            error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS);
+            let _ = self.engine_emitter.send(EngineEvent::Disconnected).await;
+            self.close().await;
+        }
     }
 
     /// Try to recover the connection by doing a full reconnect.
     /// It recreates a new RTCSession
-    async fn try_restart_connection(&self) -> EngineResult<()> {
+    async fn try_restart_connection(self: &Arc<Self>) -> EngineResult<()> {
+        let info = self.session_info.lock().clone().unwrap();
         self.close().await;
-        Ok(())
+        self.connect(&info.url, &info.token, info.options).await?;
+        self.running_handle
+            .read()
+            .await
+            .as_ref()
+            .unwrap()
+            .session
+            .wait_pc_connection()
+            .await
+
+        // TODO(theomonnom): Resend SignalClient queue
     }
 
     /// Try to restart the current session
     async fn try_resume_connection(&self) -> EngineResult<()> {
-        Ok(())
+        let handle = self.running_handle.read().await;
+        handle.as_ref().unwrap().session.restart().await?;
+        handle.as_ref().unwrap().session.wait_pc_connection().await
     }
 }

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

@@ -21,7 +21,7 @@ pub type OnOfferHandler = Box<
         + Sync,
 >;
 
-pub(crate) struct PCTransport {
+pub struct PCTransport {
     signal_target: SignalTarget,
     peer_connection: PeerConnection,
     pending_candidates: Vec<IceCandidate>,

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

@@ -28,14 +28,17 @@ pub enum RTCEvent {
     },
     DataChannel {
         data_channel: DataChannel,
+        target: SignalTarget,
     },
     // TODO (theomonnom): Move Offer to PCTransport
     Offer {
         offer: SessionDescription,
+        target: SignalTarget,
     },
     AddTrack {
         rtp_receiver: RtpReceiver,
         streams: Vec<MediaStream>,
+        target: SignalTarget,
     },
     Data {
         data: Vec<u8>,
@@ -91,28 +94,22 @@ fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler
 }
 
 pub fn forward_pc_events(transport: &mut PCTransport, rtc_emitter: RTCEmitter) {
+    let signal_target = transport.signal_target();
     transport
         .peer_connection()
-        .on_ice_candidate(on_ice_candidate(
+        .on_ice_candidate(on_ice_candidate(signal_target, rtc_emitter.clone()));
-            transport.signal_target(),
-            rtc_emitter.clone(),
-        ));
-
-    transport.peer_connection().on_data_channel(on_data_channel(
-        transport.signal_target(),
-        rtc_emitter.clone(),
-    ));
 
     transport
         .peer_connection()
-        .on_add_track(on_add_track(transport.signal_target(), rtc_emitter.clone()));
+        .on_data_channel(on_data_channel(signal_target, rtc_emitter.clone()));
 
     transport
         .peer_connection()
-        .on_connection_change(on_connection_change(
+        .on_add_track(on_add_track(signal_target, rtc_emitter.clone()));
-            transport.signal_target(),
+
-            rtc_emitter.clone(),
+    transport
-        ));
+        .peer_connection()
+        .on_connection_change(on_connection_change(signal_target, rtc_emitter.clone()));
 
     transport.on_offer(on_offer(transport.signal_target(), rtc_emitter.clone()));
 }

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

@@ -1,6 +1,7 @@
 use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle};
 use livekit_webrtc::rtp_receiver::RtpReceiver;
 use parking_lot::Mutex;
+use std::convert::TryInto;
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -65,6 +66,21 @@ pub enum PCState {
     Closed,
 }
 
+impl TryInto<PCState> for u8 {
+    type Error = &'static str;
+
+    fn try_into(self) -> Result<PCState, Self::Error> {
+        match self {
+            0 => Ok(PCState::New),
+            1 => Ok(PCState::Connected),
+            2 => Ok(PCState::Disconnected),
+            3 => Ok(PCState::Reconnecting),
+            4 => Ok(PCState::Closed),
+            _ => Err("invalid PCState"),
+        }
+    }
+}
+
 #[derive(Serialize, Deserialize)]
 #[allow(non_snake_case)]
 struct IceCandidateJSON {
@@ -236,7 +252,11 @@ impl RTCSession {
         self.inner.publish_data(data, kind).await
     }
 
-    pub async fn wait_pc_connectiom(&self) -> EngineResult<()> {
+    pub async fn restart(&self) -> EngineResult<()> {
+        self.inner.restart_session().await
+    }
+
+    pub async fn wait_pc_connection(&self) -> EngineResult<()> {
         self.inner.wait_pc_connection().await
     }
 }
@@ -246,6 +266,14 @@ impl RTCSession {
         &self.inner.info
     }
 
+    pub fn state(&self) -> PCState {
+        self.inner
+            .pc_state
+            .load(Ordering::SeqCst)
+            .try_into()
+            .unwrap()
+    }
+
     pub fn publisher(&self) -> &AsyncMutex<PCTransport> {
         &self.inner.publisher_pc
     }
@@ -412,10 +440,13 @@ impl SessionInner {
                     self.on_session_disconnected("pc_state failed");
                 }
             }
-            RTCEvent::DataChannel { data_channel } => {
+            RTCEvent::DataChannel {
+                data_channel,
+                target: _,
+            } => {
                 self.subscriber_dc.lock().push(data_channel);
             }
-            RTCEvent::Offer { offer } => {
+            RTCEvent::Offer { offer, target: _ } => {
                 // Send the publisher offer to the server
                 self.signal_client
                     .send(signal_request::Message::Offer(proto::SessionDescription {
@@ -427,6 +458,7 @@ impl SessionInner {
             RTCEvent::AddTrack {
                 rtp_receiver,
                 mut streams,
+                target: _,
             } => {
                 if !streams.is_empty() {
                     let _ = self.emitter.send(SessionEvent::MediaTrack {

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

@@ -36,14 +36,14 @@ pub enum SignalError {
 
 /// Events used by the RTCEngine who will handle the reconnection logic
 #[derive(Debug)]
-pub(crate) enum SignalEvent {
+pub enum SignalEvent {
     Open,
     Signal(signal_response::Message),
     Close,
 }
 
 #[derive(Debug, Clone)]
-pub(crate) struct SignalOptions {
+pub struct SignalOptions {
     pub(crate) reconnect: bool,
     pub(crate) sid: String,
     pub auto_subscribe: bool,

examples/.cargo/config

@@ -0,0 +1,2 @@
+[target.x86_64-pc-windows-msvc]
+rustflags = ["-C", "target-feature=+crt-static"]

examples/Cargo.toml

@@ -1,4 +1,4 @@
 [workspace]
 members = ["*"]
-exclude = ["target"]
+exclude = ["target", ".cargo"]
 resolver = "2"

examples/simple_room/src/main.rs

@@ -1,15 +1,10 @@
-use tracing_subscriber::prelude::*;
 mod app;
 mod events;
 mod video_grid;
 mod video_renderer;
 
 fn main() {
-    let fmt_layer = tracing_subscriber::fmt::Layer::default();
+    tracing_subscriber::fmt::init();
-
-    tracing_subscriber::registry()
-        .with(fmt_layer)
-        .init();
 
     let rt = tokio::runtime::Builder::new_multi_thread()
         .enable_all()