diff --git a/Cargo.lock b/Cargo.lock index 544be83..5ebf964 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -662,6 +662,20 @@ dependencies = [ "url", ] +[[package]] +name = "livekit-ffi" +version = "0.1.1" +dependencies = [ + "lazy_static", + "livekit", + "parking_lot", + "prost", + "prost-build", + "prost-types", + "thiserror", + "tokio", +] + [[package]] name = "livekit-utils" version = "0.1.1" diff --git a/Cargo.toml b/Cargo.toml index 36cb2f8..891654b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,8 +1,9 @@ [workspace] members = [ "livekit", - "livekit-webrtc", "livekit-utils", + "livekit-ffi", + "livekit-webrtc", "webrtc-sys" ] diff --git a/examples/Cargo.lock b/examples/Cargo.lock index a9c5032..a59cab1 100644 --- a/examples/Cargo.lock +++ b/examples/Cargo.lock @@ -4,9 +4,9 @@ version = 3 [[package]] name = "ab_glyph" -version = "0.2.18" +version = "0.2.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4dcdbc68024b653943864d436fe8a24b028095bc1cf91a8926f8241e4aaffe59" +checksum = "e5568a4aa5ba8adf5175c5c460b030e27d8893412976cc37bef0e4fbc16cfbba" dependencies = [ "ab_glyph_rasterizer", "owned_ttf_parser", @@ -129,9 +129,9 @@ dependencies = [ [[package]] name = "atomic_refcell" -version = "0.1.8" +version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "73b5e5f48b927f04e952dedc932f31995a65a0bf65ec971c74436e51bf6e970d" +checksum = "857253367827bd9d0fd973f0ef15506a96e79e41b0ad7aa691203a4e3214f6c8" [[package]] name = "autocfg" @@ -227,9 +227,9 @@ checksum = "dfb24e866b15a1af2a1b663f10c6b6b8f397a84aadb828f12e5b289ec23a3a3c" [[package]] name = "bzip2" -version = "0.4.3" +version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6afcd980b5f3a45017c57e57a2fcccbb351cc43a356ce117ef760ef8052b89b0" +checksum = "bdb116a6ef3f6c3698828873ad02c3014b3c85cadb88496095628e3ef1e347f8" dependencies = [ "bzip2-sys", "libc", @@ -1714,9 +1714,9 @@ checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" [[package]] name = "owned_ttf_parser" -version = "0.17.1" +version = "0.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "18904d3c65493a9f0d7542293d1a7f69bfdc309a6b9ef4f46dc3e58b0577edc5" +checksum = "2a5f3c7ca08b6879e7965fb25e24d1f5eeb32ea73f9ad99b3854778a38c57e93" dependencies = [ "ttf-parser", ] @@ -1827,9 +1827,9 @@ checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de" [[package]] name = "prettyplease" -version = "0.1.22" +version = "0.1.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2c8992a85d8e93a28bdf76137db888d3874e3b230dee5ed8bebac4c9f7617773" +checksum = "e97e3215779627f01ee256d2fad52f3d95e8e1c11e9fc6fd08f7cd455d5d5c78" dependencies = [ "proc-macro2", "syn", @@ -2484,9 +2484,9 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c" [[package]] name = "tokio" -version = "1.23.0" +version = "1.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eab6d665857cc6ca78d6e80303a02cea7a7851e85dfbd77cbdc09bd129f1ef46" +checksum = "1d9f76183f91ecfb55e1d7d5602bd1d979e38a3a522fe900241cf195624d67ae" dependencies = [ "autocfg", "bytes", @@ -2606,9 +2606,9 @@ dependencies = [ [[package]] name = "ttf-parser" -version = "0.17.1" +version = "0.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "375812fa44dab6df41c195cd2f7fecb488f6c09fbaafb62807488cefab642bff" +checksum = "dbf554b6e535f9a160b2ed4ea83f99000f21cbc0a693df26e258eaf2c226a151" [[package]] name = "tungstenite" diff --git a/examples/simple_room/src/app.rs b/examples/simple_room/src/app.rs index 1f6b30f..0c62e73 100644 --- a/examples/simple_room/src/app.rs +++ b/examples/simple_room/src/app.rs @@ -362,7 +362,7 @@ impl App { )); ui.label(format!( "ParticipantCount: {:?}", - room.session().participants().read().len() + 1 + room.session().participants().len() + 1 )); } } @@ -411,7 +411,6 @@ impl App { let name = self.state.room.lock().as_ref().and_then(|room| { room.session() .participants() - .read() .get(participant_sid) .map(|p| p.name()) }); diff --git a/livekit-ffi/Cargo.toml b/livekit-ffi/Cargo.toml index 25686ce..3a1ee1d 100644 --- a/livekit-ffi/Cargo.toml +++ b/livekit-ffi/Cargo.toml @@ -9,3 +9,15 @@ repository = "https://github.com/livekit/client-sdk-rust" [dependencies] livekit = { path = "../livekit", version = "0.1.1" } tokio = { version = "1.0", features = ["full"] } +parking_lot = { version = "0.12.1", features = ["send_guard"] } +prost = "0.11.0" +prost-types = "0.11.1" +lazy_static = "1.4.0" +thiserror = "1.0.38" + +[build-dependencies] +prost-build = { version = "0.11.1" } + + +[lib] +crate-type = ["cdylib", "staticlib"] diff --git a/livekit-ffi/build.rs b/livekit-ffi/build.rs new file mode 100644 index 0000000..89dfc6e --- /dev/null +++ b/livekit-ffi/build.rs @@ -0,0 +1,7 @@ +use std::io::Result; + +fn main() -> Result<()> { + prost_build::compile_protos(&["protocol/ffi.proto"], &["protocol/"])?; + Ok(()) +} + diff --git a/livekit-ffi/protocol/ffi.proto b/livekit-ffi/protocol/ffi.proto new file mode 100644 index 0000000..360a121 --- /dev/null +++ b/livekit-ffi/protocol/ffi.proto @@ -0,0 +1,224 @@ +syntax = "proto3"; + +package livekit; +option csharp_namespace = "LiveKit.Proto"; + +/// IPC + +/// # Safety +/// The foreign language is responsable for disposing an handle +/// Forgetting to dispose the handle may lead to memory leaks +/// Messages bellow can contain an FFIHandle +message FFIHandleId { uint32 id = 1; } + +message FFIRequest { + oneof message { + InitializeRequest configure = 1; + ConnectRequest async_connect = 2; + DisconnectRequest async_disconnect = 3; + } +} + +message FFIResponse { + oneof message { + ConnectResponse async_connect = 1; + RoomEvent room_event = 2; + TrackEvent track_event = 3; + ParticipantEvent participant_event = 4; + } +} + +message InitializeRequest { uint64 callback_ptr = 1; } + +message ConnectRequest { + string url = 1; + string token = 2; + RoomOptions options = 3; +} + +message DisconnectRequest { string room_sid = 1; } + +message ConnectResponse { + bool success = 1; + optional RoomInfo room = 2; +} + +/// Models + +message RoomOptions { + bool auto_subscribe = 1; + bool adaptive_stream = 2; +} + +message RoomInfo { + string sid = 1; + string name = 2; + ParticipantInfo local_participant = 3; + repeated ParticipantInfo participants = 4; +} + +message ParticipantInfo { + string sid = 1; + string name = 2; + string identity = 3; + string metadata = 4; +} + +enum TrackKind { + KIND_UNKNOWN = 0; + KIND_AUDIO = 1; + KIND_VIDEO = 2; +} + +enum StreamState { + STATE_UNKNOWN = 0; + STATE_ACTIVE = 1; + STATE_PAUSED = 2; +} + +message TrackPublicationInfo { + string sid = 1; + string name = 2; + TrackKind kind = 3; +} + +message TrackInfo { + string sid = 1; + string name = 2; + TrackKind kind = 3; + StreamState state = 4; + bool muted = 5; +} + +/// Room Events + +message RoomEvent { + string room_sid = 1; + oneof message { + ParticipantConnected participant_connected = 2; + ParticipantDisconnected participant_disconnected = 3; + TrackPublished track_published = 4; + TrackUnpublished track_unpublished = 5; + TrackSubscribed track_subscribed = 6; + TrackUnsubscribed track_unsubscribed = 7; + } +} + +message ParticipantConnected { ParticipantInfo info = 1; } + +message ParticipantDisconnected { ParticipantInfo info = 1; } + +message TrackPublished { + TrackPublicationInfo publication = 1; + string participant_sid = 2; +} + +message TrackUnpublished { + TrackPublicationInfo publication = 1; + string participant_sid = 2; +} + +// Publication isn't needed for subscription events on the FFI +// The FFI will retrieve the publication using the Track sid +message TrackSubscribed { + // TrackPublicationInfo publication = 1; + TrackInfo track = 1; + string participant_sid = 2; +} + +message TrackUnsubscribed { + // TrackPublicationInfo publication = 1; + TrackInfo track = 1; + string participant_sid = 2; +} + +message DataReceived { + FFIHandleId handle = 1; + string participant_sid = 2; + uint64 data_ptr = 3; + uint64 data_size = 4; +} + +/// Track Events + +message TrackEvent { + string track_sid = 1; + oneof message { FrameReceived frame_received = 2; } +} + +message FrameReceived { + VideoFrame frame = 1; + VideoFrameBuffer frame_buffer = 2; +} + +message VideoFrame { + int32 width = 1; + int32 height = 2; + uint32 size = 3; + uint32 id = 4; // uint16 + int64 timestamp_us = 5; + int64 ntp_time_ms = 6; + uint32 transport_frame_id = 7; + uint32 timestamp = 8; + VideoRotation rotation = 9; +} + +message VideoFrameBuffer { + FFIHandleId handle = 1; + VideoFrameBufferType buffer_type = 2; + int32 width = 3; + int32 height = 4; + oneof buffer { + PlanarYuvBuffer yuv = 5; + BiplanarYuvBuffer bi_yuv = 6; + NativeBuffer native = 7; + } +} + +message PlanarYuvBuffer { + int32 chroma_width = 1; + int32 chroma_height = 2; + int32 stride_y = 3; + int32 stride_u = 4; + int32 stride_v = 5; + + // *const u8 or *const u16 + uint64 data_y_ptr = 6; + uint64 data_u_ptr = 7; + uint64 data_v_ptr = 8; +} + +message BiplanarYuvBuffer { + int32 chroma_width = 1; + int32 chroma_height = 2; + int32 stride_y = 3; + int32 stride_uv = 4; + + uint64 data_y_ptr = 5; + uint64 data_uv_ptr = 6; +} + +message NativeBuffer { + // TODO(theomonnom): Expose graphic context? +} + +enum VideoFrameBufferType { + NATIVE = 0; + I420 = 1; + I420A = 2; + I422 = 3; + I444 = 4; + I010 = 5; + NV12 = 6; +} + +enum VideoRotation { + VIDEO_ROTATION_0 = 0; + VIDEO_ROTATION_90 = 1; + VIDEO_ROTATION_180 = 2; + VIDEO_ROTATION_270 = 3; +} + +/// Participant Events + +message ParticipantEvent { string participant_sid = 1; } diff --git a/livekit-ffi/src/lib.rs b/livekit-ffi/src/lib.rs index 8b13789..243f07f 100644 --- a/livekit-ffi/src/lib.rs +++ b/livekit-ffi/src/lib.rs @@ -1 +1,5 @@ +mod proto { + include!(concat!(env!("OUT_DIR"), "/livekit.rs")); +} +mod server; diff --git a/livekit-ffi/src/server/conversion.rs b/livekit-ffi/src/server/conversion.rs new file mode 100644 index 0000000..2881e57 --- /dev/null +++ b/livekit-ffi/src/server/conversion.rs @@ -0,0 +1,270 @@ +use crate::{proto, server::FFIHandleId}; +use livekit::{ + prelude::*, + webrtc::video_frame_buffer::{ + BiplanarYuv8Buffer, BiplanarYuvBuffer, I010Buffer, I420ABuffer, I420Buffer, I422Buffer, + I444Buffer, NV12Buffer, PlanarYuv16BBuffer, PlanarYuv8Buffer, PlanarYuvBuffer, + }, +}; +use std::sync::Arc; + +impl From for proto::FfiHandleId { + fn from(id: FFIHandleId) -> Self { + Self { id: id as u32 } + } +} + +macro_rules! impl_participant_into { + ($p:ty) => { + impl From<$p> for proto::ParticipantInfo { + fn from(p: $p) -> Self { + Self { + name: p.name(), + sid: p.sid().to_string(), + identity: p.identity().to_string(), + metadata: p.metadata(), + } + } + } + }; +} + +impl_participant_into!(&Arc); +impl_participant_into!(&Arc); +impl_participant_into!(&Participant); + +macro_rules! impl_publication_into { + ($p:ty) => { + impl From<$p> for proto::TrackPublicationInfo { + fn from(p: $p) -> Self { + Self { + name: p.name(), + sid: p.sid().to_string(), + kind: proto::TrackKind::from(p.kind()).into(), + } + } + } + }; +} + +impl_publication_into!(&LocalTrackPublication); +impl_publication_into!(&RemoteTrackPublication); +impl_publication_into!(&TrackPublication); + +macro_rules! impl_track_into { + ($t:ty) => { + impl From<$t> for proto::TrackInfo { + fn from(track: $t) -> Self { + Self { + name: track.name(), + state: proto::StreamState::from(track.stream_state()).into(), + sid: track.sid().to_string(), + kind: proto::TrackKind::from(track.kind()).into(), + muted: track.muted(), + } + } + } + }; +} + +impl_track_into!(&LocalAudioTrack); +impl_track_into!(&LocalVideoTrack); +impl_track_into!(&RemoteAudioTrack); +impl_track_into!(&RemoteVideoTrack); +impl_track_into!(&TrackHandle); +impl_track_into!(&LocalTrackHandle); +impl_track_into!(&RemoteTrackHandle); + +impl From for proto::TrackKind { + fn from(kind: TrackKind) -> Self { + match kind { + TrackKind::Unknown => proto::TrackKind::KindUnknown, + TrackKind::Audio => proto::TrackKind::KindAudio, + TrackKind::Video => proto::TrackKind::KindVideo, + } + } +} + +impl From for proto::StreamState { + fn from(state: StreamState) -> Self { + match state { + StreamState::Unknown => Self::StateUnknown, + StreamState::Active => Self::StateActive, + StreamState::Paused => Self::StatePaused, + } + } +} + +impl proto::RoomEvent { + pub fn from(room_sid: impl Into, event: RoomEvent) -> Option { + let message = match event { + RoomEvent::ParticipantConnected(participant) => Some( + proto::room_event::Message::ParticipantConnected(proto::ParticipantConnected { + info: Some((&participant).into()), + }), + ), + RoomEvent::ParticipantDisconnected(participant) => { + Some(proto::room_event::Message::ParticipantDisconnected( + proto::ParticipantDisconnected { + info: Some((&participant).into()), + }, + )) + } + RoomEvent::TrackPublished { + publication, + participant, + } => Some(proto::room_event::Message::TrackPublished( + proto::TrackPublished { + participant_sid: participant.sid().to_string(), + publication: Some((&publication).into()), + }, + )), + RoomEvent::TrackUnpublished { + publication, + participant, + } => Some(proto::room_event::Message::TrackUnpublished( + proto::TrackUnpublished { + participant_sid: participant.sid().to_string(), + publication: Some((&publication).into()), + }, + )), + RoomEvent::TrackSubscribed { + track, + publication: _, + participant, + } => Some(proto::room_event::Message::TrackSubscribed( + proto::TrackSubscribed { + participant_sid: participant.sid().to_string(), + track: Some((&track).into()), + }, + )), + RoomEvent::TrackUnsubscribed { + track, + publication: _, + participant, + } => Some(proto::room_event::Message::TrackUnsubscribed( + proto::TrackUnsubscribed { + participant_sid: participant.sid().to_string(), + track: Some((&track).into()), + }, + )), + _ => None, + }; + + message.map(|message| proto::RoomEvent { + room_sid: room_sid.into(), + message: Some(message), + }) + } +} + +impl From for proto::VideoRotation { + fn from(rotation: VideoRotation) -> proto::VideoRotation { + match rotation { + VideoRotation::VideoRotation0 => Self::VideoRotation0, + VideoRotation::VideoRotation90 => Self::VideoRotation90, + VideoRotation::VideoRotation180 => Self::VideoRotation180, + VideoRotation::VideoRotation270 => Self::VideoRotation270, + } + } +} + +impl From for proto::VideoFrame { + fn from(frame: VideoFrame) -> Self { + Self { + width: frame.width(), + height: frame.height(), + size: frame.size(), + id: frame.id() as u32, + timestamp_us: frame.timestamp_us(), + ntp_time_ms: frame.ntp_time_ms(), + transport_frame_id: frame.transport_frame_id(), + timestamp: frame.timestamp(), + rotation: proto::VideoRotation::from(frame.rotation()).into(), + } + } +} + +impl From for proto::VideoFrameBufferType { + fn from(buffer_type: VideoFrameBufferType) -> Self { + match buffer_type { + VideoFrameBufferType::Native => Self::Native, + VideoFrameBufferType::I420 => Self::I420, + VideoFrameBufferType::I420A => Self::I420a, + VideoFrameBufferType::I422 => Self::I422, + VideoFrameBufferType::I444 => Self::I444, + VideoFrameBufferType::I010 => Self::I010, + VideoFrameBufferType::NV12 => Self::Nv12, + } + } +} + +macro_rules! impl_yuv_into { + ($b:ty) => { + impl From<$b> for proto::PlanarYuvBuffer { + fn from(buffer: $b) -> Self { + Self { + chroma_width: buffer.chroma_width(), + chroma_height: buffer.chroma_height(), + stride_y: buffer.stride_y(), + stride_u: buffer.stride_u(), + stride_v: buffer.stride_v(), + data_y_ptr: buffer.data_y().as_ptr() as u64, + data_u_ptr: buffer.data_u().as_ptr() as u64, + data_v_ptr: buffer.data_v().as_ptr() as u64, + } + } + } + }; +} + +impl_yuv_into!(&I420Buffer); +impl_yuv_into!(&I420ABuffer); +impl_yuv_into!(&I422Buffer); +impl_yuv_into!(&I444Buffer); +impl_yuv_into!(&I010Buffer); + +macro_rules! impl_biyuv_into { + ($b:ty) => { + impl From<$b> for proto::BiplanarYuvBuffer { + fn from(buffer: $b) -> Self { + Self { + chroma_width: buffer.chroma_width(), + chroma_height: buffer.chroma_height(), + stride_y: buffer.stride_y(), + stride_uv: buffer.stride_uv(), + data_y_ptr: buffer.data_y().as_ptr() as u64, + data_uv_ptr: buffer.data_uv().as_ptr() as u64, + } + } + } + }; +} + +impl_biyuv_into!(&NV12Buffer); + +impl proto::VideoFrameBuffer { + pub fn from(handle_id: FFIHandleId, buffer: &VideoFrameBuffer) -> Self { + Self { + handle: Some(handle_id.into()), + buffer_type: proto::VideoFrameBufferType::from(buffer.buffer_type()).into(), + width: buffer.width(), + height: buffer.height(), + buffer: Some(match &buffer { + VideoFrameBuffer::Native(_) => { + proto::video_frame_buffer::Buffer::Native(proto::NativeBuffer {}) + } + VideoFrameBuffer::I420(i420) => proto::video_frame_buffer::Buffer::Yuv(i420.into()), + VideoFrameBuffer::I420A(i420a) => { + proto::video_frame_buffer::Buffer::Yuv(i420a.into()) + } + VideoFrameBuffer::I422(i422) => proto::video_frame_buffer::Buffer::Yuv(i422.into()), + VideoFrameBuffer::I444(i444) => proto::video_frame_buffer::Buffer::Yuv(i444.into()), + VideoFrameBuffer::I010(i010) => proto::video_frame_buffer::Buffer::Yuv(i010.into()), + VideoFrameBuffer::NV12(nv12) => { + proto::video_frame_buffer::Buffer::BiYuv(nv12.into()) + } + }), + } + } +} diff --git a/livekit-ffi/src/server/mod.rs b/livekit-ffi/src/server/mod.rs new file mode 100644 index 0000000..82d5ea7 --- /dev/null +++ b/livekit-ffi/src/server/mod.rs @@ -0,0 +1,221 @@ +use crate::{ + proto, proto::ffi_request::Message as FFIRequest, proto::ffi_response::Message as FFIResponse, +}; +use lazy_static::lazy_static; +use livekit::prelude::*; +use livekit::webrtc::media_stream::OnFrameHandler; +use parking_lot::{Mutex, RwLock}; +use prost::Message; +use std::any::Any; +use std::collections::HashMap; +use std::panic; +use std::slice; +use std::sync::atomic::AtomicU32; +use std::sync::atomic::{AtomicBool, Ordering}; +use thiserror::Error; + +mod conversion; + +#[derive(Error, Debug)] +pub enum FFIError { + #[error("the FFIServer isn't configured")] + NotConfigured, + #[error("failed to execute the ffi callback")] + CallbackFailed, +} + +pub type FFIHandleId = u32; +pub type FFIHandle = Box; + +type CallbackFn = unsafe extern "C" fn(*const u8, usize); // This "C" callback must be threadsafe + +lazy_static! { + static ref FFI_SERVER: FFIServer = FFIServer::default(); +} + +pub struct FFIConfig { + callback_fn: CallbackFn, +} + +/// To use the FFI, the foreign language and the FFI server must share +/// the same memory space +pub struct FFIServer { + // Object owned by the foreign language + // The foreign language is responsible for freeing this memory + ffi_owned: RwLock>, + next_handle: AtomicU32, // FFIHandle + + rooms: RwLock>, + async_runtime: tokio::runtime::Runtime, + initialized: AtomicBool, + config: Mutex>, +} + +impl Default for FFIServer { + fn default() -> Self { + Self { + ffi_owned: RwLock::new(HashMap::new()), + next_handle: Default::default(), + rooms: RwLock::new(HashMap::new()), + async_runtime: tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap(), + initialized: Default::default(), + config: Default::default(), + } + } +} + +impl FFIServer { + pub fn next_handle_id(&self) -> FFIHandleId { + self.next_handle.fetch_add(1, Ordering::SeqCst) as FFIHandleId + } + + pub fn insert_handle(&self, handle_id: FFIHandleId, handle: FFIHandle) { + self.ffi_owned.write().insert(handle_id, handle); + } + + pub fn release_handle(&self, handle_id: FFIHandleId) -> Option { + self.ffi_owned.write().remove(&handle_id) + } + + pub fn send_response(&self, message: FFIResponse) -> Result<(), FFIError> { + if !self.initialized.load(Ordering::SeqCst) { + Err(FFIError::NotConfigured)? + } + + let message = proto::FfiResponse { + message: Some(message), + } + .encode_to_vec(); + + let callback_fn = self.config.lock().as_ref().unwrap().callback_fn; + if let Err(err) = panic::catch_unwind(|| unsafe { + callback_fn(message.as_ptr(), message.len()); + }) { + eprintln!("panic when sending ffi response: {:?}", err); + Err(FFIError::CallbackFailed)? + } + + Ok(()) + } + + pub fn on_request_received(&self, message: FFIRequest) -> Result<(), FFIError> { + if let FFIRequest::Configure(ref init) = message { + self.initialized.store(true, Ordering::SeqCst); + *self.config.lock() = Some(FFIConfig { + callback_fn: unsafe { std::mem::transmute(init.callback_ptr) }, + }); + } + + if !self.initialized.load(Ordering::SeqCst) { + Err(FFIError::NotConfigured)? + } + + match message { + proto::ffi_request::Message::AsyncConnect(connect) => { + self.async_runtime.spawn(room_task(connect)); + } + _ => {} + }; + + Ok(()) + } +} + +#[no_mangle] +pub extern "C" fn livekit_ffi_request(data: *const u8, len: usize) { + let data = unsafe { slice::from_raw_parts(data, len) }; + let request = proto::FfiRequest::decode(data).expect("Failed to decode the FFIRequest"); + let res = FFI_SERVER.on_request_received(request.message.unwrap()); + if let Err(err) = res { + eprintln!("failed to handle ffi request: {:?}", err); + } +} + +// Connect a listen to Room events +async fn room_task(connect: proto::ConnectRequest) { + let res = Room::connect(&connect.url, &connect.token).await; + + if res.is_err() { + let _ = FFI_SERVER.send_response(FFIResponse::AsyncConnect(proto::ConnectResponse { + success: false, + room: None, + })); + return; + } + + // Send connect response before listening to events + let (room, mut events) = res.unwrap(); + let session = room.session(); + + let _ = FFI_SERVER.send_response(FFIResponse::AsyncConnect(proto::ConnectResponse { + success: true, + room: Some(proto::RoomInfo { + sid: session.sid(), + name: session.name(), + local_participant: Some((&room.session().local_participant()).into()), + participants: room + .session() + .participants() + .iter() + .map(|(_, p)| p.into()) + .collect(), + }), + })); + + // Listen to events + tokio::spawn(participant_task(Participant::Local( + session.local_participant(), + ))); + + while let Some(event) = events.recv().await { + if let Some(event) = proto::RoomEvent::from(session.sid(), event.clone()) { + let _ = FFI_SERVER.send_response(FFIResponse::RoomEvent(event)); + } + + match event { + RoomEvent::ParticipantConnected(p) => { + tokio::spawn(participant_task(Participant::Remote(p))); + } + RoomEvent::TrackSubscribed { + track, + publication, + participant, + } => { + if let RemoteTrackHandle::Video(video_track) = track { + let rtc_track = video_track.rtc_track(); + rtc_track.on_frame(on_video_frame(video_track.sid())); + } + } + _ => {} + } + } +} + +// Listen to participant events +async fn participant_task(participant: Participant) { + let mut participant_events = participant.register_observer(); + while let Some(event) = participant_events.recv().await { + // TODO convert event to proto + } +} + +fn on_video_frame(track_sid: TrackSid) -> OnFrameHandler { + Box::new(move |frame, buffer| { + let handle_id = FFI_SERVER.next_handle_id(); + let proto_buffer = proto::VideoFrameBuffer::from(handle_id, &buffer); + FFI_SERVER.insert_handle(handle_id, Box::new(buffer)); + + let _ = FFI_SERVER.send_response(FFIResponse::TrackEvent(proto::TrackEvent { + track_sid: track_sid.to_string(), + message: Some(proto::track_event::Message::FrameReceived( + proto::FrameReceived { + frame: Some(frame.into()), + frame_buffer: Some(proto_buffer), + }, + )), + })); + }) +} diff --git a/livekit-webrtc/src/video_frame.rs b/livekit-webrtc/src/video_frame.rs index fc90299..47261fb 100644 --- a/livekit-webrtc/src/video_frame.rs +++ b/livekit-webrtc/src/video_frame.rs @@ -1,9 +1,26 @@ +use crate::video_frame_buffer::VideoFrameBuffer; use cxx::UniquePtr; use webrtc_sys::video_frame as vf_sys; -pub use vf_sys::ffi::VideoRotation; +#[derive(Debug)] +pub enum VideoRotation { + VideoRotation0 = 0, + VideoRotation90 = 90, + VideoRotation180 = 180, + VideoRotation270 = 270, +} -use crate::video_frame_buffer::VideoFrameBuffer; +impl From for VideoRotation { + fn from(rotation: vf_sys::ffi::VideoRotation) -> Self { + match rotation { + vf_sys::ffi::VideoRotation::VideoRotation0 => Self::VideoRotation0, + vf_sys::ffi::VideoRotation::VideoRotation90 => Self::VideoRotation90, + vf_sys::ffi::VideoRotation::VideoRotation180 => Self::VideoRotation180, + vf_sys::ffi::VideoRotation::VideoRotation270 => Self::VideoRotation270, + _ => unreachable!(), + } + } +} pub struct VideoFrame { cxx_handle: UniquePtr, @@ -47,7 +64,7 @@ impl VideoFrame { } pub fn rotation(&self) -> VideoRotation { - self.cxx_handle.rotation() + self.cxx_handle.rotation().into() } /// # Safety diff --git a/livekit-webrtc/src/video_frame_buffer.rs b/livekit-webrtc/src/video_frame_buffer.rs index 06f65d3..ecfc83d 100644 --- a/livekit-webrtc/src/video_frame_buffer.rs +++ b/livekit-webrtc/src/video_frame_buffer.rs @@ -2,10 +2,36 @@ use cxx::UniquePtr; use livekit_utils::enum_dispatch; use std::pin::Pin; use std::slice; -use vfb_sys::ffi::VideoFrameBufferType; use webrtc_sys::video_frame_buffer as vfb_sys; +#[derive(Debug)] +pub enum VideoFrameBufferType { + Native, + I420, + I420A, + I422, + I444, + I010, + NV12, +} + +impl From for VideoFrameBufferType { + fn from(buffer_type: vfb_sys::ffi::VideoFrameBufferType) -> Self { + match buffer_type { + vfb_sys::ffi::VideoFrameBufferType::Native => Self::Native, + vfb_sys::ffi::VideoFrameBufferType::I420 => Self::I420, + vfb_sys::ffi::VideoFrameBufferType::I420A => Self::I420A, + vfb_sys::ffi::VideoFrameBufferType::I422 => Self::I422, + vfb_sys::ffi::VideoFrameBufferType::I444 => Self::I444, + vfb_sys::ffi::VideoFrameBufferType::I010 => Self::I010, + vfb_sys::ffi::VideoFrameBufferType::NV12 => Self::NV12, + _ => unreachable!(), + } + } +} + pub trait VideoFrameBufferTrait { + fn buffer_type(&self) -> VideoFrameBufferType; // Useful for the FFI fn width(&self) -> i32; fn height(&self) -> i32; fn to_i420(self) -> I420Buffer; @@ -25,6 +51,24 @@ pub trait PlanarYuv8Buffer: PlanarYuvBuffer { fn data_v(&self) -> &[u8]; } +pub trait PlanarYuv16BBuffer: PlanarYuvBuffer { + fn data_y(&self) -> &[u16]; + fn data_u(&self) -> &[u16]; + fn data_v(&self) -> &[u16]; +} + +pub trait BiplanarYuvBuffer: VideoFrameBufferTrait { + fn chroma_width(&self) -> i32; + fn chroma_height(&self) -> i32; + fn stride_y(&self) -> i32; + fn stride_uv(&self) -> i32; +} + +pub trait BiplanarYuv8Buffer: BiplanarYuvBuffer { + fn data_y(&self) -> &[u8]; + fn data_uv(&self) -> &[u8]; +} + pub enum VideoFrameBuffer { Native(NativeBuffer), I420(I420Buffer), @@ -38,16 +82,26 @@ pub enum VideoFrameBuffer { impl VideoFrameBuffer { pub(crate) fn new(mut cxx_handle: UniquePtr) -> Self { unsafe { - match cxx_handle.buffer_type() { + match cxx_handle.buffer_type().into() { 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)), + VideoFrameBufferType::I420A => { + Self::I420A(I420ABuffer::new(cxx_handle.pin_mut().get_i420a())) + } + VideoFrameBufferType::I422 => { + Self::I422(I422Buffer::new(cxx_handle.pin_mut().get_i422())) + } + VideoFrameBufferType::I444 => { + Self::I444(I444Buffer::new(cxx_handle.pin_mut().get_i444())) + } + VideoFrameBufferType::I010 => { + Self::I010(I010Buffer::new(cxx_handle.pin_mut().get_i010())) + } + VideoFrameBufferType::NV12 => { + Self::NV12(NV12Buffer::new(cxx_handle.pin_mut().get_nv12())) + } _ => unreachable!(), // VideoFrameBufferType is represented as i32 } } @@ -57,6 +111,7 @@ impl VideoFrameBuffer { impl VideoFrameBufferTrait for VideoFrameBuffer { enum_dispatch!( [Native, I420, I420A, I422, I444, I010, NV12] + fnc!(buffer_type, &Self, [], VideoFrameBufferType); fnc!(width, &Self, [], i32); fnc!(height, &Self, [], i32); fnc!(to_i420, Self, [], I420Buffer); @@ -81,6 +136,13 @@ macro_rules! impl_video_frame_buffer { // Allow unused_unsafe when we don't do any cast ( e.g. NativeBuffer ) #[allow(unused_unsafe)] impl VideoFrameBufferTrait for $x { + fn buffer_type(&self) -> VideoFrameBufferType { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).buffer_type().into() + } + } + fn width(&self) -> i32 { let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); unsafe { @@ -179,6 +241,90 @@ macro_rules! impl_yuv8_buffer { }; } +macro_rules! impl_yuv16_buffer { + ($x:ty $(, $cast:ident)*) => { + impl PlanarYuv16BBuffer for $x { + fn data_y(&self) -> &[u16] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + slice::from_raw_parts((*ptr).data_y(), (self.width() * self.height()) as usize) + } + } + + fn data_u(&self) -> &[u16] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + let chroma_height = (self.height() + 1) / 2; + slice::from_raw_parts((*ptr).data_u(), (self.stride_u() * chroma_height) as usize) + } + } + + fn data_v(&self) -> &[u16] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + let chroma_height = (self.height() + 1) / 2; + slice::from_raw_parts((*ptr).data_v(), (self.stride_v() * chroma_height) as usize) + } + } + } + }; +} + +macro_rules! impl_biyuv_buffer { + ($x:ty $(, $cast:ident)*) => { + impl BiplanarYuvBuffer 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_uv(&self) -> i32 { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + (*ptr).stride_uv() + } + } + } + }; +} + +macro_rules! impl_biyuv8_buffer { + ($x:ty $(, $cast:ident)*) => { + impl BiplanarYuv8Buffer for $x { + fn data_y(&self) -> &[u8] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + slice::from_raw_parts((*ptr).data_y(), (self.width() * self.height()) as usize) + } + } + + fn data_uv(&self) -> &[u8] { + let ptr = recursive_cast!(&*self.cxx_handle $(, $cast)*); + unsafe { + let chroma_height = (self.height() + 1) / 2; + slice::from_raw_parts((*ptr).data_uv(), (self.stride_uv() * chroma_height) as usize) + } + } + } + }; +} + pub struct NativeBuffer { cxx_handle: UniquePtr, } @@ -188,36 +334,49 @@ pub struct I420Buffer { } pub struct I420ABuffer { - cxx_handle: UniquePtr, + cxx_handle: UniquePtr, } pub struct I422Buffer { - cxx_handle: UniquePtr, + cxx_handle: UniquePtr, } pub struct I444Buffer { - cxx_handle: UniquePtr, + cxx_handle: UniquePtr, } pub struct I010Buffer { - cxx_handle: UniquePtr, + cxx_handle: UniquePtr, } pub struct NV12Buffer { - cxx_handle: UniquePtr, + 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_video_frame_buffer!(I420ABuffer, i420a_to_yuv8, yuv8_to_yuv, yuv_to_vfb); +impl_video_frame_buffer!(I422Buffer, i422_to_yuv8, yuv8_to_yuv, yuv_to_vfb); +impl_video_frame_buffer!(I444Buffer, i444_to_yuv8, yuv8_to_yuv, yuv_to_vfb); +impl_video_frame_buffer!(I010Buffer, i010_to_yuv16b, yuv16b_to_yuv, yuv_to_vfb); +impl_video_frame_buffer!(NV12Buffer, nv12_to_biyuv8, biyuv8_to_biyuv, biyuv_to_vfb); impl_yuv_buffer!(I420Buffer, i420_to_yuv8, yuv8_to_yuv); +impl_yuv_buffer!(I420ABuffer, i420a_to_yuv8, yuv8_to_yuv); +impl_yuv_buffer!(I422Buffer, i422_to_yuv8, yuv8_to_yuv); +impl_yuv_buffer!(I444Buffer, i444_to_yuv8, yuv8_to_yuv); +impl_yuv_buffer!(I010Buffer, i010_to_yuv16b, yuv16b_to_yuv); impl_yuv8_buffer!(I420Buffer, i420_to_yuv8); +impl_yuv8_buffer!(I420ABuffer, i420a_to_yuv8); +impl_yuv8_buffer!(I422Buffer, i422_to_yuv8); +impl_yuv8_buffer!(I444Buffer, i444_to_yuv8); + +impl_yuv16_buffer!(I010Buffer, i010_to_yuv16b); + +impl_biyuv_buffer!(NV12Buffer, nv12_to_biyuv8, biyuv8_to_biyuv); + +impl_biyuv8_buffer!(NV12Buffer, nv12_to_biyuv8); impl NativeBuffer { fn new(cxx_handle: UniquePtr) -> Self { @@ -232,31 +391,31 @@ impl I420Buffer { } impl I420ABuffer { - fn new(cxx_handle: UniquePtr) -> Self { + fn new(cxx_handle: UniquePtr) -> Self { Self { cxx_handle } } } impl I422Buffer { - fn new(cxx_handle: UniquePtr) -> Self { + fn new(cxx_handle: UniquePtr) -> Self { Self { cxx_handle } } } impl I444Buffer { - fn new(cxx_handle: UniquePtr) -> Self { + fn new(cxx_handle: UniquePtr) -> Self { Self { cxx_handle } } } impl I010Buffer { - fn new(cxx_handle: UniquePtr) -> Self { + fn new(cxx_handle: UniquePtr) -> Self { Self { cxx_handle } } } impl NV12Buffer { - fn new(cxx_handle: UniquePtr) -> Self { + fn new(cxx_handle: UniquePtr) -> Self { Self { cxx_handle } } } diff --git a/livekit/src/prelude.rs b/livekit/src/prelude.rs index 6b5f2a2..4b2d8a0 100644 --- a/livekit/src/prelude.rs +++ b/livekit/src/prelude.rs @@ -23,4 +23,6 @@ pub use crate::webrtc::{ }, rtp_receiver::RtpReceiver, rtp_transceiver::RtpTransceiver, + video_frame::{VideoFrame, VideoRotation}, + video_frame_buffer::{VideoFrameBuffer, VideoFrameBufferTrait, VideoFrameBufferType}, }; diff --git a/livekit/src/room/id.rs b/livekit/src/room/id.rs index 9bd35eb..cb5c3e5 100644 --- a/livekit/src/room/id.rs +++ b/livekit/src/room/id.rs @@ -49,8 +49,12 @@ pub struct ParticipantIdentity(String); #[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] pub struct TrackSid(String); +#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] +pub struct RoomSid(String); + id_str! { ParticipantSid; ParticipantIdentity; TrackSid; + RoomSid; } diff --git a/livekit/src/room/room_session.rs b/livekit/src/room/room_session.rs index eba8db6..913fd9e 100644 --- a/livekit/src/room/room_session.rs +++ b/livekit/src/room/room_session.rs @@ -5,7 +5,7 @@ use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine}; use crate::signal_client::SignalOptions; use crate::{RoomError, RoomEvent, RoomResult, SimulateScenario}; use livekit_utils::observer::Dispatcher; -use parking_lot::{Mutex, RwLock}; +use parking_lot::{Mutex, RwLock, RwLockReadGuard}; use std::collections::HashMap; use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::Arc; @@ -149,8 +149,8 @@ impl RoomSession { self.inner.state.load(Ordering::Acquire).try_into().unwrap() } - pub fn participants(&self) -> &RwLock>> { - &self.inner.participants + pub fn participants(&self) -> RwLockReadGuard>> { + self.inner.participants.read() } pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { diff --git a/livekit/src/room/track/audio_track.rs b/livekit/src/room/track/audio_track.rs index 3ccd183..12960ff 100644 --- a/livekit/src/room/track/audio_track.rs +++ b/livekit/src/room/track/audio_track.rs @@ -1,3 +1,4 @@ +use super::impl_track_trait; use crate::prelude::*; use std::sync::Arc; @@ -27,3 +28,5 @@ impl TryFrom for AudioTrackHandle { } } } + +impl_track_trait!(AudioTrackHandle, enum_dispatch, [Local, Remote]); diff --git a/livekit/src/room/track/local_track.rs b/livekit/src/room/track/local_track.rs index 0da4b5e..82ac845 100644 --- a/livekit/src/room/track/local_track.rs +++ b/livekit/src/room/track/local_track.rs @@ -1,3 +1,4 @@ +use super::impl_track_trait; use crate::prelude::*; use std::sync::Arc; @@ -27,3 +28,5 @@ impl TryFrom for LocalTrackHandle { } } } + +impl_track_trait!(LocalTrackHandle, enum_dispatch, [Audio, Video]); diff --git a/livekit/src/room/track/mod.rs b/livekit/src/room/track/mod.rs index 89f7b64..f224ff0 100644 --- a/livekit/src/room/track/mod.rs +++ b/livekit/src/room/track/mod.rs @@ -117,6 +117,7 @@ pub trait TrackTrait { fn name(&self) -> String; fn kind(&self) -> TrackKind; fn stream_state(&self) -> StreamState; + fn muted(&self) -> bool; fn start(&self); fn stop(&self); fn register_observer(&self) -> mpsc::UnboundedReceiver; @@ -201,6 +202,7 @@ impl TrackTrait for TrackHandle { fnc!(name, &Self, [], String); fnc!(kind, &Self, [], TrackKind); fnc!(stream_state, &Self, [], StreamState); + fnc!(muted, &Self, [], bool); fnc!(start, &Self, [], ()); fnc!(stop, &Self, [], ()); fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver); @@ -246,6 +248,10 @@ macro_rules! impl_track_trait { self.shared.stream_state.load(Ordering::SeqCst).into() } + fn muted(&self) -> bool { + self.shared.muted.load(Ordering::SeqCst) + } + fn start(&self) { self.shared.start(); } @@ -263,6 +269,25 @@ macro_rules! impl_track_trait { } } }; + ($x:ident, enum_dispatch, [$($variant:ident),+]) => { + use livekit_utils::enum_dispatch; + use tokio::sync::mpsc; + + impl TrackTrait for $x { + enum_dispatch!( + [$($variant),+] + fnc!(sid, &Self, [], TrackSid); + fnc!(name, &Self, [], String); + fnc!(kind, &Self, [], TrackKind); + fnc!(stream_state, &Self, [], StreamState); + fnc!(muted, &Self, [], bool); + fnc!(start, &Self, [], ()); + fnc!(stop, &Self, [], ()); + fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver); + fnc!(set_muted, &Self, [muted: bool], ()); + ); + } + }; } pub(super) use impl_track_trait; diff --git a/livekit/src/room/track/remote_track.rs b/livekit/src/room/track/remote_track.rs index d65ec15..a42a57e 100644 --- a/livekit/src/room/track/remote_track.rs +++ b/livekit/src/room/track/remote_track.rs @@ -1,7 +1,6 @@ +use super::impl_track_trait; use crate::prelude::*; -use livekit_utils::enum_dispatch; use std::sync::Arc; -use tokio::sync::mpsc; #[derive(Clone, Debug)] pub enum RemoteTrackHandle { @@ -9,20 +8,6 @@ pub enum RemoteTrackHandle { Video(Arc), } -impl TrackTrait for RemoteTrackHandle { - enum_dispatch!( - [Audio, Video] - fnc!(sid, &Self, [], TrackSid); - fnc!(name, &Self, [], String); - fnc!(kind, &Self, [], TrackKind); - fnc!(stream_state, &Self, [], StreamState); - fnc!(start, &Self, [], ()); - fnc!(stop, &Self, [], ()); - fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver); - fnc!(set_muted, &Self, [muted: bool], ()); - ); -} - impl From for TrackHandle { fn from(remote_track: RemoteTrackHandle) -> Self { match remote_track { @@ -43,3 +28,5 @@ impl TryFrom for RemoteTrackHandle { } } } + +impl_track_trait!(RemoteTrackHandle, enum_dispatch, [Audio, Video]); diff --git a/livekit/src/room/track/video_track.rs b/livekit/src/room/track/video_track.rs index 230582a..beca465 100644 --- a/livekit/src/room/track/video_track.rs +++ b/livekit/src/room/track/video_track.rs @@ -1,3 +1,4 @@ +use super::impl_track_trait; use crate::prelude::*; use std::sync::Arc; @@ -27,3 +28,5 @@ impl TryFrom for VideoTrackHandle { } } } + +impl_track_trait!(VideoTrackHandle, enum_dispatch, [Local, Remote]); diff --git a/webrtc-sys/include/livekit/video_frame_buffer.h b/webrtc-sys/include/livekit/video_frame_buffer.h index 1e3cfe2..723ec7e 100644 --- a/webrtc-sys/include/livekit/video_frame_buffer.h +++ b/webrtc-sys/include/livekit/video_frame_buffer.h @@ -12,9 +12,12 @@ namespace livekit { -class PlanarYuvBuffer; -class PlanarYuv8Buffer; class I420Buffer; +class I420ABuffer; +class I422Buffer; +class I444Buffer; +class I010Buffer; +class NV12Buffer; class VideoFrameBuffer { public: @@ -32,13 +35,43 @@ class VideoFrameBuffer { return std::make_unique(buffer_->ToI420()); } + // const_cast is valid here because we take the ownership on the rust side 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()))); } + std::unique_ptr get_i420a() { + return std::make_unique( + rtc::scoped_refptr( + const_cast(buffer_->GetI420A()))); + } + + std::unique_ptr get_i422() { + return std::make_unique( + rtc::scoped_refptr( + const_cast(buffer_->GetI422()))); + } + + std::unique_ptr get_i444() { + return std::make_unique( + rtc::scoped_refptr( + const_cast(buffer_->GetI444()))); + } + + std::unique_ptr get_i010() { + return std::make_unique( + rtc::scoped_refptr( + const_cast(buffer_->GetI010()))); + } + + std::unique_ptr get_nv12() { + return std::make_unique( + rtc::scoped_refptr( + const_cast(buffer_->GetNV12()))); + } + protected: rtc::scoped_refptr buffer_; }; @@ -76,24 +109,136 @@ class PlanarYuv8Buffer : public PlanarYuvBuffer { } }; +class PlanarYuv16BBuffer : public PlanarYuvBuffer { + public: + explicit PlanarYuv16BBuffer( + rtc::scoped_refptr buffer) + : PlanarYuvBuffer(buffer) {} + + const uint16_t* data_y() const { return buffer()->DataY(); } + const uint16_t* data_u() const { return buffer()->DataU(); } + const uint16_t* data_v() const { return buffer()->DataV(); } + + private: + webrtc::PlanarYuv16BBuffer* buffer() const { + return static_cast(buffer_.get()); + } +}; + +class BiplanarYuvBuffer : public VideoFrameBuffer { + public: + explicit BiplanarYuvBuffer( + 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_uv() const { return buffer()->StrideUV(); } + + private: + webrtc::BiplanarYuvBuffer* buffer() const { + return static_cast(buffer_.get()); + } +}; + +class BiplanarYuv8Buffer : public BiplanarYuvBuffer { + public: + explicit BiplanarYuv8Buffer( + rtc::scoped_refptr buffer) + : BiplanarYuvBuffer(buffer) {} + + const uint8_t* data_y() const { return buffer()->DataY(); } + const uint8_t* data_uv() const { return buffer()->DataUV(); } + + private: + webrtc::BiplanarYuv8Buffer* buffer() const { + return static_cast(buffer_.get()); + } +}; + class I420Buffer : public PlanarYuv8Buffer { public: explicit I420Buffer(rtc::scoped_refptr buffer) : PlanarYuv8Buffer(buffer) {} }; +class I420ABuffer : public I420Buffer { + public: + explicit I420ABuffer(rtc::scoped_refptr buffer) + : I420Buffer(buffer) {} +}; + +class I422Buffer : public PlanarYuv8Buffer { + public: + explicit I422Buffer(rtc::scoped_refptr buffer) + : PlanarYuv8Buffer(buffer) {} +}; + +class I444Buffer : public PlanarYuv8Buffer { + public: + explicit I444Buffer(rtc::scoped_refptr buffer) + : PlanarYuv8Buffer(buffer) {} +}; + +class I010Buffer : public PlanarYuv16BBuffer { + public: + explicit I010Buffer(rtc::scoped_refptr buffer) + : PlanarYuv16BBuffer(buffer) {} +}; + +class NV12Buffer : public BiplanarYuv8Buffer { + public: + explicit NV12Buffer(rtc::scoped_refptr buffer) + : BiplanarYuv8Buffer(buffer) {} +}; + static const VideoFrameBuffer* yuv_to_vfb(const PlanarYuvBuffer* yuv) { return yuv; } +static const VideoFrameBuffer* biyuv_to_vfb(const BiplanarYuvBuffer* biyuv) { + return biyuv; +} + static const PlanarYuvBuffer* yuv8_to_yuv(const PlanarYuv8Buffer* yuv8) { return yuv8; } +static const PlanarYuvBuffer* yuv16b_to_yuv(const PlanarYuv16BBuffer* yuv16) { + return yuv16; +} + +static const BiplanarYuvBuffer* biyuv8_to_biyuv( + const BiplanarYuv8Buffer* biyuv8) { + return biyuv8; +} + static const PlanarYuv8Buffer* i420_to_yuv8(const I420Buffer* i420) { return i420; } +static const PlanarYuv8Buffer* i420a_to_yuv8(const I420ABuffer* i420a) { + return i420a; +} + +static const PlanarYuv8Buffer* i422_to_yuv8(const I422Buffer* i422) { + return i422; +} + +static const PlanarYuv8Buffer* i444_to_yuv8(const I444Buffer* i444) { + return i444; +} + +static const PlanarYuv16BBuffer* i010_to_yuv16b(const I010Buffer* i010) { + return i010; +} + +static const BiplanarYuv8Buffer* nv12_to_biyuv8(const NV12Buffer* nv12) { + return nv12; +} + static std::unique_ptr _unique_video_frame_buffer() { return nullptr; // Ignore } diff --git a/webrtc-sys/src/lib.rs b/webrtc-sys/src/lib.rs index a2abebd..0fe1736 100644 --- a/webrtc-sys/src/lib.rs +++ b/webrtc-sys/src/lib.rs @@ -15,3 +15,16 @@ pub mod yuv_helper; pub const MEDIA_TYPE_VIDEO: &str = "video"; pub const MEDIA_TYPE_AUDIO: &str = "audio"; pub const MEDIA_TYPE_DATA: &str = "data"; + +macro_rules! impl_thread_safety { + ($obj:ty, Send) => { + unsafe impl Send for $obj {} + }; + + ($obj:ty, Send + Sync) => { + unsafe impl Send for $obj {} + unsafe impl Sync for $obj {} + }; +} + +pub(crate) use impl_thread_safety; diff --git a/webrtc-sys/src/video_frame.rs b/webrtc-sys/src/video_frame.rs index def1933..3247b82 100644 --- a/webrtc-sys/src/video_frame.rs +++ b/webrtc-sys/src/video_frame.rs @@ -1,3 +1,5 @@ +use crate::impl_thread_safety; + #[cxx::bridge(namespace = "livekit")] pub mod ffi { #[derive(Debug)] @@ -30,3 +32,5 @@ pub mod ffi { fn _unique_video_frame() -> UniquePtr; // Ignore } } + +impl_thread_safety!(ffi::VideoFrame, Send + Sync); diff --git a/webrtc-sys/src/video_frame_buffer.rs b/webrtc-sys/src/video_frame_buffer.rs index ecfa4d2..420a63c 100644 --- a/webrtc-sys/src/video_frame_buffer.rs +++ b/webrtc-sys/src/video_frame_buffer.rs @@ -1,3 +1,5 @@ +use crate::impl_thread_safety; + #[cxx::bridge(namespace = "livekit")] pub mod ffi { #[derive(Debug)] @@ -18,16 +20,29 @@ pub mod ffi { type VideoFrameBuffer; type PlanarYuvBuffer; type PlanarYuv8Buffer; + type PlanarYuv16BBuffer; + type BiplanarYuvBuffer; + type BiplanarYuv8Buffer; type I420Buffer; + type I420ABuffer; + type I422Buffer; + type I444Buffer; + type I010Buffer; + type NV12Buffer; fn buffer_type(self: &VideoFrameBuffer) -> VideoFrameBufferType; fn width(self: &VideoFrameBuffer) -> i32; fn height(self: &VideoFrameBuffer) -> i32; - // Require ownership + /// # SAFETY + /// The functions 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_* + unsafe fn get_i420a(self: Pin<&mut VideoFrameBuffer>) -> UniquePtr; + unsafe fn get_i422(self: Pin<&mut VideoFrameBuffer>) -> UniquePtr; + unsafe fn get_i444(self: Pin<&mut VideoFrameBuffer>) -> UniquePtr; + unsafe fn get_i010(self: Pin<&mut VideoFrameBuffer>) -> UniquePtr; + unsafe fn get_nv12(self: Pin<&mut VideoFrameBuffer>) -> UniquePtr; fn chroma_width(self: &PlanarYuvBuffer) -> i32; fn chroma_height(self: &PlanarYuvBuffer) -> i32; @@ -39,10 +54,43 @@ pub mod ffi { fn data_u(self: &PlanarYuv8Buffer) -> *const u8; fn data_v(self: &PlanarYuv8Buffer) -> *const u8; + fn data_y(self: &PlanarYuv16BBuffer) -> *const u16; + fn data_u(self: &PlanarYuv16BBuffer) -> *const u16; + fn data_v(self: &PlanarYuv16BBuffer) -> *const u16; + + fn chroma_width(self: &BiplanarYuvBuffer) -> i32; + fn chroma_height(self: &BiplanarYuvBuffer) -> i32; + fn stride_y(self: &BiplanarYuvBuffer) -> i32; + fn stride_uv(self: &BiplanarYuvBuffer) -> i32; + + fn data_y(self: &BiplanarYuv8Buffer) -> *const u8; + fn data_uv(self: &BiplanarYuv8Buffer) -> *const u8; + unsafe fn yuv_to_vfb(yuv: *const PlanarYuvBuffer) -> *const VideoFrameBuffer; + unsafe fn biyuv_to_vfb(yuv: *const BiplanarYuvBuffer) -> *const VideoFrameBuffer; unsafe fn yuv8_to_yuv(yuv8: *const PlanarYuv8Buffer) -> *const PlanarYuvBuffer; + unsafe fn yuv16b_to_yuv(yuv16b: *const PlanarYuv16BBuffer) -> *const PlanarYuvBuffer; + unsafe fn biyuv8_to_biyuv(biyuv8: *const BiplanarYuv8Buffer) -> *const BiplanarYuvBuffer; unsafe fn i420_to_yuv8(i420: *const I420Buffer) -> *const PlanarYuv8Buffer; + unsafe fn i420a_to_yuv8(i420a: *const I420ABuffer) -> *const PlanarYuv8Buffer; + unsafe fn i422_to_yuv8(i422: *const I422Buffer) -> *const PlanarYuv8Buffer; + unsafe fn i444_to_yuv8(i444: *const I444Buffer) -> *const PlanarYuv8Buffer; + unsafe fn i010_to_yuv16b(i010: *const I010Buffer) -> *const PlanarYuv16BBuffer; + unsafe fn nv12_to_biyuv8(nv12: *const NV12Buffer) -> *const BiplanarYuv8Buffer; fn _unique_video_frame_buffer() -> UniquePtr; } } + +impl_thread_safety!(ffi::VideoFrameBuffer, Send + Sync); +impl_thread_safety!(ffi::PlanarYuvBuffer, Send + Sync); +impl_thread_safety!(ffi::PlanarYuv8Buffer, Send + Sync); +impl_thread_safety!(ffi::PlanarYuv16BBuffer, Send + Sync); +impl_thread_safety!(ffi::BiplanarYuvBuffer, Send + Sync); +impl_thread_safety!(ffi::BiplanarYuv8Buffer, Send + Sync); +impl_thread_safety!(ffi::I420Buffer, Send + Sync); +impl_thread_safety!(ffi::I420ABuffer, Send + Sync); +impl_thread_safety!(ffi::I422Buffer, Send + Sync); +impl_thread_safety!(ffi::I444Buffer, Send + Sync); +impl_thread_safety!(ffi::I010Buffer, Send + Sync); +impl_thread_safety!(ffi::NV12Buffer, Send + Sync);