From ae776f04d0352fd45b7e2d7d6d834c9e8a83d472 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Tue, 22 Nov 2022 17:59:55 +0100 Subject: [PATCH] Initial downstream tracks --- crates/livekit-core/Cargo.toml | 2 +- crates/livekit-core/src/events.rs | 140 +++++++++ crates/livekit-core/src/lib.rs | 4 + crates/livekit-core/src/room/mod.rs | 104 +++---- .../{ => participant}/local_participant.rs | 3 +- .../{participant.rs => participant/mod.rs} | 88 ++++-- .../room/participant/remote_participant.rs | 149 ++++++++++ .../mod.rs} | 56 +++- .../src/room/remote_participant.rs | 83 ------ .../src/room/track/audio_track.rs | 31 ++ crates/livekit-core/src/room/track/events.rs | 1 + .../src/room/track/local_audio_track.rs | 7 + .../src/room/track/local_track.rs | 31 ++ .../src/room/track/local_video_track.rs | 7 + crates/livekit-core/src/room/track/mod.rs | 242 +++++++-------- .../src/room/track/remote_audio_track.rs | 30 ++ .../src/room/track/remote_track.rs | 49 ++++ .../src/room/track/remote_video_track.rs | 31 ++ .../src/room/track/video_track.rs | 31 ++ crates/livekit-core/src/utils.rs | 26 ++ .../libwebrtc-sys/CMakeLists.txt | 23 -- crates/livekit-webrtc/libwebrtc-sys/build.rs | 4 +- .../libwebrtc-sys/compile_flags.txt | 8 + .../include/livekit/media_stream.h | 111 +++++-- .../include/livekit/rust_types.h | 4 + .../include/livekit/video_frame.h | 48 +++ .../include/livekit/video_frame_buffer.h | 93 ++++++ .../libwebrtc-sys/src/candidate.cpp | 2 +- .../libwebrtc-sys/src/data_channel.cpp | 2 +- .../livekit-webrtc/libwebrtc-sys/src/jsep.cpp | 2 +- .../livekit-webrtc/libwebrtc-sys/src/lib.rs | 6 + .../libwebrtc-sys/src/media_stream.cpp | 72 ++++- .../libwebrtc-sys/src/media_stream.rs | 101 ++++++- .../libwebrtc-sys/src/peer_connection.rs | 2 +- .../src/peer_connection_factory.rs | 2 +- .../libwebrtc-sys/src/rtp_receiver.cpp | 2 +- .../libwebrtc-sys/src/rtp_receiver.rs | 2 +- .../libwebrtc-sys/src/video_frame.rs | 32 ++ .../libwebrtc-sys/src/video_frame_buffer.rs | 43 +++ .../src/.media_stream.rs.rustfmt | 277 ++++++++++++++++++ crates/livekit-webrtc/src/data_channel.rs | 18 +- crates/livekit-webrtc/src/jsep.rs | 2 +- crates/livekit-webrtc/src/lib.rs | 2 + crates/livekit-webrtc/src/media_stream.rs | 243 +++++++++++++-- crates/livekit-webrtc/src/peer_connection.rs | 2 +- crates/livekit-webrtc/src/rtp_receiver.rs | 10 +- crates/livekit-webrtc/src/video_frame.rs | 50 ++++ .../livekit-webrtc/src/video_frame_buffer.rs | 18 ++ examples/simple_room/src/main.rs | 20 +- 49 files changed, 1910 insertions(+), 406 deletions(-) create mode 100644 crates/livekit-core/src/events.rs rename crates/livekit-core/src/room/{ => participant}/local_participant.rs (93%) rename crates/livekit-core/src/room/{participant.rs => participant/mod.rs} (52%) create mode 100644 crates/livekit-core/src/room/participant/remote_participant.rs rename crates/livekit-core/src/room/{track_publication.rs => publication/mod.rs} (61%) delete mode 100644 crates/livekit-core/src/room/remote_participant.rs create mode 100644 crates/livekit-core/src/room/track/audio_track.rs create mode 100644 crates/livekit-core/src/room/track/events.rs create mode 100644 crates/livekit-core/src/room/track/local_audio_track.rs create mode 100644 crates/livekit-core/src/room/track/local_track.rs create mode 100644 crates/livekit-core/src/room/track/local_video_track.rs create mode 100644 crates/livekit-core/src/room/track/remote_audio_track.rs create mode 100644 crates/livekit-core/src/room/track/remote_track.rs create mode 100644 crates/livekit-core/src/room/track/remote_video_track.rs create mode 100644 crates/livekit-core/src/room/track/video_track.rs create mode 100644 crates/livekit-core/src/utils.rs delete mode 100644 crates/livekit-webrtc/libwebrtc-sys/CMakeLists.txt create mode 100644 crates/livekit-webrtc/libwebrtc-sys/compile_flags.txt create mode 100644 crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h create mode 100644 crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame_buffer.h create mode 100644 crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs create mode 100644 crates/livekit-webrtc/libwebrtc-sys/src/video_frame_buffer.rs create mode 100644 crates/livekit-webrtc/src/.media_stream.rs.rustfmt create mode 100644 crates/livekit-webrtc/src/video_frame.rs create mode 100644 crates/livekit-webrtc/src/video_frame_buffer.rs diff --git a/crates/livekit-core/Cargo.toml b/crates/livekit-core/Cargo.toml index ed20382..a9ce7cd 100644 --- a/crates/livekit-core/Cargo.toml +++ b/crates/livekit-core/Cargo.toml @@ -10,7 +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" +parking_lot = { version = "0.12.1", features = ["send_guard"] } url = "2.2.2" futures-util = "0.3.23" thiserror = "1.0" diff --git a/crates/livekit-core/src/events.rs b/crates/livekit-core/src/events.rs new file mode 100644 index 0000000..5dfecd7 --- /dev/null +++ b/crates/livekit-core/src/events.rs @@ -0,0 +1,140 @@ +use futures_util::future::BoxFuture; +use thiserror::Error; + +type EventHandler = Box BoxFuture<'static, ()> + Send + Sync>; + +macro_rules! event_setter { + ($fnc:ident, $event:ty) => { + pub fn $fnc(&self, mut callback: F) + where + F: FnMut($event) -> Fut + Send + Sync + 'static, + Fut: Future + Send + Sync + 'static, + { + *self.$fnc.lock() = Some(Box::new(move |event| Box::pin(callback(event)))); + } + }; +} + +#[derive(Error, Debug, Clone)] +pub enum TrackError { + #[error("could not find published track with sid: {0}")] + TrackNotFound(String), +} + +pub mod room { + use super::{EventHandler, TrackError}; + use crate::room::id::TrackSid; + use crate::room::participant::remote_participant::RemoteParticipant; + use crate::room::publication::RemoteTrackPublication; + use crate::room::track::remote_track::RemoteTrackHandle; + use crate::room::RoomHandle; + use futures::future::Future; + use parking_lot::Mutex; + use std::sync::Arc; + + #[derive(Clone)] + pub struct ParticipantConnectedEvent { + pub room_handle: RoomHandle, + pub participant: Arc, + } + + #[derive(Clone)] + pub struct ParticipantDisconnectedEvent { + pub room_handle: RoomHandle, + pub participant: Arc, + } + + #[derive(Clone)] + pub struct TrackSubscribedEvent { + pub room_handle: RoomHandle, + pub track: RemoteTrackHandle, + pub publication: RemoteTrackPublication, + pub participant: Arc, + } + + #[derive(Clone)] + pub struct TrackPublishedEvent { + pub room_handle: RoomHandle, + pub publication: RemoteTrackPublication, + pub participant: Arc, + } + + #[derive(Clone)] + pub struct TrackSubscriptionFailedEvent { + pub room_handle: RoomHandle, + pub error: TrackError, + pub sid: TrackSid, + pub participant: Arc, + } + + pub(crate) type OnParticipantConnectedHandler = EventHandler; + pub(crate) type OnParticipantDisconnectedHandler = EventHandler; + pub(crate) type OnTrackSubscribedEventHandler = EventHandler; + pub(crate) type OnTrackPublishedEventHandler = EventHandler; + pub(crate) type OnTrackSubscriptionFailedHandler = EventHandler; + + #[derive(Default)] + pub struct RoomEvents { + pub(crate) on_participant_connected: Mutex>, + pub(crate) on_participant_disconnected: Mutex>, + pub(crate) on_track_subscribed: Mutex>, + pub(crate) on_track_published: Mutex>, + pub(crate) on_track_subscription_failed: Mutex>, + } + + impl RoomEvents { + event_setter!(on_participant_connected, ParticipantConnectedEvent); + event_setter!(on_participant_disconnected, ParticipantDisconnectedEvent); + event_setter!(on_track_subscribed, TrackSubscribedEvent); + event_setter!(on_track_published, TrackPublishedEvent); + event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent); + } +} + +pub mod participant { + use super::{EventHandler, TrackError}; + use crate::room::id::TrackSid; + use crate::room::participant::remote_participant::RemoteParticipant; + use crate::room::publication::RemoteTrackPublication; + use crate::room::track::remote_track::RemoteTrackHandle; + use futures::future::Future; + use parking_lot::Mutex; + use std::sync::Arc; + + #[derive(Clone)] + pub struct TrackPublishedEvent { + pub publication: RemoteTrackPublication, + pub participant: Arc, + } + + #[derive(Clone)] + pub struct TrackSubscribedEvent { + pub track: RemoteTrackHandle, + pub publication: RemoteTrackPublication, + pub participant: Arc, + } + + #[derive(Clone)] + pub struct TrackSubscriptionFailedEvent { + pub sid: TrackSid, + pub error: TrackError, + pub participant: Arc, + } + + pub(crate) type TrackPublishedHandler = EventHandler; + pub(crate) type TrackSubscribedHandler = EventHandler; + pub(crate) type TrackSubscriptionFailedHandler = EventHandler; + + #[derive(Default)] + pub struct ParticipantEvents { + pub(crate) on_track_published: Mutex>, + pub(crate) on_track_subscribed: Mutex>, + pub(crate) on_track_subscription_failed: Mutex>, + } + + impl ParticipantEvents { + event_setter!(on_track_published, TrackPublishedEvent); + event_setter!(on_track_subscribed, TrackSubscribedEvent); + event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent); + } +} diff --git a/crates/livekit-core/src/lib.rs b/crates/livekit-core/src/lib.rs index 15669d3..a9770a7 100644 --- a/crates/livekit-core/src/lib.rs +++ b/crates/livekit-core/src/lib.rs @@ -1,8 +1,12 @@ +extern crate core; + pub mod proto { include!(concat!(env!("OUT_DIR"), "/livekit.rs")); } +mod events; mod rtc_engine; mod signal_client; +mod utils; pub mod room; diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index 84f9285..c293d29 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -1,29 +1,27 @@ -use futures_util::future::BoxFuture; use parking_lot::lock_api::RwLockUpgradableReadGuard; use parking_lot::{Mutex, RwLock}; use std::collections::HashMap; -use std::future::Future; use std::sync::atomic::AtomicU8; use std::sync::Arc; +use self::id::ParticipantSid; +use self::participant::local_participant::LocalParticipant; +use self::participant::remote_participant::RemoteParticipant; +use self::participant::ParticipantInternalTrait; +use self::participant::ParticipantTrait; +use crate::events::room::{ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents}; 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 crate::proto::participant_info; 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; +pub mod id; +pub mod participant; +pub mod publication; +pub mod track; #[derive(Error, Debug)] pub enum RoomError { @@ -52,15 +50,6 @@ struct RoomInner { 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, @@ -70,13 +59,14 @@ 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(), - }), + events: Default::default(), } } + pub fn events(&self) -> Arc { + self.events.clone() + } + pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> { let (rtc_engine, engine_events) = RTCEngine::connect(url, token, SignalOptions::default()).await?; @@ -109,28 +99,6 @@ impl Room { }) } - 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, @@ -178,7 +146,10 @@ impl Room { Self::get_participant(room_inner.clone(), &participant_sid.to_string().into()); if let Some(remote_participant) = remote_participant { - + remote_participant.add_subscribed_media_track( + track_sid.to_string().into(), + rtp_receiver.track(), + ); } else { // The server should send participant updates before sending a new offer // So this should not happen. @@ -223,13 +194,14 @@ impl Room { } } 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(), - ); + let remote_participant = + Self::get_or_create_participant(room_inner.clone(), room_events.clone(), pi); + let mut handler = room_events.on_participant_connected.lock(); + if let Some(cb) = handler.as_mut() { + cb(ParticipantConnectedEvent { + room_handle: RoomHandle::from(room_inner.clone()), + participant: remote_participant.clone(), + }); } } } @@ -247,12 +219,12 @@ impl Room { // 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(), - ); + let mut handler = room_events.on_participant_disconnected.lock(); + if let Some(cb) = handler.as_mut() { + cb(ParticipantDisconnectedEvent { + room_handle: RoomHandle::from(room_inner.clone()), + participant: remote_participant.clone(), + }); } } @@ -265,6 +237,7 @@ impl Room { fn get_or_create_participant( room_inner: Arc, + room_events: Arc, pi: proto::ParticipantInfo, ) -> Arc { let participants = room_inner.participants.upgradable_read(); @@ -275,6 +248,13 @@ impl Room { } else { let mut participants = RwLockUpgradableReadGuard::upgrade(participants); let p = Arc::new(RemoteParticipant::new(pi)); + + // Forward participantevents to room events + p.internal_events().on_track_published({ + let room_events = room_events.clone(); + |event| async move {} + }); + participants.insert(sid, p.clone()); p } diff --git a/crates/livekit-core/src/room/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs similarity index 93% rename from crates/livekit-core/src/room/local_participant.rs rename to crates/livekit-core/src/room/participant/local_participant.rs index 5f7a96e..ebc877f 100644 --- a/crates/livekit-core/src/room/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -2,7 +2,6 @@ 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, @@ -10,7 +9,7 @@ pub struct LocalParticipant { } impl LocalParticipant { - pub(super) fn new(rtc_engine: Arc, info: ParticipantInfo) -> Self { + pub(crate) fn new(rtc_engine: Arc, info: ParticipantInfo) -> Self { Self { shared: ParticipantShared::new( info.sid.into(), diff --git a/crates/livekit-core/src/room/participant.rs b/crates/livekit-core/src/room/participant/mod.rs similarity index 52% rename from crates/livekit-core/src/room/participant.rs rename to crates/livekit-core/src/room/participant/mod.rs index 425d32e..9158452 100644 --- a/crates/livekit-core/src/room/participant.rs +++ b/crates/livekit-core/src/room/participant/mod.rs @@ -1,11 +1,23 @@ +use crate::events::participant::ParticipantEvents; use crate::proto::ParticipantInfo; -use crate::room::local_participant::LocalParticipant; -use crate::room::remote_participant::RemoteParticipant; +use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; +use crate::room::participant::local_participant::LocalParticipant; +use crate::room::participant::remote_participant::RemoteParticipant; +use crate::room::publication::{TrackPublication, TrackPublicationTrait}; +use crate::utils::wrap_variants; +use futures_util::future::BoxFuture; use parking_lot::{Mutex, RwLock}; use std::collections::HashMap; use std::sync::Arc; +pub mod local_participant; +pub mod remote_participant; + +type OnTrackSubscribed = Box BoxFuture<'static, ()> + Send + Sync>; + pub(super) struct ParticipantShared { + pub(super) events: Arc, + pub(super) internal_events: Arc, pub(super) sid: Mutex, pub(super) identity: Mutex, pub(super) name: Mutex, @@ -21,6 +33,8 @@ impl ParticipantShared { metadata: String, ) -> Self { Self { + events: Default::default(), + internal_events: Default::default(), sid: Mutex::new(sid), identity: Mutex::new(identity), name: Mutex::new(name), @@ -33,11 +47,20 @@ impl ParticipantShared { *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 + *self.metadata.lock() = info.metadata; // TODO(theomonnom): callback MetadataChanged + } + + pub(crate) fn add_track_publication(&self, publication: TrackPublication) { + self.tracks.write().insert(publication.sid(), publication); } } +pub(crate) trait ParticipantInternalTrait { + fn internal_events(&self) -> Arc; +} + pub trait ParticipantTrait { + fn events(&self) -> Arc; fn sid(&self) -> ParticipantSid; fn identity(&self) -> ParticipantIdentity; fn name(&self) -> String; @@ -45,42 +68,49 @@ pub trait ParticipantTrait { fn update_info(&self, info: ParticipantInfo); } -pub enum Participant { - Local(LocalParticipant), - Remote(RemoteParticipant), +#[derive(Clone)] +pub enum ParticipantHandle { + Local(Arc), + Remote(Arc), } -macro_rules! shared_getter { - ($x:ident, $ret:ident) => { - fn $x(&self) -> $ret { - match self { - Participant::Local(p) => p.$x(), - Participant::Remote(p) => p.$x(), - } - } - }; +impl ParticipantInternalTrait for ParticipantHandle { + wrap_variants!( + [Local, Remote] + fnc!(internal_events, Arc, []); + ); } -impl ParticipantTrait for Participant { - shared_getter!(sid, ParticipantSid); - shared_getter!(identity, ParticipantIdentity); - shared_getter!(name, String); - shared_getter!(metadata, String); - - fn update_info(&self, info: ParticipantInfo) { - match self { - Participant::Local(p) => p.update_info(info), - Participant::Remote(p) => p.update_info(info), - } - } +impl ParticipantTrait for ParticipantHandle { + wrap_variants!( + [Local, Remote] + fnc!(events, Arc, []); + fnc!(sid, ParticipantSid, []); + fnc!(identity, ParticipantIdentity, []); + fnc!(name, String, []); + fnc!(metadata, String, []); + fnc!(update_info, (), [info: ParticipantInfo]); + ); } macro_rules! impl_participant_trait { - ($x:ident) => { + ($x:ty) => { + use crate::events::participant::ParticipantEvents; use crate::proto::ParticipantInfo; use crate::room::id::{ParticipantIdentity, ParticipantSid}; + use std::sync::Arc; + + impl crate::room::participant::ParticipantInternalTrait for $x { + fn internal_events(&self) -> Arc { + self.shared.internal_events.clone() + } + } impl crate::room::participant::ParticipantTrait for $x { + fn events(&self) -> Arc { + self.shared.events.clone() + } + fn sid(&self) -> ParticipantSid { self.shared.sid.lock().clone() } @@ -104,6 +134,4 @@ macro_rules! impl_participant_trait { }; } -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/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs new file mode 100644 index 0000000..1266504 --- /dev/null +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -0,0 +1,149 @@ +use crate::events::participant::{TrackSubscribedEvent, TrackSubscriptionFailedEvent}; +use crate::events::TrackError; +use crate::room::id::TrackSid; +use crate::room::participant::{impl_participant_trait, ParticipantShared}; +use crate::room::publication::{RemoteTrackPublication, TrackPublication, TrackPublicationTrait}; +use crate::room::track::remote_audio_track::RemoteAudioTrack; +use crate::room::track::remote_track::RemoteTrackHandle; +use crate::room::track::remote_video_track::RemoteVideoTrack; +use crate::room::track::{TrackTrait, TrackKind}; +use livekit_webrtc::media_stream::MediaStreamTrackHandle; +use std::time::Duration; +use tokio::time::{sleep, timeout}; +use tracing::error; + +const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5); + +pub struct RemoteParticipant { + shared: ParticipantShared, +} + +impl RemoteParticipant { + pub(crate) fn new(info: ParticipantInfo) -> Self { + Self { + shared: ParticipantShared::new( + info.sid.into(), + info.identity.into(), + info.name, + info.metadata, + ), + } + } + + pub(crate) fn add_subscribed_media_track( + self: Arc, + sid: TrackSid, + media_track: MediaStreamTrackHandle, + ) { + tokio::spawn(async move { + let wait_publication = { + let participant = self.clone(); + let sid = sid.clone(); + async move { + loop { + let publication = participant.get_track_publication(&sid); + if let Some(publication) = publication { + return publication; + } + + sleep(Duration::from_millis(50)).await; + } + } + }; + + if let Ok(remote_publication) = timeout(ADD_TRACK_TIMEOUT, wait_publication).await { + let track = match remote_publication.kind() { + TrackKind::Audio => { + if let MediaStreamTrackHandle::Audio(rtc_track) = media_track { + let audio_track = RemoteAudioTrack::new( + remote_publication.sid().into(), + remote_publication.name(), + rtc_track, + ); + RemoteTrackHandle::Audio(Arc::new(audio_track)) + } else { + unreachable!(); + } + } + TrackKind::Video => { + if let MediaStreamTrackHandle::Video(rtc_track) = media_track { + let video_track = RemoteVideoTrack::new( + remote_publication.sid().into(), + remote_publication.name(), + rtc_track, + ); + RemoteTrackHandle::Video(Arc::new(video_track)) + } else { + unreachable!() + } + } + _ => unreachable!(), + }; + + self.shared + .add_track_publication(TrackPublication::Remote(remote_publication.clone())); + track.start(); + + let event = TrackSubscribedEvent { + track, + publication: remote_publication, + participant: self.clone(), + }; + + if let Some(cb) = self + .shared + .internal_events + .on_track_subscribed + .lock() + .as_mut() + { + cb(event.clone()).await; + } + + if let Some(cb) = self.shared.events.on_track_subscribed.lock().as_mut() { + cb(event).await; + } + } else { + error!("could not find published track with sid: {:?}", sid); + + let event = TrackSubscriptionFailedEvent { + sid: sid.clone(), + error: TrackError::TrackNotFound(sid.clone().to_string()), + participant: self.clone(), + }; + + if let Some(cb) = self + .shared + .internal_events + .on_track_subscription_failed + .lock() + .as_mut() + { + cb(event.clone()).await; + } + + if let Some(cb) = self + .shared + .events + .on_track_subscription_failed + .lock() + .as_mut() + { + cb(event).await; + } + } + }); + } + + fn get_track_publication(&self, sid: &TrackSid) -> Option { + self.shared.tracks.read().get(sid).map(|track| { + if let TrackPublication::Remote(remote) = track { + remote.clone() + } else { + unreachable!() + } + }) + } +} + +impl_participant_trait!(RemoteParticipant); diff --git a/crates/livekit-core/src/room/track_publication.rs b/crates/livekit-core/src/room/publication/mod.rs similarity index 61% rename from crates/livekit-core/src/room/track_publication.rs rename to crates/livekit-core/src/room/publication/mod.rs index 2c351db..5a36bbb 100644 --- a/crates/livekit-core/src/room/track_publication.rs +++ b/crates/livekit-core/src/room/publication/mod.rs @@ -1,8 +1,10 @@ -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use crate::room::id::TrackSid; +use crate::room::track::local_track::LocalTrackHandle; +use crate::room::track::remote_track::RemoteTrackHandle; +use crate::room::track::{TrackHandle, TrackKind, TrackSource}; use parking_lot::Mutex; -use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; -use crate::room::track::{TrackKind, TrackSource}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::Arc; pub trait TrackPublicationTrait { fn name(&self) -> String; @@ -13,21 +15,22 @@ pub trait TrackPublicationTrait { } pub(super) struct TrackPublicationShared { + pub(super) track: Mutex>, pub(super) name: Mutex, pub(super) sid: Mutex, - pub(super) kind: AtomicU8, // Casted to TrackKind + pub(super) kind: AtomicU8, // Casted to TrackKind pub(super) source: AtomicU8, // Casted to TrackSource - pub(super) simulcasted: AtomicBool + pub(super) simulcasted: AtomicBool, } #[derive(Clone)] pub enum TrackPublication { Local(LocalTrackPublication), - Remote(RemoteTrackPublication) + Remote(RemoteTrackPublication), } macro_rules! shared_getter { - ($x:ident, $ret:ident) => { + ($x:ident, $ret:ty) => { fn $x(&self) -> $ret { match self { TrackPublication::Local(p) => p.$x(), @@ -37,6 +40,15 @@ macro_rules! shared_getter { }; } +impl TrackPublication { + pub fn track(&self) -> Option { + match self { + TrackPublication::Local(p) => p.shared.track.lock().clone(), + TrackPublication::Remote(p) => p.shared.track.lock().clone(), + } + } +} + impl TrackPublicationTrait for TrackPublication { shared_getter!(name, String); shared_getter!(sid, TrackSid); @@ -68,19 +80,37 @@ macro_rules! impl_publication_trait { self.shared.simulcasted.load(Ordering::SeqCst) } } - } + }; } - - #[derive(Clone)] pub struct LocalTrackPublication { - shared: Arc + shared: Arc, +} + +impl LocalTrackPublication { + pub fn track(&self) -> Option { + self.shared + .track + .lock() + .clone() + .map(|local_track| local_track.try_into().unwrap()) + } } #[derive(Clone)] pub struct RemoteTrackPublication { - shared: Arc + shared: Arc, +} + +impl RemoteTrackPublication { + pub fn track(&self) -> Option { + self.shared + .track + .lock() + .clone() + .map(|track| track.try_into().unwrap()) + } } impl_publication_trait!(LocalTrackPublication); diff --git a/crates/livekit-core/src/room/remote_participant.rs b/crates/livekit-core/src/room/remote_participant.rs deleted file mode 100644 index 341c731..0000000 --- a/crates/livekit-core/src/room/remote_participant.rs +++ /dev/null @@ -1,83 +0,0 @@ -use crate::room::id::TrackSid; -use crate::room::participant::{impl_participant_trait, ParticipantShared}; -use crate::room::track::{RemoteAudioTrack, RemoteTrack, RemoteVideoTrack, TrackKind}; -use crate::room::track_publication::{ - RemoteTrackPublication, TrackPublication, TrackPublicationTrait, -}; -use livekit_webrtc::media_stream::MediaStreamTrack; -use std::time::Duration; -use tokio::time::{sleep, timeout}; - -const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5); - - -// It should be fine to add event listeners in this structure -// Registering after should be ParticipantConnected is fine to avoid missing events -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( - &self, - sid: &TrackSid, - media_track: MediaStreamTrack, - ) { - let wait_publication = async { - loop { - let publication = self.get_track_publication(sid); - if let Some(publication) = publication { - return publication; - } - - sleep(Duration::from_millis(50)).await; - } - }; - - let res = timeout(ADD_TRACK_TIMEOUT, wait_publication).await; - - if let Ok(remote_publication) = res { - let track = match remote_publication.kind() { - TrackKind::Audio => { - let audio_track = RemoteAudioTrack::new(); - RemoteTrack::Audio(audio_track) - } - TrackKind::Video => { - let video_track = RemoteVideoTrack::new(); - RemoteTrack::Video(video_track) - } - _ => unreachable!(), - }; - - - - // TODO(theomonnom): call OnTrackSubscribed here - - } else { - // TODO(theomonnom): send error - } - } - - fn get_track_publication(&self, sid: &TrackSid) -> Option { - self.shared.tracks.read().get(sid).map(|track| { - if let TrackPublication::Remote(remote) = track { - remote.clone() - } else { - unreachable!() - } - }) - } -} - -impl_participant_trait!(RemoteParticipant); diff --git a/crates/livekit-core/src/room/track/audio_track.rs b/crates/livekit-core/src/room/track/audio_track.rs new file mode 100644 index 0000000..cea9bfb --- /dev/null +++ b/crates/livekit-core/src/room/track/audio_track.rs @@ -0,0 +1,31 @@ +use crate::room::track::local_audio_track::LocalAudioTrack; +use crate::room::track::remote_audio_track::RemoteAudioTrack; +use crate::room::track::TrackHandle; +use std::sync::Arc; + +#[derive(Clone)] +pub enum AudioTrackHandle { + Local(Arc), + Remote(Arc), +} + +impl From for TrackHandle { + fn from(audio_track: AudioTrackHandle) -> Self { + match audio_track { + AudioTrackHandle::Local(local_audio) => Self::LocalAudio(local_audio), + AudioTrackHandle::Remote(remote_audio) => Self::RemoteAudio(remote_audio), + } + } +} + +impl TryFrom for AudioTrackHandle { + type Error = &'static str; + + fn try_from(track: TrackHandle) -> Result { + match track { + TrackHandle::LocalAudio(local_audio) => Ok(Self::Local(local_audio)), + TrackHandle::RemoteAudio(remote_audio) => Ok(Self::Remote(remote_audio)), + _ => Err("not a audio track"), + } + } +} diff --git a/crates/livekit-core/src/room/track/events.rs b/crates/livekit-core/src/room/track/events.rs new file mode 100644 index 0000000..86d8cad --- /dev/null +++ b/crates/livekit-core/src/room/track/events.rs @@ -0,0 +1 @@ +pub struct TrackEvents {} diff --git a/crates/livekit-core/src/room/track/local_audio_track.rs b/crates/livekit-core/src/room/track/local_audio_track.rs new file mode 100644 index 0000000..c194c34 --- /dev/null +++ b/crates/livekit-core/src/room/track/local_audio_track.rs @@ -0,0 +1,7 @@ +use crate::room::track::{impl_track_trait, TrackShared}; + +pub struct LocalAudioTrack { + shared: TrackShared, +} + +impl_track_trait!(LocalAudioTrack); diff --git a/crates/livekit-core/src/room/track/local_track.rs b/crates/livekit-core/src/room/track/local_track.rs new file mode 100644 index 0000000..64cc127 --- /dev/null +++ b/crates/livekit-core/src/room/track/local_track.rs @@ -0,0 +1,31 @@ +use crate::room::track::local_audio_track::LocalAudioTrack; +use crate::room::track::local_video_track::LocalVideoTrack; +use crate::room::track::TrackHandle; +use std::sync::Arc; + +#[derive(Clone)] +pub enum LocalTrackHandle { + Audio(Arc), + Video(Arc), +} + +impl From for TrackHandle { + fn from(local_track: LocalTrackHandle) -> Self { + match local_track { + LocalTrackHandle::Audio(local_audio) => Self::LocalAudio(local_audio), + LocalTrackHandle::Video(local_video) => Self::LocalVideo(local_video), + } + } +} + +impl TryFrom for LocalTrackHandle { + type Error = &'static str; + + fn try_from(track: TrackHandle) -> Result { + match track { + TrackHandle::LocalAudio(local_audio) => Ok(Self::Audio(local_audio)), + TrackHandle::LocalVideo(local_video) => Ok(Self::Video(local_video)), + _ => Err("not a local track"), + } + } +} diff --git a/crates/livekit-core/src/room/track/local_video_track.rs b/crates/livekit-core/src/room/track/local_video_track.rs new file mode 100644 index 0000000..cb3f0d1 --- /dev/null +++ b/crates/livekit-core/src/room/track/local_video_track.rs @@ -0,0 +1,7 @@ +use crate::room::track::{impl_track_trait, TrackShared}; + +pub struct LocalVideoTrack { + shared: TrackShared, +} + +impl_track_trait!(LocalVideoTrack); diff --git a/crates/livekit-core/src/room/track/mod.rs b/crates/livekit-core/src/room/track/mod.rs index 1627418..ff53f4d 100644 --- a/crates/livekit-core/src/room/track/mod.rs +++ b/crates/livekit-core/src/room/track/mod.rs @@ -1,3 +1,24 @@ +use crate::room::id::TrackSid; +use crate::room::track::local_audio_track::LocalAudioTrack; +use crate::room::track::local_video_track::LocalVideoTrack; +use crate::room::track::remote_audio_track::RemoteAudioTrack; +use crate::room::track::remote_video_track::RemoteVideoTrack; +use crate::utils::wrap_variants; +use livekit_webrtc::media_stream::{MediaStreamTrackHandle, MediaStreamTrackTrait}; +use parking_lot::Mutex; +use std::sync::atomic::AtomicU8; +use std::sync::Arc; + +pub mod audio_track; +pub mod events; +pub mod local_audio_track; +pub mod local_track; +pub mod local_video_track; +pub mod remote_audio_track; +pub mod remote_track; +pub mod remote_video_track; +pub mod video_track; + #[derive(Debug)] pub enum TrackKind { Unknown, @@ -22,6 +43,16 @@ pub enum StreamState { Paused, } +impl From for StreamState { + fn from(val: u8) -> Self { + match val { + 1 => Self::Active, + 2 => Self::Paused, + _ => Self::Unknown, + } + } +} + #[derive(Debug)] pub enum TrackSource { Unknown, @@ -43,135 +74,114 @@ impl From for TrackSource { } } -pub struct LocalVideoTrack {} -pub struct RemoteVideoTrack {} -pub struct LocalAudioTrack {} - - -pub struct RemoteAudioTrack { - - +pub trait TrackTrait { + fn sid(&self) -> TrackSid; + fn name(&self) -> String; + fn kind(&self) -> TrackKind; + fn stream_state(&self) -> StreamState; + fn start(&self); + fn stop(&self); } -impl RemoteVideoTrack { - pub(crate) fn new() -> Self { - Self {} +pub(super) struct TrackShared { + pub(super) sid: Mutex, + pub(super) name: Mutex, + pub(super) kind: AtomicU8, // TrackKind + pub(super) stream_state: AtomicU8, // StreamState + pub(super) rtc_track: MediaStreamTrackHandle, +} + +impl TrackShared { + pub(crate) fn new( + sid: TrackSid, + name: String, + kind: TrackKind, + rtc_track: MediaStreamTrackHandle, + ) -> Self { + Self { + sid: Mutex::new(sid), + name: Mutex::new(name), + kind: AtomicU8::new(kind as u8), + stream_state: AtomicU8::new(StreamState::Active as u8), + rtc_track: rtc_track, + } + } + + pub(crate) fn start(&self) { + self.rtc_track.set_enabled(true); + } + + pub(crate) fn stop(&self) { + self.rtc_track.set_enabled(false); } } -impl RemoteAudioTrack { - pub(crate) fn new() -> Self { - Self {} - } +#[derive(Clone)] +pub enum TrackHandle { + LocalVideo(Arc), + LocalAudio(Arc), + RemoteVideo(Arc), + RemoteAudio(Arc), } -pub enum RemoteTrack { - Audio(RemoteAudioTrack), - Video(RemoteVideoTrack), +impl TrackTrait for TrackHandle { + wrap_variants!( + [LocalVideo, LocalAudio, RemoteVideo, RemoteAudio] + fnc!(sid, TrackSid, []); + fnc!(name, String, []); + fnc!(kind, TrackKind, []); + fnc!(stream_state, StreamState, []); + fnc!(start, (), []); + fnc!(stop, (), []); + ); } -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 TrackHandle { + pub fn rtc_track(&self) -> MediaStreamTrackHandle { + match self { + Self::RemoteVideo(remote_video) => { + MediaStreamTrackHandle::Video(remote_video.rtc_track()) + } + Self::RemoteAudio(remote_audio) => { + MediaStreamTrackHandle::Audio(remote_audio.rtc_track()) + } + _ => todo!(), } } } -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), +macro_rules! impl_track_trait { + ($x:ident) => { + use crate::room::id::TrackSid; + use crate::room::track::{StreamState, TrackKind, TrackTrait}; + use std::sync::atomic::Ordering; + + impl TrackTrait for $x { + fn sid(&self) -> TrackSid { + self.shared.sid.lock().clone() + } + + fn name(&self) -> String { + self.shared.name.lock().clone() + } + + fn kind(&self) -> TrackKind { + self.shared.kind.load(Ordering::SeqCst).into() + } + + fn stream_state(&self) -> StreamState { + self.shared.stream_state.load(Ordering::SeqCst).into() + } + + fn start(&self) { + self.shared.start(); + } + + fn stop(&self) { + self.shared.stop(); + } } - } + }; } -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"), - } - } -} +pub(super) use impl_track_trait; diff --git a/crates/livekit-core/src/room/track/remote_audio_track.rs b/crates/livekit-core/src/room/track/remote_audio_track.rs new file mode 100644 index 0000000..933f307 --- /dev/null +++ b/crates/livekit-core/src/room/track/remote_audio_track.rs @@ -0,0 +1,30 @@ +use crate::room::track::{impl_track_trait, TrackShared}; +use livekit_webrtc::media_stream::{AudioTrack, MediaStreamTrackHandle}; +use std::sync::Arc; + +pub struct RemoteAudioTrack { + shared: TrackShared, +} + +impl RemoteAudioTrack { + pub(crate) fn new(sid: TrackSid, name: String, track: Arc) -> Self { + Self { + shared: TrackShared::new( + sid, + name, + TrackKind::Audio, + MediaStreamTrackHandle::Audio(track), + ), + } + } + + pub fn rtc_track(&self) -> Arc { + if let MediaStreamTrackHandle::Audio(audio) = &self.shared.rtc_track { + audio.clone() + } else { + unreachable!() + } + } +} + +impl_track_trait!(RemoteAudioTrack); diff --git a/crates/livekit-core/src/room/track/remote_track.rs b/crates/livekit-core/src/room/track/remote_track.rs new file mode 100644 index 0000000..591b443 --- /dev/null +++ b/crates/livekit-core/src/room/track/remote_track.rs @@ -0,0 +1,49 @@ +use std::sync::Arc; + +use super::{StreamState, TrackKind}; +use crate::room::id::TrackSid; +use crate::room::track::remote_audio_track::RemoteAudioTrack; +use crate::room::track::remote_video_track::RemoteVideoTrack; +use crate::room::track::TrackHandle; +use crate::utils::wrap_variants; + +use super::TrackTrait; + +#[derive(Clone)] +pub enum RemoteTrackHandle { + Audio(Arc), + Video(Arc), +} + +impl TrackTrait for RemoteTrackHandle { + wrap_variants!( + [Audio, Video] + fnc!(sid, TrackSid, []); + fnc!(name, String, []); + fnc!(kind, TrackKind, []); + fnc!(stream_state, StreamState, []); + fnc!(start, (), []); + fnc!(stop, (), []); + ); +} + +impl From for TrackHandle { + fn from(remote_track: RemoteTrackHandle) -> Self { + match remote_track { + RemoteTrackHandle::Audio(remote_audio) => Self::RemoteAudio(remote_audio), + RemoteTrackHandle::Video(remote_video) => Self::RemoteVideo(remote_video), + } + } +} + +impl TryFrom for RemoteTrackHandle { + type Error = &'static str; + + fn try_from(track: TrackHandle) -> Result { + match track { + TrackHandle::RemoteAudio(remote_audio) => Ok(Self::Audio(remote_audio)), + TrackHandle::RemoteVideo(remote_video) => Ok(Self::Video(remote_video)), + _ => Err("not a remote track"), + } + } +} diff --git a/crates/livekit-core/src/room/track/remote_video_track.rs b/crates/livekit-core/src/room/track/remote_video_track.rs new file mode 100644 index 0000000..57caf67 --- /dev/null +++ b/crates/livekit-core/src/room/track/remote_video_track.rs @@ -0,0 +1,31 @@ +use livekit_webrtc::media_stream::{MediaStreamTrackHandle, VideoTrack}; +use std::sync::Arc; + +use crate::room::track::{impl_track_trait, TrackShared}; + +pub struct RemoteVideoTrack { + shared: TrackShared, +} + +impl RemoteVideoTrack { + pub(crate) fn new(sid: TrackSid, name: String, track: Arc) -> Self { + Self { + shared: TrackShared::new( + sid, + name, + TrackKind::Video, + MediaStreamTrackHandle::Video(track), + ), + } + } + + pub fn rtc_track(&self) -> Arc { + if let MediaStreamTrackHandle::Video(video) = &self.shared.rtc_track { + video.clone() + } else { + unreachable!() + } + } +} + +impl_track_trait!(RemoteVideoTrack); diff --git a/crates/livekit-core/src/room/track/video_track.rs b/crates/livekit-core/src/room/track/video_track.rs new file mode 100644 index 0000000..4294e5e --- /dev/null +++ b/crates/livekit-core/src/room/track/video_track.rs @@ -0,0 +1,31 @@ +use crate::room::track::local_video_track::LocalVideoTrack; +use crate::room::track::remote_video_track::RemoteVideoTrack; +use crate::room::track::TrackHandle; +use std::sync::Arc; + +#[derive(Clone)] +pub enum VideoTrackHandle { + Local(Arc), + Remote(Arc), +} + +impl From for TrackHandle { + fn from(video_track: VideoTrackHandle) -> Self { + match video_track { + VideoTrackHandle::Local(local_video) => Self::LocalVideo(local_video), + VideoTrackHandle::Remote(remote_video) => Self::RemoteVideo(remote_video), + } + } +} + +impl TryFrom for VideoTrackHandle { + type Error = &'static str; + + fn try_from(track: TrackHandle) -> Result { + match track { + TrackHandle::LocalVideo(local_video) => Ok(Self::Local(local_video)), + TrackHandle::RemoteVideo(remote_video) => Ok(Self::Remote(remote_video)), + _ => Err("not a video track"), + } + } +} diff --git a/crates/livekit-core/src/utils.rs b/crates/livekit-core/src/utils.rs new file mode 100644 index 0000000..c250092 --- /dev/null +++ b/crates/livekit-core/src/utils.rs @@ -0,0 +1,26 @@ +macro_rules! wrap_variants { + // This arm is used to avoid nested loops with the arguments + // The arguments are transformed to $combined_args TokenTree + (@match $self:ident $fnc:ident $combined_args:tt [$($variant:ident),+]) => { + match $self { + $( + Self::$variant(inner) => inner.$fnc$combined_args, + )+ + } + }; + + ($fnc:ident, $ret:ty, [$($arg:ident: $t:ty),*], [$($variant:ident),+]) => { + fn $fnc(&self, $($arg: $t),*) -> $ret { + wrap_variants!(@match self $fnc ($($arg,)*) [$($variant),+]) + } + }; + + ($variants:tt $(fnc!($fnc:ident, $ret:ty, $args:tt);)+) => { + $( + wrap_variants!($fnc, $ret, $args, $variants); + )* + }; +} + +pub(crate) use wrap_variants; + diff --git a/crates/livekit-webrtc/libwebrtc-sys/CMakeLists.txt b/crates/livekit-webrtc/libwebrtc-sys/CMakeLists.txt deleted file mode 100644 index b2b7e04..0000000 --- a/crates/livekit-webrtc/libwebrtc-sys/CMakeLists.txt +++ /dev/null @@ -1,23 +0,0 @@ -# IMPORTANT NOTE -# This file is just used because some IDEs need to understand how to do autocompletion. -# This file is completely ignored by the library ( See build.rs for the build system ) - -cmake_minimum_required(VERSION 3.22) -project(livekit-webrtc) - -set(CMAKE_CXX_STANDARD 17) - -add_definitions(-DWEBRTC_WIN) - -include_directories(libwebrtc/include) -include_directories(libwebrtc/include/third_party/abseil-cpp/) -include_directories(libwebrtc/include/third_party/libc++/) -include_directories(include/) -include_directories(../../../target/cxxbridge) # Can be different - -file(GLOB_RECURSE SRC src/*.cpp) -add_library(livekit-webrtc ${SRC}) - -#include_directories(/Users/theomonnom/Library/Android/sdk/ndk/25.0.8775105/toolchains/llvm/prebuilt/darwin-x86_64/sysroot/usr/include) -#find_library(ANDROID_LIB_ANDROID android) -#target_link_libraries(client_sdk_native PRIVATE android) \ No newline at end of file diff --git a/crates/livekit-webrtc/libwebrtc-sys/build.rs b/crates/livekit-webrtc/libwebrtc-sys/build.rs index a65a667..b1fdd96 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/build.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/build.rs @@ -52,7 +52,7 @@ fn macos_link_search_path() -> Option { fn main() { // TODO Download precompiled binaries of WebRTC for the target_os - let target_os = "macos"; + let target_os = "windows"; //let target_arch = "arm64"; let libwebrtc_dir = path::PathBuf::from("libwebrtc"); @@ -79,6 +79,8 @@ fn main() { "src/rtp_transceiver.rs", "src/rtc_error.rs", "src/webrtc.rs", + "src/video_frame.rs", + "src/video_frame_buffer.rs", ]); builder.file("src/peer_connection.cpp"); diff --git a/crates/livekit-webrtc/libwebrtc-sys/compile_flags.txt b/crates/livekit-webrtc/libwebrtc-sys/compile_flags.txt new file mode 100644 index 0000000..e0fb4f0 --- /dev/null +++ b/crates/livekit-webrtc/libwebrtc-sys/compile_flags.txt @@ -0,0 +1,8 @@ +-xc++ +-std=c++17 +-Iinclude +-Ilibwebrtc/include +-Ilibwebrtc/include/third_party/abseil-cpp +-Ilibwebrtc/include/third_party/libc++ +-I../../../target/cxxbridge +-DWEBRTC_WIN diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h index c2b3164..bf5d652 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h @@ -13,26 +13,7 @@ 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 NativeVideoFrameSink; class MediaStream { public: @@ -47,6 +28,96 @@ class MediaStream { static std::unique_ptr _unique_media_stream() { return nullptr; // Ignore } + +class MediaStreamTrack { + protected: + explicit MediaStreamTrack( + rtc::scoped_refptr track); + + public: + static std::unique_ptr from( + rtc::scoped_refptr track); + + rust::String kind() const; + rust::String id() const; + + bool enabled() const; + bool set_enabled(bool enable); + + TrackState state() const; + + protected: + rtc::scoped_refptr track_; +}; + +static std::unique_ptr _unique_media_stream_track() { + return nullptr; // Ignore +} + +class AudioTrack : public MediaStreamTrack { + public: + explicit AudioTrack(rtc::scoped_refptr track); +}; + +static std::unique_ptr _unique_audio_track() { + return nullptr; // Ignore +} + +class VideoTrack : public MediaStreamTrack { + public: + explicit VideoTrack(rtc::scoped_refptr track); + + void add_sink(NativeVideoFrameSink& sink); + void remove_sink(NativeVideoFrameSink& sink); + + void set_should_receive(bool should_receive); + bool should_receive() const; + ContentHint content_hint() const; + void set_content_hint(ContentHint hint); + + private: + webrtc::VideoTrackInterface* track() const { + return static_cast(track_.get()); + } +}; + +static std::unique_ptr _unique_video_track() { + return nullptr; // Ignore +} + +class NativeVideoFrameSink + : public rtc::VideoSinkInterface { + public: + explicit NativeVideoFrameSink(rust::Box observer); + + void OnFrame(const webrtc::VideoFrame& frame) override; + void OnDiscardedFrame() override; + void OnConstraintsChanged( + const webrtc::VideoTrackSourceConstraints& constraints) override; + + private: + rust::Box observer_; +}; + +std::unique_ptr create_native_video_frame_sink( + rust::Box observer); + +const MediaStreamTrack* video_to_media(const VideoTrack* track) { + return track; +} + +const MediaStreamTrack* audio_to_media(const AudioTrack* track) { + return track; +} + +const VideoTrack* media_to_video(const MediaStreamTrack* track) { + return static_cast(track); +} + +const AudioTrack* media_to_audio(const MediaStreamTrack* track) { + return static_cast(track); +} + } // namespace livekit #endif // CLIENT_SDK_NATIVE_MEDIA_STREAM_INTERFACE_H 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 1d9d7ef..f342734 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h @@ -15,6 +15,7 @@ struct SetLocalSdpObserverWrapper; struct SetRemoteSdpObserverWrapper; struct DataChannelObserverWrapper; struct AddIceCandidateObserverWrapper; +struct VideoFrameSinkWrapper; // Shared types enum class PeerConnectionState; @@ -24,6 +25,9 @@ enum class IceGatheringState; enum class SdpType; enum class DataState; enum class TrackState; +enum class ContentHint; +enum class VideoRotation; +enum class VideoFrameBufferType; struct SdpParseError; struct RTCOfferAnswerOptions; struct RTCError; diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h new file mode 100644 index 0000000..9c719f4 --- /dev/null +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h @@ -0,0 +1,48 @@ +// +// Created by theom on 14/11/2022. +// + +#ifndef LIVEKIT_WEBRTC_VIDEO_FRAME_H +#define LIVEKIT_WEBRTC_VIDEO_FRAME_H + +#include "api/video/video_frame.h" +#include "livekit/rust_types.h" +#include "livekit/video_frame_buffer.h" + +namespace livekit { + +class VideoFrame { + public: + explicit VideoFrame(const webrtc::VideoFrame& frame) + : frame_(std::move(frame)) {} + + int width() const { return frame_.width(); } + int height() const { return frame_.height(); } + uint32_t size() const { return frame_.size(); } + uint16_t id() const { return frame_.id(); } + int64_t timestamp_us() const { return frame_.timestamp_us(); } + int64_t ntp_time_ms() const { return frame_.ntp_time_ms(); } + uint32_t transport_frame_id() const { return frame_.transport_frame_id(); } + uint32_t timestamp() const { return frame_.timestamp(); } + + VideoRotation rotation() const { + return static_cast(frame_.rotation()); + } + + // TODO(theomonnom) This shouldn't create a new shared_ptr at each call + std::shared_ptr video_frame_buffer() const { + return std::make_shared(frame_.video_frame_buffer()); + } + + private: + webrtc::VideoFrame frame_; +}; + +static std::unique_ptr _unique_video_frame() { + return nullptr; // Ignore +} + + +} // namespace livekit + +#endif // LIVEKIT_WEBRTC_VIDEO_FRAME_H diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame_buffer.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame_buffer.h new file mode 100644 index 0000000..3b4909c --- /dev/null +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame_buffer.h @@ -0,0 +1,93 @@ +// +// Created by theom on 14/11/2022. +// + +#ifndef LIVEKIT_WEBRTC_VIDEO_FRAME_BUFFER_H +#define LIVEKIT_WEBRTC_VIDEO_FRAME_BUFFER_H + +#include "api/video/video_frame_buffer.h" +#include "rust_types.h" + +namespace livekit { + +class PlanarYuvBuffer; +class PlanarYuv8Buffer; +class I420Buffer; + +class VideoFrameBuffer { + public: + explicit VideoFrameBuffer(rtc::scoped_refptr buffer) + : buffer_(std::move(buffer)) {} + + VideoFrameBufferType buffer_type() const { + return static_cast(buffer_->type()); + } + + int width() const { return buffer_->width(); } + int height() const { return buffer_->height(); } + + std::shared_ptr to_i420() { + return std::make_shared(buffer_->ToI420()); + } + + protected: + rtc::scoped_refptr buffer_; +}; + +class PlanarYuvBuffer : public VideoFrameBuffer { + public: + explicit PlanarYuvBuffer(rtc::scoped_refptr buffer) + : VideoFrameBuffer(buffer) {} + + int chroma_width() const { return buffer()->ChromaWidth(); } + int chroma_height() const { return buffer()->ChromaHeight(); } + + int stride_y() const { return buffer()->StrideY(); } + int stride_u() const { return buffer()->StrideU(); } + int stride_v() const { return buffer()->StrideV(); } + + private: + webrtc::PlanarYuvBuffer* buffer() const { + return static_cast(buffer_.get()); + } +}; + +class PlanarYuv8Buffer : public PlanarYuvBuffer { + public: + explicit PlanarYuv8Buffer(rtc::scoped_refptr buffer) + : PlanarYuvBuffer(buffer) {} + + const uint8_t* data_y() const { return buffer()->DataY(); } + const uint8_t* data_u() const { return buffer()->DataU(); } + const uint8_t* data_v() const { return buffer()->DataV(); } + + private: + webrtc::PlanarYuv8Buffer* buffer() const { + return static_cast(buffer_.get()); + } +}; + +class I420Buffer : public PlanarYuv8Buffer { + public: + explicit I420Buffer(rtc::scoped_refptr buffer) + : PlanarYuv8Buffer(buffer) {} +}; + +std::shared_ptr to_video_frame_buffer( + std::shared_ptr buffer) { + return buffer; +} + +std::shared_ptr to_yuv_buffer( + std::shared_ptr buffer) { + return buffer; +} + +std::shared_ptr to_yuv8_buffer( + std::shared_ptr buffer) { + return buffer; +} + +} // namespace livekit + +#endif // LIVEKIT_WEBRTC_VIDEO_FRAME_BUFFER_H diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/candidate.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/candidate.cpp index 970575d..09f2190 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/candidate.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/candidate.cpp @@ -7,4 +7,4 @@ namespace livekit { Candidate::Candidate(const cricket::Candidate& candidate) : candidate_(candidate) {} -} // namespace livekit \ No newline at end of file +} // namespace livekit diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp index 72f3c65..41986f1 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp @@ -87,4 +87,4 @@ std::unique_ptr create_native_data_channel_observer( rust::Box observer) { return std::make_unique(std::move(observer)); } -} // namespace livekit \ No newline at end of file +} // namespace livekit diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp index 20ba43d..6340a83 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp @@ -158,4 +158,4 @@ create_native_set_remote_sdp_observer( std::move(observer))}); } -} // namespace livekit \ No newline at end of file +} // namespace livekit diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs b/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs index ecd1f3e..ab6e52a 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/lib.rs @@ -7,4 +7,10 @@ pub mod peer_connection_factory; pub mod rtc_error; pub mod rtp_receiver; pub mod rtp_transceiver; +pub mod video_frame; +pub mod video_frame_buffer; pub mod webrtc; + +pub const MEDIA_TYPE_VIDEO: &str = "video"; +pub const MEDIA_TYPE_AUDIO: &str = "audio"; +pub const MEDIA_TYPE_DATA: &str = "data"; diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp index 04eb1f9..b4017ee 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp @@ -4,12 +4,27 @@ #include "livekit/media_stream.h" +#include "libwebrtc-sys/src/media_stream.rs.h" + namespace livekit { MediaStreamTrack::MediaStreamTrack( rtc::scoped_refptr track) : track_(std::move(track)) {} +std::unique_ptr MediaStreamTrack::from( + rtc::scoped_refptr track) { + if (track->kind() == webrtc::MediaStreamTrackInterface::kVideoKind) { + return std::make_unique( + rtc::scoped_refptr( + static_cast(track.get()))); + } else { + return std::make_unique( + rtc::scoped_refptr( + static_cast(track.get()))); + } +} + rust::String MediaStreamTrack::kind() const { return track_->kind(); } @@ -35,7 +50,60 @@ MediaStream::MediaStream( : media_stream_(std::move(stream)) {} rust::String MediaStream::id() const { - return media_stream_->id(); + return media_stream_->id(); } -} // namespace livekit \ No newline at end of file +VideoTrack::VideoTrack(rtc::scoped_refptr track) + : MediaStreamTrack(std::move(track)) {} + +void VideoTrack::add_sink(NativeVideoFrameSink& sink) { + track()->AddOrUpdateSink(&sink, rtc::VideoSinkWants()); +} + +void VideoTrack::remove_sink(NativeVideoFrameSink& sink) { + track()->RemoveSink(&sink); +} + +void VideoTrack::set_should_receive(bool should_receive) { + track()->set_should_receive(should_receive); +} + +bool VideoTrack::should_receive() const { + return track()->should_receive(); +} + +ContentHint VideoTrack::content_hint() const { + return static_cast(track()->content_hint()); +} + +void VideoTrack::set_content_hint(ContentHint hint) { + track()->set_content_hint( + static_cast(hint)); +} + +NativeVideoFrameSink::NativeVideoFrameSink( + rust::Box observer) + : observer_(std::move(observer)) {} + +void NativeVideoFrameSink::OnFrame(const webrtc::VideoFrame& frame) { + observer_->on_frame(std::make_unique(frame)); +} + +void NativeVideoFrameSink::OnDiscardedFrame() { + observer_->on_discarded_frame(); +} + +void NativeVideoFrameSink::OnConstraintsChanged( + const webrtc::VideoTrackSourceConstraints& constraints) { + VideoTrackSourceConstraints cst; + cst.min_fps = constraints.min_fps.value_or(-1); + cst.max_fps = constraints.max_fps.value_or(-1); + observer_->on_constraints_changed(cst); +} + +std::unique_ptr create_native_video_frame_sink( + rust::Box observer) { + return std::make_unique(std::move(observer)); +} + +} // namespace livekit diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.rs b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.rs index 0f46ebf..86b8215 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.rs @@ -1,3 +1,7 @@ +use cxx::UniquePtr; + +use crate::video_frame::ffi::VideoFrame; + #[cxx::bridge(namespace = "livekit")] pub mod ffi { @@ -8,11 +12,33 @@ pub mod ffi { Ended, } + #[derive(Debug)] + #[repr(i32)] + pub enum ContentHint { + None, + Fluid, + Detailed, + Text, + } + + // -1 = optional + pub struct VideoTrackSourceConstraints { + pub min_fps: f64, + pub max_fps: f64, + } + unsafe extern "C++" { include!("livekit/media_stream.h"); + include!("livekit/video_frame.h"); + type NativeVideoFrameSink; type MediaStreamTrack; type MediaStream; + type AudioTrack; + type VideoTrack; + type VideoFrame = crate::video_frame::ffi::VideoFrame; + + fn id(self: &MediaStream) -> String; fn kind(self: &MediaStreamTrack) -> String; fn id(self: &MediaStreamTrack) -> String; @@ -20,17 +46,84 @@ pub mod ffi { fn set_enabled(self: Pin<&mut MediaStreamTrack>, enable: bool) -> bool; fn state(self: &MediaStreamTrack) -> TrackState; - fn id(self: &MediaStream) -> String; + unsafe fn add_sink(self: Pin<&mut VideoTrack>, sink: Pin<&mut NativeVideoFrameSink>); + unsafe fn remove_sink(self: Pin<&mut VideoTrack>, sink: Pin<&mut NativeVideoFrameSink>); + + fn set_should_receive(self: Pin<&mut VideoTrack>, should_receive: bool); + fn should_receive(self: &VideoTrack) -> bool; + fn content_hint(self: &VideoTrack) -> ContentHint; + fn set_content_hint(self: Pin<&mut VideoTrack>, hint: ContentHint); + + fn create_native_video_frame_sink( + observer: Box, + ) -> UniquePtr; + + unsafe fn video_to_media(track: *const VideoTrack) -> *const MediaStreamTrack; + unsafe fn audio_to_media(track: *const AudioTrack) -> *const MediaStreamTrack; + unsafe fn media_to_video(track: *const MediaStreamTrack) -> *const VideoTrack; + unsafe fn media_to_audio(track: *const MediaStreamTrack) -> *const AudioTrack; fn _unique_media_stream_track() -> UniquePtr; // Ignore fn _unique_media_stream() -> UniquePtr; // Ignore + fn _unique_audio_track() -> UniquePtr; // Ignore + fn _unique_video_track() -> UniquePtr; // Ignore + } + + extern "Rust" { + type VideoFrameSinkWrapper; + + fn on_frame(self: &VideoFrameSinkWrapper, frame: UniquePtr); + fn on_discarded_frame(self: &VideoFrameSinkWrapper); + fn on_constraints_changed( + self: &VideoFrameSinkWrapper, + constraints: VideoTrackSourceConstraints, + ); } } unsafe impl Sync for ffi::MediaStreamTrack {} - unsafe impl Send for ffi::MediaStreamTrack {} - unsafe impl Sync for ffi::MediaStream {} +unsafe impl Send for ffi::MediaStream {} +unsafe impl Send for ffi::AudioTrack {} +unsafe impl Sync for ffi::AudioTrack {} +unsafe impl Send for ffi::VideoTrack {} +unsafe impl Sync for ffi::VideoTrack {} +unsafe impl Send for ffi::NativeVideoFrameSink {} +unsafe impl Sync for ffi::NativeVideoFrameSink {} -unsafe impl Send for ffi::MediaStream {} \ No newline at end of file +pub trait VideoFrameSink: Send + Sync { + fn on_frame(&self, frame: UniquePtr); + fn on_discarded_frame(&self); + fn on_constraints_changed(&self, constraints: ffi::VideoTrackSourceConstraints); +} + +pub struct VideoFrameSinkWrapper { + observer: *mut dyn VideoFrameSink, +} + +impl VideoFrameSinkWrapper { + /// # Safety + /// VideoFrameSink must lives as long as VideoSinkInterfaceWrapper does + pub unsafe fn new(observer: *mut dyn VideoFrameSink) -> Self { + Self { observer } + } + + fn on_frame(&self, frame: UniquePtr) { + unsafe { + (*self.observer).on_frame(frame); + } + } + + fn on_discarded_frame(&self) { + unsafe { + (*self.observer).on_discarded_frame(); + } + } + + fn on_constraints_changed(&self, constraints: ffi::VideoTrackSourceConstraints) { + unsafe { + (*self.observer).on_constraints_changed(constraints); + } + } +} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs index f6b914f..688b7d3 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs @@ -344,7 +344,7 @@ pub struct PeerConnectionObserverWrapper { } impl PeerConnectionObserverWrapper { - /// SAFETY + /// # Safety /// PeerConnectionObserver must lives as long as PeerConnectionObserverWrapper does pub unsafe fn new(observer: *mut dyn PeerConnectionObserver) -> Self { Self { observer } 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 6266242..2850983 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs @@ -45,7 +45,7 @@ pub mod ffi { ) -> UniquePtr; fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr; - /// SAFETY + /// # Safety /// The observer must live as long as the PeerConnection unsafe fn create_peer_connection( self: &PeerConnectionFactory, diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp index e9095c0..a7b70c1 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.cpp @@ -10,7 +10,7 @@ RtpReceiver::RtpReceiver( : receiver_(std::move(receiver)) {} std::unique_ptr RtpReceiver::track() const { - return std::make_unique(receiver_->track()); + return MediaStreamTrack::from(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 34327de..a349fa6 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/rtp_receiver.rs @@ -15,4 +15,4 @@ pub mod ffi { unsafe impl Sync for ffi::RtpReceiver {} -unsafe impl Send for ffi::RtpReceiver {} \ No newline at end of file +unsafe impl Send for ffi::RtpReceiver {} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs b/crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs new file mode 100644 index 0000000..79ba2bf --- /dev/null +++ b/crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs @@ -0,0 +1,32 @@ +#[cxx::bridge(namespace = "livekit")] +pub mod ffi { + #[derive(Debug)] + #[repr(i32)] + pub enum VideoRotation { + VideoRotation0 = 0, + VideoRotation90 = 90, + VideoRotation180 = 180, + VideoRotation270 = 270, + } + + unsafe extern "C++" { + include!("livekit/video_frame.h"); + include!("livekit/video_frame_buffer.h"); + + type VideoFrame; + type VideoFrameBuffer = crate::video_frame_buffer::ffi::VideoFrameBuffer; + + fn width(self: &VideoFrame) -> i32; + fn height(self: &VideoFrame) -> i32; + fn size(self: &VideoFrame) -> u32; + fn id(self: &VideoFrame) -> u16; + fn timestamp_us(self: &VideoFrame) -> i64; + fn ntp_time_ms(self: &VideoFrame) -> i64; + fn transport_frame_id(self: &VideoFrame) -> u32; + fn timestamp(self: &VideoFrame) -> u32; + fn rotation(self: &VideoFrame) -> VideoRotation; + fn video_frame_buffer(self: &VideoFrame) -> SharedPtr; + + fn _unique_video_frame() -> UniquePtr; // Ignore + } +} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/video_frame_buffer.rs b/crates/livekit-webrtc/libwebrtc-sys/src/video_frame_buffer.rs new file mode 100644 index 0000000..65d4fad --- /dev/null +++ b/crates/livekit-webrtc/libwebrtc-sys/src/video_frame_buffer.rs @@ -0,0 +1,43 @@ +#[cxx::bridge(namespace = "livekit")] +pub mod ffi { + #[derive(Debug)] + #[repr(i32)] + pub enum VideoFrameBufferType { + Native, + I420, + I420A, + I422, + I444, + I010, + NV12, + } + + unsafe extern "C++" { + include!("livekit/video_frame_buffer.h"); + + type VideoFrameBuffer; + type PlanarYuvBuffer; + type PlanarYuv8Buffer; + type I420Buffer; + + fn buffer_type(self: &VideoFrameBuffer) -> VideoFrameBufferType; + fn width(self: &VideoFrameBuffer) -> i32; + fn height(self: &VideoFrameBuffer) -> i32; + fn to_i420(self: Pin<&mut VideoFrameBuffer>) -> SharedPtr; + + fn chroma_width(self: &PlanarYuvBuffer) -> i32; + fn chroma_height(self: &PlanarYuvBuffer) -> i32; + fn stride_y(self: &PlanarYuvBuffer) -> i32; + fn stride_u(self: &PlanarYuvBuffer) -> i32; + fn stride_v(self: &PlanarYuvBuffer) -> i32; + + fn data_y(self: &PlanarYuv8Buffer) -> *const u8; + fn data_u(self: &PlanarYuv8Buffer) -> *const u8; + fn data_v(self: &PlanarYuv8Buffer) -> *const u8; + + fn to_video_frame_buffer(buffer: SharedPtr) + -> SharedPtr; + fn to_yuv_buffer(buffer: SharedPtr) -> SharedPtr; + fn to_yuv8_buffer(buffer: SharedPtr) -> SharedPtr; + } +} diff --git a/crates/livekit-webrtc/src/.media_stream.rs.rustfmt b/crates/livekit-webrtc/src/.media_stream.rs.rustfmt new file mode 100644 index 0000000..c5641bd --- /dev/null +++ b/crates/livekit-webrtc/src/.media_stream.rs.rustfmt @@ -0,0 +1,277 @@ +use cxx::UniquePtr; +use libwebrtc_sys::media_stream as sys_ms; +use libwebrtc_sys::MEDIA_TYPE_VIDEO; +use std::fmt::{Debug, Formatter}; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; + +pub use sys_ms::ffi::ContentHint; +pub use sys_ms::ffi::TrackState; + +use crate::video_frame::VideoFrame; + +pub trait MediaStreamTrackTrait { + fn kind(&self) -> String; + fn id(&self) -> String; + fn enabled(&self) -> bool; + fn set_enabled(&self, enabled: bool) -> bool; + fn state(&self) -> TrackState; +} + +pub enum MediaStreamTrack { + Audio(Arc), + Video(Arc), +} + +macro_rules! shared_getter { + ($x:ident, $ret:ty) => { + fn $x(&self) -> $ret { + match self { + Self::Video(inner) => inner.$x(), + Self::Audio(inner) => inner.$x(), + } + } + }; +} + +impl MediaStreamTrack { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + unsafe { + if cxx_handle.kind() == MEDIA_TYPE_VIDEO { + Self::Video(Arc::new(VideoTrack::new(UniquePtr::from_raw( + sys_ms::ffi::media_to_video(cxx_handle.into_raw()) + as *mut sys_ms::ffi::VideoTrack, + )))) + } else { + Self::Audio(Arc::new(AudioTrack::new(UniquePtr::from_raw( + sys_ms::ffi::media_to_audio(cxx_handle.into_raw()) + as *mut sys_ms::ffi::AudioTrack, + )))) + } + } + } +} + +impl Debug for MediaStreamTrack { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + f.debug_struct("MediaStreamTrack") + .field("id", &self.id()) + .field("kind", &self.kind()) + .field("enabled", &self.enabled()) + .field("state", &self.state()) + .finish() + } +} + +impl MediaStreamTrackTrait for MediaStreamTrack { + shared_getter!(kind, String); + shared_getter!(id, String); + shared_getter!(enabled, bool); + shared_getter!(state, TrackState); + + fn set_enabled(&self, enabled: bool) -> bool { + match self { + Self::Video(inner) => inner.set_enabled(enabled), + Self::Audio(inner) => inner.set_enabled(enabled), + } + } +} + +pub struct AudioTrack { + cxx_handle: Mutex>, +} + +impl AudioTrack { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + Self { + cxx_handle: Mutex::new(cxx_handle), + } + } +} + +pub struct VideoTrack { + cxx_handle: Mutex>, + observer: Box, + + // Keep alive for c++ + native_observer: UniquePtr, +} + +macro_rules! impl_media_stream_track_trait { + ($x:ty, $cast:ident) => { + impl MediaStreamTrackTrait for $x { + fn kind(&self) -> String { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).kind() } + } + + fn id(&self) -> String { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).id() } + } + + fn enabled(&self) -> bool { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).enabled() } + } + + fn set_enabled(&self, enabled: bool) -> bool { + unsafe { + let media = + sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap()) as *mut sys_ms::ffi::MediaStreamTrack; + + Pin::new_unchecked(&mut *media).set_enabled(enabled) + } + } + + fn state(&self) -> TrackState { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).state() } + } + } + }; +} + +impl_media_stream_track_trait!(VideoTrack, video_to_media); +impl_media_stream_track_trait!(AudioTrack, audio_to_media); + +pub type OnFrameHandler = Box; +pub type OnDiscardedFrameHandler = Box; +pub type OnConstraintsChanged = Box; + +#[derive(Default)] +struct InternalVideoTrackSink { + on_frame_handler: Mutex>, + on_discarded_frame_handler: Mutex>, + on_constraints_changed_handler: Mutex>, +} + +pub struct VideoTrackSourceConstraints { + pub min_fps: Option, + pub max_fps: Option, +} + +impl From for VideoTrackSourceConstraints { + fn from(cst: sys_ms::ffi::VideoTrackSourceConstraints) -> Self { + Self { + min_fps: (cst.min_fps != 1.0).then_some(cst.min_fps), + max_fps: (cst.max_fps != 1.0).then_some(cst.max_fps), + } + } +} + +impl sys_ms::VideoFrameSink for InternalVideoTrackSink { + fn on_frame(&self, frame: UniquePtr) { + if let Some(cb) = self.on_frame_handler.lock().unwrap().as_mut() { + cb(VideoFrame::new(frame)); + } + } + + fn on_discarded_frame(&self) { + if let Some(cb) = self.on_discarded_frame_handler.lock().unwrap().as_mut() { + cb(); + } + } + + fn on_constraints_changed(&self, constraints: sys_ms::ffi::VideoTrackSourceConstraints) { + if let Some(cb) = self.on_constraints_changed_handler.lock().unwrap().as_mut() { + cb(constraints.into()); + } + } +} + +impl VideoTrack { + fn new(cxx_handle: UniquePtr) -> Self { + let mut observer = Box::new(InternalVideoTrackSink::default()); + + let mut track = unsafe { + Self { + cxx_handle: Mutex::new(cxx_handle), + native_observer: sys_ms::ffi::create_native_video_frame_sink(Box::new( + sys_ms::VideoFrameSinkWrapper::new(&mut *observer), + )), + observer, + } + }; + + unsafe { + track + .cxx_handle + .lock() + .unwrap() + .pin_mut() + .add_sink(track.native_observer.pin_mut()); + } + + track + } + + pub fn set_should_receive(&mut self, should_receive: bool) { + + + + self.cxx_handle + .lock() + .unwrap() + .pin_mut() + .set_should_receive(should_receive) + } + + pub fn set_content_hint(&mut self, hint: ContentHint) { + self.cxx_handle + .lock() + .unwrap() + .pin_mut() + .set_content_hint(hint) + } + + pub fn should_receive(&self) -> bool { + self.cxx_handle.lock().unwrap().should_receive() + } + + pub fn content_hint(&self) -> ContentHint { + self.cxx_handle.lock().unwrap().content_hint() + } + + pub fn on_frame(&mut self, handler: OnFrameHandler) { + *self.observer.on_frame_handler.lock().unwrap() = Some(handler); + } + + pub fn on_discarded_frame(&mut self, handler: OnDiscardedFrameHandler) { + *self.observer.on_discarded_frame_handler.lock().unwrap() = Some(handler); + } + + pub fn on_constraints_changed(&mut self, handler: OnConstraintsChanged) { + *self.observer.on_constraints_changed_handler.lock().unwrap() = Some(handler); + } +} + +impl Drop for VideoTrack { + fn drop(&mut self) { + unsafe { + self.cxx_handle + .lock() + .unwrap() + .pin_mut() + .remove_sink(self.native_observer.pin_mut()); + } + } +} + +pub struct MediaStream { + cxx_handle: UniquePtr, +} + +impl Debug for MediaStream { + fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { + f.debug_struct("MediaStream") + .field("id", &self.id()) + .finish() + } +} + +impl MediaStream { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } + + pub fn id(&self) -> String { + self.cxx_handle.id() + } +} diff --git a/crates/livekit-webrtc/src/data_channel.rs b/crates/livekit-webrtc/src/data_channel.rs index a231ee3..284340f 100644 --- a/crates/livekit-webrtc/src/data_channel.rs +++ b/crates/livekit-webrtc/src/data_channel.rs @@ -109,13 +109,13 @@ impl Drop for DataChannel { pub type OnStateChangeHandler = Box; pub type OnMessageHandler = Box; -// data, is_binary pub type OnBufferedAmountChangeHandler = Box; +#[derive(Default)] struct InternalDataChannelObserver { - on_state_change_handler: Arc>>, - on_message_handler: Arc>>, - on_buffered_amount_change_handler: Arc>>, + on_state_change_handler: Mutex>, + on_message_handler: Mutex>, + on_buffered_amount_change_handler: Mutex>, } impl sys_dc::DataChannelObserver for InternalDataChannelObserver { @@ -144,16 +144,6 @@ impl sys_dc::DataChannelObserver for InternalDataChannelObserver { } } -impl Default for InternalDataChannelObserver { - fn default() -> Self { - Self { - on_state_change_handler: Arc::new(Default::default()), - on_message_handler: Arc::new(Default::default()), - on_buffered_amount_change_handler: Arc::new(Default::default()), - } - } -} - #[derive(Debug)] pub struct DataChannelInit { #[deprecated] diff --git a/crates/livekit-webrtc/src/jsep.rs b/crates/livekit-webrtc/src/jsep.rs index 228ba62..4638545 100644 --- a/crates/livekit-webrtc/src/jsep.rs +++ b/crates/livekit-webrtc/src/jsep.rs @@ -1,4 +1,4 @@ -use std::fmt::{Debug, Display, Formatter}; +use std::fmt::{Debug, Formatter}; use cxx::UniquePtr; diff --git a/crates/livekit-webrtc/src/lib.rs b/crates/livekit-webrtc/src/lib.rs index 4bcbcfc..684388d 100644 --- a/crates/livekit-webrtc/src/lib.rs +++ b/crates/livekit-webrtc/src/lib.rs @@ -6,4 +6,6 @@ pub mod peer_connection_factory; pub mod rtc_error; pub mod rtp_receiver; pub mod rtp_transceiver; +pub mod video_frame; +pub mod video_frame_buffer; pub mod webrtc; diff --git a/crates/livekit-webrtc/src/media_stream.rs b/crates/livekit-webrtc/src/media_stream.rs index 38bfed9..bb5981d 100644 --- a/crates/livekit-webrtc/src/media_stream.rs +++ b/crates/livekit-webrtc/src/media_stream.rs @@ -1,14 +1,59 @@ use cxx::UniquePtr; -use std::fmt::{Debug, Formatter}; - use libwebrtc_sys::media_stream as sys_ms; +use libwebrtc_sys::MEDIA_TYPE_VIDEO; +use std::fmt::{Debug, Formatter}; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; + +pub use sys_ms::ffi::ContentHint; pub use sys_ms::ffi::TrackState; -pub struct MediaStreamTrack { - cxx_handle: UniquePtr, +use crate::video_frame::VideoFrame; + +pub trait MediaStreamTrackTrait { + fn kind(&self) -> String; + fn id(&self) -> String; + fn enabled(&self) -> bool; + fn set_enabled(&self, enabled: bool) -> bool; + fn state(&self) -> TrackState; } -impl Debug for MediaStreamTrack { +#[derive(Clone)] +pub enum MediaStreamTrackHandle { + Audio(Arc), + Video(Arc), +} + +macro_rules! shared_getter { + ($x:ident, $ret:ty) => { + fn $x(&self) -> $ret { + match self { + Self::Video(inner) => inner.$x(), + Self::Audio(inner) => inner.$x(), + } + } + }; +} + +impl MediaStreamTrackHandle { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + unsafe { + if cxx_handle.kind() == MEDIA_TYPE_VIDEO { + Self::Video(VideoTrack::new(UniquePtr::from_raw( + sys_ms::ffi::media_to_video(cxx_handle.into_raw()) + as *mut sys_ms::ffi::VideoTrack, + ))) + } else { + Self::Audio(AudioTrack::new(UniquePtr::from_raw( + sys_ms::ffi::media_to_audio(cxx_handle.into_raw()) + as *mut sys_ms::ffi::AudioTrack, + ))) + } + } + } +} + +impl Debug for MediaStreamTrackHandle { fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { f.debug_struct("MediaStreamTrack") .field("id", &self.id()) @@ -19,29 +64,191 @@ impl Debug for MediaStreamTrack { } } -impl MediaStreamTrack { - pub(crate) fn new(cxx_handle: UniquePtr) -> Self { - Self { cxx_handle } +impl MediaStreamTrackTrait for MediaStreamTrackHandle { + shared_getter!(kind, String); + shared_getter!(id, String); + shared_getter!(enabled, bool); + shared_getter!(state, TrackState); + + fn set_enabled(&self, enabled: bool) -> bool { + match self { + Self::Video(inner) => inner.set_enabled(enabled), + Self::Audio(inner) => inner.set_enabled(enabled), + } + } +} + +pub struct AudioTrack { + cxx_handle: Mutex>, +} + +impl AudioTrack { + pub(crate) fn new(cxx_handle: UniquePtr) -> Arc { + Arc::new(Self { + cxx_handle: Mutex::new(cxx_handle), + }) + } +} + +pub struct VideoTrack { + cxx_handle: Mutex>, + observer: Box, + + // Keep alive for c++ + native_observer: UniquePtr, +} + +macro_rules! impl_media_stream_track_trait { + ($x:ty, $cast:ident) => { + impl MediaStreamTrackTrait for $x { + fn kind(&self) -> String { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).kind() } + } + + fn id(&self) -> String { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).id() } + } + + fn enabled(&self) -> bool { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).enabled() } + } + + fn set_enabled(&self, enabled: bool) -> bool { + unsafe { + let media = + sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap()) as *mut sys_ms::ffi::MediaStreamTrack; + + Pin::new_unchecked(&mut *media).set_enabled(enabled) + } + } + + fn state(&self) -> TrackState { + unsafe { (*sys_ms::ffi::$cast(&**self.cxx_handle.lock().unwrap())).state() } + } + } + }; +} + +impl_media_stream_track_trait!(VideoTrack, video_to_media); +impl_media_stream_track_trait!(AudioTrack, audio_to_media); + +pub type OnFrameHandler = Box; +pub type OnDiscardedFrameHandler = Box; +pub type OnConstraintsChanged = Box; + +#[derive(Default)] +struct InternalVideoTrackSink { + on_frame_handler: Mutex>, + on_discarded_frame_handler: Mutex>, + on_constraints_changed_handler: Mutex>, +} + +pub struct VideoTrackSourceConstraints { + pub min_fps: Option, + pub max_fps: Option, +} + +impl From for VideoTrackSourceConstraints { + fn from(cst: sys_ms::ffi::VideoTrackSourceConstraints) -> Self { + Self { + min_fps: (cst.min_fps != 1.0).then_some(cst.min_fps), + max_fps: (cst.max_fps != 1.0).then_some(cst.max_fps), + } + } +} + +impl sys_ms::VideoFrameSink for InternalVideoTrackSink { + fn on_frame(&self, frame: UniquePtr) { + if let Some(cb) = self.on_frame_handler.lock().unwrap().as_mut() { + cb(VideoFrame::new(frame)); + } } - fn kind(&self) -> String { - self.cxx_handle.kind() + fn on_discarded_frame(&self) { + if let Some(cb) = self.on_discarded_frame_handler.lock().unwrap().as_mut() { + cb(); + } } - fn id(&self) -> String { - self.cxx_handle.id() + fn on_constraints_changed(&self, constraints: sys_ms::ffi::VideoTrackSourceConstraints) { + if let Some(cb) = self.on_constraints_changed_handler.lock().unwrap().as_mut() { + cb(constraints.into()); + } + } +} + +impl VideoTrack { + fn new(cxx_handle: UniquePtr) -> Arc { + let mut observer = Box::new(InternalVideoTrackSink::default()); + + let mut track = unsafe { + Self { + cxx_handle: Mutex::new(cxx_handle), + native_observer: sys_ms::ffi::create_native_video_frame_sink(Box::new( + sys_ms::VideoFrameSinkWrapper::new(&mut *observer), + )), + observer, + } + }; + + unsafe { + track + .cxx_handle + .lock() + .unwrap() + .pin_mut() + .add_sink(track.native_observer.pin_mut()); + } + + Arc::new(track) } - fn enabled(&self) -> bool { - self.cxx_handle.enabled() + pub fn set_should_receive(&mut self, should_receive: bool) { + self.cxx_handle + .lock() + .unwrap() + .pin_mut() + .set_should_receive(should_receive) } - fn set_enabled(&mut self, enable: bool) -> bool { - self.cxx_handle.pin_mut().set_enabled(enable) + pub fn set_content_hint(&mut self, hint: ContentHint) { + self.cxx_handle + .lock() + .unwrap() + .pin_mut() + .set_content_hint(hint) } - fn state(&self) -> TrackState { - self.cxx_handle.state() + pub fn should_receive(&self) -> bool { + self.cxx_handle.lock().unwrap().should_receive() + } + + pub fn content_hint(&self) -> ContentHint { + self.cxx_handle.lock().unwrap().content_hint() + } + + pub fn on_frame(&mut self, handler: OnFrameHandler) { + *self.observer.on_frame_handler.lock().unwrap() = Some(handler); + } + + pub fn on_discarded_frame(&mut self, handler: OnDiscardedFrameHandler) { + *self.observer.on_discarded_frame_handler.lock().unwrap() = Some(handler); + } + + pub fn on_constraints_changed(&mut self, handler: OnConstraintsChanged) { + *self.observer.on_constraints_changed_handler.lock().unwrap() = Some(handler); + } +} + +impl Drop for VideoTrack { + fn drop(&mut self) { + unsafe { + self.cxx_handle + .lock() + .unwrap() + .pin_mut() + .remove_sink(self.native_observer.pin_mut()); + } } } diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index b95005a..52fa6c0 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -18,7 +18,7 @@ pub use libwebrtc_sys::peer_connection::ffi::SignalingState; use crate::data_channel::{DataChannel, DataChannelInit}; use crate::jsep::{IceCandidate, SessionDescription}; -use crate::media_stream::MediaStream; +use crate::media_stream::{MediaStream, VideoTrack, AudioTrack}; use crate::rtc_error::RTCError; use crate::rtp_receiver::RtpReceiver; use crate::rtp_transceiver::RtpTransceiver; diff --git a/crates/livekit-webrtc/src/rtp_receiver.rs b/crates/livekit-webrtc/src/rtp_receiver.rs index 6a4ea5b..c35e1e2 100644 --- a/crates/livekit-webrtc/src/rtp_receiver.rs +++ b/crates/livekit-webrtc/src/rtp_receiver.rs @@ -1,4 +1,4 @@ -use crate::media_stream::MediaStreamTrack; +use crate::media_stream::{MediaStreamTrackHandle, MediaStreamTrackTrait}; use cxx::UniquePtr; use libwebrtc_sys::rtp_receiver as sys_rec; use std::fmt::{Debug, Formatter}; @@ -17,10 +17,12 @@ impl Debug for RtpReceiver { impl RtpReceiver { pub(crate) fn new(cxx_handle: UniquePtr) -> Self { - Self { cxx_handle } + Self { + cxx_handle, + } } - pub fn track(&self) -> MediaStreamTrack { - MediaStreamTrack::new(self.cxx_handle.track()) + pub fn track(&self) -> MediaStreamTrackHandle { + MediaStreamTrackHandle::new(self.cxx_handle.track()) } } diff --git a/crates/livekit-webrtc/src/video_frame.rs b/crates/livekit-webrtc/src/video_frame.rs new file mode 100644 index 0000000..4926525 --- /dev/null +++ b/crates/livekit-webrtc/src/video_frame.rs @@ -0,0 +1,50 @@ +use cxx::UniquePtr; +use libwebrtc_sys::video_frame as vf_sys; + +pub use vf_sys::ffi::VideoRotation; + +pub struct VideoFrame { + cxx_handle: UniquePtr, +} + +impl VideoFrame { + pub(crate) fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } + + pub fn width(&self) -> i32 { + self.cxx_handle.width() + } + + pub fn height(&self) -> i32 { + self.cxx_handle.height() + } + + pub fn size(&self) -> u32 { + self.cxx_handle.size() + } + + pub fn id(&self) -> u16 { + self.cxx_handle.id() + } + + pub fn timestamp_us(&self) -> i64 { + self.cxx_handle.timestamp_us() + } + + pub fn ntp_time_ms(&self) -> i64 { + self.cxx_handle.ntp_time_ms() + } + + pub fn transport_frame_id(&self) -> u32 { + self.cxx_handle.transport_frame_id() + } + + pub fn timestamp(&self) -> u32 { + self.cxx_handle.timestamp() + } + + pub fn rotation(&self) -> VideoRotation { + self.cxx_handle.rotation() + } +} diff --git a/crates/livekit-webrtc/src/video_frame_buffer.rs b/crates/livekit-webrtc/src/video_frame_buffer.rs new file mode 100644 index 0000000..9c60b85 --- /dev/null +++ b/crates/livekit-webrtc/src/video_frame_buffer.rs @@ -0,0 +1,18 @@ +use cxx::UniquePtr; +use libwebrtc_sys::video_frame_buffer as vfb_sys; + +pub use vfb_sys::ffi::VideoFrameBufferType; + +pub struct VideoFrameBuffer { + cxx_handle: UniquePtr, +} + +impl VideoFrameBuffer { + pub fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } + + pub fn buffer_type(&self) -> VideoFrameBufferType { + self.cxx_handle.buffer_type() + } +} diff --git a/examples/simple_room/src/main.rs b/examples/simple_room/src/main.rs index 52b161c..3997612 100644 --- a/examples/simple_room/src/main.rs +++ b/examples/simple_room/src/main.rs @@ -1,6 +1,7 @@ -use livekit::room::Room; +use livekit::room::track::TrackTrait; +use livekit::room::{track::remote_track::RemoteTrackHandle, Room}; use std::sync::{Arc, Mutex}; -use tracing::{info, trace}; +use tracing::{event_enabled, info, trace}; const URL: &str = "ws://localhost:7880"; const TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0NzMsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NjQ4MDY0NzMsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.BgVdBnq3XFD3_BQHoe1azqjifYysubgFl6Qlzu9IQGI"; @@ -12,8 +13,19 @@ async fn main() { tracing_subscriber::fmt::init(); let room = Room::new(); - room.on_participant_connected(async |participant| { + room.events() + .on_participant_connected(|event| async move {}); + + room.events().on_track_subscribed(|event| async move { + let track = event.publication.track().unwrap(); + if let RemoteTrackHandle::Video(video_track) = track { + let rtc_track = video_track.rtc_track(); + rtc_track.on_frame(Box::new(|frame| { Box::pin(async move { + + - }) + }) })) + } + }); }