From 39fa798c7f1d2b1bca245e7235e12834f276c357 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Mon, 28 Nov 2022 15:52:37 +0100 Subject: [PATCH] trying to receive VideoFrame, progress... --- crates/livekit-core/Cargo.toml | 2 +- crates/livekit-core/src/events.rs | 2 +- crates/livekit-core/src/room/mod.rs | 73 +++-- .../src/room/participant/local_participant.rs | 12 +- .../livekit-core/src/room/participant/mod.rs | 22 +- .../room/participant/remote_participant.rs | 66 ++++- .../livekit-core/src/room/publication/mod.rs | 97 +++++-- crates/livekit-core/src/room/track/mod.rs | 27 +- crates/livekit-core/src/rtc_engine/mod.rs | 17 +- .../include/livekit/media_stream.h | 8 +- .../include/livekit/video_frame.h | 4 +- .../include/livekit/video_frame_buffer.h | 32 ++- .../libwebrtc-sys/src/media_stream.cpp | 4 + .../libwebrtc-sys/src/video_frame.rs | 2 +- .../libwebrtc-sys/src/video_frame_buffer.rs | 15 +- crates/livekit-webrtc/src/media_stream.rs | 21 +- crates/livekit-webrtc/src/peer_connection.rs | 8 +- crates/livekit-webrtc/src/video_frame.rs | 12 +- .../livekit-webrtc/src/video_frame_buffer.rs | 250 +++++++++++++++++- examples/simple_room/src/main.rs | 29 +- 20 files changed, 575 insertions(+), 128 deletions(-) diff --git a/crates/livekit-core/Cargo.toml b/crates/livekit-core/Cargo.toml index a9ce7cd..3cf11f7 100644 --- a/crates/livekit-core/Cargo.toml +++ b/crates/livekit-core/Cargo.toml @@ -21,4 +21,4 @@ lazy_static = "1.4.0" tracing = "0.1" [build-dependencies] -prost-build = { version = "0.11.1" } \ No newline at end of file +prost-build = { version = "0.11.1" } diff --git a/crates/livekit-core/src/events.rs b/crates/livekit-core/src/events.rs index 5dfecd7..c85506f 100644 --- a/crates/livekit-core/src/events.rs +++ b/crates/livekit-core/src/events.rs @@ -8,7 +8,7 @@ macro_rules! event_setter { pub fn $fnc(&self, mut callback: F) where F: FnMut($event) -> Fut + Send + Sync + 'static, - Fut: Future + Send + Sync + 'static, + Fut: Future + Send + 'static, { *self.$fnc.lock() = Some(Box::new(move |event| Box::pin(callback(event)))); } diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index c293d29..f9305ea 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -9,11 +9,13 @@ 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::events::room::{ + ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents, TrackSubscribedEvent, +}; use crate::proto; use crate::proto::participant_info; use thiserror::Error; -use tracing::error; +use tracing::{debug, error}; use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, RTCEngine}; use crate::signal_client::SignalOptions; @@ -63,10 +65,6 @@ impl Room { } } - 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?; @@ -88,11 +86,21 @@ impl Room { self.inner = Some(inner.clone()); + // Add already connected participants + for pi in join_response.other_participants { + let p = Self::create_participant(inner.clone(), self.events.clone(), pi.clone()); + p.update_info(pi).await; + } + tokio::spawn(Self::room_task(inner, self.events.clone(), engine_events)); Ok(()) } + pub fn events(&self) -> Arc { + self.events.clone() + } + pub fn get_handle(&self) -> Option { self.inner.as_ref().map(|inner| RoomHandle { inner: inner.clone(), @@ -121,6 +129,7 @@ impl Room { match event { EngineEvent::ParticipantUpdate(update) => { Self::handle_participant_update(room_inner.clone(), room_events.clone(), update) + .await } EngineEvent::AddTrack { rtp_receiver, @@ -164,7 +173,7 @@ impl Room { Ok(()) } - fn handle_participant_update( + async fn handle_participant_update( room_inner: Arc, room_events: Arc, update: proto::ParticipantUpdate, @@ -173,7 +182,7 @@ impl Room { if pi.sid == room_inner.local_participant.sid() || pi.identity == room_inner.local_participant.identity() { - room_inner.local_participant.update_info(pi); + room_inner.local_participant.clone().update_info(pi).await; continue; } @@ -190,12 +199,12 @@ impl Room { ) } else { // Participant is already connected, update the informations - remote_participant.update_info(pi); + remote_participant.update_info(pi).await; } } else { // Create a new participant and call OnConnect event let remote_participant = - Self::get_or_create_participant(room_inner.clone(), room_events.clone(), pi); + Self::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 { @@ -235,29 +244,41 @@ impl Room { room_inner.participants.read().get(sid).cloned() } - fn get_or_create_participant( + fn create_participant( room_inner: Arc, room_events: 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)); + let p = Arc::new(RemoteParticipant::new(pi.clone())); - // Forward participantevents to room events - p.internal_events().on_track_published({ + // Forward participantevents to room events + p.internal_events().on_track_subscribed({ + let room_events = room_events.clone(); + let room_inner = room_inner.clone(); + + move |event| { let room_events = room_events.clone(); - |event| async move {} - }); + let room_inner = room_inner.clone(); - participants.insert(sid, p.clone()); - p - } + async move { + if let Some(cb) = room_events.clone().on_track_subscribed.lock().as_mut() { + cb(TrackSubscribedEvent { + room_handle: RoomHandle::from(room_inner.clone()), + track: event.track, + participant: event.participant, + publication: event.publication, + }) + .await; + } + } + } + }); + + room_inner + .participants + .write() + .insert(pi.sid.into(), p.clone()); + p } } diff --git a/crates/livekit-core/src/room/participant/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs index ebc877f..7927e21 100644 --- a/crates/livekit-core/src/room/participant/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -1,5 +1,5 @@ use crate::proto::{data_packet, DataPacket, UserPacket}; -use crate::room::participant::{impl_participant_trait, ParticipantShared}; +use crate::room::participant::{impl_participant_trait, ParticipantShared, ParticipantInternalTrait}; use crate::room::RoomError; use crate::rtc_engine::RTCEngine; @@ -40,6 +40,16 @@ impl LocalParticipant { .await .map_err(Into::into) } + + pub(crate) async fn update_info(self: Arc, info: ParticipantInfo) { + self.shared.update_info(info); + } +} + +impl ParticipantInternalTrait for LocalParticipant { + fn internal_events(&self) -> Arc { + self.shared.internal_events.clone() + } } impl_participant_trait!(LocalParticipant); diff --git a/crates/livekit-core/src/room/participant/mod.rs b/crates/livekit-core/src/room/participant/mod.rs index 9158452..56c7da2 100644 --- a/crates/livekit-core/src/room/participant/mod.rs +++ b/crates/livekit-core/src/room/participant/mod.rs @@ -65,7 +65,6 @@ pub trait ParticipantTrait { fn identity(&self) -> ParticipantIdentity; fn name(&self) -> String; fn metadata(&self) -> String; - fn update_info(&self, info: ParticipantInfo); } #[derive(Clone)] @@ -74,6 +73,16 @@ pub enum ParticipantHandle { Remote(Arc), } +impl ParticipantHandle { + // TODO(theomonnom): Add async support to wrap_variants ... + pub(crate) async fn update_info(&self, info: ParticipantInfo) { + match self { + Self::Local(inner) => inner.clone().update_info(info).await, + Self::Remote(inner) => inner.clone().update_info(info).await, + } + } +} + impl ParticipantInternalTrait for ParticipantHandle { wrap_variants!( [Local, Remote] @@ -89,7 +98,6 @@ impl ParticipantTrait for ParticipantHandle { fnc!(identity, ParticipantIdentity, []); fnc!(name, String, []); fnc!(metadata, String, []); - fnc!(update_info, (), [info: ParticipantInfo]); ); } @@ -100,12 +108,6 @@ macro_rules! impl_participant_trait { 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() @@ -126,10 +128,6 @@ macro_rules! impl_participant_trait { fn metadata(&self) -> String { self.shared.metadata.lock().clone() } - - fn update_info(&self, info: ParticipantInfo) { - self.shared.update_info(info); - } } }; } diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index 1266504..12efbae 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -1,16 +1,25 @@ -use crate::events::participant::{TrackSubscribedEvent, TrackSubscriptionFailedEvent}; +use crate::events::participant::{ + TrackPublishedEvent, 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::participant::{ + impl_participant_trait, ParticipantInternalTrait, ParticipantShared, +}; +use crate::room::publication::{ + RemoteTrackPublication, TrackPublication, TrackPublicationInternalTrait, TrackPublicationTrait, +}; use crate::room::track::remote_audio_track::RemoteAudioTrack; use crate::room::track::remote_track::RemoteTrackHandle; use crate::room::track::remote_video_track::RemoteVideoTrack; -use crate::room::track::{TrackTrait, TrackKind}; +use crate::room::track::{TrackKind, TrackTrait, TrackHandle}; use livekit_webrtc::media_stream::MediaStreamTrackHandle; +use std::collections::HashSet; use std::time::Duration; use tokio::time::{sleep, timeout}; -use tracing::error; +use tracing::{info, error}; + +use super::ParticipantTrait; const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5); @@ -80,6 +89,9 @@ impl RemoteParticipant { _ => unreachable!(), }; + info!("starting track: {:?}", sid); + + remote_publication.update_track(Some(track.clone().into())); self.shared .add_track_publication(TrackPublication::Remote(remote_publication.clone())); track.start(); @@ -144,6 +156,50 @@ impl RemoteParticipant { } }) } + + pub(crate) async fn update_info(self: Arc, info: ParticipantInfo) { + self.shared.update_info(info.clone()); + + let mut valid_tracks = HashSet::::new(); + + for track in info.tracks { + if let Some(publication) = self.get_track_publication(&track.sid.clone().into()) { + publication.update_info(track.clone()); + } else { + let publication = RemoteTrackPublication::new(track.clone(), self.sid(), None); + self.shared + .add_track_publication(TrackPublication::Remote(publication.clone())); + + // This is a new track, fire publish events + let event = TrackPublishedEvent { + participant: self.clone(), + publication: publication.clone(), + }; + + if let Some(cb) = self + .shared + .internal_events + .on_track_published + .lock() + .as_mut() + { + cb(event.clone()).await; + } + + if let Some(cb) = self.shared.events.on_track_published.lock().as_mut() { + cb(event).await; + } + } + + valid_tracks.insert(track.sid.into()); + } + } +} + +impl ParticipantInternalTrait for RemoteParticipant { + fn internal_events(&self) -> Arc { + self.shared.internal_events.clone() + } } impl_participant_trait!(RemoteParticipant); diff --git a/crates/livekit-core/src/room/publication/mod.rs b/crates/livekit-core/src/room/publication/mod.rs index 5a36bbb..1820e88 100644 --- a/crates/livekit-core/src/room/publication/mod.rs +++ b/crates/livekit-core/src/room/publication/mod.rs @@ -1,11 +1,22 @@ +use crate::proto::TrackType; +use crate::proto::{TrackInfo, TrackSource as ProtoTrackSource}; +use crate::room::id::ParticipantSid; use crate::room::id::TrackSid; use crate::room::track::local_track::LocalTrackHandle; use crate::room::track::remote_track::RemoteTrackHandle; use crate::room::track::{TrackHandle, TrackKind, TrackSource}; +use crate::utils::wrap_variants; use parking_lot::Mutex; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; +use super::track::TrackDimension; + +pub(crate) trait TrackPublicationInternalTrait { + fn update_track(&self, track: Option); + fn update_info(&self, info: TrackInfo); +} + pub trait TrackPublicationTrait { fn name(&self) -> String; fn sid(&self) -> TrackSid; @@ -21,6 +32,43 @@ pub(super) struct TrackPublicationShared { pub(super) kind: AtomicU8, // Casted to TrackKind pub(super) source: AtomicU8, // Casted to TrackSource pub(super) simulcasted: AtomicBool, + pub(super) dimension: Mutex, + pub(super) mime_type: Mutex, + pub(super) participant: ParticipantSid, // TODO(theomonnom) Use WeakParticipant instead +} + +impl TrackPublicationShared { + pub fn new(info: TrackInfo, participant: ParticipantSid, track: Option) -> Arc { + Arc::new(Self { + track: Mutex::new(track), + name: Mutex::new(info.name), + sid: Mutex::new(info.sid.into()), + kind: AtomicU8::new(TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8), + source: AtomicU8::new(TrackSource::from( + ProtoTrackSource::from_i32(info.source).unwrap(), + ) as u8), + simulcasted: AtomicBool::new(info.simulcast), + dimension: Mutex::new(TrackDimension(info.width, info.height)), + mime_type: Mutex::new(info.mime_type), + participant, + }) + } + + pub fn update_info(&self, info: TrackInfo) { + *self.name.lock() = info.name; + *self.sid.lock() = info.sid.into(); + self.kind.store( + TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8, + Ordering::SeqCst, + ); + self.source.store( + TrackSource::from(ProtoTrackSource::from_i32(info.source).unwrap()) as u8, + Ordering::SeqCst, + ); + self.simulcasted.store(info.simulcast, Ordering::SeqCst); + *self.dimension.lock() = TrackDimension(info.width, info.height); + *self.mime_type.lock() = info.mime_type; + } } #[derive(Clone)] @@ -29,19 +77,9 @@ pub enum TrackPublication { Remote(RemoteTrackPublication), } -macro_rules! shared_getter { - ($x:ident, $ret:ty) => { - fn $x(&self) -> $ret { - match self { - TrackPublication::Local(p) => p.$x(), - TrackPublication::Remote(p) => p.$x(), - } - } - }; -} - impl TrackPublication { pub fn track(&self) -> Option { + // Not calling Local/Remote function here, we don't need "cast" match self { TrackPublication::Local(p) => p.shared.track.lock().clone(), TrackPublication::Remote(p) => p.shared.track.lock().clone(), @@ -49,16 +87,37 @@ impl TrackPublication { } } +impl TrackPublicationInternalTrait for TrackPublication { + wrap_variants!( + [Local, Remote] + fnc!(update_track, (), [track: Option]); + fnc!(update_info, (), [info: TrackInfo]); + ); +} + impl TrackPublicationTrait for TrackPublication { - shared_getter!(name, String); - shared_getter!(sid, TrackSid); - shared_getter!(kind, TrackKind); - shared_getter!(source, TrackSource); - shared_getter!(simulcasted, bool); + wrap_variants!( + [Local, Remote] + fnc!(sid, TrackSid, []); + fnc!(name, String, []); + fnc!(kind, TrackKind, []); + fnc!(source, TrackSource, []); + fnc!(simulcasted, bool, []); + ); } macro_rules! impl_publication_trait { ($x:ident) => { + impl TrackPublicationInternalTrait for $x { + fn update_track(&self, track: Option) { + *self.shared.track.lock() = track; + } + + fn update_info(&self, info: TrackInfo) { + self.shared.update_info(info); + } + } + impl TrackPublicationTrait for $x { fn name(&self) -> String { self.shared.name.lock().clone() @@ -104,6 +163,12 @@ pub struct RemoteTrackPublication { } impl RemoteTrackPublication { + pub fn new(info: TrackInfo, participant: ParticipantSid, track: Option) -> Self { + Self { + shared: TrackPublicationShared::new(info, participant, track), + } + } + pub fn track(&self) -> Option { self.shared .track diff --git a/crates/livekit-core/src/room/track/mod.rs b/crates/livekit-core/src/room/track/mod.rs index ff53f4d..327084f 100644 --- a/crates/livekit-core/src/room/track/mod.rs +++ b/crates/livekit-core/src/room/track/mod.rs @@ -1,3 +1,4 @@ +use crate::proto::{TrackSource as ProtoTrackSource, TrackType}; use crate::room::id::TrackSid; use crate::room::track::local_audio_track::LocalAudioTrack; use crate::room::track::local_video_track::LocalVideoTrack; @@ -36,6 +37,16 @@ impl From for TrackKind { } } +impl From for TrackKind { + fn from(r#type: TrackType) -> Self { + match r#type { + TrackType::Audio => Self::Audio, + TrackType::Video => Self::Video, + _ => Self::Unknown, + } + } +} + #[derive(Debug)] pub enum StreamState { Unknown, @@ -74,6 +85,20 @@ impl From for TrackSource { } } +impl From for TrackSource { + fn from(source: ProtoTrackSource) -> Self { + match source { + ProtoTrackSource::Camera => Self::Camera, + ProtoTrackSource::Microphone => Self::Microphone, + ProtoTrackSource::ScreenShare => Self::Screenshare, + ProtoTrackSource::ScreenShareAudio => Self::ScreenshareAudio, + ProtoTrackSource::Unknown => Self::Unknown, + } + } +} + +pub struct TrackDimension(pub u32, pub u32); + pub trait TrackTrait { fn sid(&self) -> TrackSid; fn name(&self) -> String; @@ -103,7 +128,7 @@ impl TrackShared { name: Mutex::new(name), kind: AtomicU8::new(kind as u8), stream_state: AtomicU8::new(StreamState::Active as u8), - rtc_track: rtc_track, + rtc_track, } } diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index 1410b59..4fbb08a 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -10,7 +10,7 @@ use prost::Message; use serde::{Deserialize, Serialize}; use thiserror::Error; use tokio::time::sleep; -use tracing::{debug, error, trace}; +use tracing::{debug, error, info, trace}; use crate::{proto, signal_client}; use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState}; @@ -280,6 +280,7 @@ impl RTCEngine { } RTCEvent::ConnectionChange { state, target } => { // Reconnect if we've been disconnected unexpectedly + trace!("Connection change, {:?} {:?}", state, target); let subscriber_primary = engine_inner.join_response.lock().subscriber_primary; let is_primary = subscriber_primary && target == SignalTarget::Subscriber; @@ -328,10 +329,12 @@ impl RTCEngine { target, } => { if target == SignalTarget::Subscriber { - let _ = emitter.send(EngineEvent::AddTrack { - rtp_receiver, - streams, - }); + let _ = emitter + .send(EngineEvent::AddTrack { + rtp_receiver, + streams, + }) + .await; } } RTCEvent::Data { data, binary } => { @@ -377,7 +380,7 @@ impl RTCEngine { 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); + trace!("received offer for the subscriber: {:?}", offer); let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; engine_inner @@ -438,7 +441,7 @@ impl RTCEngine { } } signal_response::Message::Update(update) => { - let _ = emitter.send(EngineEvent::ParticipantUpdate(update)); + let _ = emitter.send(EngineEvent::ParticipantUpdate(update)).await; } _ => {} } 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 bf5d652..603940a 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/media_stream.h @@ -102,19 +102,19 @@ class NativeVideoFrameSink std::unique_ptr create_native_video_frame_sink( rust::Box observer); -const MediaStreamTrack* video_to_media(const VideoTrack* track) { +static const MediaStreamTrack* video_to_media(const VideoTrack* track) { return track; } -const MediaStreamTrack* audio_to_media(const AudioTrack* track) { +static const MediaStreamTrack* audio_to_media(const AudioTrack* track) { return track; } -const VideoTrack* media_to_video(const MediaStreamTrack* track) { +static const VideoTrack* media_to_video(const MediaStreamTrack* track) { return static_cast(track); } -const AudioTrack* media_to_audio(const MediaStreamTrack* track) { +static const AudioTrack* media_to_audio(const MediaStreamTrack* track) { return static_cast(track); } diff --git a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h index 9c719f4..6a247a5 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame.h @@ -30,8 +30,8 @@ class VideoFrame { } // 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()); + std::unique_ptr video_frame_buffer() const { + return std::make_unique(frame_.video_frame_buffer()); } private: 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 index 3b4909c..1e3cfe2 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame_buffer.h +++ b/crates/livekit-webrtc/libwebrtc-sys/include/livekit/video_frame_buffer.h @@ -5,6 +5,8 @@ #ifndef LIVEKIT_WEBRTC_VIDEO_FRAME_BUFFER_H #define LIVEKIT_WEBRTC_VIDEO_FRAME_BUFFER_H +#include + #include "api/video/video_frame_buffer.h" #include "rust_types.h" @@ -26,8 +28,15 @@ class VideoFrameBuffer { int width() const { return buffer_->width(); } int height() const { return buffer_->height(); } - std::shared_ptr to_i420() { - return std::make_shared(buffer_->ToI420()); + std::unique_ptr to_i420() { + return std::make_unique(buffer_->ToI420()); + } + + std::unique_ptr get_i420() { + // const_cast is valid here because we take the ownership on the rust side + return std::make_unique( + rtc::scoped_refptr( + const_cast(buffer_->GetI420()))); } protected: @@ -73,19 +82,20 @@ class I420Buffer : public PlanarYuv8Buffer { : PlanarYuv8Buffer(buffer) {} }; -std::shared_ptr to_video_frame_buffer( - std::shared_ptr buffer) { - return buffer; +static const VideoFrameBuffer* yuv_to_vfb(const PlanarYuvBuffer* yuv) { + return yuv; } -std::shared_ptr to_yuv_buffer( - std::shared_ptr buffer) { - return buffer; +static const PlanarYuvBuffer* yuv8_to_yuv(const PlanarYuv8Buffer* yuv8) { + return yuv8; } -std::shared_ptr to_yuv8_buffer( - std::shared_ptr buffer) { - return buffer; +static const PlanarYuv8Buffer* i420_to_yuv8(const I420Buffer* i420) { + return i420; +} + +static std::unique_ptr _unique_video_frame_buffer() { + return nullptr; // Ignore } } // namespace livekit diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp index b4017ee..b2758a3 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/media_stream.cpp @@ -4,6 +4,7 @@ #include "livekit/media_stream.h" +#include "api/media_stream_interface.h" #include "libwebrtc-sys/src/media_stream.rs.h" namespace livekit { @@ -53,6 +54,9 @@ rust::String MediaStream::id() const { return media_stream_->id(); } +AudioTrack::AudioTrack(rtc::scoped_refptr track) + : MediaStreamTrack(std::move(track)) {} + VideoTrack::VideoTrack(rtc::scoped_refptr track) : MediaStreamTrack(std::move(track)) {} diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs b/crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs index 79ba2bf..def1933 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/video_frame.rs @@ -25,7 +25,7 @@ pub mod ffi { 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 video_frame_buffer(self: &VideoFrame) -> UniquePtr; 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 index 65d4fad..ecfa4d2 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/video_frame_buffer.rs +++ b/crates/livekit-webrtc/libwebrtc-sys/src/video_frame_buffer.rs @@ -23,7 +23,11 @@ pub mod ffi { fn buffer_type(self: &VideoFrameBuffer) -> VideoFrameBufferType; fn width(self: &VideoFrameBuffer) -> i32; fn height(self: &VideoFrameBuffer) -> i32; - fn to_i420(self: Pin<&mut VideoFrameBuffer>) -> SharedPtr; + + // Require ownership + unsafe fn to_i420(self: Pin<&mut VideoFrameBuffer>) -> UniquePtr; + unsafe fn get_i420(self: Pin<&mut VideoFrameBuffer>) -> UniquePtr; + // TODO(theomonnom): Bridge other get_* fn chroma_width(self: &PlanarYuvBuffer) -> i32; fn chroma_height(self: &PlanarYuvBuffer) -> i32; @@ -35,9 +39,10 @@ pub mod ffi { 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; + unsafe fn yuv_to_vfb(yuv: *const PlanarYuvBuffer) -> *const VideoFrameBuffer; + unsafe fn yuv8_to_yuv(yuv8: *const PlanarYuv8Buffer) -> *const PlanarYuvBuffer; + unsafe fn i420_to_yuv8(i420: *const I420Buffer) -> *const PlanarYuv8Buffer; + + fn _unique_video_frame_buffer() -> UniquePtr; } } diff --git a/crates/livekit-webrtc/src/media_stream.rs b/crates/livekit-webrtc/src/media_stream.rs index bb5981d..565c92b 100644 --- a/crates/livekit-webrtc/src/media_stream.rs +++ b/crates/livekit-webrtc/src/media_stream.rs @@ -9,6 +9,7 @@ pub use sys_ms::ffi::ContentHint; pub use sys_ms::ffi::TrackState; use crate::video_frame::VideoFrame; +use crate::video_frame_buffer::VideoFrameBuffer; pub trait MediaStreamTrackTrait { fn kind(&self) -> String; @@ -115,8 +116,8 @@ macro_rules! impl_media_stream_track_trait { 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; + 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) } @@ -132,7 +133,7 @@ macro_rules! impl_media_stream_track_trait { impl_media_stream_track_trait!(VideoTrack, video_to_media); impl_media_stream_track_trait!(AudioTrack, audio_to_media); -pub type OnFrameHandler = Box; +pub type OnFrameHandler = Box; pub type OnDiscardedFrameHandler = Box; pub type OnConstraintsChanged = Box; @@ -160,7 +161,9 @@ impl From for VideoTrackSourceConstrai 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)); + let frame = VideoFrame::new(frame); + let video_frame_buffer = unsafe { frame.video_frame_buffer() }; + cb(frame, video_frame_buffer); } } @@ -203,7 +206,7 @@ impl VideoTrack { Arc::new(track) } - pub fn set_should_receive(&mut self, should_receive: bool) { + pub fn set_should_receive(&self, should_receive: bool) { self.cxx_handle .lock() .unwrap() @@ -211,7 +214,7 @@ impl VideoTrack { .set_should_receive(should_receive) } - pub fn set_content_hint(&mut self, hint: ContentHint) { + pub fn set_content_hint(&self, hint: ContentHint) { self.cxx_handle .lock() .unwrap() @@ -227,15 +230,15 @@ impl VideoTrack { self.cxx_handle.lock().unwrap().content_hint() } - pub fn on_frame(&mut self, handler: OnFrameHandler) { + pub fn on_frame(&self, handler: OnFrameHandler) { *self.observer.on_frame_handler.lock().unwrap() = Some(handler); } - pub fn on_discarded_frame(&mut self, handler: OnDiscardedFrameHandler) { + pub fn on_discarded_frame(&self, handler: OnDiscardedFrameHandler) { *self.observer.on_discarded_frame_handler.lock().unwrap() = Some(handler); } - pub fn on_constraints_changed(&mut self, handler: OnConstraintsChanged) { + pub fn on_constraints_changed(&self, handler: OnConstraintsChanged) { *self.observer.on_constraints_changed_handler.lock().unwrap() = Some(handler); } } diff --git a/crates/livekit-webrtc/src/peer_connection.rs b/crates/livekit-webrtc/src/peer_connection.rs index 52fa6c0..7cec14e 100644 --- a/crates/livekit-webrtc/src/peer_connection.rs +++ b/crates/livekit-webrtc/src/peer_connection.rs @@ -459,7 +459,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver { } fn on_ice_connection_change(&self, new_state: IceConnectionState) { - trace!("on_ice_connection_change"); + trace!("on_ice_connection_change (new_state: {:?})", new_state); let mut handler = self.on_ice_connection_change_handler.lock().unwrap(); if let Some(f) = handler.as_mut() { f(new_state); @@ -467,7 +467,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver { } fn on_standardized_ice_connection_change(&self, new_state: IceConnectionState) { - trace!("on_standardized_ice_connection_change"); + trace!("on_standardized_ice_connection_change (new_state: {:?}", new_state); let mut handler = self .on_standardized_ice_connection_change_handler .lock() @@ -478,7 +478,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver { } fn on_connection_change(&self, new_state: PeerConnectionState) { - trace!("on_connection_change"); + trace!("on_connection_change (new_state: {:?})", new_state); let mut handler = self.on_connection_change_handler.lock().unwrap(); if let Some(f) = handler.as_mut() { f(new_state); @@ -486,7 +486,7 @@ impl sys_pc::PeerConnectionObserver for InternalObserver { } fn on_ice_gathering_change(&self, new_state: IceGatheringState) { - trace!("on_ice_gathering_change"); + trace!("on_ice_gathering_change (new_state: {:?}", new_state); let mut handler = self.on_ice_gathering_change_handler.lock().unwrap(); if let Some(f) = handler.as_mut() { f(new_state); diff --git a/crates/livekit-webrtc/src/video_frame.rs b/crates/livekit-webrtc/src/video_frame.rs index 4926525..c36bf79 100644 --- a/crates/livekit-webrtc/src/video_frame.rs +++ b/crates/livekit-webrtc/src/video_frame.rs @@ -3,6 +3,8 @@ use libwebrtc_sys::video_frame as vf_sys; pub use vf_sys::ffi::VideoRotation; +use crate::video_frame_buffer::VideoFrameBuffer; + pub struct VideoFrame { cxx_handle: UniquePtr, } @@ -31,7 +33,7 @@ impl VideoFrame { pub fn timestamp_us(&self) -> i64 { self.cxx_handle.timestamp_us() } - + pub fn ntp_time_ms(&self) -> i64 { self.cxx_handle.ntp_time_ms() } @@ -47,4 +49,12 @@ impl VideoFrame { pub fn rotation(&self) -> VideoRotation { self.cxx_handle.rotation() } + + /// # Safety + /// Must be called only once, this function create the safe Rust + /// wrapper around a VideoFrameBuffer. + /// Only one wrapper musts exist at a time. + pub(crate) unsafe fn video_frame_buffer(&self) -> VideoFrameBuffer { + VideoFrameBuffer::new(self.cxx_handle.video_frame_buffer()) + } } diff --git a/crates/livekit-webrtc/src/video_frame_buffer.rs b/crates/livekit-webrtc/src/video_frame_buffer.rs index 9c60b85..64cf28d 100644 --- a/crates/livekit-webrtc/src/video_frame_buffer.rs +++ b/crates/livekit-webrtc/src/video_frame_buffer.rs @@ -1,18 +1,250 @@ use cxx::UniquePtr; use libwebrtc_sys::video_frame_buffer as vfb_sys; +use std::pin::Pin; +use std::slice; +use vfb_sys::ffi::VideoFrameBufferType; -pub use vfb_sys::ffi::VideoFrameBufferType; +pub trait VideoFrameBufferTrait { + fn width(&self) -> i32; + fn height(&self) -> i32; + fn to_i420(self) -> I420Buffer; +} -pub struct VideoFrameBuffer { - cxx_handle: UniquePtr, +pub trait PlanarYuvBuffer: VideoFrameBufferTrait { + fn chroma_width(&self) -> i32; + fn chroma_height(&self) -> i32; + fn stride_y(&self) -> i32; + fn stride_u(&self) -> i32; + fn stride_v(&self) -> i32; +} + +pub trait PlanarYuv8Buffer: PlanarYuvBuffer { + fn data_y(&self) -> &[u8]; + fn data_u(&self) -> &[u8]; + fn data_v(&self) -> &[u8]; +} + +pub enum VideoFrameBuffer { + Native(NativeBuffer), + I420(I420Buffer), + I420A(I420ABuffer), + I422(I422Buffer), + I444(I444Buffer), + I010(I010Buffer), + NV12(NV12Buffer), } impl VideoFrameBuffer { - pub fn new(cxx_handle: UniquePtr) -> Self { - Self { cxx_handle } - } - - pub fn buffer_type(&self) -> VideoFrameBufferType { - self.cxx_handle.buffer_type() + pub(crate) fn new(mut cxx_handle: UniquePtr) -> Self { + unsafe { + match cxx_handle.buffer_type() { + VideoFrameBufferType::Native => Self::Native(NativeBuffer::new(cxx_handle)), + VideoFrameBufferType::I420 => { + Self::I420(I420Buffer::new(cxx_handle.pin_mut().get_i420())) + } + VideoFrameBufferType::I420A => Self::I420A(I420ABuffer::new(cxx_handle)), + VideoFrameBufferType::I422 => Self::I422(I422Buffer::new(cxx_handle)), + VideoFrameBufferType::I444 => Self::I444(I444Buffer::new(cxx_handle)), + VideoFrameBufferType::I010 => Self::I010(I010Buffer::new(cxx_handle)), + VideoFrameBufferType::NV12 => Self::NV12(NV12Buffer::new(cxx_handle)), + _ => unreachable!(), // VideoFrameBufferType is represented as i32 + } + } + } +} + +macro_rules! recursive_cast { + ($ptr:expr $(, $fnc:ident)*) => { + { + let ptr = $ptr; + $( + let ptr = unsafe { vfb_sys::ffi::$fnc(ptr) }; + )* + ptr + } + }; +} + +macro_rules! impl_video_frame_buffer { + ($x:ty $(, $cast:ident)*) => { + + // Allow unused_unsafe when we don't do any cast ( e.g. NativeBuffer ) + #[allow(unused_unsafe)] + impl VideoFrameBufferTrait for $x { + fn width(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).width() + } + } + + fn height(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).height() + } + } + + // Require ownership because libwebrtc uses the same pointers + fn to_i420(self) -> I420Buffer { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*) + as *const vfb_sys::ffi::VideoFrameBuffer + as *mut vfb_sys::ffi::VideoFrameBuffer; + + unsafe { + I420Buffer::new(Pin::new_unchecked(&mut *ptr).to_i420()) + } + } + } + }; +} + +macro_rules! impl_yuv_buffer { + ($x:ty $(, $cast:ident)*) => { + impl PlanarYuvBuffer for $x { + fn chroma_width(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).chroma_width() + } + } + + fn chroma_height(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).chroma_height() + } + } + + fn stride_y(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).stride_y() + } + } + + fn stride_u(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).stride_u() + } + } + + fn stride_v(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).stride_v() + } + } + } + }; +} + +macro_rules! impl_yuv8_buffer { + ($x:ty $(, $cast:ident)*) => { + impl PlanarYuv8Buffer for $x { + fn data_y(&self) -> &[u8] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + slice::from_raw_parts((*ptr).data_y(), self.stride_y().try_into().unwrap()) + } + } + + fn data_u(&self) -> &[u8] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + slice::from_raw_parts((*ptr).data_u(), self.stride_u().try_into().unwrap()) + } + } + + fn data_v(&self) -> &[u8] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + slice::from_raw_parts((*ptr).data_v(), self.stride_v().try_into().unwrap()) + } + } + } + }; +} + +pub struct NativeBuffer { + cxx_handle: UniquePtr, +} + +pub struct I420Buffer { + cxx_handle: UniquePtr, +} + +pub struct I420ABuffer { + cxx_handle: UniquePtr, +} + +pub struct I422Buffer { + cxx_handle: UniquePtr, +} + +pub struct I444Buffer { + cxx_handle: UniquePtr, +} + +pub struct I010Buffer { + cxx_handle: UniquePtr, +} + +pub struct NV12Buffer { + cxx_handle: UniquePtr, +} + +impl_video_frame_buffer!(NativeBuffer); +impl_video_frame_buffer!(I420Buffer, i420_to_yuv8, yuv8_to_yuv, yuv_to_vfb); +impl_video_frame_buffer!(I420ABuffer); +impl_video_frame_buffer!(I422Buffer); +impl_video_frame_buffer!(I444Buffer); +impl_video_frame_buffer!(I010Buffer); +impl_video_frame_buffer!(NV12Buffer); + +impl_yuv_buffer!(I420Buffer, i420_to_yuv8, yuv8_to_yuv); + +impl_yuv8_buffer!(I420Buffer, i420_to_yuv8); + +impl NativeBuffer { + fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } +} + +impl I420Buffer { + fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } +} + +impl I420ABuffer { + fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } +} + +impl I422Buffer { + fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } +} + +impl I444Buffer { + fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } +} + +impl I010Buffer { + fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } + } +} + +impl NV12Buffer { + fn new(cxx_handle: UniquePtr) -> Self { + Self { cxx_handle } } } diff --git a/examples/simple_room/src/main.rs b/examples/simple_room/src/main.rs index 3997612..1b49771 100644 --- a/examples/simple_room/src/main.rs +++ b/examples/simple_room/src/main.rs @@ -1,7 +1,8 @@ -use livekit::room::track::TrackTrait; +use std::time::Duration; + +use livekit::room::RoomError; use livekit::room::{track::remote_track::RemoteTrackHandle, Room}; -use std::sync::{Arc, Mutex}; -use tracing::{event_enabled, info, trace}; +use tokio::time::sleep; const URL: &str = "ws://localhost:7880"; const TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0NzMsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NjQ4MDY0NzMsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.BgVdBnq3XFD3_BQHoe1azqjifYysubgFl6Qlzu9IQGI"; @@ -9,23 +10,27 @@ const TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0N // eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY3MzAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ3ZWIiLCJuYmYiOjE2NjQ4MDY3MzAsInN1YiI6IndlYiIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.VbDoULjX1CVGZu2sPy3SvWYlVZUBXxQVPmdB9BnmlN4 #[tokio::main] -async fn main() { +async fn main() -> Result<(), RoomError> { tracing_subscriber::fmt::init(); - let room = Room::new(); + let mut room = Room::new(); room.events() - .on_participant_connected(|event| async move {}); + .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 { - - - - - }) })) + rtc_track.set_should_receive(true); + rtc_track.on_frame(Box::new(|_frame, _buffer| { + // called on libwebrtc worker_thread + println!("Received frame"); + })); } }); + + room.connect(URL, TOKEN).await?; + + sleep(Duration::from_secs(200)).await; + Ok(()) }