publish client-sdk-native (#12)

* Create README.md

* add example

* crates & examples

* fix syntax

* add LICENSE

* thirdparty LICENSE

* rearrange repo

* fix build

* egui versions

* prepare publish

* fix demo compilation

* add test ci

* Update rust.yml

* forgot runs-on

* install protoc before building

* avoid rate limit

* include submodules

* updates to readme

* cache rust builds

Co-authored-by: David Zhao <dz@livekit.io>
Co-authored-by: David Zhao <david@davidzhao.com>

livekit/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "livekit"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0"
+description = "Native SDK for LiveKit"
+
+[dependencies]
+livekit-webrtc = { path = "../livekit-webrtc", version = "0.1.0" }
+livekit-utils = { path = "../livekit-utils", version = "0.1.0" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1.0"
+tokio-tungstenite = { version = "0.17.2", features = ["native-tls"] }
+tokio = { version = "1", features = ["full"] }
+futures = "0.3"
+parking_lot = { version = "0.12.1", features = ["send_guard"] }
+url = "2.2.2"
+futures-util = "0.3.23"
+thiserror = "1.0"
+prost = "0.11.0"
+prost-types = "0.11.1"
+lazy_static = "1.4.0"
+tracing = "0.1"
+
+[build-dependencies]
+prost-build = { version = "0.11.1" }

livekit/build.rs

@@ -0,0 +1,12 @@
+use std::io::Result;
+
+fn main() -> Result<()> {
+    prost_build::compile_protos(
+        &[
+            "protocol/livekit_rtc.proto",
+            "protocol/livekit_models.proto",
+        ],
+        &["protocol/"],
+    )?;
+    Ok(())
+}

livekit/src/lib.rs

@@ -0,0 +1,13 @@
+extern crate core;
+
+pub mod proto {
+    include!(concat!(env!("OUT_DIR"), "/livekit.rs"));
+}
+
+mod rtc_engine;
+mod signal_client;
+
+pub mod room;
+pub mod webrtc {
+    pub use livekit_webrtc::*;
+}

livekit/src/room/id.rs

@@ -0,0 +1,56 @@
+use std::fmt;
+
+macro_rules! id_str {
+    ($($name:ident;)*) => {
+        $(
+            impl $name {
+                pub fn new(str: String) -> Self {
+                    Self(str)
+                }
+
+                pub fn as_str(&self) -> &str {
+                    &self.0
+                }
+            }
+
+            impl From<String> for $name {
+                fn from(str: String) -> $name {
+                    $name(str)
+                }
+            }
+
+            impl From<$name> for String {
+                fn from(id: $name) -> String {
+                    id.0
+                }
+            }
+
+            impl PartialEq<$name> for String {
+                fn eq(&self, u: &$name) -> bool {
+                    *self == *u.0
+                }
+            }
+
+            impl fmt::Display for $name {
+                fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                    f.write_str(&self.0)
+                }
+            }
+        )*
+    }
+}
+
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+pub struct ParticipantSid(String);
+
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+pub struct ParticipantIdentity(String);
+
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+pub struct TrackSid(String);
+
+id_str! {
+    ParticipantSid;
+    ParticipantIdentity;
+    TrackSid;
+}

livekit/src/room/mod.rs

@@ -0,0 +1,115 @@
+use self::participant::ConnectionQuality;
+use self::room_session::{ConnectionState, RoomSession, SessionHandle};
+use crate::proto::data_packet;
+use crate::room::id::TrackSid;
+use crate::room::participant::remote_participant::RemoteParticipant;
+use crate::room::participant::Participant;
+use crate::room::publication::RemoteTrackPublication;
+use crate::room::publication::TrackPublication;
+use crate::room::track::remote_track::RemoteTrackHandle;
+use crate::rtc_engine::EngineError;
+use std::fmt::Debug;
+use std::sync::Arc;
+use thiserror::Error;
+use tokio::sync::mpsc;
+
+pub use crate::rtc_engine::SimulateScenario;
+
+pub mod id;
+pub mod participant;
+pub mod publication;
+pub mod room_session;
+pub mod track;
+
+pub type RoomEvents = mpsc::UnboundedReceiver<RoomEvent>;
+pub type RoomEmitter = mpsc::UnboundedSender<RoomEvent>;
+pub type RoomResult<T> = Result<T, RoomError>;
+
+#[derive(Error, Debug)]
+pub enum RoomError {
+    #[error("engine : {0}")]
+    Engine(#[from] EngineError),
+    #[error("room failure: {0}")]
+    Internal(String),
+}
+
+#[derive(Error, Debug, Clone)]
+pub enum TrackError {
+    #[error("could not find published track with sid: {0}")]
+    TrackNotFound(String),
+}
+
+#[derive(Clone, Debug)]
+pub enum RoomEvent {
+    ParticipantConnected(Arc<RemoteParticipant>),
+    ParticipantDisconnected(Arc<RemoteParticipant>),
+    TrackSubscribed {
+        track: RemoteTrackHandle,
+        publication: RemoteTrackPublication,
+        participant: Arc<RemoteParticipant>,
+    },
+    TrackPublished {
+        publication: RemoteTrackPublication,
+        participant: Arc<RemoteParticipant>,
+    },
+    TrackUnpublished {
+        publication: RemoteTrackPublication,
+        participant: Arc<RemoteParticipant>,
+    },
+    TrackUnsubscribed {
+        track: RemoteTrackHandle,
+        publication: RemoteTrackPublication,
+        participant: Arc<RemoteParticipant>,
+    },
+    TrackSubscriptionFailed {
+        error: TrackError,
+        sid: TrackSid,
+        participant: Arc<RemoteParticipant>,
+    },
+    TrackMuted {
+        publication: TrackPublication,
+        participant: Participant,
+    },
+    TrackUnmuted {
+        publication: TrackPublication,
+        participant: Participant,
+    },
+    ActiveSpeakersChanged {
+        speakers: Vec<Participant>,
+    },
+    ConnectionQualityChanged {
+        quality: ConnectionQuality,
+        participant: Participant,
+    },
+    DataReceived {
+        payload: Arc<Vec<u8>>,
+        kind: data_packet::Kind,
+        participant: Arc<RemoteParticipant>,
+    },
+    ConnectionStateChanged(ConnectionState),
+    Connected,
+    Disconnected,
+    Reconnecting,
+    Reconnected,
+}
+
+#[derive(Debug)]
+pub struct Room {
+    handle: SessionHandle,
+}
+
+impl Room {
+    pub async fn connect(url: &str, token: &str) -> RoomResult<(Self, RoomEvents)> {
+        let (emitter, events) = mpsc::unbounded_channel();
+        let handle = SessionHandle::connect(emitter, url, token).await?;
+        Ok((Self { handle }, events))
+    }
+
+    pub async fn close(self) {
+        self.handle.close().await;
+    }
+
+    pub fn session(&self) -> RoomSession {
+        self.handle.session()
+    }
+}

livekit/src/room/participant/local_participant.rs

@@ -0,0 +1,76 @@
+use super::ConnectionQuality;
+use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket};
+use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid};
+use crate::room::participant::{
+    impl_participant_trait, ParticipantEvent, ParticipantInternalTrait, ParticipantShared,
+    ParticipantTrait,
+};
+use crate::room::publication::TrackPublication;
+use crate::room::RoomError;
+use crate::rtc_engine::RTCEngine;
+use parking_lot::RwLockReadGuard;
+use std::collections::HashMap;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use tokio::sync::mpsc;
+
+#[derive(Debug)]
+pub struct LocalParticipant {
+    shared: ParticipantShared,
+    rtc_engine: Arc<RTCEngine>,
+}
+
+impl LocalParticipant {
+    pub(crate) fn new(
+        rtc_engine: Arc<RTCEngine>,
+        sid: ParticipantSid,
+        identity: ParticipantIdentity,
+        name: String,
+        metadata: String,
+    ) -> Self {
+        Self {
+            shared: ParticipantShared::new(sid, identity, name, metadata),
+            rtc_engine,
+        }
+    }
+
+    pub async fn publish_data(
+        &self,
+        data: &[u8],
+        kind: data_packet::Kind,
+    ) -> Result<(), RoomError> {
+        let data = DataPacket {
+            kind: kind as i32,
+            value: Some(data_packet::Value::User(UserPacket {
+                participant_sid: self.sid().to_string(),
+                payload: data.to_vec(),
+                destination_sids: vec![],
+            })),
+        };
+
+        self.rtc_engine
+            .publish_data(&data, kind)
+            .await
+            .map_err(Into::into)
+    }
+}
+
+impl ParticipantInternalTrait for LocalParticipant {
+    fn update_info(self: &Arc<Self>, info: ParticipantInfo, _emit_events: bool) {
+        self.shared.update_info(info);
+    }
+
+    fn set_speaking(&self, speaking: bool) {
+        self.shared.set_speaking(speaking);
+    }
+
+    fn set_audio_level(&self, level: f32) {
+        self.shared.set_audio_level(level);
+    }
+
+    fn set_connection_quality(&self, quality: ConnectionQuality) {
+        self.shared.set_connection_quality(quality);
+    }
+}
+
+impl_participant_trait!(LocalParticipant);

livekit/src/room/participant/mod.rs

@@ -0,0 +1,247 @@
+use super::publication::RemoteTrackPublication;
+use super::TrackError;
+use crate::proto;
+use crate::proto::ParticipantInfo;
+use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid};
+use crate::room::participant::local_participant::LocalParticipant;
+use crate::room::participant::remote_participant::RemoteParticipant;
+use crate::room::publication::{TrackPublication, TrackPublicationTrait};
+use crate::room::track::remote_track::RemoteTrackHandle;
+use livekit_utils::enum_dispatch;
+use livekit_utils::observer::Dispatcher;
+use parking_lot::{Mutex, RwLock, RwLockReadGuard};
+use proto::data_packet;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering};
+use std::sync::Arc;
+use tokio::sync::mpsc;
+
+pub mod local_participant;
+pub mod remote_participant;
+
+#[derive(Debug, Clone)]
+pub enum ParticipantEvent {
+    TrackPublished {
+        publication: RemoteTrackPublication,
+    },
+    TrackUnpublished {
+        publication: RemoteTrackPublication,
+    },
+    TrackSubscribed {
+        track: RemoteTrackHandle,
+        publication: RemoteTrackPublication,
+    },
+    TrackUnsubscribed {
+        track: RemoteTrackHandle,
+        publication: RemoteTrackPublication,
+    },
+    TrackSubscriptionFailed {
+        error: TrackError,
+        sid: TrackSid,
+    },
+    DataReceived {
+        payload: Arc<Vec<u8>>,
+        kind: data_packet::Kind,
+    },
+    SpeakingChanged {
+        speaking: bool,
+    },
+    TrackMuted {
+        publication: TrackPublication,
+    },
+    TrackUnmuted {
+        publication: TrackPublication,
+    },
+    ConnectionQualityChanged {
+        quality: ConnectionQuality,
+    },
+}
+
+#[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>,
+    pub(super) identity: Mutex<ParticipantIdentity>,
+    pub(super) name: Mutex<String>,
+    pub(super) metadata: Mutex<String>,
+    pub(super) tracks: RwLock<HashMap<TrackSid, TrackPublication>>,
+    pub(super) speaking: AtomicBool,
+    pub(super) audio_level: AtomicU32,
+    pub(super) connection_quality: AtomicU8,
+    pub(super) dispatcher: Mutex<Dispatcher<ParticipantEvent>>,
+}
+
+impl ParticipantShared {
+    pub(super) fn new(
+        sid: ParticipantSid,
+        identity: ParticipantIdentity,
+        name: String,
+        metadata: String,
+    ) -> Self {
+        Self {
+            sid: Mutex::new(sid),
+            identity: Mutex::new(identity),
+            name: Mutex::new(name),
+            metadata: Mutex::new(metadata),
+            tracks: Default::default(),
+            speaking: Default::default(),
+            audio_level: Default::default(),
+            connection_quality: AtomicU8::new(ConnectionQuality::Unknown as u8),
+            dispatcher: Default::default(),
+        }
+    }
+
+    pub(crate) fn update_info(&self, info: ParticipantInfo) {
+        *self.sid.lock() = info.sid.into();
+        *self.identity.lock() = info.identity.into();
+        *self.name.lock() = info.name;
+        *self.metadata.lock() = info.metadata; // TODO(theomonnom): callback MetadataChanged
+    }
+
+    pub(crate) fn set_speaking(&self, speaking: bool) {
+        self.speaking.store(speaking, Ordering::SeqCst);
+    }
+
+    pub(crate) fn set_audio_level(&self, audio_level: f32) {
+        self.audio_level
+            .store(audio_level.to_bits(), Ordering::SeqCst)
+    }
+
+    pub(crate) fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
+        self.dispatcher.lock().register()
+    }
+
+    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);
+    }
+}
+
+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, emit_events: bool);
+}
+
+pub trait ParticipantTrait {
+    fn sid(&self) -> ParticipantSid;
+    fn identity(&self) -> ParticipantIdentity;
+    fn name(&self) -> String;
+    fn metadata(&self) -> String;
+    fn is_speaking(&self) -> bool;
+    fn audio_level(&self) -> f32;
+    fn connection_quality(&self) -> ConnectionQuality;
+    fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>>;
+    fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent>;
+}
+
+#[derive(Debug, Clone)]
+pub enum Participant {
+    Local(Arc<LocalParticipant>),
+    Remote(Arc<RemoteParticipant>),
+}
+
+// TODO(theomonnom): Should I provide a WeakParticipant here ?
+
+impl Participant {
+    enum_dispatch!(
+        [Local, Remote]
+        fnc!(pub(crate), update_info, &Self, [info: ParticipantInfo, emit_events: bool], ());
+        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], ());
+    );
+}
+
+impl ParticipantTrait for Participant {
+    enum_dispatch!(
+        [Local, Remote]
+        fnc!(sid, &Self, [], ParticipantSid);
+        fnc!(identity, &Self, [], ParticipantIdentity);
+        fnc!(name, &Self, [], String);
+        fnc!(metadata, &Self, [], String);
+        fnc!(is_speaking, &Self, [], bool);
+        fnc!(audio_level, &Self, [], f32);
+        fnc!(connection_quality, &Self, [], ConnectionQuality);
+        fnc!(tracks, &Self, [], RwLockReadGuard<HashMap<TrackSid, TrackPublication>>);
+        fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver<ParticipantEvent>);
+    );
+}
+
+macro_rules! impl_participant_trait {
+    ($x:ty) => {
+        impl crate::room::participant::ParticipantTrait for $x {
+            fn sid(&self) -> ParticipantSid {
+                self.shared.sid.lock().clone()
+            }
+
+            fn identity(&self) -> ParticipantIdentity {
+                self.shared.identity.lock().clone()
+            }
+
+            fn name(&self) -> String {
+                self.shared.name.lock().clone()
+            }
+
+            fn metadata(&self) -> String {
+                self.shared.metadata.lock().clone()
+            }
+
+            fn is_speaking(&self) -> bool {
+                self.shared.speaking.load(Ordering::SeqCst)
+            }
+
+            fn audio_level(&self) -> f32 {
+                f32::from_bits(self.shared.audio_level.load(Ordering::SeqCst))
+            }
+
+            fn connection_quality(&self) -> ConnectionQuality {
+                self.shared.connection_quality.load(Ordering::SeqCst).into()
+            }
+
+            fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>> {
+                self.shared.tracks.read()
+            }
+
+            fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
+                self.shared.register_observer()
+            }
+        }
+    };
+}
+
+pub(super) use impl_participant_trait;

