Initial Room

Switching computer

Cargo.lock

@@ -494,6 +494,7 @@ dependencies = [
  "futures-util",
  "lazy_static",
  "livekit-webrtc",
+ "parking_lot",
  "prost",
  "prost-build",
  "prost-types",

crates/livekit-core/Cargo.toml

@@ -10,6 +10,7 @@ serde_json = "1.0"
 tokio-tungstenite = { version = "0.17.2", features = ["native-tls"] }
 tokio = { version = "1", features = ["full"] }
 futures = "0.3"
+parking_lot = "0.12.1"
 url = "2.2.2"
 futures-util = "0.3.23"
 thiserror = "1.0"

crates/livekit-core/src/event.rs

@@ -1,43 +0,0 @@
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use futures::Stream;
-use tokio::sync::mpsc;
-
-/// Using unbounded channels to prevent users from blocking internal logic ( e.g: ws heartbeat )
-/// Users must listen to all events to avoid the process from running out of memory
-
-#[derive(Clone, Debug)]
-pub struct Emitter<T> {
-    tx: mpsc::UnboundedSender<T>,
-}
-
-impl<T> Emitter<T> {
-    pub fn new() -> (Self, mpsc::UnboundedReceiver<T>) {
-        let (tx, rx) = mpsc::unbounded_channel();
-
-        (Self { tx }, rx)
-    }
-
-    pub fn event(&self, event: T) {
-        let _ = self.tx.send(event);
-    }
-}
-
-#[derive(Debug)]
-pub struct Events<T> {
-    rx: mpsc::UnboundedReceiver<T>,
-}
-
-impl<T> Events<T> {
-    pub fn new(rx: mpsc::UnboundedReceiver<T>) -> Self {
-        Self { rx }
-    }
-}
-
-impl<T> Stream for Events<T> {
-    type Item = T;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
-        self.rx.poll_recv(cx)
-    }
-}

crates/livekit-core/src/lib.rs

@@ -1,12 +1,11 @@
+extern crate core;
+extern crate core;
+
 pub mod proto {
     include!(concat!(env!("OUT_DIR"), "/livekit.rs"));
 }
 
-mod lk_runtime;
-mod signal_client;
-mod pc_transport;
 mod rtc_engine;
-mod local_participant;
-mod event;
+mod signal_client;
 
-pub mod room;
+pub mod room;

crates/livekit-core/src/local_participant.rs

@@ -1,43 +0,0 @@
-use std::sync::Arc;
-use tokio::sync::Mutex;
-
-use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket};
-use crate::room::RoomError;
-use crate::rtc_engine::RTCEngine;
-
-pub struct LocalParticipant {
-    sid: String,
-    identity: String,
-    name: String,
-
-    engine: Arc<Mutex<RTCEngine>>,
-}
-
-impl LocalParticipant {
-    pub(crate) fn from(info: ParticipantInfo, engine: Arc<Mutex<RTCEngine>>) -> Self {
-        Self {
-            sid: info.sid,
-            identity: info.identity,
-            name: info.name,
-            engine,
-        }
-    }
-
-    pub(crate) fn update(info: ParticipantInfo) {
-        // TODO(theomonnom)
-    }
-
-    // TODO(theomonnom) Add the destinations parameter
-    pub async fn publish_data(&mut 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.clone(),
-                payload: data.to_vec(),
-                destination_sids: vec![], // TODO(theomonnom)
-            })),
-        };
-
-        self.engine.lock().await.publish_data(&data, kind).await.map_err(Into::into)
-    }
-}

crates/livekit-core/src/room.rs

@@ -1,78 +0,0 @@
-use std::sync::Arc;
-
-use thiserror::Error;
-use tokio::sync::Mutex;
-
-use crate::local_participant::LocalParticipant;
-use crate::rtc_engine;
-use crate::rtc_engine::{EngineError, RTCEngine};
-
-#[derive(Error, Debug)]
-pub enum RoomError {
-    #[error("internal RTCEngine failure")]
-    Engine(#[from] EngineError),
-}
-
-#[derive(Debug)]
-pub enum RoomEvent { 
-    
-}
-
-pub struct Room {
-    sid: String,
-    name: String,
-    local_participant: LocalParticipant,
-    internal: Arc<RoomInternal>
-}
-
-#[tracing::instrument(skip(url, token))]
-pub async fn connect(url: &str, token: &str) -> Result<Room, RoomError> {
-    let engine = rtc_engine::connect(url, token).await?;
-    let join = engine.join_response().await;
-    let engine = Arc::new(Mutex::new(engine));
-    let local_participant = LocalParticipant::from(join.participant.unwrap(), engine.clone());
-    let internal = Arc::new(RoomInternal::new(engine));
-    let room_info = join.room.unwrap();
-
-    tokio::spawn(async move {
-
-    });
-
-    Ok(Room {
-        sid: room_info.sid,
-        name: room_info.name,
-        local_participant,
-        internal,
-    })
-}
-
-impl Room {
-    pub fn local_participant(&self) -> &LocalParticipant {
-        &self.local_participant
-    }
-
-    pub fn local_participant_mut(&mut self) -> &mut LocalParticipant {
-        &mut self.local_participant
-    }
-
-    pub fn sid(&self) -> &str {
-        &self.sid
-    }
-
-    pub fn name(&self) -> &str {
-        &self.name
-    }
-}
-
-
-struct RoomInternal {
-    engine: Arc<Mutex<RTCEngine>>,
-}
-
-impl RoomInternal {
-    pub fn new(engine: Arc<Mutex<RTCEngine>>) -> Self {
-        Self {
-            engine
-        }
-    }
-}

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

@@ -0,0 +1,38 @@
+macro_rules! id_str {
+    ($($name:ident;)*) => {
+        $(
+            impl From<String> for $name {
+                fn from(str: String) -> $name {
+                    $name(str)
+                }
+            }
+
+            impl PartialEq<$name> for String {
+                fn eq(&self, u: &$name) -> bool {
+                    *self == *u.0
+                }
+            }
+
+            impl From<$name> for String {
+                fn from(id: $name) -> String {
+                    id.0
+                }
+            }
+        )*
+    }
+}
+
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+pub struct ParticipantSid(pub String);
+
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+pub struct ParticipantIdentity(pub String);
+
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+pub struct TrackSid(pub String);
+
+id_str! {
+    ParticipantSid;
+    ParticipantIdentity;
+    TrackSid;
+}

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

@@ -0,0 +1,46 @@
+use crate::proto::{data_packet, DataPacket, UserPacket};
+use crate::room::participant::{impl_participant_trait, ParticipantShared};
+use crate::room::RoomError;
+use crate::rtc_engine::RTCEngine;
+use std::sync::Arc;
+
+pub struct LocalParticipant {
+    shared: ParticipantShared,
+    rtc_engine: Arc<RTCEngine>,
+}
+
+impl LocalParticipant {
+    pub(super) fn new(rtc_engine: Arc<RTCEngine>, info: ParticipantInfo) -> Self {
+        Self {
+            shared: ParticipantShared::new(
+                info.sid.into(),
+                info.identity.into(),
+                info.name,
+                info.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: "".to_string(), /*self.sid().to_owned()*/
+                payload: data.to_vec(),
+                destination_sids: vec![],
+            })),
+        };
+
+        self.rtc_engine
+            .publish_data(&data, kind)
+            .await
+            .map_err(Into::into)
+    }
+}
+
+impl_participant_trait!(LocalParticipant);

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

