diff --git a/Cargo.lock b/Cargo.lock index b50b8ab..fe5be1c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -494,6 +494,7 @@ dependencies = [ "futures-util", "lazy_static", "livekit-webrtc", + "parking_lot", "prost", "prost-build", "prost-types", diff --git a/crates/livekit-core/Cargo.toml b/crates/livekit-core/Cargo.toml index ab79746..ed20382 100644 --- a/crates/livekit-core/Cargo.toml +++ b/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" diff --git a/crates/livekit-core/src/event.rs b/crates/livekit-core/src/event.rs deleted file mode 100644 index 680a014..0000000 --- a/crates/livekit-core/src/event.rs +++ /dev/null @@ -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 { - tx: mpsc::UnboundedSender, -} - -impl Emitter { - pub fn new() -> (Self, mpsc::UnboundedReceiver) { - 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 { - rx: mpsc::UnboundedReceiver, -} - -impl Events { - pub fn new(rx: mpsc::UnboundedReceiver) -> Self { - Self { rx } - } -} - -impl Stream for Events { - type Item = T; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> { - self.rx.poll_recv(cx) - } -} diff --git a/crates/livekit-core/src/lib.rs b/crates/livekit-core/src/lib.rs index 8b972d1..bfc8c9a 100644 --- a/crates/livekit-core/src/lib.rs +++ b/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; \ No newline at end of file +pub mod room; diff --git a/crates/livekit-core/src/local_participant.rs b/crates/livekit-core/src/local_participant.rs deleted file mode 100644 index 4ffc925..0000000 --- a/crates/livekit-core/src/local_participant.rs +++ /dev/null @@ -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>, -} - -impl LocalParticipant { - pub(crate) fn from(info: ParticipantInfo, engine: Arc>) -> 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) - } -} diff --git a/crates/livekit-core/src/room.rs b/crates/livekit-core/src/room.rs deleted file mode 100644 index d2070a2..0000000 --- a/crates/livekit-core/src/room.rs +++ /dev/null @@ -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 -} - -#[tracing::instrument(skip(url, token))] -pub async fn connect(url: &str, token: &str) -> Result { - 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>, -} - -impl RoomInternal { - pub fn new(engine: Arc>) -> Self { - Self { - engine - } - } -} \ No newline at end of file diff --git a/crates/livekit-core/src/room/id.rs b/crates/livekit-core/src/room/id.rs new file mode 100644 index 0000000..f46198a --- /dev/null +++ b/crates/livekit-core/src/room/id.rs @@ -0,0 +1,38 @@ +macro_rules! id_str { + ($($name:ident;)*) => { + $( + impl From 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; +} diff --git a/crates/livekit-core/src/room/local_participant.rs b/crates/livekit-core/src/room/local_participant.rs new file mode 100644 index 0000000..5f7a96e --- /dev/null +++ b/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, +} + +impl LocalParticipant { + pub(super) fn new(rtc_engine: Arc, 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); diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs new file mode 100644 index 0000000..853d992 --- /dev/null +++ b/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 = Result; + +pub enum ConnectionState { + Disconnected, + Connecting, + Connected, + Reconnecting, +} + +struct RoomInner { + state: AtomicU8, // ConnectionState + sid: Mutex, + name: Mutex, + participants: RwLock>>, + rtc_engine: Arc, + local_participant: Arc, +} + +type OnParticipantConnectedHandler = + Box) -> BoxFuture<'static, ()> + Send + Sync>; +type OnParticipantDisconnectedHandler = OnParticipantConnectedHandler; + +struct RoomEvents { + on_participant_connected_handler: Mutex>, + on_participant_disconnected_handler: Mutex>, +} + +pub struct Room { + inner: Option>, + events: Arc, +} + +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 { + self.inner.as_ref().map(|inner| RoomHandle { + inner: inner.clone(), + }) + } + + pub fn on_participant_connected(&self, mut callback: F) + where + F: FnMut(RoomHandle, Arc) -> Fut + Send + Sync + 'static, + Fut: Future + 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(&self, mut callback: F) + where + F: FnMut(RoomHandle, Arc) -> Fut + Send + Sync + 'static, + Fut: Future + 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, + room_events: Arc, + 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, + room_events: Arc, + 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, + room_events: Arc, + 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, + room_events: Arc, + remote_participant: Arc, + ) { + 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, + sid: &ParticipantSid, + ) -> Option> { + room_inner.participants.read().get(sid).cloned() + } + + fn get_or_create_participant( + room_inner: Arc, + pi: proto::ParticipantInfo, + ) -> Arc { + 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, +} + +impl RoomHandle { + fn from(room_inner: Arc) -> 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 { + 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 + } +} diff --git a/crates/livekit-core/src/room/participant.rs b/crates/livekit-core/src/room/participant.rs new file mode 100644 index 0000000..b36c60c --- /dev/null +++ b/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, + pub(super) identity: Mutex, + pub(super) name: Mutex, + pub(super) metadata: Mutex, + pub(super) tracks: RwLock>>, +} + +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; diff --git a/crates/livekit-core/src/room/remote_participant.rs b/crates/livekit-core/src/room/remote_participant.rs new file mode 100644 index 0000000..5751cdd --- /dev/null +++ b/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 { + let track = self.shared.tracks.read().get(&sid.to_string().into()).unwrap().clone(); + + + None + } +} + +impl_participant_trait!(RemoteParticipant); diff --git a/crates/livekit-core/src/room/track.rs b/crates/livekit-core/src/room/track.rs new file mode 100644 index 0000000..8b88b74 --- /dev/null +++ b/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 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 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 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 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 for VideoTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + 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 for AudioTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + 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 for LocalTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + 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 for RemoteTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + 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"), + } + } +} diff --git a/crates/livekit-core/src/room/track_publication.rs b/crates/livekit-core/src/room/track_publication.rs new file mode 100644 index 0000000..e2c6361 --- /dev/null +++ b/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, + pub(super) sid: Mutex, + 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 +} + +#[derive(Clone)] +pub struct RemoteTrackPublication { + shared: Arc +} + +#[derive(Clone)] +pub enum TrackPublication { + Local(LocalTrackPublication), + Remote(RemoteTrackPublication) +} \ No newline at end of file diff --git a/crates/livekit-core/src/rtc_engine/engine_internal.rs b/crates/livekit-core/src/rtc_engine/engine_internal.rs deleted file mode 100644 index 7c98870..0000000 --- a/crates/livekit-core/src/rtc_engine/engine_internal.rs +++ /dev/null @@ -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, - binary: bool, - }, -} - -pub(crate) struct EngineInternal { - pub(super) publisher_pc: Arc>, - pub(super) subscriber_pc: Arc>, - pub(super) lossy_dc: Arc>, - pub(super) reliable_dc: Arc>, - pub(super) lossy_dc_sub: Arc>>, - pub(super) reliable_dc_sub: Arc>>, - - pub(super) msg_sender: mpsc::Sender, - pub(super) join_response: Mutex, - 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, - sender: mpsc::Sender, - join: JoinResponse, - ) -> Result { - 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) { - 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, - 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, - mut receiver: mpsc::Receiver, - signal_client: Arc, - ) { - 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, - signal: signal_response::Message, - signal_client: Arc, - ) -> 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, - msg: InternalMessage, - signal_client: Arc, - ) -> 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) -> 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> { - if kind == data_packet::Kind::Reliable { - self.reliable_dc.clone() - } else { - self.lossy_dc.clone() - } - } -} diff --git a/crates/livekit-core/src/lk_runtime.rs b/crates/livekit-core/src/rtc_engine/lk_runtime.rs similarity index 100% rename from crates/livekit-core/src/lk_runtime.rs rename to crates/livekit-core/src/rtc_engine/lk_runtime.rs diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index 180097f..86fd80f 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/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> = Mutex::new(Weak::new()); } +pub(crate) type EngineEmitter = mpsc::Sender; +pub(crate) type EngineEvents = mpsc::Receiver; +pub(crate) type EngineResult = Result; + 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, + }, } #[derive(Debug)] -pub enum EngineEvent { - DataReceived(Packet), +struct EngineInner { + has_published: AtomicBool, + join_response: Mutex, + pc_state: AtomicU8, // Casted to PCState enum + + publisher_pc: AsyncMutex, + subscriber_pc: AsyncMutex, + + // Publisher data channels + // Used to send data to other participants ( The SFU forward the messages ) + lossy_dc: Mutex, + reliable_dc: Mutex, + + // 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>, + sub_lossy_dc: Mutex>, } #[derive(Debug)] pub struct RTCEngine { signal_client: Arc, - internal: Arc, + engine_inner: Arc, #[allow(unused)] lk_runtime: Arc, // 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 { - // 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, + engine_inner: Arc, + 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, + engine_inner: Arc, + 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, + engine_inner: Arc, + 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, + engine_inner: Arc, + 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, + 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 { + if kind == data_packet::Kind::Reliable { + &self.engine_inner.reliable_dc + } else { + &self.engine_inner.lossy_dc + } } } diff --git a/crates/livekit-core/src/pc_transport.rs b/crates/livekit-core/src/rtc_engine/pc_transport.rs similarity index 90% rename from crates/livekit-core/src/pc_transport.rs rename to crates/livekit-core/src/rtc_engine/pc_transport.rs index ca4f2b2..8d9676a 100644 --- a/crates/livekit-core/src/pc_transport.rs +++ b/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 Pin + Send + 'static>>) + Send + Sync>; +pub type OnOfferHandler = Box< + dyn (FnMut(SessionDescription) -> Pin + 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; diff --git a/crates/livekit-core/src/rtc_engine/rtc_events.rs b/crates/livekit-core/src/rtc_engine/rtc_events.rs new file mode 100644 index 0000000..78712d7 --- /dev/null +++ b/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; +pub(super) type RTCEvents = mpsc::UnboundedReceiver; + +#[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, + target: SignalTarget, + }, + Data { + data: Vec, + 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, + }); + }) +} diff --git a/crates/livekit-core/src/signal_client/mod.rs b/crates/livekit-core/src/signal_client/mod.rs index b02e526..9e9d7c4 100644 --- a/crates/livekit-core/src/signal_client/mod.rs +++ b/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; -type SignalEvents = Events; -type SignalResult = Result; +pub(crate) type SignalEmitter = mpsc::Sender; +pub(crate) type SignalEvents = mpsc::Receiver; +pub(crate) type SignalResult = Result; + +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 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, + ) -> SignalResult { + 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()))? + } +} diff --git a/crates/livekit-core/src/signal_client/signal_stream.rs b/crates/livekit-core/src/signal_client/signal_stream.rs index 5c0d7bd..eb0680d 100644 --- a/crates/livekit-core/src/signal_client/signal_stream.rs +++ b/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::(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, mut ws_writer: SplitSink, 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, mut ws_reader: SplitStream, 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 diff --git a/crates/livekit-webrtc/libwebrtc-sys/build.rs b/crates/livekit-webrtc/libwebrtc-sys/build.rs index 909f297..c6d95cc 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/build.rs +++ b/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(); } diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h new file mode 100644 index 0000000..c2b3164 --- /dev/null +++ b/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 + +#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 track); + + rust::String kind() const; + rust::String id() const; + + bool enabled() const; + bool set_enabled(bool enable); + + TrackState state() const; + + private: + rtc::scoped_refptr track_; +}; + +static std::unique_ptr _unique_media_stream_track() { + return nullptr; // Ignore +} + +class MediaStream { + public: + explicit MediaStream(rtc::scoped_refptr stream); + + rust::String id() const; + + private: + rtc::scoped_refptr media_stream_; +}; + +static std::unique_ptr _unique_media_stream() { + return nullptr; // Ignore +} +} // namespace livekit + +#endif // CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream_interface.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream_interface.h deleted file mode 100644 index 41f71c4..0000000 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream_interface.h +++ /dev/null @@ -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 - -#include "api/media_stream_interface.h" - -namespace livekit { - -class MediaStreamInterface { - public: - explicit MediaStreamInterface( - rtc::scoped_refptr stream); - - private: - rtc::scoped_refptr media_stream_; -}; - -static std::unique_ptr _unique_media_stream() { - return nullptr; // Ignore -} -} // namespace livekit - -#endif // CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rtp_receiver.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rtp_receiver.h index 636fb79..f541cf5 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rtp_receiver.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rtp_receiver.h @@ -8,6 +8,7 @@ #include #include "api/rtp_receiver_interface.h" +#include "livekit/media_stream.h" namespace livekit { @@ -16,6 +17,8 @@ class RtpReceiver { explicit RtpReceiver( rtc::scoped_refptr receiver); + std::unique_ptr track() const; + private: rtc::scoped_refptr receiver_; }; diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h index 4f766a1..1d9d7ef 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h +++ b/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; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs index 7b15d29..4ba65c2 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs @@ -68,11 +68,18 @@ pub mod ffi { observer: Box, ) -> UniquePtr; - fn create_ice_candidate(sdp_mid: String, sdp_mline_index: i32, sdp: String) -> Result>; - fn create_session_description(sdp_type: SdpType, sdp: String) -> Result>; + fn create_ice_candidate( + sdp_mid: String, + sdp_mline_index: i32, + sdp: String, + ) -> Result>; + fn create_session_description( + sdp_type: SdpType, + sdp: String, + ) -> Result>; fn _unique_ice_candidate() -> UniquePtr; // Ignore - fn _unique_session_description() -> UniquePtr; // Ignore + fn _unique_session_description() -> UniquePtr; // 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 } } } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs b/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs index 9959239..ecd1f3e 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs +++ b/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; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp new file mode 100644 index 0000000..04eb1f9 --- /dev/null +++ b/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 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(track_->state()); +} + +MediaStream::MediaStream( + rtc::scoped_refptr stream) + : media_stream_(std::move(stream)) {} + +rust::String MediaStream::id() const { + return media_stream_->id(); +} + +} // namespace livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.rs b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.rs new file mode 100644 index 0000000..0f46ebf --- /dev/null +++ b/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; // Ignore + fn _unique_media_stream() -> UniquePtr; // 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 {} \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp deleted file mode 100644 index 8f5936c..0000000 --- a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.cpp +++ /dev/null @@ -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 stream) - : media_stream_(std::move(stream)) {} -} // namespace livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs deleted file mode 100644 index 1999583..0000000 --- a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream_interface.rs +++ /dev/null @@ -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; // Ignore - } -} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp index 203c164..1ad9738 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp +++ b/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 stream) { - observer_->on_add_stream(std::make_unique(stream)); + observer_->on_add_stream(std::make_unique(stream)); } void NativePeerConnectionObserver::OnRemoveStream( rtc::scoped_refptr stream) { - observer_->on_remove_stream(std::make_unique(stream)); + observer_->on_remove_stream(std::make_unique(stream)); } void NativePeerConnectionObserver::OnDataChannel( @@ -241,7 +242,7 @@ void NativePeerConnectionObserver::OnAddTrack( rust::Vec vec; for (const auto& item : streams) { - vec.push_back(MediaStreamPtr{std::make_unique(item)}); + vec.push_back(MediaStreamPtr{std::make_unique(item)}); } observer_->on_add_track(std::make_unique(receiver), diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index 6b82f5f..f6b914f 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/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, + pub ptr: UniquePtr, } 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, - ); - fn on_remove_stream( - self: &PeerConnectionObserverWrapper, - stream: UniquePtr, - ); + fn on_add_stream(self: &PeerConnectionObserverWrapper, stream: UniquePtr); + fn on_remove_stream(self: &PeerConnectionObserverWrapper, stream: UniquePtr); fn on_data_channel( self: &PeerConnectionObserverWrapper, data_channel: UniquePtr, @@ -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); - fn on_remove_stream(&self, stream: UniquePtr); + fn on_add_stream(&self, stream: UniquePtr); + fn on_remove_stream(&self, stream: UniquePtr); fn on_data_channel(&self, data_channel: UniquePtr); 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>); 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, - streams: Vec>, - ); + fn on_add_track(&self, receiver: UniquePtr, streams: Vec>); fn on_track(&self, transceiver: UniquePtr); fn on_remove_track(&self, receiver: UniquePtr); fn on_interesting_usage(&self, usage_pattern: i32); @@ -366,13 +356,13 @@ impl PeerConnectionObserverWrapper { } } - fn on_add_stream(&self, stream: UniquePtr) { + fn on_add_stream(&self, stream: UniquePtr) { unsafe { (*self.observer).on_add_stream(stream); } } - fn on_remove_stream(&self, stream: UniquePtr) { + fn on_remove_stream(&self, stream: UniquePtr) { unsafe { (*self.observer).on_remove_stream(stream); } diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs index 690f605..6266242 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs +++ b/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) -> UniquePtr; + fn create_peer_connection_factory( + runtime: SharedPtr, + ) -> UniquePtr; fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr; /// SAFETY diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp index b413a32..e9095c0 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp @@ -8,4 +8,9 @@ namespace livekit { RtpReceiver::RtpReceiver( rtc::scoped_refptr receiver) : receiver_(std::move(receiver)) {} + +std::unique_ptr RtpReceiver::track() const { + return std::make_unique(receiver_->track()); +} + } // namespace livekit \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.rs b/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.rs index 6b6cc92..34327de 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.rs +++ b/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; + fn _unique_rtp_receiver() -> UniquePtr; // Ignore } } + +unsafe impl Sync for ffi::RtpReceiver {} + +unsafe impl Send for ffi::RtpReceiver {} \ No newline at end of file diff --git a/crates/livekit-webrtc/src/data_channel.rs b/crates/livekit-webrtc/src/data_channel.rs index 9af7620..a45c6ca 100644 --- a/crates/livekit-webrtc/src/data_channel.rs +++ b/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; pub type OnMessageHandler = Box; // data, is_binary diff --git a/crates/livekit-webrtc/src/jsep.rs b/crates/livekit-webrtc/src/jsep.rs index 8117830..228ba62 100644 --- a/crates/livekit-webrtc/src/jsep.rs +++ b/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 { - 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 { + 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)), diff --git a/crates/livekit-webrtc/src/media_stream.rs b/crates/livekit-webrtc/src/media_stream.rs index 9a693e7..3fff376 100644 --- a/crates/livekit-webrtc/src/media_stream.rs +++ b/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, +} + +impl MediaStreamTrack { + pub(crate) fn new(cxx_handle: UniquePtr) -> 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, +} + +impl MediaStream { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + Self { + cxx_handle + } + } + + pub fn id(&self) -> String { + self.cxx_handle.id() + } +} \ No newline at end of file diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index 1b0dedd..4d44676 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -402,10 +402,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver { } } - fn on_add_stream( - &self, - stream: UniquePtr, - ) { + fn on_add_stream(&self, stream: UniquePtr) { 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, - ) { + fn on_remove_stream(&self, stream: UniquePtr) { 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, - streams: Vec>, + streams: Vec>, ) { 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) } } diff --git a/crates/livekit-webrtc/src/peer_connection_factory.rs b/crates/livekit-webrtc/src/peer_connection_factory.rs index 09f06ff..90291b3 100644 --- a/crates/livekit-webrtc/src/peer_connection_factory.rs +++ b/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 diff --git a/crates/livekit-webrtc/src/rtc_error.rs b/crates/livekit-webrtc/src/rtc_error.rs index e054cf4..86c81e1 100644 --- a/crates/livekit-webrtc/src/rtc_error.rs +++ b/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; - diff --git a/crates/livekit-webrtc/src/rtp_receiver.rs b/crates/livekit-webrtc/src/rtp_receiver.rs index d7f431b..0bcfee3 100644 --- a/crates/livekit-webrtc/src/rtp_receiver.rs +++ b/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, +} + +impl RtpReceiver { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } + + pub fn track(&self) -> MediaStreamTrack { + MediaStreamTrack::new(self.cxx_handle.track()) + } +} diff --git a/examples/Cargo.lock b/examples/Cargo.lock index 9a656ed..c82ce3c 100644 --- a/examples/Cargo.lock +++ b/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", diff --git a/examples/simple_room/Cargo.toml b/examples/simple_room/Cargo.toml index 85f7276..1f13b23 100644 --- a/examples/simple_room/Cargo.toml +++ b/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 = "../.." } \ No newline at end of file +livekit = { path = "../.." } +futures = "0.3" \ No newline at end of file diff --git a/examples/simple_room/src/main.rs b/examples/simple_room/src/main.rs index 1c09fbe..52b161c 100644 --- a/examples/simple_room/src/main.rs +++ b/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(()) + + }) } diff --git a/src/lib.rs b/src/lib.rs index fe7ff25..ef89eee 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,2 +1,2 @@ // export everything inside livekit-core -pub use livekit_core::*; \ No newline at end of file +pub use livekit_core::*;