feat: basic ffi server to support other languages (#34)

* wip

* wip

* wip

* wip

not a fan of the way I do handles ..

* errors and async semantic

* fix example

livekit-ffi/src/lib.rs

@@ -1 +1,5 @@
+mod proto {
+    include!(concat!(env!("OUT_DIR"), "/livekit.rs"));
+}
 
+mod server;

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<FFIHandleId> 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<LocalParticipant>);
+impl_participant_into!(&Arc<RemoteParticipant>);
+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<TrackKind> 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<StreamState> 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<String>, event: RoomEvent) -> Option<Self> {
+        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<VideoRotation> 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<VideoFrame> 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<VideoFrameBufferType> 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())
+                }
+            }),
+        }
+    }
+}

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<dyn Any + Send + Sync>;
+
+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<HashMap<FFIHandleId, FFIHandle>>,
+    next_handle: AtomicU32, // FFIHandle
+
+    rooms: RwLock<HashMap<RoomSid, Room>>,
+    async_runtime: tokio::runtime::Runtime,
+    initialized: AtomicBool,
+    config: Mutex<Option<FFIConfig>>,
+}
+
+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<FFIHandle> {
+        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),
+                },
+            )),
+        }));
+    })
+}