@@ -0,0 +1,316 @@
+use futures_util::future::BoxFuture;
+use parking_lot::lock_api::RwLockUpgradableReadGuard;
+use parking_lot::{Mutex, RwLock};
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::future::Future;
+use std::sync::atomic::AtomicU8;
+use std::sync::Arc;
+
+use crate::proto;
+use crate::proto::{participant_info, ParticipantInfo};
+use crate::room::id::{ParticipantIdentity, ParticipantSid};
+use crate::room::local_participant::LocalParticipant;
+use crate::room::participant::ParticipantTrait;
+use crate::room::remote_participant::RemoteParticipant;
+use thiserror::Error;
+use tracing::error;
+
+use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, RTCEngine};
+use crate::signal_client::SignalOptions;
+
+mod id;
+mod local_participant;
+mod participant;
+mod remote_participant;
+mod track;
+mod track_publication;
+
+#[derive(Error, Debug)]
+pub enum RoomError {
+    #[error("internal RTCEngine failure")]
+    Engine(#[from] EngineError),
+    #[error("internal Room failure")]
+    Internal(String),
+}
+
+type RoomResult<T> = Result<T, RoomError>;
+
+pub enum ConnectionState {
+    Disconnected,
+    Connecting,
+    Connected,
+    Reconnecting,
+}
+
+struct RoomInner {
+    state: AtomicU8, // ConnectionState
+    sid: Mutex<String>,
+    name: Mutex<String>,
+    participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
+    rtc_engine: Arc<RTCEngine>,
+    local_participant: Arc<LocalParticipant>,
+}
+
+type OnParticipantConnectedHandler =
+    Box<dyn FnMut(RoomHandle, Arc<RemoteParticipant>) -> BoxFuture<'static, ()> + Send + Sync>;
+type OnParticipantDisconnectedHandler = OnParticipantConnectedHandler;
+
+struct RoomEvents {
+    on_participant_connected_handler: Mutex<Option<OnParticipantConnectedHandler>>,
+    on_participant_disconnected_handler: Mutex<Option<OnParticipantDisconnectedHandler>>,
+}
+
+pub struct Room {
+    inner: Option<Arc<RoomInner>>,
+    events: Arc<RoomEvents>,
+}
+
+impl Room {
+    pub fn new() -> Room {
+        Self {
+            inner: None,
+            events: Arc::new(RoomEvents {
+                on_participant_connected_handler: Default::default(),
+                on_participant_disconnected_handler: Default::default(),
+            }),
+        }
+    }
+
+    pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> {
+        let (rtc_engine, engine_events) =
+            RTCEngine::connect(url, token, SignalOptions::default()).await?;
+        let rtc_engine = Arc::new(rtc_engine);
+        let join_response = rtc_engine.join_response();
+        let local_participant = Arc::new(LocalParticipant::new(
+            rtc_engine.clone(),
+            join_response.participant.unwrap().clone(),
+        ));
+        let room_info = join_response.room.unwrap();
+        let inner = Arc::new(RoomInner {
+            state: AtomicU8::new(ConnectionState::Connecting as u8),
+            sid: Mutex::new(room_info.sid),
+            name: Mutex::new(room_info.name),
+            participants: Default::default(),
+            rtc_engine,
+            local_participant,
+        });
+
+        self.inner = Some(inner.clone());
+
+        tokio::spawn(Self::room_task(inner, self.events.clone(), engine_events));
+
+        Ok(())
+    }
+
+    pub fn get_handle(&self) -> Option<RoomHandle> {
+        self.inner.as_ref().map(|inner| RoomHandle {
+            inner: inner.clone(),
+        })
+    }
+
+    pub fn on_participant_connected<F, Fut>(&self, mut callback: F)
+    where
+        F: FnMut(RoomHandle, Arc<RemoteParticipant>) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = ()> + Send + Sync + 'static,
+    {
+        *self.events.on_participant_connected_handler.lock() =
+            Some(Box::new(move |handle, participant| {
+                Box::pin(callback(handle, participant))
+            }));
+    }
+
+    pub fn on_participant_disconnected<F, Fut>(&self, mut callback: F)
+    where
+        F: FnMut(RoomHandle, Arc<RemoteParticipant>) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = ()> + Send + Sync + 'static,
+    {
+        *self.events.on_participant_disconnected_handler.lock() =
+            Some(Box::new(move |handle, participant| {
+                Box::pin(callback(handle, participant))
+            }));
+    }
+
+    async fn room_task(
+        room_inner: Arc<RoomInner>,
+        room_events: Arc<RoomEvents>,
+        mut engine_events: EngineEvents,
+    ) {
+        while let Some(event) = engine_events.recv().await {
+            if let Err(err) =
+                Self::handle_event(room_inner.clone(), room_events.clone(), event).await
+            {
+                error!("failed to handle engine event: {:?}", err);
+            }
+        }
+    }
+
+    async fn handle_event(
+        room_inner: Arc<RoomInner>,
+        room_events: Arc<RoomEvents>,
+        event: EngineEvent,
+    ) -> RoomResult<()> {
+        match event {
+            EngineEvent::ParticipantUpdate(update) => {
+                Self::handle_participant_update(room_inner.clone(), room_events.clone(), update)
+            }
+            EngineEvent::AddTrack {
+                rtp_receiver,
+                streams,
+            } => {
+                if streams.is_empty() {
+                    Err(RoomError::Internal(
+                        "AddTrack event with empty streams".to_string(),
+                    ))?;
+                }
+
+                let first_stream_id = streams.first().unwrap().id();
+                let stream_id = unpack_stream_id(&first_stream_id);
+                if stream_id.is_none() {
+                    Err(RoomError::Internal(format!(
+                        "AddTrack event with invalid track_id: {:?}",
+                        first_stream_id
+                    )))?;
+                }
+
+                let (participant_sid, track_sid) = stream_id.unwrap();
+                let remote_participant =
+                    Self::get_participant(room_inner.clone(), &participant_sid.to_string().into());
+
+                if let Some(remote_participant) = remote_participant {
+
+                } else {
+                    // The server should send participant updates before sending a new offer
+                    // So this should not happen.
+                    Err(RoomError::Internal(format!(
+                        "AddTrack event with invalid participant_sid: {:?}",
+                        participant_sid
+                    )))?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    fn handle_participant_update(
+        room_inner: Arc<RoomInner>,
+        room_events: Arc<RoomEvents>,
+        update: proto::ParticipantUpdate,
+    ) {
+        for pi in update.participants {
+            if pi.sid == room_inner.local_participant.sid()
+                || pi.identity == room_inner.local_participant.identity()
+            {
+                room_inner.local_participant.update_info(pi);
+                continue;
+            }
+
+            let remote_participant =
+                Self::get_participant(room_inner.clone(), &pi.sid.clone().into());
+
+            if let Some(remote_participant) = remote_participant {
+                if pi.state == participant_info::State::Disconnected as i32 {
+                    // Participant disconencted
+                    Self::handle_participant_disconnect(
+                        room_inner.clone(),
+                        room_events.clone(),
+                        remote_participant,
+                    )
+                } else {
+                    // Participant is already connected, update the informations
+                    remote_participant.update_info(pi);
+                }
+            } else {
+                // Create a new participant and call OnConnect event
+                let remote_participant = Self::get_or_create_participant(room_inner.clone(), pi);
+                let mut handler = room_events.on_participant_connected_handler.lock();
+                if let Some(callback) = handler.as_mut() {
+                    callback(
+                        RoomHandle::from(room_inner.clone()),
+                        remote_participant.clone(),
+                    );
+                }
+            }
+        }
+    }
+
+    fn handle_participant_disconnect(
+        room_inner: Arc<RoomInner>,
+        room_events: Arc<RoomEvents>,
+        remote_participant: Arc<RemoteParticipant>,
+    ) {
+        room_inner
+            .participants
+            .write()
+            .remove(&remote_participant.sid());
+
+        // TODO(theomonnom): Unpublish all tracks
+
+        let mut handler = room_events.on_participant_disconnected_handler.lock();
+        if let Some(callback) = handler.as_mut() {
+            callback(
+                RoomHandle::from(room_inner.clone()),
+                remote_participant.clone(),
+            );
+        }
+    }
+
+    fn get_participant(
+        room_inner: Arc<RoomInner>,
+        sid: &ParticipantSid,
+    ) -> Option<Arc<RemoteParticipant>> {
+        room_inner.participants.read().get(sid).cloned()
+    }
+
+    fn get_or_create_participant(
+        room_inner: Arc<RoomInner>,
+        pi: proto::ParticipantInfo,
+    ) -> Arc<RemoteParticipant> {
+        let participants = room_inner.participants.upgradable_read();
+        let sid = pi.sid.clone().into();
+        if let Some(p) = participants.get(&sid) {
+            p.update_info(pi);
+            p.clone()
+        } else {
+            let mut participants = RwLockUpgradableReadGuard::upgrade(participants);
+            let p = Arc::new(RemoteParticipant::new(pi));
+            participants.insert(sid, p.clone());
+            p
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct RoomHandle {
+    inner: Arc<RoomInner>,
+}
+
+impl RoomHandle {
+    fn from(room_inner: Arc<RoomInner>) -> Self {
+        Self { inner: room_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()
+    }
+}
+
+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
+    }
+}

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

@@ -0,0 +1,109 @@
+use crate::proto::ParticipantInfo;
+use crate::room::local_participant::LocalParticipant;
+use crate::room::remote_participant::RemoteParticipant;
+use parking_lot::{Mutex, RwLock};
+use std::collections::HashMap;
+use std::sync::Arc;
+
+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, Arc<TrackPublication>>>,
+}
+
+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(),
+        }
+    }
+
+    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
+    }
+}
+
+pub trait ParticipantTrait {
+    fn sid(&self) -> ParticipantSid;
+    fn identity(&self) -> ParticipantIdentity;
+    fn name(&self) -> String;
+    fn metadata(&self) -> String;
+    fn update_info(&self, info: ParticipantInfo);
+}
+
+pub enum Participant {
+    Local(LocalParticipant),
+    Remote(RemoteParticipant),
+}
+
+macro_rules! shared_method {
+    ($x:ident, $ret:ident) => {
+        fn $x(&self) -> $ret {
+            match self {
+                Participant::Local(p) => p.$x(),
+                Participant::Remote(p) => p.$x(),
+            }
+        }
+    };
+}
+
+impl ParticipantTrait for Participant {
+    shared_method!(sid, ParticipantSid);
+    shared_method!(identity, ParticipantIdentity);
+    shared_method!(name, String);
+    shared_method!(metadata, String);
+
+    fn update_info(&self, info: ParticipantInfo) {
+        match self {
+            Participant::Local(p) => p.update_info(info),
+            Participant::Remote(p) => p.update_info(info),
+        }
+    }
+}
+
+macro_rules! impl_participant_trait {
+    ($x:ident) => {
+        use crate::proto::ParticipantInfo;
+        use crate::room::id::{ParticipantIdentity, ParticipantSid};
+
+        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 update_info(&self, info: ParticipantInfo) {
+                self.shared.update_info(info);
+            }
+        }
+    };
+}
+
+use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid};
+use crate::room::track_publication::TrackPublication;
+pub(super) use impl_participant_trait;

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

@@ -0,0 +1,34 @@
+use std::sync::Arc;
+use crate::room::participant::{impl_participant_trait, ParticipantShared};
+use crate::room::track_publication::RemoteTrackPublication;
+
+pub struct RemoteParticipant {
+    shared: ParticipantShared,
+}
+
+impl RemoteParticipant {
+    pub(super) fn new(info: ParticipantInfo) -> Self {
+        Self {
+            shared: ParticipantShared::new(
+                info.sid.into(),
+                info.identity.into(),
+                info.name,
+                info.metadata,
+            ),
+        }
+    }
+
+    pub(super) async fn add_subscribed_media_track() {
+
+
+    }
+
+    fn get_track_publication(&self, sid: &str) -> Option<RemoteTrackPublication> {
+        let track = self.shared.tracks.read().get(&sid.to_string().into()).unwrap().clone();
+        
+        
+        None
+    }
+}
+
+impl_participant_trait!(RemoteParticipant);

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

@@ -0,0 +1,135 @@
+pub enum TrackKind {
+    Audio,
+    Video
+}
+
+pub enum StreamState {
+    Active,
+    Paused,
+    Unknown
+}
+
+pub enum TrackSource {
+    Camera,
+    Microphone,
+    Screenshare,
+    ScreenshareAudio,
+    Unknown,
+}
+
+pub struct LocalVideoTrack {}
+pub struct RemoteVideoTrack {}
+pub struct LocalAudioTrack {}
+pub struct RemoteAudioTrack {}
+
+
+pub enum RemoteTrack {
+    Audio(RemoteAudioTrack),
+    Video(RemoteVideoTrack),
+}
+
+pub enum LocalTrack {
+    Audio(LocalAudioTrack),
+    Video(LocalVideoTrack),
+}
+
+pub enum VideoTrack {
+    Local(LocalVideoTrack),
+    Remote(RemoteVideoTrack),
+}
+
+pub enum AudioTrack {
+    Local(LocalAudioTrack),
+    Remote(RemoteAudioTrack),
+}
+
+pub enum Track {
+    LocalVideo(LocalVideoTrack),
+    LocalAudio(LocalAudioTrack),
+    RemoteVideo(RemoteVideoTrack),
+    RemoteAudio(RemoteAudioTrack),
+}
+
+impl From<VideoTrack> for Track {
+    fn from(video_track: VideoTrack) -> Self {
+        match video_track {
+            VideoTrack::Local(local_video) => Self::LocalVideo(local_video),
+            VideoTrack::Remote(remote_video) => Self::RemoteVideo(remote_video),
+        }
+    }
+}
+
+impl From<AudioTrack> for Track {
+    fn from(audio_track: AudioTrack) -> Self {
+        match audio_track {
+            AudioTrack::Local(local_audio) => Self::LocalAudio(local_audio),
+            AudioTrack::Remote(remote_audio) => Self::RemoteAudio(remote_audio),
+        }
+    }
+}
+
+impl From<LocalTrack> for Track {
+    fn from(local_track: LocalTrack) -> Self {
+        match local_track {
+            LocalTrack::Audio(local_audio) => Self::LocalAudio(local_audio),
+            LocalTrack::Video(local_video) => Self::LocalVideo(local_video),
+        }
+    }
+}
+
+impl From<RemoteTrack> for Track {
+    fn from(remote_track: RemoteTrack) -> Self {
+        match remote_track {
+            RemoteTrack::Audio(remote_audio) => Self::RemoteAudio(remote_audio),
+            RemoteTrack::Video(remote_video) => Self::RemoteVideo(remote_video),
+        }
+    }
+}
+
+impl TryFrom<Track> for VideoTrack {
+    type Error = &'static str;
+
+    fn try_from(track: Track) -> Result<Self, Self::Error> {
+        match track {
+            Track::LocalVideo(local_video) => Ok(Self::Local(local_video)),
+            Track::RemoteVideo(remote_video) => Ok(Self::Remote(remote_video)),
+            _ => Err("not a video track"),
+        }
+    }
+}
+
+impl TryFrom<Track> for AudioTrack {
+    type Error = &'static str;
+
+    fn try_from(track: Track) -> Result<Self, Self::Error> {
+        match track {
+            Track::LocalAudio(local_audio) => Ok(Self::Local(local_audio)),
+            Track::RemoteAudio(remote_audio) => Ok(Self::Remote(remote_audio)),
+            _ => Err("not a audio track"),
+        }
+    }
+}
+
+impl TryFrom<Track> for LocalTrack {
+    type Error = &'static str;
+
+    fn try_from(track: Track) -> Result<Self, Self::Error> {
+        match track {
+            Track::LocalAudio(local_audio) => Ok(Self::Audio(local_audio)),
+            Track::LocalVideo(local_video) => Ok(Self::Video(local_video)),
+            _ => Err("not a local track"),
+        }
+    }
+}
+
+impl TryFrom<Track> for RemoteTrack {
+    type Error = &'static str;
+
+    fn try_from(track: Track) -> Result<Self, Self::Error> {
+        match track {
+            Track::RemoteAudio(remote_audio) => Ok(Self::Audio(remote_audio)),
+            Track::RemoteVideo(remote_video) => Ok(Self::Video(remote_video)),
+            _ => Err("not a remote track"),
+        }
+    }
+}

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

@@ -0,0 +1,28 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU8};
+use parking_lot::Mutex;
+use crate::room::id::TrackSid;
+
+pub(super) struct TrackPublicationShared {
+    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
+}
+
+#[derive(Clone)]
+pub struct LocalTrackPublication {
+    shared: Arc<TrackPublicationShared>
+}
+
+#[derive(Clone)]
+pub struct RemoteTrackPublication {
+    shared: Arc<TrackPublicationShared>
+}
+
+#[derive(Clone)]
+pub enum TrackPublication {
+    Local(LocalTrackPublication),
+    Remote(RemoteTrackPublication)
+}

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