livekit/src/room/participant/remote_participant.rs

@@ -0,0 +1,222 @@
+use super::ConnectionQuality;
+use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket};
+use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid};
+use crate::room::participant::{
+    impl_participant_trait, ParticipantEvent, ParticipantInternalTrait, ParticipantShared,
+    ParticipantTrait,
+};
+use crate::room::publication::{
+    RemoteTrackPublication, TrackPublication, TrackPublicationInternalTrait, TrackPublicationTrait,
+};
+use crate::room::track::remote_audio_track::RemoteAudioTrack;
+use crate::room::track::remote_track::RemoteTrackHandle;
+use crate::room::track::remote_video_track::RemoteVideoTrack;
+use crate::room::track::{TrackKind, TrackTrait};
+use crate::room::TrackError;
+use livekit_webrtc::media_stream::MediaStreamTrackHandle;
+use parking_lot::RwLockReadGuard;
+use std::collections::HashMap;
+use std::collections::HashSet;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc;
+use tokio::time::timeout;
+use tracing::{debug, error, instrument, Level};
+
+const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[derive(Debug)]
+pub struct RemoteParticipant {
+    shared: ParticipantShared,
+}
+
+impl RemoteParticipant {
+    pub(crate) fn new(
+        sid: ParticipantSid,
+        identity: ParticipantIdentity,
+        name: String,
+        metadata: String,
+    ) -> Self {
+        Self {
+            shared: ParticipantShared::new(sid, identity, name, metadata),
+        }
+    }
+
+    fn get_track_publication(&self, sid: &TrackSid) -> Option<RemoteTrackPublication> {
+        self.shared.tracks.read().get(sid).map(|track| {
+            if let TrackPublication::Remote(remote) = track {
+                remote.clone()
+            } else {
+                unreachable!()
+            }
+        })
+    }
+
+    /// Called by the RoomSession when receiving data by the RTCSession
+    /// It is just used to emit the Data event on the participant dispatcher.
+    pub(crate) fn on_data_received(&self, data: Arc<Vec<u8>>, kind: data_packet::Kind) {
+        self.shared
+            .dispatcher
+            .lock()
+            .dispatch(&ParticipantEvent::DataReceived {
+                payload: data,
+                kind,
+            });
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    pub(crate) async fn add_subscribed_media_track(
+        self: Arc<Self>,
+        sid: TrackSid,
+        media_track: MediaStreamTrackHandle,
+    ) {
+        let wait_publication = {
+            let participant = self.clone();
+            let sid = sid.clone();
+            async move {
+                loop {
+                    let publication = participant.get_track_publication(&sid);
+                    if let Some(publication) = publication {
+                        return publication;
+                    }
+
+                    tokio::task::yield_now().await;
+                }
+            }
+        };
+
+        if let Ok(remote_publication) = timeout(ADD_TRACK_TIMEOUT, wait_publication).await {
+            let track = match remote_publication.kind() {
+                TrackKind::Audio => {
+                    if let MediaStreamTrackHandle::Audio(rtc_track) = media_track {
+                        let audio_track = RemoteAudioTrack::new(
+                            remote_publication.sid().into(),
+                            remote_publication.name(),
+                            rtc_track,
+                        );
+                        RemoteTrackHandle::Audio(Arc::new(audio_track))
+                    } else {
+                        unreachable!();
+                    }
+                }
+                TrackKind::Video => {
+                    if let MediaStreamTrackHandle::Video(rtc_track) = media_track {
+                        let video_track = RemoteVideoTrack::new(
+                            remote_publication.sid().into(),
+                            remote_publication.name(),
+                            rtc_track,
+                        );
+                        RemoteTrackHandle::Video(Arc::new(video_track))
+                    } else {
+                        unreachable!()
+                    }
+                }
+                _ => unreachable!(),
+            };
+
+            debug!("starting track: {:?}", sid);
+
+            remote_publication.update_track(Some(track.clone().into()));
+            self.shared
+                .add_track_publication(TrackPublication::Remote(remote_publication.clone()));
+            track.start();
+
+            self.shared
+                .dispatcher
+                .lock()
+                .dispatch(&ParticipantEvent::TrackSubscribed {
+                    track: track,
+                    publication: remote_publication,
+                });
+        } else {
+            error!("could not find published track with sid: {:?}", sid);
+
+            self.shared
+                .dispatcher
+                .lock()
+                .dispatch(&ParticipantEvent::TrackSubscriptionFailed {
+                    sid: sid.clone(),
+                    error: TrackError::TrackNotFound(sid.clone().to_string()),
+                });
+        }
+    }
+
+    pub(crate) fn unpublish_track(self: &Arc<Self>, sid: &TrackSid, emit_events: bool) {
+        if let Some(publication) = self.get_track_publication(sid) {
+            // Unsubscribe to the track if needed
+            if let Some(track) = publication.track() {
+                track.stop();
+
+                self.shared
+                    .dispatcher
+                    .lock()
+                    .dispatch(&ParticipantEvent::TrackUnsubscribed {
+                        track: track.clone(),
+                        publication: publication.clone(),
+                    });
+            }
+
+            if emit_events {
+                self.shared
+                    .dispatcher
+                    .lock()
+                    .dispatch(&ParticipantEvent::TrackUnpublished {
+                        publication: publication.clone(),
+                    });
+            }
+
+            publication.update_track(None);
+        }
+    }
+}
+
+impl ParticipantInternalTrait for RemoteParticipant {
+    fn update_info(self: &Arc<Self>, info: ParticipantInfo, emit_events: bool) {
+        self.shared.update_info(info.clone());
+
+        let mut valid_tracks = HashSet::<TrackSid>::new();
+        for track in info.tracks {
+            if let Some(publication) = self.get_track_publication(&track.sid.clone().into()) {
+                publication.update_info(track.clone());
+            } else {
+                let publication = RemoteTrackPublication::new(track.clone(), self.sid(), None);
+                self.shared
+                    .add_track_publication(TrackPublication::Remote(publication.clone()));
+
+                // This is a new track, dispatch publish event
+                if emit_events {
+                    self.shared
+                        .dispatcher
+                        .lock()
+                        .dispatch(&ParticipantEvent::TrackPublished { publication });
+                }
+            }
+
+            valid_tracks.insert(track.sid.into());
+        }
+
+        // remove tracks that are no longer valid
+        for (sid, _) in self.shared.tracks.read().iter() {
+            if valid_tracks.contains(sid) {
+                continue;
+            }
+
+            self.unpublish_track(sid, emit_events);
+        }
+    }
+
+    fn set_speaking(&self, speaking: bool) {
+        self.shared.set_speaking(speaking);
+    }
+
+    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);

livekit/src/room/publication/mod.rs

@@ -0,0 +1,262 @@
+use crate::proto::TrackType;
+use crate::proto::{TrackInfo, TrackSource as ProtoTrackSource};
+use crate::room::id::ParticipantSid;
+use crate::room::id::TrackSid;
+use crate::room::track::local_track::LocalTrackHandle;
+use crate::room::track::remote_track::RemoteTrackHandle;
+use crate::room::track::{TrackHandle, TrackKind, TrackSource, TrackTrait};
+use livekit_utils::enum_dispatch;
+use livekit_utils::observer::Dispatcher;
+use parking_lot::Mutex;
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
+use std::sync::Arc;
+use tokio::sync::{mpsc, oneshot};
+
+use super::track::{TrackDimension, TrackEvent};
+
+pub(crate) trait TrackPublicationInternalTrait {
+    fn update_track(&self, track: Option<TrackHandle>);
+    fn update_info(&self, info: TrackInfo);
+}
+
+pub trait TrackPublicationTrait {
+    fn name(&self) -> String;
+    fn sid(&self) -> TrackSid;
+    fn kind(&self) -> TrackKind;
+    fn source(&self) -> TrackSource;
+    fn muted(&self) -> bool;
+    fn simulcasted(&self) -> bool;
+}
+
+#[derive(Debug)]
+pub(super) struct TrackPublicationShared {
+    pub(super) track: Mutex<Option<TrackHandle>>,
+    pub(super) name: Mutex<String>,
+    pub(super) sid: Mutex<TrackSid>,
+    pub(super) kind: AtomicU8,   // Casted to TrackKind
+    pub(super) source: AtomicU8, // Casted to TrackSource
+    pub(super) simulcasted: AtomicBool,
+    pub(super) dimension: Mutex<TrackDimension>,
+    pub(super) mime_type: Mutex<String>,
+    pub(super) muted: AtomicBool,
+    pub(super) participant: ParticipantSid,
+    pub(super) dispatcher: Mutex<Dispatcher<TrackEvent>>,
+    pub(super) close_sender: Mutex<Option<oneshot::Sender<()>>>,
+}
+
+impl TrackPublicationShared {
+    pub fn new(
+        info: TrackInfo,
+        participant: ParticipantSid,
+        track: Option<TrackHandle>,
+    ) -> Arc<Self> {
+        Arc::new(Self {
+            track: Mutex::new(track),
+            name: Mutex::new(info.name),
+            sid: Mutex::new(info.sid.into()),
+            kind: AtomicU8::new(TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8),
+            source: AtomicU8::new(TrackSource::from(
+                ProtoTrackSource::from_i32(info.source).unwrap(),
+            ) as u8),
+            simulcasted: AtomicBool::new(info.simulcast),
+            dimension: Mutex::new(TrackDimension(info.width, info.height)),
+            mime_type: Mutex::new(info.mime_type),
+            muted: AtomicBool::new(info.muted),
+            dispatcher: Default::default(),
+            close_sender: Default::default(),
+            participant,
+        })
+    }
+
+    pub fn update_track(self: &Arc<Self>, track: Option<TrackHandle>) {
+        let mut old_track = self.track.lock();
+
+        if let Some(close_sender) = self.close_sender.lock().take() {
+            let _ = close_sender.send(());
+        }
+
+        *old_track = track.clone();
+        if let Some(track) = track {
+            let (close_sender, close_receiver) = oneshot::channel();
+            self.close_sender.lock().replace(close_sender);
+
+            let track_receiver = track.register_observer();
+            tokio::spawn(
+                self.clone()
+                    .publication_task(close_receiver, track_receiver),
+            );
+        }
+    }
+
+    /// Task used to forward TrackHandle's events to the TrackPublications's dispatcher
+    async fn publication_task(
+        self: Arc<Self>,
+        mut close_receiver: oneshot::Receiver<()>,
+        mut track_receiver: mpsc::UnboundedReceiver<TrackEvent>,
+    ) {
+        loop {
+            tokio::select! {
+                Some(event) = track_receiver.recv() => {
+                    self.dispatcher.lock().dispatch(&event);
+                }
+                _ = &mut close_receiver => {
+                    break;
+                }
+            }
+        }
+    }
+
+    pub fn update_info(&self, info: TrackInfo) {
+        *self.name.lock() = info.name;
+        *self.sid.lock() = info.sid.into();
+        *self.dimension.lock() = TrackDimension(info.width, info.height);
+        *self.mime_type.lock() = info.mime_type;
+        self.kind.store(
+            TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8,
+            Ordering::SeqCst,
+        );
+        self.source.store(
+            TrackSource::from(ProtoTrackSource::from_i32(info.source).unwrap()) as u8,
+            Ordering::SeqCst,
+        );
+        self.simulcasted.store(info.simulcast, Ordering::SeqCst);
+        self.muted.store(info.muted, Ordering::SeqCst);
+
+        if let Some(track) = self.track.lock().as_ref() {
+            track.set_muted(info.muted);
+        }
+    }
+}
+
+impl Drop for TrackPublicationShared {
+    fn drop(&mut self) {
+        if let Some(close_sender) = self.close_sender.lock().take() {
+            let _ = close_sender.send(());
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub enum TrackPublication {
+    Local(LocalTrackPublication),
+    Remote(RemoteTrackPublication),
+}
+
+impl TrackPublication {
+    pub fn track(&self) -> Option<TrackHandle> {
+        // Not calling Local/Remote function here, we don't need "cast"
+        match self {
+            TrackPublication::Local(p) => p.shared.track.lock().clone(),
+            TrackPublication::Remote(p) => p.shared.track.lock().clone(),
+        }
+    }
+}
+
+impl TrackPublicationInternalTrait for TrackPublication {
+    enum_dispatch!(
+        [Local, Remote]
+        fnc!(update_track, &Self, [track: Option<TrackHandle>], ());
+        fnc!(update_info, &Self, [info: TrackInfo], ());
+    );
+}
+
+impl TrackPublicationTrait for TrackPublication {
+    enum_dispatch!(
+        [Local, Remote]
+        fnc!(sid, &Self, [], TrackSid);
+        fnc!(name, &Self, [], String);
+        fnc!(kind, &Self, [], TrackKind);
+        fnc!(source, &Self, [], TrackSource);
+        fnc!(muted, &Self, [], bool);
+        fnc!(simulcasted, &Self, [], bool);
+    );
+}
+
+macro_rules! impl_publication_trait {
+    ($x:ident) => {
+        impl TrackPublicationTrait for $x {
+            fn name(&self) -> String {
+                self.shared.name.lock().clone()
+            }
+
+            fn sid(&self) -> TrackSid {
+                self.shared.sid.lock().clone()
+            }
+
+            fn kind(&self) -> TrackKind {
+                self.shared.kind.load(Ordering::SeqCst).into()
+            }
+
+            fn source(&self) -> TrackSource {
+                self.shared.source.load(Ordering::SeqCst).into()
+            }
+
+            fn simulcasted(&self) -> bool {
+                self.shared.simulcasted.load(Ordering::SeqCst)
+            }
+
+            fn muted(&self) -> bool {
+                self.shared.muted.load(Ordering::SeqCst)
+            }
+        }
+    };
+}
+
+#[derive(Clone, Debug)]
+pub struct LocalTrackPublication {
+    shared: Arc<TrackPublicationShared>,
+}
+
+impl LocalTrackPublication {
+    pub fn track(&self) -> Option<LocalTrackHandle> {
+        self.shared
+            .track
+            .lock()
+            .clone()
+            .map(|local_track| local_track.try_into().unwrap())
+    }
+}
+
+impl TrackPublicationInternalTrait for LocalTrackPublication {
+    fn update_track(&self, track: Option<TrackHandle>) {
+        self.shared.update_track(track);
+    }
+
+    fn update_info(&self, info: TrackInfo) {
+        self.shared.update_info(info);
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct RemoteTrackPublication {
+    shared: Arc<TrackPublicationShared>,
+}
+
+impl RemoteTrackPublication {
+    pub fn new(info: TrackInfo, participant: ParticipantSid, track: Option<TrackHandle>) -> Self {
+        Self {
+            shared: TrackPublicationShared::new(info, participant, track),
+        }
+    }
+
+    pub fn track(&self) -> Option<RemoteTrackHandle> {
+        self.shared
+            .track
+            .lock()
+            .clone()
+            .map(|track| track.try_into().unwrap())
+    }
+}
+
+impl TrackPublicationInternalTrait for RemoteTrackPublication {
+    fn update_track(&self, track: Option<TrackHandle>) {
+        self.shared.update_track(track);
+    }
+
+    fn update_info(&self, info: TrackInfo) {
+        self.shared.update_info(info);
+    }
+}
+
+impl_publication_trait!(LocalTrackPublication);
+impl_publication_trait!(RemoteTrackPublication);

livekit/src/room/room_session.rs

@@ -0,0 +1,571 @@
+use super::id::{ParticipantIdentity, ParticipantSid};
+use super::participant::local_participant::LocalParticipant;
+use super::participant::remote_participant::RemoteParticipant;
+use super::participant::{ConnectionQuality, Participant, ParticipantEvent};
+use super::participant::{ParticipantInternalTrait, ParticipantTrait};
+use super::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario};
+use crate::proto::{self, participant_info, SpeakerInfo};
+use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine};
+use crate::signal_client::SignalOptions;
+use parking_lot::{Mutex, RwLock};
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::Arc;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+use tracing::{error, instrument, Level};
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum ConnectionState {
+    Disconnected,
+    Connected,
+    Reconnecting,
+    Unknown,
+}
+
+impl From<u8> for ConnectionState {
+    fn from(value: u8) -> Self {
+        match value {
+            0 => ConnectionState::Disconnected,
+            1 => ConnectionState::Connected,
+            2 => ConnectionState::Reconnecting,
+            _ => ConnectionState::Unknown,
+        }
+    }
+}
+
+/// Internal representation of a RoomSession
+#[derive(Debug)]
+struct SessionInner {
+    state: AtomicU8, // ConnectionState
+    sid: Mutex<String>,
+    name: Mutex<String>,
+    participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
+    participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
+    active_speakers: RwLock<Vec<Participant>>,
+    rtc_engine: Arc<RTCEngine>,
+    local_participant: Arc<LocalParticipant>,
+    room_emitter: RoomEmitter,
+}
+
+#[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(room_emitter: RoomEmitter, 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 = Arc::new(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),
+            name: Mutex::new(room_info.name),
+            participants: Default::default(),
+            participants_tasks: Default::default(),
+            active_speakers: Default::default(),
+            rtc_engine,
+            local_participant,
+            room_emitter,
+        });
+
+        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(), false);
+        }
+
+        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::from(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 session(&self) -> RoomSession {
+        self.session.clone()
+    }
+}
+
+impl RoomSession {
+    fn from(inner: Arc<SessionInner>) -> Self {
+        Self { inner }
+    }
+
+    pub fn sid(&self) -> String {
+        self.inner.sid.lock().clone()
+    }
+
+    pub fn name(&self) -> String {
+        self.inner.name.lock().clone()
+    }
+
+    pub fn local_participant(&self) -> Arc<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) -> &RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>> {
+        &self.inner.participants
+    }
+
+    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() => {
+                    match res {
+                        Some(event) => {
+                            if let Err(err) = self.on_engine_event(event).await {
+                                error!("failed to handle engine event: {:?}", err);
+                            }
+                        },
+                        _ => panic!("engine_events has been closed unexpectedly")
+                    };
+                },
+                 _ = &mut close_receiver => {
+                    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() => {
+                    match res {
+                        Some(event) => {
+                            if let Err(err) = self.on_participant_event(&participant, event).await {
+                                error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
+                            }
+                        },
+                        _ => panic!("participant_events has been closed unexpectedly")
+                    };
+                },
+                _ = &mut close_rx => {
+                    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 } => {
+                    let _ = self.room_emitter.send(RoomEvent::TrackPublished {
+                        participant: remote_participant.clone(),
+                        publication,
+                    });
+                }
+                ParticipantEvent::TrackUnpublished { publication } => {
+                    let _ = self.room_emitter.send(RoomEvent::TrackUnpublished {
+                        participant: remote_participant.clone(),
+                        publication,
+                    });
+                }
+                ParticipantEvent::TrackSubscribed { track, publication } => {
+                    let _ = self.room_emitter.send(RoomEvent::TrackSubscribed {
+                        participant: remote_participant.clone(),
+                        track,
+                        publication,
+                    });
+                }
+                ParticipantEvent::TrackUnsubscribed { track, publication } => {
+                    let _ = self.room_emitter.send(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(update) => self.handle_participant_update(update),
+            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) {
+                    let _ = self.room_emitter.send(RoomEvent::Reconnecting);
+                }
+            }
+            EngineEvent::Resumed => {
+                self.update_connection_state(ConnectionState::Connected);
+                let _ = self.room_emitter.send(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()) {
+                    let _ = self.room_emitter.send(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);
+        let _ = self
+            .room_emitter
+            .send(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>, update: proto::ParticipantUpdate) {
+        for pi in update.participants {
+            if pi.sid == self.local_participant.sid()
+                || pi.identity == self.local_participant.identity()
+            {
+                self.local_participant.clone().update_info(pi, true);
+                continue;
+            }
+
+            let remote_participant = self.get_participant(&pi.sid.clone().into());
+
+            if let Some(remote_participant) = remote_participant {
+                if pi.state == participant_info::State::Disconnected as i32 {
+                    // Participant disconnected
+                    self.clone()
+                        .handle_participant_disconnect(remote_participant)
+                } else {
+                    // Participant is already connected, update the it
+                    remote_participant.update_info(pi.clone(), true);
+                }
+            } else {
+                // Create a new participant
+                let remote_participant = {
+                    let pi = pi.clone();
+                    self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
+                };
+
+                let _ = self
+                    .room_emitter
+                    .send(RoomEvent::ParticipantConnected(remote_participant.clone()));
+
+                remote_participant.update_info(pi.clone(), true);
+            }
+        }
+    }
+
+    /// 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();
+
+        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
+            .room_emitter
+            .send(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);
+            let _ = self.room_emitter.send(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) {
+            let _ = self.room_emitter.send(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);
+        let _ = self.room_emitter.send(RoomEvent::Reconnected);
+
+        if let Some(pi) = join_response.participant {
+            self.local_participant.update_info(pi, true); // The sid may have changed
+        }
+
+        self.handle_participant_update(proto::ParticipantUpdate {
+            participants: 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);
+        let _ = self.room_emitter.send(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,
+    ) -> Arc<RemoteParticipant> {
+        let participant = Arc::new(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: Arc<RemoteParticipant>) {
+        tokio::spawn(async move {
+            for (sid, _) in &*remote_participant.tracks() {
+                remote_participant.unpublish_track(&sid, true);
+            }
+
+            // 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());
+
+            let _ = self.room_emitter.send(RoomEvent::ParticipantDisconnected(
+                remote_participant.clone(),
+            ));
+        });
+    }
+
+    fn get_participant(&self, sid: &ParticipantSid) -> Option<Arc<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
+    }
+}

livekit/src/room/track/audio_track.rs

@@ -0,0 +1,31 @@
+use crate::room::track::local_audio_track::LocalAudioTrack;
+use crate::room::track::remote_audio_track::RemoteAudioTrack;
+use crate::room::track::TrackHandle;
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub enum AudioTrackHandle {
+    Local(Arc<LocalAudioTrack>),
+    Remote(Arc<RemoteAudioTrack>),
+}
+
+impl From<AudioTrackHandle> for TrackHandle {
+    fn from(audio_track: AudioTrackHandle) -> Self {
+        match audio_track {
+            AudioTrackHandle::Local(local_audio) => Self::LocalAudio(local_audio),
+            AudioTrackHandle::Remote(remote_audio) => Self::RemoteAudio(remote_audio),
+        }
+    }
+}
+
+impl TryFrom<TrackHandle> for AudioTrackHandle {
+    type Error = &'static str;
+
+    fn try_from(track: TrackHandle) -> Result<Self, Self::Error> {
+        match track {
+            TrackHandle::LocalAudio(local_audio) => Ok(Self::Local(local_audio)),
+            TrackHandle::RemoteAudio(remote_audio) => Ok(Self::Remote(remote_audio)),
+            _ => Err("not a audio track"),
+        }
+    }
+}

livekit/src/room/track/local_audio_track.rs

@@ -0,0 +1,8 @@
+use crate::room::track::{impl_track_trait, TrackShared};
+
+#[derive(Debug)]
+pub struct LocalAudioTrack {
+    shared: TrackShared,
+}
+
+impl_track_trait!(LocalAudioTrack);

livekit/src/room/track/local_track.rs

@@ -0,0 +1,31 @@
+use crate::room::track::local_audio_track::LocalAudioTrack;
+use crate::room::track::local_video_track::LocalVideoTrack;
+use crate::room::track::TrackHandle;
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub enum LocalTrackHandle {
+    Audio(Arc<LocalAudioTrack>),
+    Video(Arc<LocalVideoTrack>),
+}
+
+impl From<LocalTrackHandle> for TrackHandle {
+    fn from(local_track: LocalTrackHandle) -> Self {
+        match local_track {
+            LocalTrackHandle::Audio(local_audio) => Self::LocalAudio(local_audio),
+            LocalTrackHandle::Video(local_video) => Self::LocalVideo(local_video),
+        }
+    }
+}
+
+impl TryFrom<TrackHandle> for LocalTrackHandle {
+    type Error = &'static str;
+
+    fn try_from(track: TrackHandle) -> Result<Self, Self::Error> {
+        match track {
+            TrackHandle::LocalAudio(local_audio) => Ok(Self::Audio(local_audio)),
+            TrackHandle::LocalVideo(local_video) => Ok(Self::Video(local_video)),
+            _ => Err("not a local track"),
+        }
+    }
+}

livekit/src/room/track/local_video_track.rs

@@ -0,0 +1,8 @@
+use crate::room::track::{impl_track_trait, TrackShared};
+
+#[derive(Debug)]
+pub struct LocalVideoTrack {
+    shared: TrackShared,
+}
+
+impl_track_trait!(LocalVideoTrack);

livekit/src/room/track/mod.rs

@@ -0,0 +1,257 @@
+use crate::proto::{TrackSource as ProtoTrackSource, TrackType};
+use crate::room::id::TrackSid;
+use crate::room::track::local_audio_track::LocalAudioTrack;
+use crate::room::track::local_video_track::LocalVideoTrack;
+use crate::room::track::remote_audio_track::RemoteAudioTrack;
+use crate::room::track::remote_video_track::RemoteVideoTrack;
+use livekit_utils::enum_dispatch;
+use livekit_utils::observer::Dispatcher;
+use livekit_webrtc::media_stream::{MediaStreamTrackHandle, MediaStreamTrackTrait};
+use parking_lot::Mutex;
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
+use std::sync::Arc;
+use tokio::sync::mpsc;
+
+pub mod audio_track;
+pub mod local_audio_track;
+pub mod local_track;
+pub mod local_video_track;
+pub mod remote_audio_track;
+pub mod remote_track;
+pub mod remote_video_track;
+pub mod video_track;
+
+#[derive(Debug)]
+pub enum TrackKind {
+    Unknown,
+    Audio,
+    Video,
+}
+
+impl From<u8> for TrackKind {
+    fn from(val: u8) -> Self {
+        match val {
+            1 => Self::Audio,
+            2 => Self::Video,
+            _ => Self::Unknown,
+        }
+    }
+}
+
+impl From<TrackType> for TrackKind {
+    fn from(r#type: TrackType) -> Self {
+        match r#type {
+            TrackType::Audio => Self::Audio,
+            TrackType::Video => Self::Video,
+            _ => Self::Unknown,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum StreamState {
+    Unknown,
+    Active,
+    Paused,
+}
+
+impl From<u8> for StreamState {
+    fn from(val: u8) -> Self {
+        match val {
+            1 => Self::Active,
+            2 => Self::Paused,
+            _ => Self::Unknown,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum TrackSource {
+    Unknown,
+    Camera,
+    Microphone,
+    Screenshare,
+    ScreenshareAudio,
+}
+
+impl From<u8> for TrackSource {
+    fn from(val: u8) -> Self {
+        match val {
+            1 => Self::Camera,
+            2 => Self::Microphone,
+            3 => Self::Screenshare,
+            4 => Self::ScreenshareAudio,
+            _ => Self::Unknown,
+        }
+    }
+}
+
+impl From<ProtoTrackSource> for TrackSource {
+    fn from(source: ProtoTrackSource) -> Self {
+        match source {
+            ProtoTrackSource::Camera => Self::Camera,
+            ProtoTrackSource::Microphone => Self::Microphone,
+            ProtoTrackSource::ScreenShare => Self::Screenshare,
+            ProtoTrackSource::ScreenShareAudio => Self::ScreenshareAudio,
+            ProtoTrackSource::Unknown => Self::Unknown,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct TrackDimension(pub u32, pub u32);
+
+pub trait TrackTrait {
+    fn sid(&self) -> TrackSid;
+    fn name(&self) -> String;
+    fn kind(&self) -> TrackKind;
+    fn stream_state(&self) -> StreamState;
+    fn start(&self);
+    fn stop(&self);
+    fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent>;
+    fn set_muted(&self, muted: bool);
+}
+
+#[derive(Debug, Clone)]
+pub enum TrackEvent {
+    Mute,
+    Unmute,
+}
+
+#[derive(Debug)]
+pub(super) struct TrackShared {
+    pub(super) sid: Mutex<TrackSid>,
+    pub(super) name: Mutex<String>,
+    pub(super) kind: AtomicU8,         // TrackKind
+    pub(super) stream_state: AtomicU8, // StreamState
+    pub(super) muted: AtomicBool,
+    pub(super) rtc_track: MediaStreamTrackHandle,
+    pub(super) dispatcher: Mutex<Dispatcher<TrackEvent>>,
+}
+
+impl TrackShared {
+    pub(crate) fn new(
+        sid: TrackSid,
+        name: String,
+        kind: TrackKind,
+        rtc_track: MediaStreamTrackHandle,
+    ) -> Self {
+        Self {
+            sid: Mutex::new(sid),
+            name: Mutex::new(name),
+            kind: AtomicU8::new(kind as u8),
+            stream_state: AtomicU8::new(StreamState::Active as u8),
+            muted: AtomicBool::new(false),
+            rtc_track,
+            dispatcher: Default::default(),
+        }
+    }
+
+    pub(crate) fn start(&self) {
+        self.rtc_track.set_enabled(true);
+    }
+
+    pub(crate) fn stop(&self) {
+        self.rtc_track.set_enabled(false);
+    }
+
+    pub(crate) fn set_muted(&self, muted: bool) {
+        if self.muted.load(Ordering::SeqCst) == muted {
+            return;
+        }
+
+        self.muted.store(muted, Ordering::SeqCst);
+        self.rtc_track.set_enabled(!muted);
+
+        self.dispatcher.lock().dispatch(if muted {
+            &TrackEvent::Mute
+        } else {
+            &TrackEvent::Unmute
+        });
+    }
+
+    pub(crate) fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
+        self.dispatcher.lock().register()
+    }
+}
+
+#[derive(Clone, Debug)]
+pub enum TrackHandle {
+    LocalVideo(Arc<LocalVideoTrack>),
+    LocalAudio(Arc<LocalAudioTrack>),
+    RemoteVideo(Arc<RemoteVideoTrack>),
+    RemoteAudio(Arc<RemoteAudioTrack>),
+}
+
+impl TrackTrait for TrackHandle {
+    enum_dispatch!(
+        [LocalVideo, LocalAudio, RemoteVideo, RemoteAudio]
+        fnc!(sid, &Self, [], TrackSid);
+        fnc!(name, &Self, [], String);
+        fnc!(kind, &Self, [], TrackKind);
+        fnc!(stream_state, &Self, [], StreamState);
+        fnc!(start, &Self, [], ());
+        fnc!(stop, &Self, [], ());
+        fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver<TrackEvent>);
+        fnc!(set_muted, &Self, [muted: bool], ());
+    );
+}
+
+impl TrackHandle {
+    pub fn rtc_track(&self) -> MediaStreamTrackHandle {
+        match self {
+            Self::RemoteVideo(remote_video) => {
+                MediaStreamTrackHandle::Video(remote_video.rtc_track())
+            }
+            Self::RemoteAudio(remote_audio) => {
+                MediaStreamTrackHandle::Audio(remote_audio.rtc_track())
+            }
+            _ => todo!(),
+        }
+    }
+}
+
+macro_rules! impl_track_trait {
+    ($x:ident) => {
+        use std::sync::atomic::Ordering;
+        use tokio::sync::mpsc;
+        use $crate::room::id::TrackSid;
+        use $crate::room::track::{StreamState, TrackEvent, TrackKind, TrackTrait};
+
+        impl TrackTrait for $x {
+            fn sid(&self) -> TrackSid {
+                self.shared.sid.lock().clone()
+            }
+
+            fn name(&self) -> String {
+                self.shared.name.lock().clone()
+            }
+
+            fn kind(&self) -> TrackKind {
+                self.shared.kind.load(Ordering::SeqCst).into()
+            }
+
+            fn stream_state(&self) -> StreamState {
+                self.shared.stream_state.load(Ordering::SeqCst).into()
+            }
+
+            fn start(&self) {
+                self.shared.start();
+            }
+
+            fn stop(&self) {
+                self.shared.stop();
+            }
+
+            fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
+                self.shared.register_observer()
+            }
+
+            fn set_muted(&self, muted: bool) {
+                self.shared.set_muted(muted);
+            }
+        }
+    };
+}
+
+pub(super) use impl_track_trait;

livekit/src/room/track/remote_audio_track.rs

@@ -0,0 +1,31 @@
+use crate::room::track::{impl_track_trait, TrackShared};
+use livekit_webrtc::media_stream::{AudioTrack, MediaStreamTrackHandle};
+use std::sync::Arc;
+
+#[derive(Debug)]
+pub struct RemoteAudioTrack {
+    shared: TrackShared,
+}
+
+impl RemoteAudioTrack {
+    pub(crate) fn new(sid: TrackSid, name: String, track: Arc<AudioTrack>) -> Self {
+        Self {
+            shared: TrackShared::new(
+                sid,
+                name,
+                TrackKind::Audio,
+                MediaStreamTrackHandle::Audio(track),
+            ),
+        }
+    }
+
+    pub fn rtc_track(&self) -> Arc<AudioTrack> {
+        if let MediaStreamTrackHandle::Audio(audio) = &self.shared.rtc_track {
+            audio.clone()
+        } else {
+            unreachable!()
+        }
+    }
+}
+
+impl_track_trait!(RemoteAudioTrack);

livekit/src/room/track/remote_track.rs

@@ -0,0 +1,51 @@
+use std::sync::Arc;
+use super::{StreamState, TrackKind};
+use crate::room::id::TrackSid;
+use crate::room::track::remote_audio_track::RemoteAudioTrack;
+use crate::room::track::remote_video_track::RemoteVideoTrack;
+use crate::room::track::{TrackHandle, TrackEvent};
+use tokio::sync::mpsc;
+use livekit_utils::enum_dispatch;
+
+use super::TrackTrait;
+
+#[derive(Clone, Debug)]
+pub enum RemoteTrackHandle {
+    Audio(Arc<RemoteAudioTrack>),
+    Video(Arc<RemoteVideoTrack>),
+}
+
+impl TrackTrait for RemoteTrackHandle {
+    enum_dispatch!(
+        [Audio, Video]
+        fnc!(sid, &Self, [], TrackSid);
+        fnc!(name, &Self, [], String);
+        fnc!(kind, &Self, [], TrackKind);
+        fnc!(stream_state, &Self, [], StreamState);
+        fnc!(start, &Self, [], ());
+        fnc!(stop, &Self, [], ());
+        fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver<TrackEvent>);
+        fnc!(set_muted, &Self, [muted: bool], ());
+    );
+}
+
+impl From<RemoteTrackHandle> for TrackHandle {
+    fn from(remote_track: RemoteTrackHandle) -> Self {
+        match remote_track {
+            RemoteTrackHandle::Audio(remote_audio) => Self::RemoteAudio(remote_audio),
+            RemoteTrackHandle::Video(remote_video) => Self::RemoteVideo(remote_video),
+        }
+    }
+}
+
+impl TryFrom<TrackHandle> for RemoteTrackHandle {
+    type Error = &'static str;
+
+    fn try_from(track: TrackHandle) -> Result<Self, Self::Error> {
+        match track {
+            TrackHandle::RemoteAudio(remote_audio) => Ok(Self::Audio(remote_audio)),
+            TrackHandle::RemoteVideo(remote_video) => Ok(Self::Video(remote_video)),
+            _ => Err("not a remote track"),
+        }
+    }
+}

livekit/src/room/track/remote_video_track.rs

@@ -0,0 +1,32 @@
+use livekit_webrtc::media_stream::{MediaStreamTrackHandle, VideoTrack};
+use std::sync::Arc;
+
+use crate::room::track::{impl_track_trait, TrackShared};
+
+#[derive(Debug)]
+pub struct RemoteVideoTrack {
+    shared: TrackShared,
+}
+
+impl RemoteVideoTrack {
+    pub(crate) fn new(sid: TrackSid, name: String, track: Arc<VideoTrack>) -> Self {
+        Self {
+            shared: TrackShared::new(
+                sid,
+                name,
+                TrackKind::Video,
+                MediaStreamTrackHandle::Video(track),
+            ),
+        }
+    }
+
+    pub fn rtc_track(&self) -> Arc<VideoTrack> {
+        if let MediaStreamTrackHandle::Video(video) = &self.shared.rtc_track {
+            video.clone()
+        } else {
+            unreachable!()
+        }
+    }
+}
+
+impl_track_trait!(RemoteVideoTrack);

livekit/src/room/track/video_track.rs

@@ -0,0 +1,31 @@
+use crate::room::track::local_video_track::LocalVideoTrack;
+use crate::room::track::remote_video_track::RemoteVideoTrack;
+use crate::room::track::TrackHandle;
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub enum VideoTrackHandle {
+    Local(Arc<LocalVideoTrack>),
+    Remote(Arc<RemoteVideoTrack>),
+}
+
+impl From<VideoTrackHandle> for TrackHandle {
+    fn from(video_track: VideoTrackHandle) -> Self {
+        match video_track {
+            VideoTrackHandle::Local(local_video) => Self::LocalVideo(local_video),
+            VideoTrackHandle::Remote(remote_video) => Self::RemoteVideo(remote_video),
+        }
+    }
+}
+
+impl TryFrom<TrackHandle> for VideoTrackHandle {
+    type Error = &'static str;
+
+    fn try_from(track: TrackHandle) -> Result<Self, Self::Error> {
+        match track {
+            TrackHandle::LocalVideo(local_video) => Ok(Self::Local(local_video)),
+            TrackHandle::RemoteVideo(remote_video) => Ok(Self::Remote(remote_video)),
+            _ => Err("not a video track"),
+        }
+    }
+}

livekit/src/rtc_engine/lk_runtime.rs

@@ -0,0 +1,37 @@
+use std::fmt::{Debug, Formatter};
+
+use tracing::trace;
+
+use livekit_webrtc::peer_connection_factory::PeerConnectionFactory;
+use livekit_webrtc::webrtc::RTCRuntime;
+
+/// SAFETY: The order of initialization and deletion is important for LKRuntime.
+/// See the C++ constructors & destructors of these fields
+
+pub struct LKRuntime {
+    pub pc_factory: PeerConnectionFactory,
+    pub rtc_runtime: RTCRuntime,
+}
+
+impl Debug for LKRuntime {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "LKRuntime")
+    }
+}
+
+impl Default for LKRuntime {
+    fn default() -> Self {
+        trace!("LKRuntime::default()");
+        let rtc_runtime = RTCRuntime::new();
+        Self {
+            pc_factory: PeerConnectionFactory::new(rtc_runtime.clone()),
+            rtc_runtime,
+        }
+    }
+}
+
+impl Drop for LKRuntime {
+    fn drop(&mut self) {
+        trace!("LKRuntime::drop()");
+    }
+}

livekit/src/rtc_engine/mod.rs

@@ -0,0 +1,471 @@
+use futures::future::BoxFuture;
+use futures::FutureExt;
+use livekit_webrtc::data_channel::DataSendError;
+use livekit_webrtc::jsep::SdpParseError;
+use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle};
+use livekit_webrtc::rtc_error::RTCError;
+use livekit_webrtc::rtp_receiver::RtpReceiver;
+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 tokio::time::{interval, Interval};
+
+use lazy_static::lazy_static;
+use tokio::sync::{mpsc, oneshot};
+use tracing::{error, info, warn};
+
+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};
+
+use self::rtc_session::{RTCSession, SessionEvent, SessionEvents, SessionInfo};
+
+mod lk_runtime;
+mod pc_transport;
+mod rtc_events;
+mod rtc_session;
+
+pub(crate) type EngineEmitter = mpsc::Sender<EngineEvent>;
+pub(crate) type EngineEvents = mpsc::Receiver<EngineEvent>;
+pub(crate) type EngineResult<T> = Result<T, EngineError>;
+
+#[derive(Debug, Clone, Eq, PartialEq)]
+#[repr(u8)]
+pub enum SimulateScenario {
+    SignalReconnect,
+    Speaker,
+    NodeFailure,
+    ServerLeave,
+    Migration,
+    ForceTcp,
+    ForceTls,
+}
+
+#[derive(Error, Debug)]
+pub enum EngineError {
+    #[error("signal failure: {0}")]
+    Signal(#[from] SignalError),
+    #[error("internal webrtc failure")]
+    Rtc(#[from] RTCError),
+    #[error("failed to parse sdp")]
+    Parse(#[from] SdpParseError),
+    #[error("serde error")]
+    Serde(#[from] serde_json::Error),
+    #[error("failed to send data to the datachannel")]
+    Data(#[from] DataSendError),
+    #[error("connection error: {0}")]
+    Connection(String),
+    #[error("decode error")]
+    Decode(#[from] prost::DecodeError),
+    #[error("internal error: {0}")]
+    Internal(String), // Unexpected error
+}
+
+#[derive(Debug)]
+pub enum EngineEvent {
+    ParticipantUpdate(ParticipantUpdate),
+    MediaTrack {
+        track: MediaStreamTrackHandle,
+        stream: MediaStream,
+        receiver: RtpReceiver,
+    },
+    Data {
+        participant_sid: String,
+        payload: Vec<u8>,
+        kind: data_packet::Kind,
+    },
+    SpeakersChanged {
+        speakers: Vec<SpeakerInfo>,
+    },
+    ConnectionQuality {
+        updates: Vec<proto::ConnectionQualityInfo>,
+    },
+    Resuming,
+    Resumed,
+    Restarting,
+    Restarted,
+    Disconnected,
+}
+
+pub const RECONNECT_ATTEMPTS: u32 = 10;
+pub const RECONNECT_INTERVAL: Duration = Duration::from_secs(5);
+
+lazy_static! {
+    // Share one LKRuntime across all RTCEngine instances
+    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
+}
+///
+/// Represents a running RTCSession with the ability to close the session
+/// and the engine_task
+#[derive(Debug)]
+struct EngineHandle {
+    session: RTCSession,
+    engine_task: JoinHandle<()>,
+    close_sender: oneshot::Sender<()>,
+}
+
+#[derive(Debug)]
+struct EngineInner {
+    lk_runtime: Arc<LKRuntime>,
+    session_info: Mutex<Option<SessionInfo>>, // Last/Current Sessioninfo
+    running_handle: AsyncRwLock<Option<EngineHandle>>,
+    opened: AtomicBool,
+    engine_emitter: EngineEmitter,
+
+    // Reconnecting fields
+    reconnecting: AtomicBool,
+    full_reconnect: AtomicBool,
+    reconnect_interval: Mutex<Interval>,
+}
+
+#[derive(Debug)]
+pub struct RTCEngine {
+    inner: Arc<EngineInner>,
+}
+
+impl RTCEngine {
+    pub fn new() -> (Self, EngineEvents) {
+        let mut lk_runtime = None;
+        {
+            let mut lk_runtime_ref = LK_RUNTIME.lock();
+            lk_runtime = lk_runtime_ref.upgrade();
+
+            if lk_runtime.is_none() {
+                let new_runtime = Arc::new(LKRuntime::default());
+                *lk_runtime_ref = Arc::downgrade(&new_runtime);
+                lk_runtime = Some(new_runtime);
+            }
+        }
+
+        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(),
+            opened: Default::default(),
+            engine_emitter,
+            reconnecting: Default::default(),
+            full_reconnect: Default::default(),
+            reconnect_interval: Mutex::new(interval(RECONNECT_INTERVAL)),
+        });
+
+        (Self { inner }, engine_events)
+    }
+
+    #[tracing::instrument]
+    pub async fn connect(
+        &self,
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+    ) -> EngineResult<()> {
+        self.inner.connect(url, token, options).await
+    }
+
+    #[tracing::instrument]
+    pub async fn close(&self) {
+        self.inner.close().await
+    }
+
+    #[tracing::instrument(skip(data))]
+    pub async fn publish_data(
+        &self,
+        data: &DataPacket,
+        kind: data_packet::Kind,
+    ) -> EngineResult<()> {
+        self.inner.wait_reconnection().await?;
+        self.inner
+            .running_handle
+            .read()
+            .await
+            .as_ref()
+            .unwrap()
+            .session
+            .publish_data(data, kind)
+            .await
+    }
+
+    pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
+        self.inner.wait_reconnection().await?;
+        self.inner
+            .running_handle
+            .read()
+            .await
+            .as_ref()
+            .unwrap()
+            .session
+            .simulate_scenario(scenario)
+            .await;
+        Ok(())
+    }
+
+    pub fn join_response(&self) -> Option<JoinResponse> {
+        if let Some(info) = self.inner.session_info.lock().as_ref() {
+            Some(info.join_response.clone())
+        } else {
+            None
+        }
+    }
+}
+
+impl EngineInner {
+    async fn engine_task(
+        self: Arc<Self>,
+        mut session_events: SessionEvents,
+        mut close_receiver: oneshot::Receiver<()>,
+    ) {
+        loop {
+            tokio::select! {
+                res = session_events.recv() => {
+                    if let Some(event) = res {
+                        if let Err(err) = self.on_session_event(event).await {
+                            error!("failed to handle session event: {:?}", err);
+                        }
+                    } else {
+                        panic!("rtc_sessions has been closed unexpectedly");
+                    }
+                },
+                 _ = &mut close_receiver => {
+                    break;
+                }
+            }
+        }
+    }
+
+    async fn on_session_event(self: &Arc<Self>, event: SessionEvent) -> EngineResult<()> {
+        match event {
+            SessionEvent::Close {
+                source,
+                reason,
+                can_reconnect,
+                retry_now,
+                full_reconnect,
+            } => {
+                info!("received session close: {}, {:?}", source, reason);
+                if can_reconnect {
+                    self.clone().try_reconnect(retry_now, full_reconnect);
+                } else {
+                    self.close().await;
+                }
+            }
+            SessionEvent::Data {
+                participant_sid,
+                payload,
+                kind,
+            } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::Data {
+                        participant_sid,
+                        payload,
+                        kind,
+                    })
+                    .await;
+            }
+            SessionEvent::MediaTrack {
+                track,
+                stream,
+                receiver,
+            } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::MediaTrack {
+                        track,
+                        stream,
+                        receiver,
+                    })
+                    .await;
+            }
+            SessionEvent::SpeakersChanged { speakers } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::SpeakersChanged { speakers })
+                    .await;
+            }
+            SessionEvent::ConnectionQuality { updates } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::ConnectionQuality { updates })
+                    .await;
+            }
+            SessionEvent::Connected => {}
+        }
+        Ok(())
+    }
+
+    fn connect<'a>(
+        self: &'a Arc<Self>,
+        url: &'a str,
+        token: &'a str,
+        options: SignalOptions,
+    ) -> BoxFuture<'a, EngineResult<()>> {
+        async {
+            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(())
+        }
+        .boxed()
+    }
+
+    async fn terminate_session(&self) {
+        if let Some(handle) = self.running_handle.write().await.take() {
+            handle.session.close().await;
+            let _ = handle.close_sender.send(());
+            let _ = handle.engine_task.await;
+        }
+    }
+
+    async fn close(&self) {
+        self.opened.store(false, Ordering::SeqCst);
+        self.terminate_session().await;
+        let _ = self.engine_emitter.send(EngineEvent::Disconnected).await;
+    }
+
+    // Wait for the reconnection task to finish
+    // Return directly if no open RTCSession
+    async fn wait_reconnection(&self) -> EngineResult<()> {
+        if !self.opened.load(Ordering::SeqCst) {
+            Err(EngineError::Connection("not opened".to_owned()))?
+        }
+
+        while self.reconnecting.load(Ordering::Acquire) {
+            tokio::task::yield_now().await;
+        }
+
+        if self.running_handle.read().await.is_none() {
+            Err(EngineError::Connection("reconnection failed".to_owned()))?
+        }
+
+        Ok(())
+    }
+
+    /// Start the reconnect task if not already started
+    fn try_reconnect(self: Arc<Self>, retry_now: bool, full_reconnect: bool) {
+        if !self.opened.load(Ordering::SeqCst) {
+            return;
+        }
+
+        if self.reconnecting.load(Ordering::SeqCst) {
+            if retry_now {
+                self.reconnect_interval.lock().reset();
+                self.full_reconnect.store(full_reconnect, Ordering::SeqCst);
+            }
+            return;
+        }
+
+        warn!("reconnecting RTCEngine...");
+
+        self.reconnecting.store(true, Ordering::SeqCst);
+        self.full_reconnect.store(full_reconnect, Ordering::SeqCst);
+        self.reconnect_interval.lock().reset();
+        tokio::spawn({
+            let inner = self.clone();
+            async move {
+                let res = inner.reconnect_task().await;
+                inner.reconnecting.store(false, Ordering::SeqCst);
+
+                if res.is_ok() {
+                    warn!("RTCEngine successfully reconnected")
+                } else {
+                    error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS);
+                    inner.close().await;
+                }
+            }
+        });
+    }
+
+    /// 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 reconnect_task(self: &Arc<Self>) -> EngineResult<()> {
+        for i in 0..RECONNECT_ATTEMPTS {
+            if !self.opened.load(Ordering::Acquire) {
+                // The user closed the RTCEngine, cancel the reconnection task
+                return Ok(());
+            }
+
+            if self.full_reconnect.load(Ordering::SeqCst) {
+                if i == 0 {
+                    let _ = self.engine_emitter.send(EngineEvent::Restarting).await;
+                }
+
+                info!("restarting connection... attempt: {}", i);
+                if let Err(err) = self.try_restart_connection().await {
+                    error!("restarting connection failed: {}", err);
+                } else {
+                    let _ = self.engine_emitter.send(EngineEvent::Restarted).await;
+                    return Ok(());
+                }
+            } else {
+                if i == 0 {
+                    let _ = self.engine_emitter.send(EngineEvent::Resuming).await;
+                }
+
+                info!("resuming connection... attempt: {}", i);
+                if let Err(err) = self.try_resume_connection().await {
+                    error!("resuming connection failed: {}", err);
+                    if let EngineError::Signal(_) = err {
+                        self.full_reconnect.store(true, Ordering::SeqCst);
+                    }
+                } else {
+                    let _ = self.engine_emitter.send(EngineEvent::Resumed).await;
+                    return Ok(());
+                }
+            }
+
+            self.reconnect_interval.lock().tick().await;
+        }
+
+        Err(EngineError::Connection("failed to reconnect".to_owned()))
+    }
+
+    /// Try to recover the connection by doing a full reconnect.
+    /// It recreates a new RTCSession
+    async fn try_restart_connection(self: &Arc<Self>) -> EngineResult<()> {
+        let info = self.session_info.lock().clone().unwrap();
+        self.terminate_session().await;
+        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<()> {
+        let handle = self.running_handle.read().await;
+        handle.as_ref().unwrap().session.restart().await?;
+        handle.as_ref().unwrap().session.wait_pc_connection().await
+    }
+}

