finish proto of livekit-ffi (#36)

* ffi req

* handle req_id

* remaining room events

* fix

* Update ffi.proto

* use sid instead of info

* state -> stream_state

* "s"

* participant is optional on data received

* add ReleaseHandleRequest

* Update ffi.proto

* add DataPacketKind

* VideoSinkInfo

* append info

It looks nicer on the FFI languages

* wip

* keep it simple

* Update ffi.proto

* update server to latest proto

* to_i420

* i420_to_abgr

* to_argb

* add publications to ParticipantInfo

It'll be used when we join a Room

* cleanup + DisposeRequest

* fix compilation

* send the whole buffer info with to_i420
This commit is contained in:
Théo Monnom
2023-02-06 00:57:43 +01:00
committed by GitHub
parent 1610f86316
commit e023b34fd1
14 changed files with 832 additions and 239 deletions
+181 -52
View File
@@ -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; }
+74 -15
View File
@@ -10,7 +10,7 @@ use std::sync::Arc;
impl From<FFIHandleId> 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<LocalParticipant>);
impl_participant_into!(&Arc<RemoteParticipant>);
impl_participant_into!(&Participant);
impl From<TrackSource> 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<VideoRotation> for proto::VideoRotation {
}
}
impl From<VideoFrame> for proto::VideoFrame {
impl From<VideoFrame> for proto::VideoFrameInfo {
fn from(frame: VideoFrame) -> Self {
Self {
width: frame.width(),
@@ -201,7 +225,7 @@ impl From<VideoFrameBufferType> 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<proto::VideoFormatType> 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(),
}
}
}
+168 -118
View File
@@ -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<dyn Any + Send + Sync>;
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<HashMap<FFIHandleId, FFIHandle>>,
next_handle: AtomicU32, // FFIHandle
next_handle_id: AtomicU64, // FFIHandleId
next_async_id: AtomicU64,
rooms: RwLock<HashMap<RoomSid, Room>>,
rooms: RwLock<HashMap<RoomSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
async_runtime: tokio::runtime::Runtime,
initialized: AtomicBool,
config: Mutex<Option<FFIConfig>>,
@@ -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<u64>,
) -> 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::<VideoFrameBuffer>() {
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::<VideoFrameBuffer>() {
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
}
+119
View File
@@ -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<livekit::RoomEvent>,
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,
);
})
}
//
+3 -3
View File
@@ -122,13 +122,13 @@ impl_media_stream_track_trait!(AudioTrack, audio_to_media);
pub type OnFrameHandler = Box<dyn FnMut(VideoFrame, VideoFrameBuffer) + Send + Sync>;
pub type OnDiscardedFrameHandler = Box<dyn FnMut() + Send + Sync>;
pub type OnConstraintsChanged = Box<dyn FnMut(VideoTrackSourceConstraints) + Send + Sync>;
pub type OnConstraintsChangedHandler = Box<dyn FnMut(VideoTrackSourceConstraints) + Send + Sync>;
#[derive(Default)]
struct InternalVideoTrackSink {
on_frame_handler: Mutex<Option<OnFrameHandler>>,
on_discarded_frame_handler: Mutex<Option<OnDiscardedFrameHandler>>,
on_constraints_changed_handler: Mutex<Option<OnConstraintsChanged>>,
on_constraints_changed_handler: Mutex<Option<OnConstraintsChangedHandler>>,
}
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);
}
}
+3 -1
View File
@@ -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;
+78 -1
View File
@@ -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<vfb_sys::ffi::VideoFrameBufferType> 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 {
+85 -30
View File
@@ -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);
+5 -2
View File
@@ -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,
},
};
+2 -2
View File
@@ -55,12 +55,12 @@ pub enum RoomEvent {
participant: Arc<RemoteParticipant>,
},
TrackMuted {
publication: TrackPublication,
participant: Participant,
publication: TrackPublication,
},
TrackUnmuted {
publication: TrackPublication,
participant: Participant,
publication: TrackPublication,
},
ActiveSpeakersChanged {
speakers: Vec<Participant>,
+14 -2
View File
@@ -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)
}
+9 -3
View File
@@ -37,8 +37,9 @@ impl From<u8> for ConnectionState {
#[derive(Debug)]
struct SessionInner {
state: AtomicU8, // ConnectionState
sid: Mutex<String>,
sid: Mutex<RoomSid>,
name: Mutex<String>,
metadata: Mutex<String>,
participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
active_speakers: RwLock<Vec<Participant>>,
@@ -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<LocalParticipant> {
self.inner.local_participant.clone()
}
+52 -10
View File
@@ -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
+39
View File
@@ -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,
);
}
}