@@ -1,576 +0,0 @@
-use std::fmt::{Debug, Formatter};
-use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
-use std::sync::Arc;
-use std::time::Duration;
-
-use prost::Message;
-use serde::{Deserialize, Serialize};
-use tokio::sync::{mpsc, Mutex};
-use tokio::time;
-use tracing::{event, Level};
-
-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::{
-    ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
-};
-
-use crate::lk_runtime::LKRuntime;
-use crate::pc_transport::PCTransport;
-use crate::proto;
-use crate::proto::data_packet::Value;
-use crate::proto::{
-    data_packet, signal_request, signal_response, DataPacket, JoinResponse, SignalTarget,
-    TrickleRequest,
-};
-use crate::rtc_engine::{EngineError, MAX_ICE_CONNECT_TIMEOUT};
-use crate::signal_client::SignalClient;
-
-const LOSSY_DC_LABEL: &str = "_lossy";
-const RELIABLE_DC_LABEL: &str = "_reliable";
-
-// Used to communicate IceCandidate with the server
-#[derive(Serialize, Deserialize)]
-#[allow(non_snake_case)]
-struct IceCandidateJSON {
-    sdpMid: String,
-    sdpMLineIndex: i32,
-    candidate: String,
-}
-
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-pub(crate) enum PCState {
-    New,
-    Connected,
-    Disconnected,
-    Reconnecting,
-    Closed,
-}
-
-#[derive(Debug)]
-pub(crate) enum InternalMessage {
-    IceCandidate {
-        ice_candidate: IceCandidate,
-        publisher: bool,
-    },
-    ConnectionChange {
-        state: PeerConnectionState,
-        primary: bool,
-    },
-    PrimaryDataChannel {
-        data_channel: DataChannel,
-    },
-    PublisherOffer {
-        offer: SessionDescription,
-    },
-    Data {
-        data: Vec<u8>,
-        binary: bool,
-    },
-}
-
-pub(crate) struct EngineInternal {
-    pub(super) publisher_pc: Arc<Mutex<PCTransport>>,
-    pub(super) subscriber_pc: Arc<Mutex<PCTransport>>,
-    pub(super) lossy_dc: Arc<Mutex<DataChannel>>,
-    pub(super) reliable_dc: Arc<Mutex<DataChannel>>,
-    pub(super) lossy_dc_sub: Arc<Mutex<Option<DataChannel>>>,
-    pub(super) reliable_dc_sub: Arc<Mutex<Option<DataChannel>>>,
-
-    pub(super) msg_sender: mpsc::Sender<InternalMessage>,
-    pub(super) join_response: Mutex<JoinResponse>,
-    pub(super) pc_state: AtomicU8, // casted to PCState
-    pub(super) has_published: AtomicBool,
-}
-
-impl Debug for EngineInternal {
-    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "EngineInternal")
-    }
-}
-
-impl EngineInternal {
-    /// Configure the PeerConnections
-    ///
-    /// This is called on connect & on full reconnect.
-    /// Create the PeerConnections & the DataChannels.
-    /// Register listeners and send the internal messages
-    /// to the event_loop.
-    #[tracing::instrument]
-    pub(super) fn configure(
-        lk_runtime: Arc<LKRuntime>,
-        sender: mpsc::Sender<InternalMessage>,
-        join: JoinResponse,
-    ) -> Result<Self, EngineError> {
-        let rtc_config = RTCConfiguration {
-            ice_servers: {
-                let mut servers = vec![];
-                for is in join.ice_servers.clone() {
-                    servers.push(ICEServer {
-                        urls: is.urls,
-                        username: is.username,
-                        password: is.credential,
-                    })
-                }
-                servers
-            },
-            continual_gathering_policy: ContinualGatheringPolicy::GatherContinually,
-            ice_transport_type: IceTransportsType::All,
-        };
-
-        let mut publisher_pc = PCTransport::new(
-            lk_runtime
-                .pc_factory
-                .create_peer_connection(rtc_config.clone())?,
-        );
-        let mut subscriber_pc =
-            PCTransport::new(lk_runtime.pc_factory.create_peer_connection(rtc_config)?);
-
-        publisher_pc.peer_connection().on_ice_candidate(Box::new({
-            let sender = sender.clone();
-            move |ice_candidate| {
-                let _ = sender.blocking_send(InternalMessage::IceCandidate {
-                    ice_candidate,
-                    publisher: true,
-                });
-            }
-        }));
-
-        subscriber_pc.peer_connection().on_ice_candidate(Box::new({
-            let sender = sender.clone();
-            move |ice_candidate| {
-                let _ = sender.blocking_send(InternalMessage::IceCandidate {
-                    ice_candidate,
-                    publisher: false,
-                });
-            }
-        }));
-
-        publisher_pc.on_offer({
-            let sender = sender.clone();
-            Box::new(move |offer| {
-                let sender = sender.clone();
-
-                tokio::spawn(async move {
-                    let _ = sender.send(InternalMessage::PublisherOffer { offer }).await;
-                });
-
-                Box::pin(async move {})
-            })
-        });
-
-        let mut primary_pc = &mut publisher_pc;
-        let mut secondary_pc = &mut subscriber_pc;
-        if join.subscriber_primary {
-            primary_pc = &mut subscriber_pc;
-            secondary_pc = &mut publisher_pc;
-
-            primary_pc.peer_connection().on_data_channel(Box::new({
-                let sender = sender.clone();
-                move |data_channel| {
-                    let _ =
-                        sender.blocking_send(InternalMessage::PrimaryDataChannel { data_channel });
-                }
-            }));
-        }
-
-        primary_pc.peer_connection().on_connection_change(Box::new({
-            let sender = sender.clone();
-            move |state| {
-                let _ = sender.blocking_send(InternalMessage::ConnectionChange {
-                    state,
-                    primary: true,
-                });
-            }
-        }));
-
-        secondary_pc
-            .peer_connection()
-            .on_connection_change(Box::new({
-                let sender = sender.clone();
-                move |state| {
-                    let _ = sender.blocking_send(InternalMessage::ConnectionChange {
-                        state,
-                        primary: false,
-                    });
-                }
-            }));
-
-        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()
-            },
-        )?;
-
-        Self::configure_dc(&mut lossy_dc, sender.clone());
-        Self::configure_dc(&mut reliable_dc, sender.clone());
-
-        Ok(Self {
-            publisher_pc: Arc::new(Mutex::new(publisher_pc)),
-            subscriber_pc: Arc::new(Mutex::new(subscriber_pc)),
-            lossy_dc: Arc::new(Mutex::new(lossy_dc)),
-            reliable_dc: Arc::new(Mutex::new(reliable_dc)),
-            lossy_dc_sub: Default::default(),
-            reliable_dc_sub: Default::default(),
-            msg_sender: sender,
-            join_response: Mutex::new(join),
-            pc_state: AtomicU8::new(PCState::New as u8),
-            has_published: AtomicBool::new(false),
-        })
-    }
-
-    /// Send InternalMessage when a datachannel receives data
-    #[tracing::instrument]
-    fn configure_dc(data_channel: &mut DataChannel, sender: mpsc::Sender<InternalMessage>) {
-        data_channel.on_message(Box::new(move |data, binary| {
-            let _ = sender.blocking_send(InternalMessage::Data {
-                data: data.to_vec(),
-                binary,
-            });
-        }));
-    }
-
-    /// Ensure the publisher PeerConnection is connected
-    ///
-    /// When subscriber_primary is enabled, only the subscriber PeerConnection is negotiated.
-    /// This allows for faster connection when we don't need the publisher
-    #[tracing::instrument]
-    pub(super) async fn ensure_publisher_connected(
-        self: &Arc<Self>,
-        kind: data_packet::Kind,
-    ) -> Result<(), EngineError> {
-        if !self.join_response.lock().await.subscriber_primary {
-            return Ok(());
-        }
-
-        let publisher = &self.publisher_pc;
-        {
-            let mut publisher = publisher.lock().await;
-            if !publisher.is_connected()
-                && publisher.peer_connection().ice_connection_state()
-                    != IceConnectionState::IceConnectionChecking
-            {
-                tokio::spawn({
-                    let internal = self.clone();
-                    async move {
-                        let _ = internal.negotiate_publisher().await;
-                    }
-                });
-            }
-        }
-
-        let dc = self.data_channel(kind);
-        if dc.lock().await.state() == DataState::Open {
-            return Ok(());
-        }
-
-        let res = time::timeout(MAX_ICE_CONNECT_TIMEOUT, async move {
-            let mut interval = time::interval(Duration::from_millis(50));
-
-            loop {
-                if publisher.lock().await.is_connected()
-                    && dc.lock().await.state() == DataState::Open
-                {
-                    break;
-                }
-
-                interval.tick().await;
-            }
-        })
-        .await;
-
-        if res.is_err() {
-            let err =
-                EngineError::Connection("could not establish publisher connection".to_string());
-            event!(Level::ERROR, error = ?err);
-            Err(err)
-        } else {
-            Ok(())
-        }
-    }
-
-    /// Run the event_loop of the RTCEngine
-    #[tracing::instrument]
-    pub(super) async fn run(
-        self: &Arc<Self>,
-        mut receiver: mpsc::Receiver<InternalMessage>,
-        signal_client: Arc<SignalClient>,
-    ) {
-        loop {
-            tokio::select! {
-                signal = signal_client.recv() => {
-                    match signal {
-                        Some(signal) => {
-                            if let Err(err) = self.handle_signal(signal, signal_client.clone()).await {
-                                event!(
-                                    Level::ERROR,
-                                    "failed to handle signal: {:?}",
-                                    err,
-                                );
-                            }
-                        }
-                        None => {
-                            // TODO(theomonnom) Trigger reconnect
-                        }
-                    }
-                },
-                Some(msg) = receiver.recv() => {
-                    if let Err(err) = self.handle_message(msg, signal_client.clone()).await {
-                        event!(
-                            Level::ERROR,
-                            "failed to handle engine message: {:?}",
-                            err,
-                        );
-                    }
-                },
-            }
-        }
-    }
-
-    /// Handle SignalResponse messages coming from the server
-    ///
-    /// Run the needed livekit-protocol
-    #[tracing::instrument]
-    async fn handle_signal(
-        self: &Arc<Self>,
-        signal: signal_response::Message,
-        signal_client: Arc<SignalClient>,
-    ) -> Result<(), EngineError> {
-        match signal {
-            signal_response::Message::Answer(answer) => {
-                event!(Level::TRACE, "received answer for publisher: {:?}", answer);
-                let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
-                self.publisher_pc
-                    .lock()
-                    .await
-                    .set_remote_description(sdp)
-                    .await?;
-            }
-            signal_response::Message::Offer(offer) => {
-                // Handle the subscriber offer & send an answer to livekit-server
-                // We always get an offer from the server when connecting
-                event!(Level::TRACE, "received offer for subscriber: {:?}", offer);
-                let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
-
-                self.subscriber_pc
-                    .lock()
-                    .await
-                    .set_remote_description(sdp)
-                    .await?;
-                let answer = self
-                    .subscriber_pc
-                    .lock()
-                    .await
-                    .peer_connection()
-                    .create_answer(RTCOfferAnswerOptions::default())
-                    .await?;
-                self.subscriber_pc
-                    .lock()
-                    .await
-                    .peer_connection()
-                    .set_local_description(answer.clone())
-                    .await?;
-
-                tokio::spawn(async move {
-                    let _ = signal_client
-                        .send(signal_request::Message::Answer(proto::SessionDescription {
-                            r#type: "answer".to_string(),
-                            sdp: answer.to_string(),
-                        }))
-                        .await;
-                });
-            }
-            signal_response::Message::Trickle(trickle) => {
-                // Add the IceCandidate received from the livekit-server
-                let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?;
-                let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?;
-
-                event!(
-                    Level::TRACE,
-                    "received ice_candidate ({:?}) - {:?}",
-                    SignalTarget::from_i32(trickle.target).unwrap(),
-                    ice
-                );
-
-                if trickle.target == SignalTarget::Publisher as i32 {
-                    self.publisher_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice)
-                        .await?;
-                } else {
-                    self.subscriber_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice)
-                        .await?;
-                }
-            }
-            _ => {}
-        }
-
-        Ok(())
-    }
-
-    /// Handle libwebrtc messages
-    ///
-    /// Every message used inside this function comes from libwebrtc.
-    /// The messages are received in [EngineInternal](#run)
-    /// We're not handling the messages inside the signaling_thread, to return
-    /// as quickly as possible.
-    #[tracing::instrument]
-    async fn handle_message(
-        self: &Arc<Self>,
-        msg: InternalMessage,
-        signal_client: Arc<SignalClient>,
-    ) -> Result<(), EngineError> {
-        match msg {
-            InternalMessage::IceCandidate {
-                ice_candidate,
-                publisher,
-            } => {
-                // Send the IceCandidate to livekit-server
-                // Note that ContinualGatheringPolicy is set to GatherContinually
-                let json = serde_json::to_string(&IceCandidateJSON {
-                    sdpMid: ice_candidate.sdp_mid(),
-                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
-                    candidate: ice_candidate.candidate(),
-                })?;
-
-                let target = if publisher {
-                    SignalTarget::Publisher
-                } else {
-                    SignalTarget::Subscriber
-                };
-
-                event!(
-                    Level::TRACE,
-                    "sending ice_candidate ({:?}) - {:?}",
-                    target,
-                    ice_candidate
-                );
-
-                tokio::spawn(async move {
-                    let _ = signal_client
-                        .send(signal_request::Message::Trickle(TrickleRequest {
-                            candidate_init: json,
-                            target: target as i32,
-                        }))
-                        .await;
-                });
-            }
-            InternalMessage::ConnectionChange { state, primary } => {
-                // PeerConnectionState changed
-                // Reconnect if we've been disconnected unexpectedly
-                // If connected for the first time, send OnConnect event
-                if primary && state == PeerConnectionState::Connected {
-                    let old_state = self.pc_state.load(Ordering::SeqCst);
-                    self.pc_state
-                        .store(PCState::Connected as u8, Ordering::SeqCst);
-
-                    if old_state == PCState::New as u8 {
-                        // TODO(theomonnom) OnConnected
-                    }
-                } else if state == PeerConnectionState::Failed {
-                    self.pc_state
-                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
-
-                    // TODO(theomonnom) handle Disconnect
-                }
-            }
-            InternalMessage::PrimaryDataChannel { mut data_channel } => {
-                // Received datachannel from the primary PeerConnection.
-                // If subscriber_primary is enabled, the datachannel is used for downstream data
-                let reliable = data_channel.label() == RELIABLE_DC_LABEL;
-                Self::configure_dc(&mut data_channel, self.msg_sender.clone());
-
-                event!(
-                    Level::TRACE,
-                    "received primary data_channel - {:?}",
-                    data_channel
-                );
-
-                if reliable {
-                    *self.reliable_dc_sub.lock().await = Some(data_channel);
-                } else {
-                    *self.lossy_dc_sub.lock().await = Some(data_channel);
-                }
-            }
-            InternalMessage::PublisherOffer { offer } => {
-                // Send the publisher offer to livekit-server
-                event!(Level::TRACE, "sending publisher offer - {:?}", offer);
-
-                tokio::spawn(async move {
-                    let _ = signal_client
-                        .send(signal_request::Message::Offer(proto::SessionDescription {
-                            r#type: "offer".to_string(),
-                            sdp: offer.to_string(),
-                        }))
-                        .await;
-                });
-            }
-            InternalMessage::Data { data, binary } => {
-                // Received data from a datachannel
-                // If this is a Speaker DataPacket, update the active speakers
-                // Send SpeakersChanged/OnData event
-                if !binary {
-                    return Err(EngineError::Internal(
-                        "text messages aren't supported".to_string(),
-                    ));
-                }
-
-                let data = DataPacket::decode(&*data)?;
-                match data.value.unwrap() {
-                    Value::User(user) => {
-                        /*let mut handler = self.on_data_handler.lock().await;
-                        if let Some(f) = &mut *handler {
-                            f(Packet {
-                                data: user,
-                                kind: data_packet::Kind::from_i32(data.kind).unwrap(),
-                            })
-                                .await;
-                        }*/
-                    }
-                    Value::Speaker(_) => {
-                        // TODO(theomonnonm)
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    #[tracing::instrument]
-    async fn negotiate_publisher(self: &Arc<Self>) -> Result<(), EngineError> {
-        self.has_published.store(true, Ordering::SeqCst);
-        if let Err(err) = self.publisher_pc.lock().await.negotiate().await {
-            event!(Level::ERROR, "failed to negotiate the publisher: {:?}", err,);
-            Err(EngineError::Rtc(err))
-        } else {
-            Ok(())
-        }
-    }
-
-    pub(super) fn data_channel(&self, kind: data_packet::Kind) -> Arc<Mutex<DataChannel>> {
-        if kind == data_packet::Kind::Reliable {
-            self.reliable_dc.clone()
-        } else {
-            self.lossy_dc.clone()
-        }
-    }
-}

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

@@ -1,29 +1,72 @@
-use crate::lk_runtime::LKRuntime;
-use crate::proto::{data_packet, signal_response, DataPacket, JoinResponse, UserPacket};
-use crate::rtc_engine::engine_internal::EngineInternal;
-use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalOptions};
-use futures_util::{FutureExt, StreamExt};
-use lazy_static::lazy_static;
-use livekit_webrtc::data_channel::DataSendError;
-use livekit_webrtc::jsep::SdpParseError;
-use livekit_webrtc::rtc_error::RTCError;
-use prost::Message;
+use core::num::flt2dec::Sign;
+use parking_lot::Mutex;
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::sync::{Arc, Weak};
 use std::time::Duration;
-use thiserror::Error;
-use tokio::sync::{mpsc, Mutex};
-use tokio::time;
-use tracing::{event, Level};
 
-mod engine_internal;
+use tokio::sync::{mpsc, Mutex as AsyncMutex};
+
+use lazy_static::lazy_static;
+use prost::Message;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio::time::sleep;
+use tracing::{debug, error, trace};
+
+use crate::{proto, signal_client};
+use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState};
+use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription};
+use livekit_webrtc::media_stream::MediaStream;
+use livekit_webrtc::peer_connection::{
+    IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
+};
+use livekit_webrtc::peer_connection_factory::RTCConfiguration;
+use livekit_webrtc::rtc_error::RTCError;
+use livekit_webrtc::rtp_receiver::RtpReceiver;
+
+use crate::proto::data_packet::Value;
+use crate::proto::{
+    data_packet, signal_request, signal_response, DataPacket, JoinResponse, ParticipantUpdate,
+    SignalTarget, TrickleRequest,
+};
+use crate::rtc_engine::lk_runtime::LKRuntime;
+use crate::rtc_engine::pc_transport::PCTransport;
+use crate::rtc_engine::rtc_events::{RTCEmitter, RTCEvent, RTCEvents};
+use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions};
+
+mod lk_runtime;
+mod pc_transport;
+mod rtc_events;
 
 lazy_static! {
     // Share one LKRuntime across all RTCEngine instances
     static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
 }
 