livekit/src/rtc_engine/pc_transport.rs

@@ -0,0 +1,178 @@
+use std::fmt::{Debug, Formatter};
+use std::future::Future;
+use std::pin::Pin;
+use std::time::Duration;
+
+use tracing::{event, Level};
+
+use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
+use livekit_webrtc::peer_connection::{
+    IceConnectionState, PeerConnection, RTCOfferAnswerOptions, SignalingState,
+};
+use livekit_webrtc::rtc_error::RTCError;
+
+use crate::proto::SignalTarget;
+
+const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150);
+
+pub type OnOfferHandler = Box<
+    dyn (FnMut(SessionDescription) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>)
+        + Send
+        + Sync,
+>;
+
+pub struct PCTransport {
+    signal_target: SignalTarget,
+    peer_connection: PeerConnection,
+    pending_candidates: Vec<IceCandidate>,
+    on_offer_handler: Option<OnOfferHandler>,
+    renegotiate: bool,
+    restarting_ice: bool,
+}
+
+impl Debug for PCTransport {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        f.write_str("PCTransport")
+    }
+}
+
+impl PCTransport {
+    pub fn new(peer_connection: PeerConnection, signal_target: SignalTarget) -> Self {
+        Self {
+            signal_target,
+            peer_connection,
+            pending_candidates: Vec::default(),
+            on_offer_handler: None,
+            restarting_ice: false,
+            renegotiate: false,
+        }
+    }
+
+    pub fn is_connected(&self) -> bool {
+        self.peer_connection.ice_connection_state() == IceConnectionState::IceConnectionConnected
+            || self.peer_connection.ice_connection_state()
+                == IceConnectionState::IceConnectionCompleted
+    }
+
+    pub fn peer_connection(&mut self) -> &mut PeerConnection {
+        &mut self.peer_connection
+    }
+
+    pub fn signal_target(&self) -> SignalTarget {
+        self.signal_target.clone()
+    }
+
+    pub fn on_offer(&mut self, handler: OnOfferHandler) {
+        self.on_offer_handler = Some(handler);
+    }
+
+    pub fn prepare_ice_restart(&mut self) {
+        self.restarting_ice = true;
+    }
+
+    pub fn close(&mut self) {
+        self.peer_connection.close();
+    }
+
+    #[tracing::instrument(level = Level::DEBUG)]
+    pub async fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) -> Result<(), RTCError> {
+        if self.peer_connection.remote_description().is_some() && !self.restarting_ice {
+            self.peer_connection
+                .add_ice_candidate(ice_candidate)
+                .await?;
+
+            return Ok(());
+        }
+
+        self.pending_candidates.push(ice_candidate);
+        Ok(())
+    }
+
+    #[tracing::instrument(level = Level::DEBUG)]
+    pub async fn set_remote_description(
+        &mut self,
+        remote_description: SessionDescription,
+    ) -> Result<(), RTCError> {
+        self.peer_connection
+            .set_remote_description(remote_description)
+            .await?;
+
+        for ic in self.pending_candidates.drain(..) {
+            self.peer_connection.add_ice_candidate(ic).await?;
+        }
+        self.restarting_ice = false;
+
+        if self.renegotiate {
+            self.renegotiate = false;
+            self.create_and_send_offer(RTCOfferAnswerOptions::default())
+                .await?;
+        }
+
+        Ok(())
+    }
+
+    #[tracing::instrument(level = Level::DEBUG)]
+    pub async fn negotiate(&mut self) -> Result<(), RTCError> {
+        // TODO(theomonnom) Debounce here with NEGOTIATION_FREQUENCY
+        self.create_and_send_offer(RTCOfferAnswerOptions::default())
+            .await
+    }
+
+    #[tracing::instrument(level = Level::DEBUG)]
+    pub async fn create_anwser(
+        &mut self,
+        offer: SessionDescription,
+        options: RTCOfferAnswerOptions,
+    ) -> Result<SessionDescription, RTCError> {
+        self.set_remote_description(offer).await?;
+        let answer = self
+            .peer_connection()
+            .create_answer(RTCOfferAnswerOptions::default())
+            .await?;
+        self.peer_connection()
+            .set_local_description(answer.clone())
+            .await?;
+
+        Ok(answer)
+    }
+
+    #[tracing::instrument(level = Level::DEBUG)]
+    pub async fn create_and_send_offer(
+        &mut self,
+        options: RTCOfferAnswerOptions,
+    ) -> Result<(), RTCError> {
+        if self.on_offer_handler.is_none() {
+            return Ok(());
+        }
+
+        if options.ice_restart {
+            event!(Level::TRACE, "restarting ICE");
+            self.restarting_ice = true;
+        }
+
+        if self.peer_connection.signaling_state() == SignalingState::HaveLocalOffer {
+            if options.ice_restart {
+                if let Some(remote_description) = self.peer_connection.remote_description() {
+                    self.peer_connection
+                        .set_remote_description(remote_description)
+                        .await?;
+                } else {
+                    event!(
+                        Level::ERROR,
+                        "trying to restart ICE when the pc doesn't have remote description"
+                    );
+                }
+            } else {
+                self.renegotiate = true;
+                return Ok(());
+            }
+        }
+
+        let offer = self.peer_connection.create_offer(options).await?;
+        self.peer_connection
+            .set_local_description(offer.clone())
+            .await?;
+        self.on_offer_handler.as_mut().unwrap()(offer).await;
+        Ok(())
+    }
+}

