diff --git a/livekit-ffi/protocol/ffi.proto b/livekit-ffi/protocol/ffi.proto index 360a121..c8c4381 100644 --- a/livekit-ffi/protocol/ffi.proto +++ b/livekit-ffi/protocol/ffi.proto @@ -8,27 +8,50 @@ option csharp_namespace = "LiveKit.Proto"; /// # 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; } +/// Messages in this file can contain an FFIHandle +message FFIHandleId { uint64 id = 1; } +/// This is the input of livekit_ffi_request function message FFIRequest { oneof message { - InitializeRequest configure = 1; - ConnectRequest async_connect = 2; - DisconnectRequest async_disconnect = 3; + InitializeRequest initialize = 1; + // Stop all rooms synchronously (Do we need async here?). + // e.g: This is used for the Unity Editor after each assemblies reload. + DisposeRequest dispose = 2; + ConnectRequest async_connect = 3; + DisconnectRequest async_disconnect = 4; + ToI420Request to_i420 = 5; + ToARGBRequest to_argb = 6; } } +/// This is the output of livekit_ffi_request function. +/// The message field is mostly used to send result of a synchronous operation +/// to the foreign language. message FFIResponse { + optional uint64 async_id = 1; + oneof message { ToI420Response to_i420 = 2; } +} + +/// This message is used to receive the result of asynchronous requests +/// Or to receive events of a Room +message FFIEvent { + // Used if the message is used to send the async result to the foreign + // language + optional uint64 async_id = 1; oneof message { - ConnectResponse async_connect = 1; - RoomEvent room_event = 2; - TrackEvent track_event = 3; - ParticipantEvent participant_event = 4; + ConnectEvent connect_event = 2; + RoomEvent room_event = 3; + TrackEvent track_event = 4; + ParticipantEvent participant_event = 5; } } -message InitializeRequest { uint64 callback_ptr = 1; } +// Setup the callback where the foreign language can receive events +// and responses to asynchronous requests +message InitializeRequest { uint64 event_callback_ptr = 1; } + +message DisposeRequest {} message ConnectRequest { string url = 1; @@ -38,13 +61,34 @@ message ConnectRequest { message DisconnectRequest { string room_sid = 1; } -message ConnectResponse { +/// Convert a VideoFrameBuffer to a I420Buffer +message ToI420Request { + FFIHandleId buffer = 1; // NOTE: This buffer will be dropped! +} + +message ToARGBRequest { + FFIHandleId buffer = 1; + uint64 dst_ptr = 2; + VideoFormatType dst_format = 3; + int32 dst_stride = 4; + int32 dst_width = 5; + int32 dst_height = 6; +} + +message ConnectEvent { bool success = 1; optional RoomInfo room = 2; } +message ToI420Response { optional VideoFrameBufferInfo new_buffer = 1; } + /// Models +message Dimension { + uint32 width = 1; + uint32 height = 2; +} + message RoomOptions { bool auto_subscribe = 1; bool adaptive_stream = 2; @@ -53,8 +97,9 @@ message RoomOptions { message RoomInfo { string sid = 1; string name = 2; - ParticipantInfo local_participant = 3; - repeated ParticipantInfo participants = 4; + string metadata = 3; + ParticipantInfo local_participant = 4; + repeated ParticipantInfo participants = 5; } message ParticipantInfo { @@ -62,6 +107,31 @@ message ParticipantInfo { string name = 2; string identity = 3; string metadata = 4; + repeated TrackPublicationInfo publications = 5; +} + +message TrackPublicationInfo { + string sid = 1; + string name = 2; + TrackKind kind = 3; + TrackSource source = 4; + bool simulcasted = 5; + Dimension dimension = 6; + string mime_type = 7; + bool muted = 8; +} + +message TrackInfo { + string sid = 1; + string name = 2; + TrackKind kind = 3; + StreamState stream_state = 4; + bool muted = 5; +} + +message VideoSinkInfo { + string track_sid = 1; + // More info? } enum TrackKind { @@ -70,24 +140,50 @@ enum TrackKind { KIND_VIDEO = 2; } +enum TrackSource { + SOURCE_UNKNOWN = 0; + SOURCE_CAMERA = 1; + SOURCE_MICROPHONE = 2; + SOURCE_SCREENSHARE = 3; + SOURCE_SCREENSHARE_AUDIO = 4; +} + +enum ConnectionQuality { + QUALITY_POOR = 0; + QUALITY_GOOD = 1; + QUALITY_EXCELLENT = 2; +} + +enum ConnectionState { + CONN_DISCONNECTED = 0; + CONN_CONNECTED = 1; + CONN_RECONNECTING = 2; + CONN_UNKNOWN = 3; +} + enum StreamState { STATE_UNKNOWN = 0; STATE_ACTIVE = 1; STATE_PAUSED = 2; } -message TrackPublicationInfo { - string sid = 1; - string name = 2; - TrackKind kind = 3; +enum VideoRotation { + VIDEO_ROTATION_0 = 0; + VIDEO_ROTATION_90 = 1; + VIDEO_ROTATION_180 = 2; + VIDEO_ROTATION_270 = 3; } -message TrackInfo { - string sid = 1; - string name = 2; - TrackKind kind = 3; - StreamState state = 4; - bool muted = 5; +enum DataPacketKind { + KIND_UNRELIABLE = 0; + KIND_RELIABLE = 1; +} + +enum VideoFormatType { + FORMAT_ARGB = 0; + FORMAT_BGRA = 1; + FORMAT_ABGR = 2; + FORMAT_RGBA = 3; } /// Room Events @@ -101,6 +197,16 @@ message RoomEvent { TrackUnpublished track_unpublished = 5; TrackSubscribed track_subscribed = 6; TrackUnsubscribed track_unsubscribed = 7; + TrackMuted track_muted = 8; + TrackUnmuted track_unmuted = 9; + ActiveSpeakersChanged speakers_changed = 10; + ConnectionQualityChanged connection_quality_changed = 11; + DataReceived data_received = 12; + ConnectionStateChanged connection_state_changed = 13; + Connected connected = 14; + Disconnected disconnected = 15; + Reconnecting reconnecting = 16; + Reconnected reconnected = 17; } } @@ -109,36 +215,61 @@ message ParticipantConnected { ParticipantInfo info = 1; } message ParticipantDisconnected { ParticipantInfo info = 1; } message TrackPublished { - TrackPublicationInfo publication = 1; - string participant_sid = 2; + string participant_sid = 1; + TrackPublicationInfo publication = 2; } message TrackUnpublished { - TrackPublicationInfo publication = 1; - string participant_sid = 2; + string participant_sid = 1; + string publication_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; + string participant_sid = 1; + TrackInfo track = 2; + VideoSinkInfo sink = 3; } message TrackUnsubscribed { - // TrackPublicationInfo publication = 1; - TrackInfo track = 1; - string participant_sid = 2; + // The FFI language can dispose/remove the VideoSink here + string participant_sid = 1; + string track_sid = 2; +} + +message TrackMuted { + string participant_sid = 1; + string track_sid = 2; +} + +message TrackUnmuted { + string participant_sid = 1; + string track_sid = 2; +} + +message ActiveSpeakersChanged { repeated string participant_sids = 1; } + +message ConnectionQualityChanged { + string participant_sid = 1; + ConnectionQuality quality = 2; } message DataReceived { FFIHandleId handle = 1; - string participant_sid = 2; + optional string participant_sid = 2; uint64 data_ptr = 3; uint64 data_size = 4; + DataPacketKind kind = 5; } +message ConnectionStateChanged { ConnectionState state = 1; } + +message Connected {} +message Disconnected {} +message Reconnecting {} +message Reconnected {} + /// Track Events message TrackEvent { @@ -147,11 +278,11 @@ message TrackEvent { } message FrameReceived { - VideoFrame frame = 1; - VideoFrameBuffer frame_buffer = 2; + VideoFrameInfo frame = 1; + VideoFrameBufferInfo frame_buffer = 2; } -message VideoFrame { +message VideoFrameInfo { int32 width = 1; int32 height = 2; uint32 size = 3; @@ -163,19 +294,19 @@ message VideoFrame { VideoRotation rotation = 9; } -message VideoFrameBuffer { +message VideoFrameBufferInfo { 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; + PlanarYuvBufferInfo yuv = 5; + BiplanarYuvBufferInfo bi_yuv = 6; + NativeBufferInfo native = 7; } } -message PlanarYuvBuffer { +message PlanarYuvBufferInfo { int32 chroma_width = 1; int32 chroma_height = 2; int32 stride_y = 3; @@ -188,7 +319,7 @@ message PlanarYuvBuffer { uint64 data_v_ptr = 8; } -message BiplanarYuvBuffer { +message BiplanarYuvBufferInfo { int32 chroma_width = 1; int32 chroma_height = 2; int32 stride_y = 3; @@ -198,7 +329,7 @@ message BiplanarYuvBuffer { uint64 data_uv_ptr = 6; } -message NativeBuffer { +message NativeBufferInfo { // TODO(theomonnom): Expose graphic context? } @@ -212,13 +343,11 @@ enum VideoFrameBufferType { 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; } +message ParticipantEvent { + string participant_sid = 1; + oneof message { IsSpeakingChanged speaking_changed = 2; } +} + +message IsSpeakingChanged { bool speaking = 1; } diff --git a/livekit-ffi/src/server/conversion.rs b/livekit-ffi/src/server/conversion.rs index 2881e57..7257024 100644 --- a/livekit-ffi/src/server/conversion.rs +++ b/livekit-ffi/src/server/conversion.rs @@ -10,7 +10,7 @@ use std::sync::Arc; impl From for proto::FfiHandleId { fn from(id: FFIHandleId) -> Self { - Self { id: id as u32 } + Self { id: id as u64 } } } @@ -23,6 +23,7 @@ macro_rules! impl_participant_into { sid: p.sid().to_string(), identity: p.identity().to_string(), metadata: p.metadata(), + publications: p.tracks().iter().map(|(_, p)| p.into()).collect(), } } } @@ -33,6 +34,18 @@ impl_participant_into!(&Arc); impl_participant_into!(&Arc); impl_participant_into!(&Participant); +impl From for proto::TrackSource { + fn from(source: TrackSource) -> proto::TrackSource { + match source { + TrackSource::Unknown => proto::TrackSource::SourceUnknown, + TrackSource::Camera => proto::TrackSource::SourceCamera, + TrackSource::Microphone => proto::TrackSource::SourceMicrophone, + TrackSource::Screenshare => proto::TrackSource::SourceScreenshare, + TrackSource::ScreenshareAudio => proto::TrackSource::SourceScreenshareAudio, + } + } +} + macro_rules! impl_publication_into { ($p:ty) => { impl From<$p> for proto::TrackPublicationInfo { @@ -41,6 +54,14 @@ macro_rules! impl_publication_into { name: p.name(), sid: p.sid().to_string(), kind: proto::TrackKind::from(p.kind()).into(), + source: proto::TrackSource::from(p.source()).into(), + dimension: Some(proto::Dimension { + width: p.dimension().0, + height: p.dimension().1, + }), + mime_type: p.mime_type(), + simulcasted: p.simulcasted(), + muted: p.muted(), } } } @@ -57,7 +78,7 @@ macro_rules! impl_track_into { fn from(track: $t) -> Self { Self { name: track.name(), - state: proto::StreamState::from(track.stream_state()).into(), + stream_state: proto::StreamState::from(track.stream_state()).into(), sid: track.sid().to_string(), kind: proto::TrackKind::from(track.kind()).into(), muted: track.muted(), @@ -125,7 +146,7 @@ impl proto::RoomEvent { } => Some(proto::room_event::Message::TrackUnpublished( proto::TrackUnpublished { participant_sid: participant.sid().to_string(), - publication: Some((&publication).into()), + publication_sid: publication.sid().into(), }, )), RoomEvent::TrackSubscribed { @@ -136,6 +157,9 @@ impl proto::RoomEvent { proto::TrackSubscribed { participant_sid: participant.sid().to_string(), track: Some((&track).into()), + sink: Some(proto::VideoSinkInfo { + track_sid: track.sid().to_string(), + }), }, )), RoomEvent::TrackUnsubscribed { @@ -145,7 +169,7 @@ impl proto::RoomEvent { } => Some(proto::room_event::Message::TrackUnsubscribed( proto::TrackUnsubscribed { participant_sid: participant.sid().to_string(), - track: Some((&track).into()), + track_sid: track.sid().to_string(), }, )), _ => None, @@ -169,7 +193,7 @@ impl From for proto::VideoRotation { } } -impl From for proto::VideoFrame { +impl From for proto::VideoFrameInfo { fn from(frame: VideoFrame) -> Self { Self { width: frame.width(), @@ -201,7 +225,7 @@ impl From for proto::VideoFrameBufferType { macro_rules! impl_yuv_into { ($b:ty) => { - impl From<$b> for proto::PlanarYuvBuffer { + impl From<$b> for proto::PlanarYuvBufferInfo { fn from(buffer: $b) -> Self { Self { chroma_width: buffer.chroma_width(), @@ -226,7 +250,7 @@ impl_yuv_into!(&I010Buffer); macro_rules! impl_biyuv_into { ($b:ty) => { - impl From<$b> for proto::BiplanarYuvBuffer { + impl From<$b> for proto::BiplanarYuvBufferInfo { fn from(buffer: $b) -> Self { Self { chroma_width: buffer.chroma_width(), @@ -243,7 +267,7 @@ macro_rules! impl_biyuv_into { impl_biyuv_into!(&NV12Buffer); -impl proto::VideoFrameBuffer { +impl proto::VideoFrameBufferInfo { pub fn from(handle_id: FFIHandleId, buffer: &VideoFrameBuffer) -> Self { Self { handle: Some(handle_id.into()), @@ -252,19 +276,54 @@ impl proto::VideoFrameBuffer { height: buffer.height(), buffer: Some(match &buffer { VideoFrameBuffer::Native(_) => { - proto::video_frame_buffer::Buffer::Native(proto::NativeBuffer {}) + proto::video_frame_buffer_info::Buffer::Native(proto::NativeBufferInfo {}) + } + VideoFrameBuffer::I420(i420) => { + proto::video_frame_buffer_info::Buffer::Yuv(i420.into()) } - VideoFrameBuffer::I420(i420) => proto::video_frame_buffer::Buffer::Yuv(i420.into()), VideoFrameBuffer::I420A(i420a) => { - proto::video_frame_buffer::Buffer::Yuv(i420a.into()) + proto::video_frame_buffer_info::Buffer::Yuv(i420a.into()) + } + VideoFrameBuffer::I422(i422) => { + proto::video_frame_buffer_info::Buffer::Yuv(i422.into()) + } + VideoFrameBuffer::I444(i444) => { + proto::video_frame_buffer_info::Buffer::Yuv(i444.into()) + } + VideoFrameBuffer::I010(i010) => { + proto::video_frame_buffer_info::Buffer::Yuv(i010.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()) + proto::video_frame_buffer_info::Buffer::BiYuv(nv12.into()) } }), } } } + +impl From for VideoFormatType { + fn from(format: proto::VideoFormatType) -> Self { + match format { + proto::VideoFormatType::FormatArgb => Self::ARGB, + proto::VideoFormatType::FormatBgra => Self::BGRA, + proto::VideoFormatType::FormatAbgr => Self::ABGR, + proto::VideoFormatType::FormatRgba => Self::RGBA, + } + } +} + +impl From<&RoomSession> for proto::RoomInfo { + fn from(session: &RoomSession) -> Self { + Self { + sid: session.sid().into(), + name: session.name(), + metadata: session.metadata(), + local_participant: Some((&session.local_participant()).into()), + participants: session + .participants() + .iter() + .map(|(_, p)| p.into()) + .collect(), + } + } +} diff --git a/livekit-ffi/src/server/mod.rs b/livekit-ffi/src/server/mod.rs index 82d5ea7..27744a3 100644 --- a/livekit-ffi/src/server/mod.rs +++ b/livekit-ffi/src/server/mod.rs @@ -1,30 +1,30 @@ -use crate::{ - proto, proto::ffi_request::Message as FFIRequest, proto::ffi_response::Message as FFIResponse, -}; +use crate::proto; 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::AtomicU64; use std::sync::atomic::{AtomicBool, Ordering}; use thiserror::Error; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; mod conversion; +mod room; #[derive(Error, Debug)] pub enum FFIError { #[error("the FFIServer isn't configured")] NotConfigured, - #[error("failed to execute the ffi callback")] + #[error("failed to execute the FFICallback")] CallbackFailed, } -pub type FFIHandleId = u32; +pub type FFIHandleId = usize; pub type FFIHandle = Box; type CallbackFn = unsafe extern "C" fn(*const u8, usize); // This "C" callback must be threadsafe @@ -42,10 +42,13 @@ pub struct FFIConfig { pub struct FFIServer { // Object owned by the foreign language // The foreign language is responsible for freeing this memory + // + // NOTE: For VideoBuffers, we always store the enum VideoFrameBuffer ffi_owned: RwLock>, - next_handle: AtomicU32, // FFIHandle + next_handle_id: AtomicU64, // FFIHandleId + next_async_id: AtomicU64, - rooms: RwLock>, + rooms: RwLock, oneshot::Sender<()>)>>, async_runtime: tokio::runtime::Runtime, initialized: AtomicBool, config: Mutex>, @@ -55,7 +58,8 @@ impl Default for FFIServer { fn default() -> Self { Self { ffi_owned: RwLock::new(HashMap::new()), - next_handle: Default::default(), + next_handle_id: AtomicU64::new(1), // 0 is considered invalid + next_async_id: AtomicU64::new(1), rooms: RwLock::new(HashMap::new()), async_runtime: tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -68,8 +72,45 @@ impl Default for FFIServer { } impl FFIServer { + pub fn initialize(&self, init: &proto::InitializeRequest) { + if self.initialized() { + self.dispose(); + } + + self.initialized.store(true, Ordering::SeqCst); + *self.config.lock() = Some(FFIConfig { + callback_fn: unsafe { std::mem::transmute(init.event_callback_ptr) }, + }); + } + + pub fn dispose(&self) { + self.initialized.store(false, Ordering::SeqCst); + *self.config.lock() = None; + self.async_runtime.block_on(self.close()); + } + + pub async fn close(&self) { + // Close all rooms + for (k, (handle, shutdown_tx)) in self.rooms.write().drain() { + let _ = shutdown_tx.send(()); + let _ = handle.await; + } + } + + pub fn add_room(&self, sid: RoomSid, handle: (JoinHandle<()>, oneshot::Sender<()>)) { + self.rooms.write().insert(sid, handle); + } + + pub fn initialized(&self) -> bool { + self.initialized.load(Ordering::SeqCst) + } + pub fn next_handle_id(&self) -> FFIHandleId { - self.next_handle.fetch_add(1, Ordering::SeqCst) as FFIHandleId + self.next_handle_id.fetch_add(1, Ordering::SeqCst) as FFIHandleId + } + + pub fn next_async_id(&self) -> u64 { + self.next_async_id.fetch_add(1, Ordering::SeqCst) } pub fn insert_handle(&self, handle_id: FFIHandleId, handle: FFIHandle) { @@ -80,142 +121,151 @@ impl FFIServer { self.ffi_owned.write().remove(&handle_id) } - pub fn send_response(&self, message: FFIResponse) -> Result<(), FFIError> { - if !self.initialized.load(Ordering::SeqCst) { + pub fn send_event( + &self, + message: proto::ffi_event::Message, + async_id: Option, + ) -> Result<(), FFIError> { + let config = self.config.lock(); + + if !self.initialized() { Err(FFIError::NotConfigured)? } - let message = proto::FfiResponse { + let message = proto::FfiEvent { + async_id, message: Some(message), } .encode_to_vec(); - let callback_fn = self.config.lock().as_ref().unwrap().callback_fn; + let config = config.as_ref().unwrap(); if let Err(err) = panic::catch_unwind(|| unsafe { - callback_fn(message.as_ptr(), message.len()); + (config.callback_fn)(message.as_ptr(), message.len()); }) { - eprintln!("panic when sending ffi response: {:?}", err); + eprintln!("panic when sending ffi event: {:?}", 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)? - } - + pub fn handle_request(&self, message: proto::ffi_request::Message) -> proto::FfiResponse { match message { proto::ffi_request::Message::AsyncConnect(connect) => { - self.async_runtime.spawn(room_task(connect)); + let async_id = self.next_async_id(); + self.async_runtime + .spawn(room::create_room(&FFI_SERVER, async_id, connect)); + + return proto::FfiResponse { + async_id: Some(async_id), + ..Default::default() + }; } - _ => {} - }; + proto::ffi_request::Message::ToI420(to_i420) => { + let mut buffer_info = None; + let buffer = self.release_handle(to_i420.buffer.unwrap().id as FFIHandleId); - Ok(()) - } -} + if let Some(buffer) = buffer { + if let Ok(buffer) = buffer.downcast::() { + let handle_id = self.next_handle_id(); + let i420 = VideoFrameBuffer::I420(buffer.to_i420()); + buffer_info = Some(proto::VideoFrameBufferInfo::from(handle_id, &i420)); + self.insert_handle(handle_id, Box::new(i420)); + } + } -#[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))); + return proto::FfiResponse { + message: Some(proto::ffi_response::Message::ToI420( + proto::ToI420Response { + new_buffer: buffer_info, + }, + )), + ..Default::default() + }; } - 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())); + proto::ffi_request::Message::ToArgb(to_argb) => { + let ffi_owned = self.ffi_owned.read(); + let buffer = ffi_owned.get(&(to_argb.buffer.unwrap().id as FFIHandleId)); + + if let Some(buffer) = buffer { + if let Some(buffer) = buffer.downcast_ref::() { + let dst_buf = unsafe { + slice::from_raw_parts_mut( + to_argb.dst_ptr as *mut u8, + (to_argb.dst_stride * to_argb.dst_height) as usize, + ) + }; + + if let Err(err) = buffer.to_argb( + proto::VideoFormatType::from_i32(to_argb.dst_format) + .unwrap() + .into(), + dst_buf, + to_argb.dst_stride, + to_argb.dst_width, + to_argb.dst_height, + ) { + eprintln!("failed to convert videoframe to argb: {:?}", err); + } + } } } _ => {} } + + proto::FfiResponse::default() } } -// 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 +/// This function is threadsafe, this is useful to run synchronous requests in another thread (e.g +/// color conversion) +#[no_mangle] +pub extern "C" fn livekit_ffi_request( + data: *const u8, + len: usize, + data_ptr: *mut *const u8, + data_len: *mut usize, +) -> FFIHandleId { + let data = unsafe { slice::from_raw_parts(data, len) }; + let res = proto::FfiRequest::decode(data); + if let Err(ref err) = res { + eprintln!("failed to decode FfiRequest: {:?}", err); + return 0; } + + if res.as_ref().unwrap().message.is_none() { + eprintln!("request message is empty"); + return 0; + } + + let message = res.unwrap().message.unwrap(); + if let proto::ffi_request::Message::Initialize(ref init) = message { + FFI_SERVER.initialize(init); + } + + if let proto::ffi_request::Message::Dispose(_) = message { + FFI_SERVER.dispose(); + } + + if !FFI_SERVER.initialized() { + eprintln!("the FFIServer isn't initialized"); + return 0; + } + + let res = FFI_SERVER.handle_request(message); + let buf = res.encode_to_vec(); + + unsafe { + *data_ptr = buf.as_ptr(); + *data_len = buf.len(); + } + + let handle_id = FFI_SERVER.next_handle_id(); + FFI_SERVER.insert_handle(handle_id, Box::new(buf)); + handle_id } -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), - }, - )), - })); - }) +#[no_mangle] +pub extern "C" fn livekit_ffi_drop_handle(handle_id: FFIHandleId) -> bool { + FFI_SERVER.release_handle(handle_id).is_some() // Free the memory } diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs new file mode 100644 index 0000000..b1d200d --- /dev/null +++ b/livekit-ffi/src/server/room.rs @@ -0,0 +1,119 @@ +use crate::proto::{self}; +use crate::server::FFIServer; +use livekit::prelude::*; +use tokio::sync::{mpsc, oneshot}; + +pub async fn create_room( + server: &'static FFIServer, + async_id: u64, + connect: proto::ConnectRequest, +) { + let res = Room::connect(&connect.url, &connect.token).await; + if let Err(err) = &res { + // Failed to connect to the room + let _ = server.send_event( + proto::ffi_event::Message::ConnectEvent(proto::ConnectEvent { + success: false, + room: None, + }), + Some(async_id), + ); + return; + } + + let (room, events) = res.unwrap(); + let session = room.session(); + + // Successfully connected to the room + let _ = server.send_event( + proto::ffi_event::Message::ConnectEvent(proto::ConnectEvent { + success: true, + room: Some((&session).into()), + }), + Some(async_id), + ); + + // Add the room to the server and listen to the incoming events + let (close_tx, close_rx) = oneshot::channel(); + let room_handle = tokio::spawn(room_task(server, room, events, close_rx)); + server.add_room(session.sid(), (room_handle, close_tx)); +} + +async fn room_task( + server: &'static FFIServer, + room: Room, + mut events: mpsc::UnboundedReceiver, + mut close_rx: oneshot::Receiver<()>, +) { + let session = room.session(); + + tokio::spawn(participant_task(Participant::Local( + session.local_participant(), + ))); + + loop { + tokio::select! { + Some(event) = events.recv() => { + if let Some(event) = proto::RoomEvent::from(session.sid(), event.clone()) { + let _ = server.send_event(proto::ffi_event::Message::RoomEvent(event), None); + } + + 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(server, video_track.sid())); + } + } + _ => {} + } + }, + _ = &mut close_rx => { + break; + } + }; + } + + room.close().await; +} + +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(server: &'static FFIServer, track_sid: TrackSid) -> OnFrameHandler { + // TODO(theomonnom): Should I use VideoSinkInfo here? (It'll help to have a more verbose + // lifetime) + + Box::new(move |frame, buffer| { + // Frame received, create a new FFIHandle from the video buffer. + let handle_id = server.next_handle_id(); + let proto_buffer = proto::VideoFrameBufferInfo::from(handle_id, &buffer); + server.insert_handle(handle_id, Box::new(buffer)); + + // Send the received frame to the FFI language. + let _ = server.send_event( + proto::ffi_event::Message::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), + }, + )), + }), + None, + ); + }) +} +// diff --git a/livekit-webrtc/src/media_stream.rs b/livekit-webrtc/src/media_stream.rs index 41abd49..118d527 100644 --- a/livekit-webrtc/src/media_stream.rs +++ b/livekit-webrtc/src/media_stream.rs @@ -122,13 +122,13 @@ impl_media_stream_track_trait!(AudioTrack, audio_to_media); pub type OnFrameHandler = Box; pub type OnDiscardedFrameHandler = Box; -pub type OnConstraintsChanged = Box; +pub type OnConstraintsChangedHandler = Box; #[derive(Default)] struct InternalVideoTrackSink { on_frame_handler: Mutex>, on_discarded_frame_handler: Mutex>, - on_constraints_changed_handler: Mutex>, + on_constraints_changed_handler: Mutex>, } pub struct VideoTrackSourceConstraints { @@ -225,7 +225,7 @@ impl VideoTrack { *self.observer.on_discarded_frame_handler.lock().unwrap() = Some(handler); } - pub fn on_constraints_changed(&self, handler: OnConstraintsChanged) { + pub fn on_constraints_changed(&self, handler: OnConstraintsChangedHandler) { *self.observer.on_constraints_changed_handler.lock().unwrap() = Some(handler); } } diff --git a/livekit-webrtc/src/prelude.rs b/livekit-webrtc/src/prelude.rs index c78e396..c58dfbc 100644 --- a/livekit-webrtc/src/prelude.rs +++ b/livekit-webrtc/src/prelude.rs @@ -1,7 +1,8 @@ pub use crate::data_channel::{DataChannel, DataChannelInit, DataState}; pub use crate::jsep::{IceCandidate, SessionDescription}; pub use crate::media_stream::{ - AudioTrack, MediaStream, MediaStreamTrackHandle, MediaStreamTrackTrait, VideoTrack, + AudioTrack, MediaStream, MediaStreamTrackHandle, MediaStreamTrackTrait, + OnConstraintsChangedHandler, OnDiscardedFrameHandler, OnFrameHandler, VideoTrack, }; pub use crate::peer_connection::{ IceConnectionState, IceGatheringState, PeerConnection, PeerConnectionState, @@ -16,3 +17,4 @@ pub use crate::rtp_transceiver::RtpTransceiver; pub use crate::video_frame::{VideoFrame, VideoRotation}; pub use crate::video_frame_buffer::*; pub use crate::webrtc::RTCRuntime; +pub use crate::yuv_helper::ConvertError; diff --git a/livekit-webrtc/src/video_frame_buffer.rs b/livekit-webrtc/src/video_frame_buffer.rs index ecfc83d..d873b54 100644 --- a/livekit-webrtc/src/video_frame_buffer.rs +++ b/livekit-webrtc/src/video_frame_buffer.rs @@ -4,6 +4,8 @@ use std::pin::Pin; use std::slice; use webrtc_sys::video_frame_buffer as vfb_sys; +use crate::yuv_helper::{self, ConvertError}; + #[derive(Debug)] pub enum VideoFrameBufferType { Native, @@ -15,6 +17,15 @@ pub enum VideoFrameBufferType { NV12, } +// types to convert to +#[derive(Debug)] +pub enum VideoFormatType { + ARGB, + BGRA, + ABGR, + RGBA, +} + impl From for VideoFrameBufferType { fn from(buffer_type: vfb_sys::ffi::VideoFrameBufferType) -> Self { match buffer_type { @@ -102,10 +113,76 @@ impl VideoFrameBuffer { VideoFrameBufferType::NV12 => { Self::NV12(NV12Buffer::new(cxx_handle.pin_mut().get_nv12())) } - _ => unreachable!(), // VideoFrameBufferType is represented as i32 } } } + + pub fn to_argb( + &self, + format: VideoFormatType, + dst: &mut [u8], + dst_stride: i32, + dst_width: i32, + dst_height: i32, + ) -> Result<(), ConvertError> { + match self { + Self::I420(i420) => match format { + VideoFormatType::ARGB => yuv_helper::i420_to_argb( + i420.data_y(), + i420.stride_y(), + i420.data_u(), + i420.stride_u(), + i420.data_v(), + i420.stride_v(), + dst, + dst_stride, + dst_width, + dst_height, + )?, + VideoFormatType::BGRA => yuv_helper::i420_to_bgra( + i420.data_y(), + i420.stride_y(), + i420.data_u(), + i420.stride_u(), + i420.data_v(), + i420.stride_v(), + dst, + dst_stride, + dst_width, + dst_height, + )?, + VideoFormatType::ABGR => yuv_helper::i420_to_abgr( + i420.data_y(), + i420.stride_y(), + i420.data_u(), + i420.stride_u(), + i420.data_v(), + i420.stride_v(), + dst, + dst_stride, + dst_width, + dst_height, + )?, + VideoFormatType::RGBA => yuv_helper::i420_to_rgba( + i420.data_y(), + i420.stride_y(), + i420.data_u(), + i420.stride_u(), + i420.data_v(), + i420.stride_v(), + dst, + dst_stride, + dst_width, + dst_height, + )?, + }, + _ => { + // TODO(theomonnom): Support other buffer types + } + }; + + Ok(()) + } } impl VideoFrameBufferTrait for VideoFrameBuffer { diff --git a/livekit-webrtc/src/yuv_helper.rs b/livekit-webrtc/src/yuv_helper.rs index 37d2886..87d5543 100644 --- a/livekit-webrtc/src/yuv_helper.rs +++ b/livekit-webrtc/src/yuv_helper.rs @@ -1,43 +1,98 @@ -use std::convert::TryInto; +use thiserror::Error; use webrtc_sys::yuv_helper as yuv_sys; -pub fn i420_to_abgr( +#[derive(Error, Debug)] +pub enum ConvertError { + #[error("conversion failed: {0}")] + Convert(&'static str), +} + +fn i420_safety( src_y: &[u8], src_stride_y: i32, src_u: &[u8], src_stride_u: i32, src_v: &[u8], src_stride_v: i32, - dst_abgr: &mut [u8], - dst_stride_abgr: i32, - width: i32, + dst: &mut [u8], + dst_stride: i32, + _width: i32, height: i32, -) { - // Assert minimum capacity for safety - let chroma_height = (height + 1) / 2; // the buffer should be padded? - let min_y: usize = (src_stride_y * height).try_into().unwrap(); - let min_u: usize = (src_stride_u * chroma_height).try_into().unwrap(); - let min_v: usize = (src_stride_v * chroma_height).try_into().unwrap(); - let min_abgr: usize = (dst_stride_abgr * height).try_into().unwrap(); +) -> Result<(), ConvertError> { + let chroma_height = (height + 1) / 2; + let min_y = (src_stride_y * height) as usize; + let min_u = (src_stride_u * chroma_height) as usize; + let min_v = (src_stride_v * chroma_height) as usize; + let min_dst = (dst_stride * height) as usize; - assert!(src_y.len() >= min_y); - assert!(src_u.len() >= min_u); - assert!(src_v.len() >= min_v); - assert!(dst_abgr.len() >= min_abgr); - - unsafe { - yuv_sys::ffi::i420_to_abgr( - src_y.as_ptr(), - src_stride_y, - src_u.as_ptr(), - src_stride_u, - src_v.as_ptr(), - src_stride_v, - dst_abgr.as_mut_ptr(), - dst_stride_abgr, - width, - height, - ); + if src_y.len() < min_y { + return Err(ConvertError::Convert("src_y isn't large enough")); } + + if src_u.len() < min_u { + return Err(ConvertError::Convert("src_u isn't large enough")); + } + + if src_v.len() < min_v { + return Err(ConvertError::Convert("src_v isn't large enough")); + } + + if dst.len() < min_dst { + return Err(ConvertError::Convert("dst isn't large enough")); + } + + Ok(()) } + +macro_rules! i420_to_x { + ($x:ident) => { + pub fn $x( + src_y: &[u8], + src_stride_y: i32, + src_u: &[u8], + src_stride_u: i32, + src_v: &[u8], + src_stride_v: i32, + dst: &mut [u8], + dst_stride: i32, + width: i32, + height: i32, + ) -> Result<(), ConvertError> { + i420_safety( + src_y, + src_stride_y, + src_u, + src_stride_u, + src_v, + src_stride_v, + dst, + dst_stride, + width, + height, + )?; + + unsafe { + yuv_sys::ffi::$x( + src_y.as_ptr(), + src_stride_y, + src_u.as_ptr(), + src_stride_u, + src_v.as_ptr(), + src_stride_v, + dst.as_mut_ptr(), + dst_stride, + width, + height, + ); + } + + Ok(()) + } + }; +} + +i420_to_x!(i420_to_argb); +i420_to_x!(i420_to_bgra); +i420_to_x!(i420_to_abgr); +i420_to_x!(i420_to_rgba); diff --git a/livekit/src/prelude.rs b/livekit/src/prelude.rs index 4b2d8a0..9958e97 100644 --- a/livekit/src/prelude.rs +++ b/livekit/src/prelude.rs @@ -19,10 +19,13 @@ pub use crate::id::*; pub use crate::webrtc::{ data_channel::DataChannel, media_stream::{ - AudioTrack, MediaStream, MediaStreamTrackHandle, MediaStreamTrackTrait, VideoTrack, + AudioTrack, MediaStream, MediaStreamTrackHandle, MediaStreamTrackTrait, + OnConstraintsChangedHandler, OnDiscardedFrameHandler, OnFrameHandler, VideoTrack, }, rtp_receiver::RtpReceiver, rtp_transceiver::RtpTransceiver, video_frame::{VideoFrame, VideoRotation}, - video_frame_buffer::{VideoFrameBuffer, VideoFrameBufferTrait, VideoFrameBufferType}, + video_frame_buffer::{ + VideoFormatType, VideoFrameBuffer, VideoFrameBufferTrait, VideoFrameBufferType, + }, }; diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index 3546ec1..56f05f7 100644 --- a/livekit/src/room/mod.rs +++ b/livekit/src/room/mod.rs @@ -55,12 +55,12 @@ pub enum RoomEvent { participant: Arc, }, TrackMuted { - publication: TrackPublication, participant: Participant, + publication: TrackPublication, }, TrackUnmuted { - publication: TrackPublication, participant: Participant, + publication: TrackPublication, }, ActiveSpeakersChanged { speakers: Vec, diff --git a/livekit/src/room/publication/mod.rs b/livekit/src/room/publication/mod.rs index 2ab3742..dc383e7 100644 --- a/livekit/src/room/publication/mod.rs +++ b/livekit/src/room/publication/mod.rs @@ -19,8 +19,10 @@ pub trait TrackPublicationTrait { fn sid(&self) -> TrackSid; fn kind(&self) -> TrackKind; fn source(&self) -> TrackSource; - fn muted(&self) -> bool; fn simulcasted(&self) -> bool; + fn dimension(&self) -> TrackDimension; + fn mime_type(&self) -> String; + fn muted(&self) -> bool; } #[derive(Debug)] @@ -164,8 +166,10 @@ impl TrackPublicationTrait for TrackPublication { fnc!(name, &Self, [], String); fnc!(kind, &Self, [], TrackKind); fnc!(source, &Self, [], TrackSource); - fnc!(muted, &Self, [], bool); fnc!(simulcasted, &Self, [], bool); + fnc!(dimension, &Self, [], TrackDimension); + fnc!(mime_type, &Self, [], String); + fnc!(muted, &Self, [], bool); ); } @@ -192,6 +196,14 @@ macro_rules! impl_publication_trait { self.shared.simulcasted.load(Ordering::SeqCst) } + fn dimension(&self) -> TrackDimension { + self.shared.dimension.lock().clone() + } + + fn mime_type(&self) -> String { + self.shared.mime_type.lock().clone() + } + fn muted(&self) -> bool { self.shared.muted.load(Ordering::SeqCst) } diff --git a/livekit/src/room/room_session.rs b/livekit/src/room/room_session.rs index 913fd9e..8d4f87d 100644 --- a/livekit/src/room/room_session.rs +++ b/livekit/src/room/room_session.rs @@ -37,8 +37,9 @@ impl From for ConnectionState { #[derive(Debug)] struct SessionInner { state: AtomicU8, // ConnectionState - sid: Mutex, + sid: Mutex, name: Mutex, + metadata: Mutex, participants: RwLock>>, participants_tasks: RwLock, oneshot::Sender<()>)>>, active_speakers: RwLock>, @@ -82,8 +83,9 @@ impl SessionHandle { let room_info = join_response.room.unwrap(); let inner = Arc::new(SessionInner { state: AtomicU8::new(ConnectionState::Disconnected as u8), - sid: Mutex::new(room_info.sid), + sid: Mutex::new(room_info.sid.into()), name: Mutex::new(room_info.name), + metadata: Mutex::new(room_info.metadata), participants: Default::default(), participants_tasks: Default::default(), active_speakers: Default::default(), @@ -133,7 +135,7 @@ impl RoomSession { Self { inner } } - pub fn sid(&self) -> String { + pub fn sid(&self) -> RoomSid { self.inner.sid.lock().clone() } @@ -141,6 +143,10 @@ impl RoomSession { self.inner.name.lock().clone() } + pub fn metadata(&self) -> String { + self.inner.metadata.lock().clone() + } + pub fn local_participant(&self) -> Arc { self.inner.local_participant.clone() } diff --git a/webrtc-sys/include/livekit/yuv_helper.h b/webrtc-sys/include/livekit/yuv_helper.h index 1378cdd..64e02df 100644 --- a/webrtc-sys/include/livekit/yuv_helper.h +++ b/webrtc-sys/include/livekit/yuv_helper.h @@ -11,18 +11,60 @@ namespace livekit { +static void i420_to_argb(const uint8_t* src_y, + int src_stride_y, + const uint8_t* src_u, + int src_stride_u, + const uint8_t* src_v, + int src_stride_v, + uint8_t* dst_argb, + int dst_stride_argb, + int width, + int height) { + webrtc::I420ToARGB(src_y, src_stride_y, src_u, src_stride_u, src_v, + src_stride_v, dst_argb, dst_stride_argb, width, height); +} + +static void i420_to_bgra(const uint8_t* src_y, + int src_stride_y, + const uint8_t* src_u, + int src_stride_u, + const uint8_t* src_v, + int src_stride_v, + uint8_t* dst_bgra, + int dst_stride_bgra, + int width, + int height) { + webrtc::I420ToBGRA(src_y, src_stride_y, src_u, src_stride_u, src_v, + src_stride_v, dst_bgra, dst_stride_bgra, width, height); +} + static void i420_to_abgr(const uint8_t* src_y, - int src_stride_y, - const uint8_t* src_u, - int src_stride_u, - const uint8_t* src_v, - int src_stride_v, - uint8_t* dst_rgba, - int dst_stride_abgr, - int width, - int height) { + int src_stride_y, + const uint8_t* src_u, + int src_stride_u, + const uint8_t* src_v, + int src_stride_v, + uint8_t* dst_abgr, + int dst_stride_abgr, + int width, + int height) { webrtc::I420ToABGR(src_y, src_stride_y, src_u, src_stride_u, src_v, - src_stride_v, dst_rgba, dst_stride_abgr, width, height); + src_stride_v, dst_abgr, dst_stride_abgr, width, height); +} + +static void i420_to_rgba(const uint8_t* src_y, + int src_stride_y, + const uint8_t* src_u, + int src_stride_u, + const uint8_t* src_v, + int src_stride_v, + uint8_t* dst_rgba, + int dst_stride_rgba, + int width, + int height) { + webrtc::I420ToRGBA(src_y, src_stride_y, src_u, src_stride_u, src_v, + src_stride_v, dst_rgba, dst_stride_rgba, width, height); } } // namespace livekit diff --git a/webrtc-sys/src/yuv_helper.rs b/webrtc-sys/src/yuv_helper.rs index 69da044..420d25d 100644 --- a/webrtc-sys/src/yuv_helper.rs +++ b/webrtc-sys/src/yuv_helper.rs @@ -3,6 +3,32 @@ pub mod ffi { unsafe extern "C++" { include!("livekit/yuv_helper.h"); + unsafe fn i420_to_argb( + src_y: *const u8, + src_stride_y: i32, + src_u: *const u8, + src_stride_u: i32, + src_v: *const u8, + src_stride_v: i32, + dst_argb: *mut u8, + dst_stride_argb: i32, + width: i32, + height: i32, + ); + + unsafe fn i420_to_bgra( + src_y: *const u8, + src_stride_y: i32, + src_u: *const u8, + src_stride_u: i32, + src_v: *const u8, + src_stride_v: i32, + dst_bgra: *mut u8, + dst_stride_bgra: i32, + width: i32, + height: i32, + ); + unsafe fn i420_to_abgr( src_y: *const u8, src_stride_y: i32, @@ -15,5 +41,18 @@ pub mod ffi { width: i32, height: i32, ); + + unsafe fn i420_to_rgba( + src_y: *const u8, + src_stride_y: i32, + src_u: *const u8, + src_stride_u: i32, + src_v: *const u8, + src_stride_v: i32, + dst_rgba: *mut u8, + dst_stride_rgba: i32, + width: i32, + height: i32, + ); } }