+pub(crate) type EngineEmitter = mpsc::Sender<EngineEvent>;
+pub(crate) type EngineEvents = mpsc::Receiver<EngineEvent>;
+pub(crate) type EngineResult<T> = Result<T, EngineError>;
+
 pub(crate) const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
-pub(crate) const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
+pub(crate) const LOSSY_DC_LABEL: &str = "_lossy";
+pub(crate) const RELIABLE_DC_LABEL: &str = "_reliable";
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub(crate) enum PCState {
+    New,
+    Connected,
+    Disconnected,
+    Reconnecting,
+    Closed,
+}
+
+#[derive(Serialize, Deserialize)]
+#[allow(non_snake_case)]
+struct IceCandidateJSON {
+    sdpMid: String,
+    sdpMLineIndex: i32,
+    candidate: String,
+}
 
 #[derive(Error, Debug)]
 pub enum EngineError {
@@ -46,108 +89,542 @@ pub enum EngineError {
 }
 
 #[derive(Debug)]
-pub struct Packet {
-    pub data: UserPacket,
-    pub kind: data_packet::Kind,
+pub(crate) enum EngineEvent {
+    ParticipantUpdate(ParticipantUpdate),
+    AddTrack {
+        rtp_receiver: RtpReceiver,
+        streams: Vec<MediaStream>,
+    },
 }
 
 #[derive(Debug)]
-pub enum EngineEvent {
-    DataReceived(Packet),
+struct EngineInner {
+    has_published: AtomicBool,
+    join_response: Mutex<JoinResponse>,
+    pc_state: AtomicU8, // Casted to PCState enum
+
+    publisher_pc: AsyncMutex<PCTransport>,
+    subscriber_pc: AsyncMutex<PCTransport>,
+
+    // Publisher data channels
+    // Used to send data to other participants ( The SFU forward the messages )
+    lossy_dc: Mutex<DataChannel>,
+    reliable_dc: Mutex<DataChannel>,
+
+    // Subscriber data channels
+    // These fields are never used, we just keep a strong reference to them,
+    // so we can receive data from other participants
+    sub_reliable_dc: Mutex<Option<DataChannel>>,
+    sub_lossy_dc: Mutex<Option<DataChannel>>,
 }
 
 #[derive(Debug)]
 pub struct RTCEngine {
     signal_client: Arc<SignalClient>,
-    internal: Arc<EngineInternal>,
+    engine_inner: Arc<EngineInner>,
 
     #[allow(unused)]
     lk_runtime: Arc<LKRuntime>, // Keep a reference while we're using the RTCEngine
 }
 
-#[tracing::instrument(skip(url, token))]
-pub async fn connect(
-    url: &str,
-    token: &str,
-    options: SignalOptions,
-) -> Result<RTCEngine, EngineError> {
-    // Acquire an existing/a new LKRuntime
-    let mut lk_runtime_ref = LK_RUNTIME.lock().await;
-    let mut lk_runtime = lk_runtime_ref.upgrade();
+impl RTCEngine {
+    #[tracing::instrument(skip(url, token))]
+    pub(crate) async fn connect(
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+    ) -> EngineResult<(RTCEngine, 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::new());
-        *lk_runtime_ref = Arc::downgrade(&new_runtime);
-        lk_runtime = Some(new_runtime);
-    }
-    let lk_runtime = lk_runtime.unwrap();
-    let (signal_client, mut signal_events) = SignalClient::connect(url, token, options).await?;
-    let signal_client = Arc::new(signal_client);
-
-    let join_response = time::timeout(JOIN_RESPONSE_TIMEOUT, async move {
-        while let Some(event) = signal_events.next().await {
-            match event {
-                SignalEvent::Signal(signal_response::Message::Join(join)) => return join,
-                _ => {
-                    // Should we try a reconnect on close here?
-                    continue;
-                }
+            if lk_runtime.is_none() {
+                let new_runtime = Arc::new(LKRuntime::new());
+                *lk_runtime_ref = Arc::downgrade(&new_runtime);
+                lk_runtime = Some(new_runtime);
             }
         }
 
-        unreachable!();
-    })
-    .await
-    .map_err(|_| EngineError::Internal("failed to receive JoinResponse".to_string()))?;
+        let lk_runtime = lk_runtime.unwrap();
+        let (signal_client, mut signal_events) = SignalClient::connect(url, token, options).await?;
 
-    event!(Level::DEBUG, "received JoinResponse: {:?}", join_response);
+        let join_response = signal_client::utils::next_join_response(&mut signal_events).await?;
+        debug!("received JoinResponse: {:?}", join_response);
 
-    let (sender, receiver) = mpsc::channel(8);
-    let internal = Arc::new(EngineInternal::configure(
-        lk_runtime.clone(),
-        sender,
-        join_response.clone(),
-    )?);
+        let (engine_inner, rtc_events) =
+            Self::configure_engine(lk_runtime.clone(), join_response.clone())?;
+        let engine_inner = Arc::new(engine_inner);
+        let signal_client = Arc::new(signal_client);
 
-    if !join_response.subscriber_primary {
-        internal.publisher_pc.lock().await.negotiate().await?;
+        let (emitter, events) = mpsc::channel(8);
+
+        tokio::spawn(Self::signal_task(
+            signal_client.clone(),
+            engine_inner.clone(),
+            signal_events,
+            emitter.clone(),
+        ));
+
+        tokio::spawn(Self::engine_task(
+            signal_client.clone(),
+            engine_inner.clone(),
+            rtc_events,
+            emitter.clone(),
+        ));
+
+        let rtc_engine = Self {
+            signal_client,
+            engine_inner,
+            lk_runtime,
+        };
+
+        if !join_response.subscriber_primary {
+            rtc_engine.negotiate_publisher().await?;
+        }
+
+        Ok((rtc_engine, events))
     }
 
-    tokio::spawn({
-        let signal_client = signal_client.clone();
-        let internal = internal.clone();
-
-        async move {
-            internal.run(receiver, signal_client).await;
-        }
-    });
-
-    Ok(RTCEngine {
-        lk_runtime,
-        signal_client,
-        internal,
-    })
-}
-
-impl RTCEngine {
-    /// Send data to other participants in the Room
     #[tracing::instrument]
     pub async fn publish_data(
-        &mut self,
+        &self,
         data: &DataPacket,
         kind: data_packet::Kind,
     ) -> Result<(), EngineError> {
-        self.internal.ensure_publisher_connected(kind).await?;
-        self.internal
-            .data_channel(kind)
+        self.ensure_publisher_connected(kind).await?;
+        self.data_channel(kind)
             .lock()
-            .await
             .send(&data.encode_to_vec(), true)
             .map_err(Into::into)
     }
 
-    /// Return the last received JoinResponse
-    pub async fn join_response(&self) -> JoinResponse {
-        self.internal.join_response.lock().await.clone()
+    pub fn join_response(&self) -> JoinResponse {
+        self.engine_inner.join_response.lock().clone()
+    }
+
+    async fn engine_task(
+        signal_client: Arc<SignalClient>,
+        engine_inner: Arc<EngineInner>,
+        mut rtc_events: RTCEvents,
+        emitter: EngineEmitter,
+    ) {
+        while let Some(event) = rtc_events.recv().await {
+            if let Err(err) = Self::handle_rtc(
+                event,
+                signal_client.clone(),
+                engine_inner.clone(),
+                emitter.clone(),
+            )
+            .await
+            {
+                error!("failed to handle rtc event: {:?}", err);
+            }
+        }
+    }
+
+    async fn signal_task(
+        signal_client: Arc<SignalClient>,
+        engine_inner: Arc<EngineInner>,
+        mut signal_events: SignalEvents,
+        emitter: EngineEmitter,
+    ) {
+        while let Some(signal) = signal_events.recv().await {
+            match signal {
+                SignalEvent::Open => {}
+                SignalEvent::Signal(signal) => {
+                    if let Err(err) = Self::handle_signal(
+                        signal,
+                        signal_client.clone(),
+                        engine_inner.clone(),
+                        emitter.clone(),
+                    )
+                    .await
+                    {
+                        error!("failed to handle signal: {:?}", err);
+                    }
+                }
+                SignalEvent::Close => {
+                    // Try reconnect if this isn't expected
+                }
+            }
+        }
+    }
+
+    async fn handle_rtc(
+        event: RTCEvent,
+        signal_client: Arc<SignalClient>,
+        engine_inner: Arc<EngineInner>,
+        emitter: EngineEmitter,
+    ) -> EngineResult<()> {
+        match event {
+            RTCEvent::IceCandidate {
+                ice_candidate,
+                target,
+            } => {
+                let json = serde_json::to_string(&IceCandidateJSON {
+                    sdpMid: ice_candidate.sdp_mid(),
+                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
+                    candidate: ice_candidate.candidate(),
+                })?;
+
+                trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate);
+
+                tokio::spawn(async move {
+                    signal_client
+                        .send(signal_request::Message::Trickle(TrickleRequest {
+                            candidate_init: json,
+                            target: target as i32,
+                        }))
+                        .await;
+                });
+            }
+            RTCEvent::ConnectionChange { state, target } => {
+                // Reconnect if we've been disconnected unexpectedly
+                let subscriber_primary = engine_inner.join_response.lock().subscriber_primary;
+                let is_primary = subscriber_primary && target == SignalTarget::Subscriber;
+
+                if is_primary && state == PeerConnectionState::Disconnected {
+                    let old_state = engine_inner
+                        .pc_state
+                        .swap(PCState::Connected as u8, Ordering::SeqCst);
+                    if old_state == PCState::New as u8 {
+                        // TODO(theomonnom) Handle disconnect
+                    }
+                } else if state == PeerConnectionState::Failed {
+                    engine_inner
+                        .pc_state
+                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
+                    // TODO(theomonnom) Handle disconnect
+                }
+            }
+            RTCEvent::DataChannel {
+                data_channel,
+                target,
+            } => {
+                if target == SignalTarget::Subscriber {
+                    if data_channel.label() == RELIABLE_DC_LABEL {
+                        *engine_inner.sub_reliable_dc.lock() = Some(data_channel);
+                    } else {
+                        *engine_inner.sub_lossy_dc.lock() = Some(data_channel);
+                    }
+                }
+            }
+            RTCEvent::Offer { offer, target } => {
+                if target == SignalTarget::Publisher {
+                    // Send the publisher offer to the server
+                    tokio::spawn(async move {
+                        signal_client
+                            .send(signal_request::Message::Offer(proto::SessionDescription {
+                                r#type: "offer".to_string(),
+                                sdp: offer.to_string(),
+                            }))
+                            .await;
+                    });
+                }
+            }
+            RTCEvent::AddTrack {
+                rtp_receiver,
+                streams,
+                target,
+            } => {
+                if target == SignalTarget::Subscriber {
+                    let _ = emitter.send(EngineEvent::AddTrack {
+                        rtp_receiver,
+                        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) => {
+                        // TODO(theomonnom) Send event
+                    }
+                    Value::Speaker(_) => {
+                        // TODO(theomonnonm)
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn handle_signal(
+        event: signal_response::Message,
+        signal_client: Arc<SignalClient>,
+        engine_inner: Arc<EngineInner>,
+        emitter: EngineEmitter,
+    ) -> EngineResult<()> {
+        match event {
+            signal_response::Message::Answer(answer) => {
+                trace!("received answer from the publisher: {:?}", answer);
+
+                let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
+                engine_inner
+                    .publisher_pc
+                    .lock()
+                    .await
+                    .set_remote_description(sdp)
+                    .await?;
+            }
+            signal_response::Message::Offer(offer) => {
+                // Handle the subscriber offer & send an answer to livekit-server
+                // We always get an offer from the server when connecting
+                trace!("received offer from the publisher: {:?}", offer);
+                let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
+
+                engine_inner
+                    .subscriber_pc
+                    .lock()
+                    .await
+                    .set_remote_description(sdp)
+                    .await?;
+                let answer = engine_inner
+                    .subscriber_pc
+                    .lock()
+                    .await
+                    .peer_connection()
+                    .create_answer(RTCOfferAnswerOptions::default())
+                    .await?;
+                engine_inner
+                    .subscriber_pc
+                    .lock()
+                    .await
+                    .peer_connection()
+                    .set_local_description(answer.clone())
+                    .await?;
+
+                tokio::spawn(async move {
+                    signal_client
+                        .send(signal_request::Message::Answer(proto::SessionDescription {
+                            r#type: "answer".to_string(),
+                            sdp: answer.to_string(),
+                        }))
+                        .await;
+                });
+            }
+            signal_response::Message::Trickle(trickle) => {
+                // Add the IceCandidate received from the livekit-server
+                let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?;
+                let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?;
+
+                trace!(
+                    "received ice_candidate {:?} - {:?}",
+                    SignalTarget::from_i32(trickle.target).unwrap(),
+                    ice
+                );
+
+                if trickle.target == SignalTarget::Publisher as i32 {
+                    engine_inner
+                        .publisher_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
+                } else {
+                    engine_inner
+                        .subscriber_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
+                }
+            }
+            signal_response::Message::Update(update) => {
+                let _ = emitter.send(EngineEvent::ParticipantUpdate(update));
+            }
+            _ => {}
+        }
+
+        Ok(())
+    }
+
+    async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> {
+        if !self.join_response().subscriber_primary {
+            return Ok(());
+        }
+
+        let publisher = &self.engine_inner.publisher_pc;
+        {
+            let mut publisher = publisher.lock().await;
+            if !publisher.is_connected()
+                && publisher.peer_connection().ice_connection_state()
+                    != IceConnectionState::IceConnectionChecking
+            {
+                let _ = self.negotiate_publisher().await;
+            }
+        }
+
+        let dc = self.data_channel(kind);
+        if dc.lock().state() == DataState::Open {
+            return Ok(());
+        }
+
+        // Wait until the PeerConnection is connected
+        let wait_connected = async move {
+            while publisher.lock().await.is_connected() && dc.lock().state() == DataState::Open {
+                sleep(Duration::from_millis(50)).await;
+            }
+        };
+
+        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)
+            }
+        }
+    }
+
+    async fn negotiate_publisher(&self) -> EngineResult<()> {
+        self.engine_inner
+            .has_published
+            .store(true, Ordering::SeqCst);
+        if let Err(err) = self
+            .engine_inner
+            .publisher_pc
+            .lock()
+            .await
+            .negotiate()
+            .await
+        {
+            error!("failed to negotiate the publisher: {:?}", err);
+            Err(err)?
+        } else {
+            Ok(())
+        }
+    }
+
+    fn configure_engine(
+        lk_runtime: Arc<LKRuntime>,
+        join_response: JoinResponse,
+    ) -> EngineResult<(EngineInner, RTCEvents)> {
+        let (rtc_emitter, 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())?,
+        );
+
+        let mut subscriber_pc = PCTransport::new(
+            lk_runtime
+                .pc_factory
+                .create_peer_connection(rtc_config.clone())?,
+        );
+
+        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()
+            },
+        )?;
+
+        publisher_pc
+            .peer_connection()
+            .on_ice_candidate(rtc_events::on_ice_candidate(
+                SignalTarget::Publisher,
+                rtc_emitter.clone(),
+            ));
+        subscriber_pc
+            .peer_connection()
+            .on_ice_candidate(rtc_events::on_ice_candidate(
+                SignalTarget::Subscriber,
+                rtc_emitter.clone(),
+            ));
+
+        publisher_pc.on_offer(rtc_events::on_offer(
+            SignalTarget::Publisher,
+            rtc_emitter.clone(),
+        ));
+        subscriber_pc.on_offer(rtc_events::on_offer(
+            SignalTarget::Subscriber,
+            rtc_emitter.clone(),
+        ));
+
+        publisher_pc
+            .peer_connection()
+            .on_data_channel(rtc_events::on_data_channel(
+                SignalTarget::Publisher,
+                rtc_emitter.clone(),
+            ));
+        subscriber_pc
+            .peer_connection()
+            .on_data_channel(rtc_events::on_data_channel(
+                SignalTarget::Subscriber,
+                rtc_emitter.clone(),
+            ));
+
+        publisher_pc
+            .peer_connection()
+            .on_add_track(rtc_events::on_add_track(
+                SignalTarget::Publisher,
+                rtc_emitter.clone(),
+            ));
+        subscriber_pc
+            .peer_connection()
+            .on_add_track(rtc_events::on_add_track(
+                SignalTarget::Subscriber,
+                rtc_emitter.clone(),
+            ));
+
+        publisher_pc
+            .peer_connection()
+            .on_connection_change(rtc_events::on_connection_change(
+                SignalTarget::Publisher,
+                rtc_emitter.clone(),
+            ));
+        subscriber_pc
+            .peer_connection()
+            .on_connection_change(rtc_events::on_connection_change(
+                SignalTarget::Subscriber,
+                rtc_emitter.clone(),
+            ));
+
+        lossy_dc.on_message(rtc_events::on_message(rtc_emitter.clone()));
+        reliable_dc.on_message(rtc_events::on_message(rtc_emitter.clone()));
+
+        Ok((
+            EngineInner {
+                has_published: AtomicBool::new(false),
+                join_response: Mutex::new(join_response),
+                pc_state: AtomicU8::new(PCState::New as u8),
+                publisher_pc: AsyncMutex::new(publisher_pc),
+                subscriber_pc: AsyncMutex::new(subscriber_pc),
+                lossy_dc: Mutex::new(lossy_dc),
+                reliable_dc: Mutex::new(reliable_dc),
+                sub_lossy_dc: Mutex::new(None),
+                sub_reliable_dc: Mutex::new(None),
+            },
+            events,
+        ))
+    }
+
+    fn data_channel(&self, kind: data_packet::Kind) -> &Mutex<DataChannel> {
+        if kind == data_packet::Kind::Reliable {
+            &self.engine_inner.reliable_dc
+        } else {
+            &self.engine_inner.lossy_dc
+        }
     }
 }

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

@@ -13,7 +13,11 @@ use livekit_webrtc::rtc_error::RTCError;
 
 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 type OnOfferHandler = Box<
+    dyn (FnMut(SessionDescription) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>)
+        + Send
+        + Sync,
+>;
 
 pub struct PCTransport {
     peer_connection: PeerConnection,
@@ -43,7 +47,7 @@ impl PCTransport {
     pub fn is_connected(&self) -> bool {
         self.peer_connection.ice_connection_state() == IceConnectionState::IceConnectionConnected
             || self.peer_connection.ice_connection_state()
-            == IceConnectionState::IceConnectionCompleted
+                == IceConnectionState::IceConnectionCompleted
     }
 
     pub fn peer_connection(&mut self) -> &mut PeerConnection {
@@ -118,7 +122,10 @@ impl PCTransport {
                         .set_remote_description(remote_description)
                         .await?;
                 } else {
-                    event!(Level::ERROR, "trying to restart ICE when the pc doesn't have remote description");
+                    event!(
+                        Level::ERROR,
+                        "trying to restart ICE when the pc doesn't have remote description"
+                    );
                 }
             } else {
                 self.renegotiate = true;

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

@@ -0,0 +1,103 @@
+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, OnIceCandidateHandler,
+    PeerConnectionState,
+};
+use livekit_webrtc::rtp_receiver::RtpReceiver;
+use tokio::sync::mpsc;
+
+use crate::proto::SignalTarget;
+use crate::rtc_engine::pc_transport::OnOfferHandler;
+
+pub(super) type RTCEmitter = mpsc::UnboundedSender<RTCEvent>;
+pub(super) type RTCEvents = mpsc::UnboundedReceiver<RTCEvent>;
+
+#[derive(Debug)]
+pub(super) enum RTCEvent {
+    IceCandidate {
+        ice_candidate: IceCandidate,
+        target: SignalTarget,
+    },
+    ConnectionChange {
+        state: PeerConnectionState,
+        target: SignalTarget,
+    },
+    DataChannel {
+        data_channel: DataChannel,
+        target: SignalTarget,
+    },
+    Offer {
+        offer: SessionDescription,
+        target: SignalTarget,
+    },
+    AddTrack {
+        rtp_receiver: RtpReceiver,
+        streams: Vec<MediaStream>,
+        target: SignalTarget,
+    },
+    Data {
+        data: Vec<u8>,
+        binary: bool,
+    },
+}
+
+/// Handlers used to forward event to a channel
+/// Every callback here is called on the signaling thread
+
+pub(super) fn on_connection_change(
+    target: SignalTarget,
+    emitter: RTCEmitter,
+) -> OnConnectionChangeHandler {
+    Box::new(move |state| {
+        let _ = emitter.send(RTCEvent::ConnectionChange { state, target });
+    })
+}
+
+pub(super) fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler {
+    Box::new(move |ice_candidate| {
+        let _ = emitter.send(RTCEvent::IceCandidate {
+            ice_candidate,
+            target,
+        });
+    })
+}
+
+pub(super) fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler {
+    Box::new(move |offer| {
+        let _ = emitter.send(RTCEvent::Offer { offer, target });
+
+        Box::pin(async {})
+    })
+}
+
+pub(super) 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,
+        });
+    })
+}
+
+pub(super) fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler {
+    Box::new(move |rtp_receiver, streams| {
+        let _ = emitter.send(RTCEvent::AddTrack {
+            rtp_receiver,
+            streams,
+            target,
+        });
+    })
+}
+
+pub(super) fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
+    Box::new(move |data, binary| {
+        let _ = emitter.send(RTCEvent::Data {
+            data: data.to_vec(),
+            binary,
+        });
+    })
+}

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