livekit/src/rtc_engine/rtc_events.rs

@@ -0,0 +1,145 @@
+use livekit_webrtc::data_channel::{DataChannel, OnMessageHandler};
+use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
+use livekit_webrtc::media_stream::MediaStream;
+use livekit_webrtc::peer_connection::{
+    OnAddTrackHandler, OnConnectionChangeHandler, OnDataChannelHandler, OnIceCandidateErrorHandler,
+    OnIceCandidateHandler, PeerConnectionState,
+};
+use livekit_webrtc::rtp_receiver::RtpReceiver;
+use tokio::sync::mpsc;
+use tracing::error;
+
+use crate::proto::SignalTarget;
+use crate::rtc_engine::pc_transport::OnOfferHandler;
+
+use super::pc_transport::PCTransport;
+
+pub type RTCEmitter = mpsc::UnboundedSender<RTCEvent>;
+pub type RTCEvents = mpsc::UnboundedReceiver<RTCEvent>;
+
+#[derive(Debug)]
+pub enum RTCEvent {
+    IceCandidate {
+        ice_candidate: IceCandidate,
+        target: SignalTarget,
+    },
+    ConnectionChange {
+        state: PeerConnectionState,
+        target: SignalTarget,
+    },
+    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>,
+        binary: bool,
+    },
+}
+
+/// Handlers used to forward events to a channel
+/// Every callback here is called on the signaling thread
+
+fn on_connection_change(target: SignalTarget, emitter: RTCEmitter) -> OnConnectionChangeHandler {
+    Box::new(move |state| {
+        let _ = emitter.send(RTCEvent::ConnectionChange { state, target });
+    })
+}
+
+fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler {
+    Box::new(move |ice_candidate| {
+        let _ = emitter.send(RTCEvent::IceCandidate {
+            ice_candidate,
+            target,
+        });
+    })
+}
+
+fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler {
+    Box::new(move |offer| {
+        let _ = emitter.send(RTCEvent::Offer { offer, target });
+
+        Box::pin(async {})
+    })
+}
+
+fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler {
+    Box::new(move |mut data_channel| {
+        data_channel.on_message(on_message(emitter.clone()));
+
+        let _ = emitter.send(RTCEvent::DataChannel {
+            data_channel,
+            target,
+        });
+    })
+}
+
+fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler {
+    Box::new(move |rtp_receiver, streams| {
+        let _ = emitter.send(RTCEvent::AddTrack {
+            rtp_receiver,
+            streams,
+            target,
+        });
+    })
+}
+
+fn on_ice_candidate_error(
+    target: SignalTarget,
+    _emitter: RTCEmitter,
+) -> OnIceCandidateErrorHandler {
+    Box::new(move |address, port, url, error_code, error_text| {
+        error!(
+            "ICE candidate error ({:?}): address: {} - port: {} - url: {} - error_code: {} - error_text: {}",
+            target, address, port, url, error_code, error_text
+        );
+    })
+}
+
+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(signal_target, rtc_emitter.clone()));
+
+    transport
+        .peer_connection()
+        .on_data_channel(on_data_channel(signal_target, rtc_emitter.clone()));
+
+    transport
+        .peer_connection()
+        .on_add_track(on_add_track(signal_target, rtc_emitter.clone()));
+
+    transport
+        .peer_connection()
+        .on_connection_change(on_connection_change(signal_target, rtc_emitter.clone()));
+
+    transport
+        .peer_connection()
+        .on_ice_candidate_error(on_ice_candidate_error(signal_target, rtc_emitter.clone()));
+
+    transport.on_offer(on_offer(transport.signal_target(), rtc_emitter.clone()));
+}
+
+fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
+    Box::new(move |data, binary| {
+        let _ = emitter.send(RTCEvent::Data {
+            data: data.to_vec(),
+            binary,
+        });
+    })
+}
+
+pub fn forward_dc_events(dc: &mut DataChannel, rtc_emitter: RTCEmitter) {
+    dc.on_message(on_message(rtc_emitter.clone()));
+}