@@ -1,18 +1,23 @@
-use core::num::flt2dec::Sign;
 use std::fmt::Debug;
+use std::time::Duration;
 
+use livekit_webrtc::peer_connection_factory::{
+    ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
+};
 use thiserror::Error;
+use tokio::sync::mpsc;
 use tokio_tungstenite::tungstenite::Error as WsError;
 
-use crate::event::{Emitter, Events};
-use crate::proto::{signal_request, signal_response};
+use crate::proto::{signal_request, signal_response, JoinResponse};
 use crate::signal_client::signal_stream::SignalStream;
 
 mod signal_stream;
 
-type SignalEmitter = Emitter<SignalEvent>;
-type SignalEvents = Events<SignalEvent>;
-type SignalResult<T> = Result<T, SignalError>;
+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 {
@@ -22,10 +27,12 @@ pub enum SignalError {
     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(Clone, Debug)]
+#[derive(Debug)]
 pub(crate) enum SignalEvent {
     Open,
     Signal(signal_response::Message),
@@ -40,6 +47,17 @@ pub(crate) struct SignalOptions {
     adaptive_stream: bool,
 }
 
+impl Default for SignalOptions {
+    fn default() -> Self {
+        Self {
+            reconnect: false,
+            auto_subscribe: true,
+            sid: "".to_string(),
+            adaptive_stream: true,
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct SignalClient {
     stream: SignalStream,
@@ -47,15 +65,16 @@ pub struct SignalClient {
 }
 
 impl SignalClient {
-    pub async fn connect(
+    pub(crate) async fn connect(
         url: &str,
         token: &str,
         options: SignalOptions,
     ) -> SignalResult<(Self, SignalEvents)> {
-        // TODO(theomonnom) Retry initial connection
-        let (emitter, receiver) = SignalEmitter::new();
-        let events = SignalEvents::new(receiver);
+        let (emitter, events) = mpsc::channel(8);
         let stream = SignalStream::connect(url, token, options, emitter.clone()).await?;
+
+        // TODO(theomonnom) Retry initial connection
+
         Ok((Self { stream, emitter }, events))
     }
 
@@ -69,3 +88,60 @@ impl SignalClient {
         // TODO(theomonnom) Close & recreate SignalStream, also send the queue if needed
     }
 }
+
+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::sync::mpsc;
+    use tokio::time::timeout;
+    use tokio_tungstenite::tungstenite::Error as WsError;
+    use tracing::{event, Level};
+
+    pub(crate) async fn next_join_response(
+        receiver: &mut mpsc::Receiver<SignalEvent>,
+    ) -> 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()))?
+    }
+}

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

@@ -1,13 +1,13 @@
-use futures_util::{SinkExt, StreamExt};
 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::{connect_async, MaybeTlsStream, WebSocketStream};
-use tokio_tungstenite::tungstenite::Message;
-use tokio_tungstenite::tungstenite::protocol::CloseFrame;
 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};
@@ -72,7 +72,7 @@ impl SignalStream {
         event!(Level::DEBUG, "connecting to websocket: {}", lk_url);
         let (ws_stream, _) = connect_async(lk_url).await?;
         event!(Level::DEBUG, "connected to websocket");
-        emitter.event(SignalEvent::Open);
+        let _ = emitter.send(SignalEvent::Open).await;
 
         let (ws_writer, ws_reader) = ws_stream.split();
         let (internal_tx, internal_rx) = mpsc::channel::<InternalMessage>(8);
@@ -119,7 +119,7 @@ impl SignalStream {
 
     /// This task is used to send messages to the websocket
     /// It is also responsible for closing the connection
-    pub async fn handle_write(
+    async fn handle_write(
         mut internal_rx: mpsc::Receiver<InternalMessage>,
         mut ws_writer: SplitSink<WebSocket, Message>,
         emitter: SignalEmitter,
@@ -136,7 +136,7 @@ impl SignalStream {
                         SignalRequest {
                             message: Some(signal),
                         }
-                            .encode_to_vec(),
+                        .encode_to_vec(),
                     );
 
                     if let Err(err) = ws_writer.send(data).await {
@@ -163,14 +163,14 @@ impl SignalStream {
         }
 
         let _ = ws_writer.close().await;
-        emitter.event(SignalEvent::Close);
+        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 )
-    pub async fn handle_read(
+    async fn handle_read(
         internal_tx: mpsc::Sender<InternalMessage>,
         mut ws_reader: SplitStream<WebSocket>,
         emitter: SignalEmitter,
@@ -181,8 +181,9 @@ impl SignalStream {
                     let res = SignalResponse::decode(data.as_slice())
                         .expect("failed to decode SignalResponse");
 
-                    event!(Level::TRACE, "received SignalResponse: {:?}", res);
-                    emitter.event(SignalEvent::Signal(res.message.unwrap()));
+                    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

crates/livekit-webrtc/libwebrtc-sys/build.rs

@@ -71,7 +71,7 @@ fn main() {
     let mut builder = cxx_build::bridges(&[
         "src/peer_connection.rs",
         "src/peer_connection_factory.rs",
-        "src/media_stream_interface.rs",
+        "src/media_stream.rs",
         "src/data_channel.rs",
         "src/jsep.rs",
         "src/candidate.rs",
@@ -83,7 +83,7 @@ fn main() {
 
     builder.file("src/peer_connection.cpp");
     builder.file("src/peer_connection_factory.cpp");
-    builder.file("src/media_stream_interface.cpp");
+    builder.file("src/media_stream.cpp");
     builder.file("src/data_channel.cpp");
     builder.file("src/jsep.cpp");
     builder.file("src/candidate.cpp");
@@ -203,7 +203,7 @@ fn main() {
             let jni_regex = Regex::new(r"(Java_org_webrtc.*)").unwrap();
             let content = &String::from_utf8_lossy(&readelf_output.stdout);
             let mut jni_symbols = Vec::new();
-            jni_regex.captures_iter(&content).for_each(|cap| {
+            jni_regex.captures_iter(content).for_each(|cap| {
                 jni_symbols.push(cap.get(1).unwrap().as_str());
             });
 
@@ -215,7 +215,7 @@ fn main() {
 
             write!(vs_file, "JNI_WEBRTC {{\n\tglobal: ").unwrap();
             write!(vs_file, "JNI_OnLoad; ").unwrap();
-            for x in &jni_symbols {
+            for x in jni_symbols {
                 println!("cargo:rustc-link-arg=-Wl,--undefined={}", x);
                 write!(vs_file, "{}; ", x).unwrap();
             }

crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h

@@ -0,0 +1,52 @@
+//
+// Created by Théo Monnom on 31/08/2022.
+//
+
+#ifndef CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H
+#define CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H
+
+#include <memory>
+
+#include "api/media_stream_interface.h"
+#include "livekit/rust_types.h"
+#include "rust/cxx.h"
+
+namespace livekit {
+
+class MediaStreamTrack {
+ public:
+  explicit MediaStreamTrack(
+      rtc::scoped_refptr<webrtc::MediaStreamTrackInterface> track);
+
+  rust::String kind() const;
+  rust::String id() const;
+
+  bool enabled() const;
+  bool set_enabled(bool enable);
+
+  TrackState state() const;
+
+ private:
+  rtc::scoped_refptr<webrtc::MediaStreamTrackInterface> track_;
+};
+
+static std::unique_ptr<MediaStreamTrack> _unique_media_stream_track() {
+  return nullptr;  // Ignore
+}
+
+class MediaStream {
+ public:
+  explicit MediaStream(rtc::scoped_refptr<webrtc::MediaStreamInterface> stream);
+
+  rust::String id() const;
+
+ private:
+  rtc::scoped_refptr<webrtc::MediaStreamInterface> media_stream_;
+};
+
+static std::unique_ptr<MediaStream> _unique_media_stream() {
+  return nullptr;  // Ignore
+}
+}  // namespace livekit
+
+#endif  // CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H

crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream_interface.h

@@ -1,28 +0,0 @@
-//
-// Created by Théo Monnom on 31/08/2022.
-//
-
-#ifndef CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H
-#define CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H
-
-#include <memory>
-
-#include "api/media_stream_interface.h"
-
-namespace livekit {
-
-class MediaStreamInterface {
- public:
-  explicit MediaStreamInterface(
-      rtc::scoped_refptr<webrtc::MediaStreamInterface> stream);
-
- private:
-  rtc::scoped_refptr<webrtc::MediaStreamInterface> media_stream_;
-};
-
-static std::unique_ptr<MediaStreamInterface> _unique_media_stream() {
-  return nullptr;  // Ignore
-}
-}  // namespace livekit
-
-#endif  // CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H

crates/livekit-webrtc/libwebrtc-sys/include/livekit/rtp_receiver.h

@@ -8,6 +8,7 @@
 #include <memory>
 
 #include "api/rtp_receiver_interface.h"
+#include "livekit/media_stream.h"
 
 namespace livekit {
 
@@ -16,6 +17,8 @@ class RtpReceiver {
   explicit RtpReceiver(
       rtc::scoped_refptr<webrtc::RtpReceiverInterface> receiver);
 
+  std::unique_ptr<MediaStreamTrack> track() const;
+
  private:
   rtc::scoped_refptr<webrtc::RtpReceiverInterface> receiver_;
 };

crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h

@@ -23,6 +23,7 @@ enum class IceConnectionState;
 enum class IceGatheringState;
 enum class SdpType;
 enum class DataState;
+enum class TrackState;
 struct SdpParseError;
 struct RTCOfferAnswerOptions;
 struct RTCError;

crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs

@@ -68,11 +68,18 @@ pub mod ffi {
             observer: Box<SetRemoteSdpObserverWrapper>,
         ) -> UniquePtr<NativeSetRemoteSdpObserverHandle>;
 
-        fn create_ice_candidate(sdp_mid: String, sdp_mline_index: i32, sdp: String) -> Result<UniquePtr<IceCandidate>>;
-        fn create_session_description(sdp_type: SdpType, sdp: String) -> Result<UniquePtr<SessionDescription>>;
+        fn create_ice_candidate(
+            sdp_mid: String,
+            sdp_mline_index: i32,
+            sdp: String,
+        ) -> Result<UniquePtr<IceCandidate>>;
+        fn create_session_description(
+            sdp_type: SdpType,
+            sdp: String,
+        ) -> Result<UniquePtr<SessionDescription>>;
 
         fn _unique_ice_candidate() -> UniquePtr<IceCandidate>; // Ignore
-    fn _unique_session_description() -> UniquePtr<SessionDescription>; // Ignore
+        fn _unique_session_description() -> UniquePtr<SessionDescription>; // Ignore
     }
 }
 
@@ -80,7 +87,11 @@ impl Error for ffi::SdpParseError {}
 
 impl Display for ffi::SdpParseError {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "SdpParseError occurred {}: {}", self.line, self.description)
+        write!(
+            f,
+            "SdpParseError occurred {}: {}",
+            self.line, self.description
+        )
     }
 }
 
@@ -101,10 +112,7 @@ impl ffi::SdpParseError {
         let line = String::from(&value[8..line_length]);
         let description = String::from(&value[line_length..]);
 
-        Self {
-            line,
-            description,
-        }
+        Self { line, description }
     }
 }
 

crates/livekit-webrtc/libwebrtc-sys/src/lib.rs

@@ -1,7 +1,7 @@
 pub mod candidate;
 pub mod data_channel;
 pub mod jsep;
-pub mod media_stream_interface;
+pub mod media_stream;
 pub mod peer_connection;
 pub mod peer_connection_factory;
 pub mod rtc_error;

crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp

@@ -0,0 +1,41 @@
+//
+// Created by Théo Monnom on 31/08/2022.
+//
+
+#include "livekit/media_stream.h"
+
+namespace livekit {
+
+MediaStreamTrack::MediaStreamTrack(
+    rtc::scoped_refptr<webrtc::MediaStreamTrackInterface> track)
+    : track_(std::move(track)) {}
+
+rust::String MediaStreamTrack::kind() const {
+  return track_->kind();
+}
+
+rust::String MediaStreamTrack::id() const {
+  return track_->id();
+}
+
+bool MediaStreamTrack::enabled() const {
+  return track_->enabled();
+}
+
+bool MediaStreamTrack::set_enabled(bool enable) {
+  return track_->set_enabled(enable);
+}
+
+TrackState MediaStreamTrack::state() const {
+  return static_cast<TrackState>(track_->state());
+}
+
+MediaStream::MediaStream(
+    rtc::scoped_refptr<webrtc::MediaStreamInterface> stream)
+    : media_stream_(std::move(stream)) {}
+
+rust::String MediaStream::id() const {
+    return media_stream_->id();
+}
+
+}  // namespace livekit

crates/livekit-webrtc/libwebrtc-sys/src/media_stream.rs

@@ -0,0 +1,36 @@
+#[cxx::bridge(namespace = "livekit")]
+pub mod ffi {
+
+    #[derive(Debug)]
+    #[repr(i32)]
+    pub enum TrackState {
+        Live,
+        Ended,
+    }
+
+    unsafe extern "C++" {
+        include!("livekit/media_stream.h");
+
+        type MediaStreamTrack;
+        type MediaStream;
+
+        fn kind(self: &MediaStreamTrack) -> String;
+        fn id(self: &MediaStreamTrack) -> String;
+        fn enabled(self: &MediaStreamTrack) -> bool;
+        fn set_enabled(self: Pin<&mut MediaStreamTrack>, enable: bool) -> bool;
+        fn state(self: &MediaStreamTrack) -> TrackState;
+
+        fn id(self: &MediaStream) -> String;
+
+        fn _unique_media_stream_track() -> UniquePtr<MediaStreamTrack>; // Ignore
+        fn _unique_media_stream() -> UniquePtr<MediaStream>; // Ignore
+    }
+}
+
+unsafe impl Sync for ffi::MediaStreamTrack {}
+
+unsafe impl Send for ffi::MediaStreamTrack {}
+
+unsafe impl Sync for ffi::MediaStream {}
+
+unsafe impl Send for ffi::MediaStream {}

crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp

@@ -1,12 +0,0 @@
-//
-// Created by Théo Monnom on 31/08/2022.
-//
-
-#include "livekit/media_stream_interface.h"
-
-namespace livekit {
-
-MediaStreamInterface::MediaStreamInterface(
-    rtc::scoped_refptr<webrtc::MediaStreamInterface> stream)
-    : media_stream_(std::move(stream)) {}
-}  // namespace livekit

crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs

@@ -1,10 +0,0 @@
-#[cxx::bridge(namespace = "livekit")]
-pub mod ffi {
-    unsafe extern "C++" {
-        include!("livekit/media_stream_interface.h");
-
-        type MediaStreamInterface;
-
-        fn _unique_media_stream() -> UniquePtr<MediaStreamInterface>; // Ignore
-    }
-}

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp

@@ -3,6 +3,7 @@
 //
 
 #include "livekit/peer_connection.h"
+#include "livekit/media_stream.h"
 
 #include "libwebrtc-sys/src/peer_connection.rs.h"
 #include "livekit/rtc_error.h"
@@ -142,12 +143,12 @@ void NativePeerConnectionObserver::OnSignalingChange(
 
 void NativePeerConnectionObserver::OnAddStream(
     rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) {
-  observer_->on_add_stream(std::make_unique<MediaStreamInterface>(stream));
+  observer_->on_add_stream(std::make_unique<MediaStream>(stream));
 }
 
 void NativePeerConnectionObserver::OnRemoveStream(
     rtc::scoped_refptr<webrtc::MediaStreamInterface> stream) {
-  observer_->on_remove_stream(std::make_unique<MediaStreamInterface>(stream));
+  observer_->on_remove_stream(std::make_unique<MediaStream>(stream));
 }
 
 void NativePeerConnectionObserver::OnDataChannel(
@@ -241,7 +242,7 @@ void NativePeerConnectionObserver::OnAddTrack(
   rust::Vec<MediaStreamPtr> vec;
 
   for (const auto& item : streams) {
-    vec.push_back(MediaStreamPtr{std::make_unique<MediaStreamInterface>(item)});
+    vec.push_back(MediaStreamPtr{std::make_unique<MediaStream>(item)});
   }
 
   observer_->on_add_track(std::make_unique<RtpReceiver>(receiver),

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs

@@ -6,7 +6,7 @@ use cxx::UniquePtr;
 use crate::candidate::ffi::Candidate;
 use crate::data_channel::ffi::DataChannel;
 use crate::jsep::ffi::IceCandidate;
-use crate::media_stream_interface::ffi::MediaStreamInterface;
+use crate::media_stream::ffi::MediaStream;
 use crate::rtc_error::ffi::RTCError;
 use crate::rtp_receiver::ffi::RtpReceiver;
 use crate::rtp_transceiver::ffi::RtpTransceiver;
@@ -83,7 +83,7 @@ pub mod ffi {
     // Wrapper to opaque C++ objects
     // https://github.com/dtolnay/cxx/issues/741
     struct MediaStreamPtr {
-        pub ptr: UniquePtr<MediaStreamInterface>,
+        pub ptr: UniquePtr<MediaStream>,
     }
 
     struct CandidatePtr {
@@ -96,7 +96,7 @@ pub mod ffi {
         include!("livekit/data_channel.h");
         include!("livekit/rtp_receiver.h");
         include!("livekit/rtp_transceiver.h");
-        include!("livekit/media_stream_interface.h");
+        include!("livekit/media_stream.h");
         include!("livekit/candidate.h");
         include!("libwebrtc-sys/src/rtc_error.rs.h");
 
@@ -106,7 +106,7 @@ pub mod ffi {
         type DataChannel = crate::data_channel::ffi::DataChannel;
         type RtpReceiver = crate::rtp_receiver::ffi::RtpReceiver;
         type RtpTransceiver = crate::rtp_transceiver::ffi::RtpTransceiver;
-        type MediaStreamInterface = crate::media_stream_interface::ffi::MediaStreamInterface;
+        type MediaStream = crate::media_stream::ffi::MediaStream;
         type NativeCreateSdpObserverHandle = crate::jsep::ffi::NativeCreateSdpObserverHandle;
         type NativeSetLocalSdpObserverHandle = crate::jsep::ffi::NativeSetLocalSdpObserverHandle;
         type NativeSetRemoteSdpObserverHandle = crate::jsep::ffi::NativeSetRemoteSdpObserverHandle;
@@ -194,14 +194,8 @@ pub mod ffi {
         type PeerConnectionObserverWrapper;
 
         fn on_signaling_change(self: &PeerConnectionObserverWrapper, new_state: SignalingState);
-        fn on_add_stream(
-            self: &PeerConnectionObserverWrapper,
-            stream: UniquePtr<MediaStreamInterface>,
-        );
-        fn on_remove_stream(
-            self: &PeerConnectionObserverWrapper,
-            stream: UniquePtr<MediaStreamInterface>,
-        );
+        fn on_add_stream(self: &PeerConnectionObserverWrapper, stream: UniquePtr<MediaStream>);
+        fn on_remove_stream(self: &PeerConnectionObserverWrapper, stream: UniquePtr<MediaStream>);
         fn on_data_channel(
             self: &PeerConnectionObserverWrapper,
             data_channel: UniquePtr<DataChannel>,
@@ -317,8 +311,8 @@ impl AddIceCandidateObserverWrapper {
 
 pub trait PeerConnectionObserver: Send + Sync {
     fn on_signaling_change(&self, new_state: ffi::SignalingState);
-    fn on_add_stream(&self, stream: UniquePtr<MediaStreamInterface>);
-    fn on_remove_stream(&self, stream: UniquePtr<MediaStreamInterface>);
+    fn on_add_stream(&self, stream: UniquePtr<MediaStream>);
+    fn on_remove_stream(&self, stream: UniquePtr<MediaStream>);
     fn on_data_channel(&self, data_channel: UniquePtr<DataChannel>);
     fn on_renegotiation_needed(&self);
     fn on_negotiation_needed_event(&self, event: u32);
@@ -338,11 +332,7 @@ pub trait PeerConnectionObserver: Send + Sync {
     fn on_ice_candidates_removed(&self, removed: Vec<UniquePtr<Candidate>>);
     fn on_ice_connection_receiving_change(&self, receiving: bool);
     fn on_ice_selected_candidate_pair_changed(&self, event: ffi::CandidatePairChangeEvent);
-    fn on_add_track(
-        &self,
-        receiver: UniquePtr<RtpReceiver>,
-        streams: Vec<UniquePtr<MediaStreamInterface>>,
-    );
+    fn on_add_track(&self, receiver: UniquePtr<RtpReceiver>, streams: Vec<UniquePtr<MediaStream>>);
     fn on_track(&self, transceiver: UniquePtr<RtpTransceiver>);
     fn on_remove_track(&self, receiver: UniquePtr<RtpReceiver>);
     fn on_interesting_usage(&self, usage_pattern: i32);
@@ -366,13 +356,13 @@ impl PeerConnectionObserverWrapper {
         }
     }
 
-    fn on_add_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
+    fn on_add_stream(&self, stream: UniquePtr<MediaStream>) {
         unsafe {
             (*self.observer).on_add_stream(stream);
         }
     }
 
-    fn on_remove_stream(&self, stream: UniquePtr<MediaStreamInterface>) {
+    fn on_remove_stream(&self, stream: UniquePtr<MediaStream>) {
         unsafe {
             (*self.observer).on_remove_stream(stream);
         }

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs

@@ -35,12 +35,14 @@ pub mod ffi {
 
         type PeerConnection = crate::peer_connection::ffi::PeerConnection;
         type NativePeerConnectionObserver =
-        crate::peer_connection::ffi::NativePeerConnectionObserver;
+            crate::peer_connection::ffi::NativePeerConnectionObserver;
         type PeerConnectionFactory;
         type NativeRTCConfiguration;
         type RTCRuntime = crate::webrtc::ffi::RTCRuntime;
 
-        fn create_peer_connection_factory(runtime: SharedPtr<RTCRuntime>) -> UniquePtr<PeerConnectionFactory>;
+        fn create_peer_connection_factory(
+            runtime: SharedPtr<RTCRuntime>,
+        ) -> UniquePtr<PeerConnectionFactory>;
         fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr<NativeRTCConfiguration>;
 
         /// SAFETY

crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp

@@ -8,4 +8,9 @@ namespace livekit {
 RtpReceiver::RtpReceiver(
     rtc::scoped_refptr<webrtc::RtpReceiverInterface> receiver)
     : receiver_(std::move(receiver)) {}
+
+std::unique_ptr<MediaStreamTrack> RtpReceiver::track() const {
+  return std::make_unique<MediaStreamTrack>(receiver_->track());
+}
+
 }  // namespace livekit

crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.rs

@@ -2,9 +2,17 @@
 pub mod ffi {
     unsafe extern "C++" {
         include!("livekit/rtp_receiver.h");
+        include!("livekit/media_stream.h");
 
+        type MediaStreamTrack = crate::media_stream::ffi::MediaStreamTrack;
         type RtpReceiver;
 
+        fn track(self: &RtpReceiver) -> UniquePtr<MediaStreamTrack>;
+
         fn _unique_rtp_receiver() -> UniquePtr<RtpReceiver>; // Ignore
     }
 }
+
+unsafe impl Sync for ffi::RtpReceiver {}
+
+unsafe impl Send for ffi::RtpReceiver {}

crates/livekit-webrtc/src/data_channel.rs

@@ -99,6 +99,12 @@ impl DataChannel {
     }
 }
 
+impl Drop for DataChannel {
+    fn drop(&mut self) {
+        self.cxx_handle.pin_mut().unregister_observer();
+    }
+}
+
 pub type OnStateChangeHandler = Box<dyn FnMut() + Send + Sync>;
 pub type OnMessageHandler = Box<dyn FnMut(&[u8], bool) + Send + Sync>;
 // data, is_binary

crates/livekit-webrtc/src/jsep.rs

@@ -17,8 +17,16 @@ impl Debug for IceCandidate {
 }
 
 impl IceCandidate {
-    pub fn from(sdp_mid: &str, sdp_mline_index: i32, sdp: &str) -> Result<IceCandidate, SdpParseError> {
-        let res = sys_jsep::ffi::create_ice_candidate(sdp_mid.to_string(), sdp_mline_index, sdp.to_string());
+    pub fn from(
+        sdp_mid: &str,
+        sdp_mline_index: i32,
+        sdp: &str,
+    ) -> Result<IceCandidate, SdpParseError> {
+        let res = sys_jsep::ffi::create_ice_candidate(
+            sdp_mid.to_string(),
+            sdp_mline_index,
+            sdp.to_string(),
+        );
 
         match res {
             Ok(cxx_handle) => Ok(IceCandidate::new(cxx_handle)),

crates/livekit-webrtc/src/media_stream.rs

@@ -1,2 +1,51 @@
-#[derive(Debug)]
-pub struct MediaStream {}
+use cxx::UniquePtr;
+
+use libwebrtc_sys::media_stream as sys_ms;
+
+pub use sys_ms::ffi::TrackState;
+
+pub struct MediaStreamTrack {
+    cxx_handle: UniquePtr<sys_ms::ffi::MediaStreamTrack>,
+}
+
+impl MediaStreamTrack {
+    pub(crate) fn new(cxx_handle: UniquePtr<sys_ms::ffi::MediaStreamTrack>) -> Self {
+        Self { cxx_handle }
+    }
+
+    fn kind(&self) -> String {
+        self.cxx_handle.kind()
+    }
+
+    fn id(&self) -> String {
+        self.cxx_handle.id()
+    }
+
+    fn enabled(&self) -> bool {
+        self.cxx_handle.enabled()
+    }
+
+    fn set_enabled(&mut self, enable: bool) -> bool {
+        self.cxx_handle.pin_mut().set_enabled(enable)
+    }
+
+    fn state(&self) -> TrackState {
+        self.cxx_handle.state()
+    }
+}
+
+pub struct MediaStream {
+    cxx_handle: UniquePtr<sys_ms::ffi::MediaStream>,
+}
+
+impl MediaStream {
+    pub(crate) fn new(cxx_handle: UniquePtr<sys_ms::ffi::MediaStream>) -> Self {
+        Self {
+            cxx_handle
+        }
+    }
+
+    pub fn id(&self) -> String {
+        self.cxx_handle.id()
+    }
+}

crates/livekit-webrtc/src/peer_connection.rs

@@ -402,10 +402,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_add_stream(
-        &self,
-        stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
-    ) {
+    fn on_add_stream(&self, stream: UniquePtr<libwebrtc_sys::media_stream::ffi::MediaStream>) {
         trace!("on_add_stream");
         let mut handler = self.on_add_stream_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -413,10 +410,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
         }
     }
 
-    fn on_remove_stream(
-        &self,
-        stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
-    ) {
+    fn on_remove_stream(&self, stream: UniquePtr<libwebrtc_sys::media_stream::ffi::MediaStream>) {
         trace!("on_remove_stream");
         let mut handler = self.on_remove_stream_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
@@ -548,12 +542,13 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
     fn on_add_track(
         &self,
         receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>,
-        streams: Vec<UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>>,
+        streams: Vec<UniquePtr<libwebrtc_sys::media_stream::ffi::MediaStream>>,
     ) {
         trace!("on_add_track");
         let mut handler = self.on_add_track_handler.lock().unwrap();
         if let Some(f) = handler.as_mut() {
-            // TODO(theomonnom)
+            let streams = streams.into_iter().map(MediaStream::new).collect();
+            f(RtpReceiver::new(receiver), streams)
         }
     }
 

crates/livekit-webrtc/src/peer_connection_factory.rs

@@ -18,7 +18,9 @@ pub struct PeerConnectionFactory {
 impl PeerConnectionFactory {
     pub fn new(rtc_runtime: RTCRuntime) -> Self {
         Self {
-            cxx_handle: sys_factory::ffi::create_peer_connection_factory(rtc_runtime.clone().release()),
+            cxx_handle: sys_factory::ffi::create_peer_connection_factory(
+                rtc_runtime.clone().release(),
+            ),
             rtc_runtime,
         }
     }
@@ -31,8 +33,9 @@ impl PeerConnectionFactory {
 
         unsafe {
             let mut observer = Box::new(InternalObserver::default());
-            let mut native_observer = sys_pc::ffi::create_native_peer_connection_observer(self.rtc_runtime.clone().release(),
-                                                                                          Box::new(sys_pc::PeerConnectionObserverWrapper::new(&mut *observer)),
+            let mut native_observer = sys_pc::ffi::create_native_peer_connection_observer(
+                self.rtc_runtime.clone().release(),
+                Box::new(sys_pc::PeerConnectionObserverWrapper::new(&mut *observer)),
             );
 
             let res = self

crates/livekit-webrtc/src/rtc_error.rs

@@ -1,3 +1,2 @@
 // TODO(theomonnom) Wrap the RTCError ffi so we can use Option(u16)
 pub use libwebrtc_sys::rtc_error::ffi::RTCError;
-

crates/livekit-webrtc/src/rtp_receiver.rs

@@ -1,2 +1,17 @@
-#[derive(Debug)]
-pub struct RtpReceiver {}
+use crate::media_stream::MediaStreamTrack;
+use cxx::UniquePtr;
+use libwebrtc_sys::rtp_receiver as sys_rec;
+
+pub struct RtpReceiver {
+    cxx_handle: UniquePtr<sys_rec::ffi::RtpReceiver>,
+}
+
+impl RtpReceiver {
+    pub(crate) fn new(cxx_handle: UniquePtr<sys_rec::ffi::RtpReceiver>) -> Self {
+        Self { cxx_handle }
+    }
+
+    pub fn track(&self) -> MediaStreamTrack {
+        MediaStreamTrack::new(self.cxx_handle.track())
+    }
+}

examples/Cargo.lock

@@ -472,6 +472,7 @@ dependencies = [
  "futures-util",
  "lazy_static",
  "livekit-webrtc",
+ "parking_lot",
  "prost",
  "prost-build",
  "prost-types",
@@ -931,6 +932,7 @@ dependencies = [
 name = "simple_room"
 version = "0.1.0"
 dependencies = [
+ "futures",
  "livekit",
  "tokio",
  "tracing",

examples/simple_room/Cargo.toml

@@ -7,4 +7,5 @@ edition = "2021"
 tokio = { version = "1", features = ["full"] }
 tracing = "0.1"
 tracing-subscriber = "0.3"
-livekit = { path = "../.." }
+livekit = { path = "../.." }
+futures = "0.3"

examples/simple_room/src/main.rs

@@ -1,7 +1,6 @@
-use std::time::Duration;
-use tokio::time::sleep;
-use livekit::proto::data_packet;
-use livekit::room;
+use livekit::room::Room;
+use std::sync::{Arc, Mutex};
+use tracing::{info, trace};
 
 const URL: &str = "ws://localhost:7880";
 const TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0NzMsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NjQ4MDY0NzMsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.BgVdBnq3XFD3_BQHoe1azqjifYysubgFl6Qlzu9IQGI";
@@ -9,14 +8,12 @@ const TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0N
 // eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY3MzAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ3ZWIiLCJuYmYiOjE2NjQ4MDY3MzAsInN1YiI6IndlYiIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.VbDoULjX1CVGZu2sPy3SvWYlVZUBXxQVPmdB9BnmlN4
 
 #[tokio::main]
-async fn main() -> Result<(), room::RoomError> {
+async fn main() {
     tracing_subscriber::fmt::init();
 
-    let mut room = room::connect(URL, TOKEN).await?;
-    room.local_participant()
-        .publish_data(b"some data", data_packet::Kind::Reliable)
-        .await?;
+    let room = Room::new();
+    room.on_participant_connected(async |participant| {
 
-    sleep(Duration::from_secs(120)).await;
-    Ok(())
+
+    })
 }

src/lib.rs

@@ -1,2 +1,2 @@
 // export everything inside livekit-core
-pub use livekit_core::*;
+pub use livekit_core::*;