livekit/src/rtc_engine/rtc_session.rs

@@ -0,0 +1,742 @@
+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;
+use tokio::task::JoinHandle;
+
+use tokio::sync::{mpsc, watch, Mutex as AsyncMutex};
+use tokio::time::sleep;
+
+use prost::Message;
+use serde::{Deserialize, Serialize};
+use tracing::{debug, error, trace, warn};
+
+use crate::{proto, signal_client};
+use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataState};
+use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
+use livekit_webrtc::peer_connection::{
+    IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
+};
+use livekit_webrtc::peer_connection_factory::RTCConfiguration;
+
+use crate::proto::data_packet::Value;
+use crate::proto::{
+    data_packet, signal_request, signal_response, CandidateProtocol, DataPacket, DisconnectReason,
+    JoinResponse, SignalTarget, TrickleRequest,
+};
+use crate::rtc_engine::lk_runtime::LKRuntime;
+use crate::rtc_engine::pc_transport::PCTransport;
+use crate::rtc_engine::rtc_events::{RTCEvent, RTCEvents};
+use crate::signal_client::{SignalClient, SignalEvent, SignalEvents, SignalOptions};
+
+use super::{rtc_events, EngineError, EngineResult, SimulateScenario};
+
+pub const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+pub const LOSSY_DC_LABEL: &str = "_lossy";
+pub const RELIABLE_DC_LABEL: &str = "_reliable";
+
+pub type SessionEmitter = mpsc::UnboundedSender<SessionEvent>;
+pub type SessionEvents = mpsc::UnboundedReceiver<SessionEvent>;
+
+#[derive(Debug)]
+pub enum SessionEvent {
+    Data {
+        participant_sid: String,
+        payload: Vec<u8>,
+        kind: proto::data_packet::Kind,
+    },
+    MediaTrack {
+        track: MediaStreamTrackHandle,
+        stream: MediaStream,
+        receiver: RtpReceiver,
+    },
+    SpeakersChanged {
+        speakers: Vec<proto::SpeakerInfo>,
+    },
+    ConnectionQuality {
+        updates: Vec<proto::ConnectionQualityInfo>,
+    },
+    // TODO(theomonnom): Move entirely the reconnection logic on mod.rs
+    Close {
+        source: String,
+        reason: DisconnectReason,
+        can_reconnect: bool,
+        full_reconnect: bool,
+        retry_now: bool,
+    },
+    Connected,
+}
+
+#[repr(u8)]
+pub enum PCState {
+    New,
+    Connected,
+    Disconnected,
+    Reconnecting,
+    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 {
+    sdpMid: String,
+    sdpMLineIndex: i32,
+    candidate: String,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct SessionInfo {
+    pub url: String,
+    pub token: String,
+    pub options: SignalOptions,
+    pub join_response: JoinResponse,
+}
+
+/// Fields shared with rtc_task and signal_task
+#[derive(Debug)]
+struct SessionInner {
+    info: SessionInfo,
+    signal_client: Arc<SignalClient>,
+    pc_state: AtomicU8, // PCState
+    has_published: AtomicBool,
+
+    publisher_pc: AsyncMutex<PCTransport>,
+    subscriber_pc: AsyncMutex<PCTransport>,
+
+    // Publisher data channels
+    // used to send data to other participants ( The SFU forwards the messages )
+    lossy_dc: DataChannel,
+    reliable_dc: DataChannel,
+
+    // Keep a strong reference to the subscriber datachannels,
+    // so we can receive data from other participants
+    subscriber_dc: Mutex<Vec<DataChannel>>,
+
+    emitter: SessionEmitter,
+}
+/// This struct holds a WebRTC session
+/// The session changes at every reconnection
+///
+/// RTCSession is also responsable for the signaling and the negotation
+#[derive(Debug)]
+pub struct RTCSession {
+    lk_runtime: Arc<LKRuntime>,
+    inner: Arc<SessionInner>,
+    close_emitter: watch::Sender<bool>, // false = is_running
+    signal_task: JoinHandle<()>,
+    rtc_task: JoinHandle<()>,
+}
+
+impl RTCSession {
+    pub async fn connect(
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+        lk_runtime: Arc<LKRuntime>,
+        session_emitter: SessionEmitter,
+    ) -> EngineResult<Self> {
+        // Connect to the SignalClient
+        let (signal_client, mut signal_events) = SignalClient::new();
+        let signal_client = Arc::new(signal_client);
+        signal_client.connect(url, token, options.clone()).await?;
+        let join_response = signal_client::utils::next_join_response(&mut signal_events).await?;
+        debug!("received JoinResponse: {:?}", join_response);
+
+        let (rtc_emitter, rtc_events) = mpsc::unbounded_channel();
+        let rtc_config = RTCConfiguration::from(join_response.clone());
+
+        let mut publisher_pc = PCTransport::new(
+            lk_runtime
+                .pc_factory
+                .create_peer_connection(rtc_config.clone())?,
+            SignalTarget::Publisher,
+        );
+
+        let mut subscriber_pc = PCTransport::new(
+            lk_runtime
+                .pc_factory
+                .create_peer_connection(rtc_config.clone())?,
+            SignalTarget::Subscriber,
+        );
+
+        let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(
+            LOSSY_DC_LABEL,
+            DataChannelInit {
+                ordered: true,
+                max_retransmits: Some(0),
+                ..DataChannelInit::default()
+            },
+        )?;
+
+        let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(
+            RELIABLE_DC_LABEL,
+            DataChannelInit {
+                ordered: true,
+                ..DataChannelInit::default()
+            },
+        )?;
+
+        // Forward events received in the Signaling Thread to our rtc channel
+        rtc_events::forward_pc_events(&mut publisher_pc, rtc_emitter.clone());
+        rtc_events::forward_pc_events(&mut subscriber_pc, rtc_emitter.clone());
+        rtc_events::forward_dc_events(&mut lossy_dc, rtc_emitter.clone());
+        rtc_events::forward_dc_events(&mut reliable_dc, rtc_emitter.clone());
+
+        let session_info = SessionInfo {
+            url: url.to_owned(),
+            token: token.to_owned(),
+            options,
+            join_response,
+        };
+
+        let (close_emitter, close_receiver) = watch::channel(false);
+        let inner = Arc::new(SessionInner {
+            info: session_info,
+            pc_state: AtomicU8::new(PCState::New as u8),
+            has_published: Default::default(),
+            signal_client,
+            publisher_pc: AsyncMutex::new(publisher_pc),
+            subscriber_pc: AsyncMutex::new(subscriber_pc),
+            lossy_dc,
+            reliable_dc,
+            subscriber_dc: Default::default(),
+            emitter: session_emitter,
+        });
+
+        // Start session tasks
+        let signal_task = tokio::spawn(
+            inner
+                .clone()
+                .signal_task(signal_events, close_receiver.clone()),
+        );
+        let rtc_task = tokio::spawn(inner.clone().rtc_task(rtc_events, close_receiver.clone()));
+
+        if !inner.info.join_response.subscriber_primary {
+            inner.negotiate_publisher().await?;
+        }
+
+        let session = Self {
+            lk_runtime,
+            inner: inner.clone(),
+            close_emitter,
+            signal_task,
+            rtc_task,
+        };
+
+        Ok(session)
+    }
+
+    /// Close the PeerConnections and the SignalClient
+    #[tracing::instrument]
+    pub async fn close(self) {
+        // Close the tasks
+        let _ = self.close_emitter.send(true);
+        let _ = self.rtc_task.await;
+        let _ = self.signal_task.await;
+        self.inner.close().await;
+    }
+
+    pub async fn publish_data(
+        &self,
+        data: &DataPacket,
+        kind: data_packet::Kind,
+    ) -> Result<(), EngineError> {
+        self.inner.publish_data(data, kind).await
+    }
+
+    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
+    }
+
+    pub async fn simulate_scenario(&self, scenario: SimulateScenario) {
+        self.inner.simulate_scenario(scenario).await
+    }
+}
+
+impl RTCSession {
+    pub fn info(&self) -> &SessionInfo {
+        &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
+    }
+
+    pub fn subscriber(&self) -> &AsyncMutex<PCTransport> {
+        &self.inner.subscriber_pc
+    }
+
+    pub fn signal_client(&self) -> &Arc<SignalClient> {
+        &self.inner.signal_client
+    }
+
+    pub fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel {
+        &self.inner.data_channel(kind)
+    }
+}
+
+impl SessionInner {
+    async fn rtc_task(
+        self: Arc<Self>,
+        mut rtc_events: RTCEvents,
+        mut close_receiver: watch::Receiver<bool>,
+    ) {
+        loop {
+            tokio::select! {
+                res = rtc_events.recv() => {
+                    if let Some(event) = res {
+                        if let Err(err) = self.on_rtc_event(event).await {
+                            error!("failed to handle rtc event: {:?}", err);
+                        }
+                    } else {
+                        panic!("rtc_events has been closed unexpectedly");
+                    }
+                },
+                 _ = close_receiver.changed() => {
+                    break;
+                }
+            }
+        }
+    }
+
+    async fn signal_task(
+        self: Arc<Self>,
+        mut signal_events: SignalEvents,
+        mut close_receiver: watch::Receiver<bool>,
+    ) {
+        loop {
+            tokio::select! {
+                res = signal_events.recv() => {
+                    if let Some(signal) = res {
+                        match signal {
+                            SignalEvent::Open => {}
+                            SignalEvent::Signal(signal) => {
+                                if let Err(err) = self.on_signal_event(signal).await {
+                                    error!("failed to handle signal: {:?}", err);
+                                }
+                            }
+                            SignalEvent::Close => {
+                                self.on_session_disconnected("SignalClient closed", DisconnectReason::UnknownReason, true, false, false);
+                            }
+                        }
+                    } else {
+                        panic!("signal_events has been closed unexpectedly");
+                    }
+
+                },
+                _ = close_receiver.changed() => {
+                    break;
+                }
+            }
+        }
+    }
+
+    async fn on_signal_event(&self, event: signal_response::Message) -> EngineResult<()> {
+        match event {
+            signal_response::Message::Answer(answer) => {
+                trace!("received publisher answer: {:?}", answer);
+                let answer = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
+                self.publisher_pc
+                    .lock()
+                    .await
+                    .set_remote_description(answer)
+                    .await?;
+            }
+            signal_response::Message::Offer(offer) => {
+                trace!("received subscriber offer: {:?}", offer);
+                let offer = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
+                let answer = self
+                    .subscriber_pc
+                    .lock()
+                    .await
+                    .create_anwser(offer, RTCOfferAnswerOptions::default())
+                    .await?;
+
+                self.signal_client
+                    .send(signal_request::Message::Answer(proto::SessionDescription {
+                        r#type: "answer".to_string(),
+                        sdp: answer.to_string(),
+                    }))
+                    .await;
+            }
+            signal_response::Message::Trickle(trickle) => {
+                let target = SignalTarget::from_i32(trickle.target).unwrap();
+                let ice_candidate = {
+                    let json = serde_json::from_str::<IceCandidateJSON>(&trickle.candidate_init)?;
+                    IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?
+                };
+
+                trace!("received ice_candidate {:?} {:?}", target, ice_candidate);
+
+                if target == SignalTarget::Publisher {
+                    self.publisher_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice_candidate)
+                        .await?;
+                } else {
+                    self.subscriber_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice_candidate)
+                        .await?;
+                }
+            }
+            signal_response::Message::Leave(leave) => {
+                self.on_session_disconnected(
+                    "received leave",
+                    leave.reason(),
+                    leave.can_reconnect,
+                    true,
+                    true,
+                );
+            }
+            signal_response::Message::SpeakersChanged(speaker) => {
+                let _ = self.emitter.send(SessionEvent::SpeakersChanged {
+                    speakers: speaker.speakers,
+                });
+            }
+            signal_response::Message::ConnectionQuality(quality) => {
+                let _ = self.emitter.send(SessionEvent::ConnectionQuality {
+                    updates: quality.updates,
+                });
+            }
+            _ => {}
+        }
+
+        Ok(())
+    }
+
+    async fn on_rtc_event(&self, event: RTCEvent) -> EngineResult<()> {
+        match event {
+            RTCEvent::IceCandidate {
+                ice_candidate,
+                target,
+            } => {
+                self.signal_client
+                    .send(signal_request::Message::Trickle(TrickleRequest {
+                        candidate_init: serde_json::to_string(&IceCandidateJSON {
+                            sdpMid: ice_candidate.sdp_mid(),
+                            sdpMLineIndex: ice_candidate.sdp_mline_index(),
+                            candidate: ice_candidate.candidate(),
+                        })?,
+                        target: target as i32,
+                    }))
+                    .await;
+            }
+            RTCEvent::ConnectionChange { state, target } => {
+                trace!("connection change, {:?} {:?}", state, target);
+                let is_primary = self.info.join_response.subscriber_primary
+                    && target == SignalTarget::Subscriber;
+
+                if is_primary && state == PeerConnectionState::Connected {
+                    let old_state = self
+                        .pc_state
+                        .swap(PCState::Connected as u8, Ordering::SeqCst);
+                    if old_state == PCState::New as u8 {
+                        let _ = self.emitter.send(SessionEvent::Connected);
+                    }
+                } else if state == PeerConnectionState::Failed {
+                    self.pc_state
+                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
+
+                    self.on_session_disconnected(
+                        "pc_state failed",
+                        DisconnectReason::UnknownReason,
+                        true,
+                        false,
+                        false,
+                    );
+                }
+            }
+            RTCEvent::DataChannel {
+                data_channel,
+                target: _,
+            } => {
+                self.subscriber_dc.lock().push(data_channel);
+            }
+            RTCEvent::Offer { offer, target: _ } => {
+                // Send the publisher offer to the server
+                self.signal_client
+                    .send(signal_request::Message::Offer(proto::SessionDescription {
+                        r#type: "offer".to_string(),
+                        sdp: offer.to_string(),
+                    }))
+                    .await;
+            }
+            RTCEvent::AddTrack {
+                rtp_receiver,
+                mut streams,
+                target: _,
+            } => {
+                if !streams.is_empty() {
+                    let _ = self.emitter.send(SessionEvent::MediaTrack {
+                        track: rtp_receiver.track(),
+                        stream: streams.remove(0),
+                        receiver: rtp_receiver,
+                    });
+                } else {
+                    warn!("AddTrack event with no streams");
+                }
+            }
+            RTCEvent::Data { data, binary } => {
+                if !binary {
+                    Err(EngineError::Internal(
+                        "text messages aren't supported".to_string(),
+                    ))?;
+                }
+
+                let data = DataPacket::decode(&*data)?;
+                match data.value.unwrap() {
+                    Value::User(user) => {
+                        let _ = self.emitter.send(SessionEvent::Data {
+                            participant_sid: user.participant_sid,
+                            payload: user.payload,
+                            kind: data_packet::Kind::from_i32(data.kind).unwrap(),
+                        });
+                    }
+                    Value::Speaker(_) => {}
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Called when the SignalClient or one of the PeerConnection has lost the connection
+    /// The RTCEngine may try a reconnect.
+    fn on_session_disconnected(
+        &self,
+        source: &str,
+        reason: DisconnectReason,
+        can_reconnect: bool,
+        retry_now: bool,
+        full_reconnect: bool,
+    ) {
+        let _ = self.emitter.send(SessionEvent::Close {
+            source: source.to_owned(),
+            reason,
+            can_reconnect,
+            retry_now,
+            full_reconnect,
+        });
+    }
+
+    #[tracing::instrument]
+    async fn close(&self) {
+        self.signal_client.close().await;
+        self.publisher_pc.lock().await.close();
+        self.subscriber_pc.lock().await.close();
+    }
+
+    #[tracing::instrument]
+    async fn simulate_scenario(&self, scenario: SimulateScenario) {
+        match scenario {
+            SimulateScenario::SignalReconnect => {
+                self.signal_client.close().await;
+            }
+            SimulateScenario::Speaker => {
+                self.signal_client
+                    .send(signal_request::Message::Simulate(proto::SimulateScenario {
+                        scenario: Some(proto::simulate_scenario::Scenario::SpeakerUpdate(3)),
+                    }))
+                    .await;
+            }
+            SimulateScenario::NodeFailure => {
+                self.signal_client
+                    .send(signal_request::Message::Simulate(proto::SimulateScenario {
+                        scenario: Some(proto::simulate_scenario::Scenario::NodeFailure(true)),
+                    }))
+                    .await;
+            }
+            SimulateScenario::ServerLeave => {
+                self.signal_client
+                    .send(signal_request::Message::Simulate(proto::SimulateScenario {
+                        scenario: Some(proto::simulate_scenario::Scenario::ServerLeave(true)),
+                    }))
+                    .await;
+            }
+            SimulateScenario::Migration => {
+                self.signal_client
+                    .send(signal_request::Message::Simulate(proto::SimulateScenario {
+                        scenario: Some(proto::simulate_scenario::Scenario::Migration(true)),
+                    }))
+                    .await;
+            }
+            SimulateScenario::ForceTcp => {
+                self.signal_client
+                    .send(signal_request::Message::Simulate(proto::SimulateScenario {
+                        scenario: Some(
+                            proto::simulate_scenario::Scenario::SwitchCandidateProtocol(
+                                CandidateProtocol::Tcp as i32,
+                            ),
+                        ),
+                    }))
+                    .await;
+            }
+            SimulateScenario::ForceTls => {
+                self.signal_client
+                    .send(signal_request::Message::Simulate(proto::SimulateScenario {
+                        scenario: Some(
+                            proto::simulate_scenario::Scenario::SwitchCandidateProtocol(
+                                CandidateProtocol::Tls as i32,
+                            ),
+                        ),
+                    }))
+                    .await;
+            }
+        }
+    }
+
+    #[tracing::instrument(skip(data))]
+    async fn publish_data(
+        &self,
+        data: &DataPacket,
+        kind: data_packet::Kind,
+    ) -> Result<(), EngineError> {
+        self.ensure_publisher_connected(kind).await?;
+        self.data_channel(kind)
+            .send(&data.encode_to_vec(), true)
+            .map_err(Into::into)
+    }
+
+    /// Try to restart the session by doing an ICE Restart (The SignalClient is also restarted)
+    /// This reconnection if more seemless than the full reconnection implemented in ['RTCEngine']
+    async fn restart_session(&self) -> EngineResult<()> {
+        self.signal_client.close().await;
+
+        let mut options = self.info.options.clone();
+        options.sid = self.info.join_response.participant.clone().unwrap().sid;
+        options.reconnect = true;
+
+        self.signal_client
+            .connect(&self.info.url, &self.info.token, options)
+            .await?;
+
+        self.subscriber_pc.lock().await.prepare_ice_restart();
+
+        if self.has_published.load(Ordering::Acquire) {
+            self.publisher_pc
+                .lock()
+                .await
+                .create_and_send_offer(RTCOfferAnswerOptions {
+                    ice_restart: true,
+                    ..Default::default()
+                })
+                .await?;
+        }
+
+        self.wait_pc_connection().await?;
+        self.signal_client.flush_queue().await;
+
+        Ok(())
+    }
+
+    // Wait for PCState to become PCState::Connected
+    // Timeout after ['MAX_ICE_CONNECT_TIMEOUT']
+    async fn wait_pc_connection(&self) -> EngineResult<()> {
+        let wait_connected = async move {
+            while self.pc_state.load(Ordering::Acquire) != PCState::Connected as u8 {
+                tokio::task::yield_now().await;
+            }
+        };
+
+        tokio::select! {
+            _ = wait_connected => Ok(()),
+            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
+                let err = EngineError::Connection("wait_pc_connection timed out".to_string());
+                Err(err)
+            }
+        }
+    }
+
+    /// Start publisher negotiation
+    async fn negotiate_publisher(&self) -> EngineResult<()> {
+        self.has_published.store(true, Ordering::Release);
+        let res = self.publisher_pc.lock().await.negotiate().await;
+        if let Err(err) = &res {
+            error!("failed to negotiate the publisher: {:?}", err);
+        }
+        res.map_err(Into::into)
+    }
+
+    /// Ensure the Publisher PC is connected, if not, start the negotiation
+    /// This is required when sending data to the server
+    async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> {
+        if !self.info.join_response.subscriber_primary {
+            return Ok(());
+        }
+
+        if !self.publisher_pc.lock().await.is_connected()
+            && self
+                .publisher_pc
+                .lock()
+                .await
+                .peer_connection()
+                .ice_connection_state()
+                != IceConnectionState::IceConnectionChecking
+        {
+            let _ = self.negotiate_publisher().await;
+        }
+
+        let dc = self.data_channel(kind);
+        if dc.state() == DataState::Open {
+            return Ok(());
+        }
+
+        // Wait until the PeerConnection is connected
+        let wait_connected = async {
+            while self.publisher_pc.lock().await.is_connected() && dc.state() == DataState::Open {
+                tokio::task::yield_now().await;
+            }
+        };
+
+        // TODO(theomonnom) Avoid 15 seconds deadlock on the RTCEngine by recv close here
+        tokio::select! {
+            _ = wait_connected => Ok(()),
+            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
+                let err = EngineError::Connection("could not establish publisher connection: timeout".to_string());
+                error!(error = ?err);
+                Err(err)
+            }
+        }
+    }
+
+    fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel {
+        if kind == data_packet::Kind::Reliable {
+            &self.reliable_dc
+        } else {
+            &self.lossy_dc
+        }
+    }
+}

livekit/src/signal_client/mod.rs

@@ -0,0 +1,179 @@
+use std::fmt::Debug;
+
+use std::time::Duration;
+
+use livekit_webrtc::peer_connection_factory::{
+    ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
+};
+use parking_lot::RwLock;
+use thiserror::Error;
+use tokio::sync::mpsc;
+use tokio_tungstenite::tungstenite::Error as WsError;
+
+use crate::proto::{signal_request, signal_response, JoinResponse};
+use crate::signal_client::signal_stream::SignalStream;
+use tracing::{instrument, Level};
+
+mod signal_stream;
+
+pub(crate) type SignalEmitter = mpsc::Sender<SignalEvent>;
+pub(crate) type SignalEvents = mpsc::Receiver<SignalEvent>;
+pub(crate) type SignalResult<T> = Result<T, SignalError>;
+
+pub const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[derive(Error, Debug)]
+pub enum SignalError {
+    #[error("ws failure: {0}")]
+    WsError(#[from] WsError),
+    #[error("failed to parse the url")]
+    UrlParse(#[from] url::ParseError),
+    #[error("failed to decode messages from server")]
+    ProtoParse(#[from] prost::DecodeError),
+    #[error("{0}")]
+    Timeout(String),
+}
+
+/// Events used by the RTCEngine who will handle the reconnection logic
+#[derive(Debug)]
+pub enum SignalEvent {
+    Open,
+    Signal(signal_response::Message),
+    Close,
+}
+
+#[derive(Debug, Clone)]
+pub struct SignalOptions {
+    pub(crate) reconnect: bool,
+    pub(crate) sid: String,
+    pub auto_subscribe: bool,
+    pub adaptive_stream: bool,
+}
+
+impl Default for SignalOptions {
+    fn default() -> Self {
+        Self {
+            reconnect: false,
+            auto_subscribe: true,
+            sid: "".to_string(),
+            adaptive_stream: false,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct SignalClient {
+    stream: RwLock<Option<SignalStream>>,
+    emitter: SignalEmitter,
+}
+
+impl SignalClient {
+    pub fn new() -> (Self, SignalEvents) {
+        let (emitter, events) = mpsc::channel(8);
+        (
+            Self {
+                stream: Default::default(),
+                emitter,
+            },
+            events,
+        )
+    }
+
+    #[instrument(level = Level::DEBUG, skip(url, token, options))]
+    pub async fn connect(
+        &self,
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+    ) -> SignalResult<()> {
+        let stream = SignalStream::connect(url, token, options, self.emitter.clone()).await?;
+        *self.stream.write() = Some(stream);
+        Ok(())
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    pub async fn close(&self) {
+        if let Some(stream) = self.stream.write().take() {
+            stream.close().await;
+        }
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    pub async fn send(&self, signal: signal_request::Message) {
+        if let Some(stream) = self.stream.read().as_ref() {
+            if stream.send(signal).await.is_ok() {
+                return;
+            }
+        }
+
+        // TODO(theomonnom): enqueue message
+    }
+
+    pub async fn clear_queue(&self) {
+        // TODO(theomonnom): impl
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    pub async fn flush_queue(&self) {
+        // TODO(theomonnom): impl
+    }
+}
+
+impl From<JoinResponse> for RTCConfiguration {
+    fn from(join_response: JoinResponse) -> Self {
+        Self {
+            ice_servers: {
+                let mut servers = vec![];
+                for ice_server in join_response.ice_servers.clone() {
+                    servers.push(ICEServer {
+                        urls: ice_server.urls,
+                        username: ice_server.username,
+                        password: ice_server.credential,
+                    })
+                }
+                servers
+            },
+            continual_gathering_policy: ContinualGatheringPolicy::GatherContinually,
+            ice_transport_type: IceTransportsType::All,
+        }
+    }
+}
+
+pub mod utils {
+    use crate::proto::{signal_response, JoinResponse};
+    use crate::signal_client::{SignalError, SignalEvent, SignalResult, JOIN_RESPONSE_TIMEOUT};
+    use tokio::time::timeout;
+    use tokio_tungstenite::tungstenite::Error as WsError;
+    use tracing::{event, instrument, Level};
+
+    use super::SignalEvents;
+
+    #[instrument(level = Level::DEBUG, skip(receiver))]
+    pub(crate) async fn next_join_response(
+        receiver: &mut SignalEvents,
+    ) -> SignalResult<JoinResponse> {
+        let join = async {
+            while let Some(event) = receiver.recv().await {
+                match event {
+                    SignalEvent::Signal(signal_response::Message::Join(join)) => return Ok(join),
+                    SignalEvent::Close => break,
+                    SignalEvent::Open => continue,
+                    _ => {
+                        event!(
+                            Level::WARN,
+                            "received unexpected message while waiting for JoinResponse: {:?}",
+                            event
+                        );
+                        continue;
+                    }
+                }
+            }
+
+            Err(WsError::ConnectionClosed)?
+        };
+
+        timeout(JOIN_RESPONSE_TIMEOUT, join)
+            .await
+            .map_err(|_| SignalError::Timeout("failed to receive JoinResponse".to_string()))?
+    }
+}

livekit/src/signal_client/signal_stream.rs

@@ -0,0 +1,209 @@
+use futures_util::stream::{SplitSink, SplitStream};
+use futures_util::{SinkExt, StreamExt};
+use prost::Message as ProstMessage;
+use tokio::net::TcpStream;
+use tokio::sync::{mpsc, oneshot};
+use tokio::task::JoinHandle;
+
+use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
+use tokio_tungstenite::tungstenite::protocol::CloseFrame;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
+use tracing::{event, Level};
+
+use crate::proto::{signal_request, SignalRequest, SignalResponse};
+use crate::signal_client::{SignalEmitter, SignalEvent, SignalOptions, SignalResult};
+
+pub const PROTOCOL_VERSION: u32 = 8;
+
+type WebSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
+
+#[derive(Debug)]
+enum InternalMessage {
+    Signal {
+        signal: signal_request::Message,
+        response_chn: oneshot::Sender<SignalResult<()>>,
+    },
+    Pong {
+        ping_data: Vec<u8>,
+    },
+    Close {
+        close_frame: Option<CloseFrame<'static>>,
+    },
+}
+
+/// SignalStream hold the WebSocket connection
+///
+/// It is replaced by [SignalClient] at each reconnection.
+#[derive(Debug)]
+pub(crate) struct SignalStream {
+    internal_tx: mpsc::Sender<InternalMessage>,
+    read_handle: JoinHandle<()>,
+    write_handle: JoinHandle<()>,
+}
+
+impl SignalStream {
+    /// Connect to livekit websocket.
+    /// Return SignalError if the connections failed
+    ///
+    /// SignalStream will never try to reconnect if the connection has been
+    /// closed.
+    pub(super) async fn connect(
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+        emitter: SignalEmitter,
+    ) -> SignalResult<Self> {
+        let mut lk_url = url::Url::parse(url)?;
+        lk_url.set_path("/rtc");
+        lk_url
+            .query_pairs_mut()
+            .append_pair("access_token", token)
+            .append_pair("protocol", PROTOCOL_VERSION.to_string().as_str())
+            .append_pair("reconnect", if options.reconnect { "1" } else { "0" })
+            .append_pair("sid", &options.sid)
+            .append_pair(
+                "auto_subscribe",
+                if options.auto_subscribe { "1" } else { "0" },
+            )
+            .append_pair(
+                "adaptive_stream",
+                if options.adaptive_stream { "1" } else { "0" },
+            );
+
+        event!(Level::INFO, "connecting to SignalClient: {}", lk_url);
+        let (ws_stream, _) = connect_async(lk_url).await?;
+        let _ = emitter.send(SignalEvent::Open).await;
+
+        let (ws_writer, ws_reader) = ws_stream.split();
+        let (internal_tx, internal_rx) = mpsc::channel::<InternalMessage>(8);
+
+        let write_handle = tokio::spawn(Self::write_task(internal_rx, ws_writer, emitter.clone()));
+        let read_handle = tokio::spawn(Self::read_task(internal_tx.clone(), ws_reader, emitter));
+
+        Ok(Self {
+            internal_tx,
+            read_handle,
+            write_handle,
+        })
+    }
+
+    /// Close the websocket
+    /// It sends a CloseFrame to the server before closing
+    pub async fn close(self) {
+        let _ = self
+            .internal_tx
+            .send(InternalMessage::Close {
+                close_frame: Some(CloseFrame {
+                    code: CloseCode::Normal,
+                    reason: "disconnected by client".into(),
+                }),
+            })
+            .await;
+
+        let _ = self.write_handle.await;
+        let _ = self.read_handle.await;
+    }
+
+    /// Send a SignalRequest to the websocket
+    /// It also waits for the message to be sent
+    pub async fn send(&self, signal: signal_request::Message) -> SignalResult<()> {
+        let (send, recv) = oneshot::channel();
+        let msg = InternalMessage::Signal {
+            signal,
+            response_chn: send,
+        };
+        let _ = self.internal_tx.send(msg).await;
+        recv.await.expect("channel closed")
+    }
+
+    /// This task is used to send messages to the websocket
+    /// It is also responsible for closing the connection
+    async fn write_task(
+        mut internal_rx: mpsc::Receiver<InternalMessage>,
+        mut ws_writer: SplitSink<WebSocket, Message>,
+        emitter: SignalEmitter,
+    ) {
+        while let Some(msg) = internal_rx.recv().await {
+            match msg {
+                InternalMessage::Signal {
+                    signal,
+                    response_chn,
+                } => {
+                    event!(Level::TRACE, "sending SignalRequest: {:?}", signal);
+
+                    let data = Message::Binary(
+                        SignalRequest {
+                            message: Some(signal),
+                        }
+                        .encode_to_vec(),
+                    );
+
+                    if let Err(err) = ws_writer.send(data).await {
+                        event!(Level::ERROR, "failed to send signal: {:?}", err);
+                        let _ = response_chn.send(Err(err.into()));
+                        break;
+                    }
+
+                    let _ = response_chn.send(Ok(()));
+                }
+                InternalMessage::Pong { ping_data } => {
+                    if let Err(err) = ws_writer.send(Message::Pong(ping_data)).await {
+                        event!(Level::ERROR, "failed to send pong message: {:?}", err);
+                    }
+                }
+                InternalMessage::Close { close_frame } => {
+                    if let Some(close_frame) = close_frame {
+                        let _ = ws_writer.send(Message::Close(Some(close_frame))).await;
+                        let _ = ws_writer.flush().await;
+                    }
+                    break;
+                }
+            }
+        }
+
+        let _ = ws_writer.close().await;
+        let _ = emitter.send(SignalEvent::Close).await;
+    }
+
+    /// This task is used to read incoming messages from the websocket
+    /// and dispatch them through the EventEmitter.
+    ///
+    /// It can also send messages to [handle_write] task ( Used e.g. answer to pings )
+    async fn read_task(
+        internal_tx: mpsc::Sender<InternalMessage>,
+        mut ws_reader: SplitStream<WebSocket>,
+        emitter: SignalEmitter,
+    ) {
+        while let Some(msg) = ws_reader.next().await {
+            match msg {
+                Ok(Message::Binary(data)) => {
+                    let res = SignalResponse::decode(data.as_slice())
+                        .expect("failed to decode SignalResponse");
+
+                    let msg = res.message.unwrap();
+                    event!(Level::TRACE, "received SignalResponse: {:?}", msg);
+                    let _ = emitter.send(SignalEvent::Signal(msg)).await;
+                }
+                Ok(Message::Ping(data)) => {
+                    let _ = internal_tx
+                        .send(InternalMessage::Pong { ping_data: data })
+                        .await;
+                    continue;
+                }
+                Ok(Message::Close(close)) => {
+                    event!(Level::DEBUG, "server closed the connection: {:?}", close);
+                    break;
+                }
+                _ => {
+                    event!(Level::ERROR, "unhandled websocket message {:?}", msg);
+                    break;
+                }
+            }
+        }
+
+        let _ = internal_tx
+            .send(InternalMessage::Close { close_frame: None })
+            .await;
+    }
+}