talksik/client-sdk-rust
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: add publishing & audio to ffi (#46)

Browse Source 
- Divide FFI protocol into multiple files
- Audio support
- AV Streams/Sources
- Create tracks
- Use Dashmap to store handles
- Add a capture test
     - Ignored by GHA atm
This commit is contained in:
Théo Monnom
2023-05-08 23:12:37 +02:00
committed by GitHub
parent a5caf12902
commit 1c12e749f2
58 changed files with 3046 additions and 1188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
livekit-ffi/Cargo.toml
+8
View File
@@ -16,9 +16,17 @@ prost-types = "0.11.1"
lazy_static = "1.4.0"
thiserror = "1.0.38"
futures-util = "0.3.23"
log = "0.4.17"
dashmap = "5.4.0"
[build-dependencies]
prost-build = { version = "0.11.1" }
[dev-dependencies]
livekit-api = { path = "../livekit-api", version = "0.1.0" }
[lib]
crate-type = ["cdylib", "staticlib"]
[profile.release]
opt-level = "z"
livekit-ffi/build.rs
+12 -2
View File
@@ -1,7 +1,17 @@
use std::io::Result;
fn main() -> Result<()> {
prost_build::compile_protos(&["protocol/ffi.proto"], &["protocol/"])?;
prost_build::compile_protos(
&[
"protocol/ffi.proto",
"protocol/handle.proto",
"protocol/room.proto",
"protocol/track.proto",
"protocol/participant.proto",
"protocol/video_frame.proto",
"protocol/audio_frame.proto",
],
&["protocol/"],
)?;
Ok(())
}
livekit-ffi/protocol/audio_frame.proto
+86
View File
@@ -0,0 +1,86 @@
syntax = "proto3";
package livekit;
option csharp_namespace = "LiveKit.Proto";
import "handle.proto";
// Allocate a new AudioFrameBuffer
// This is not necessary required because the data structure is fairly simple
// But keep the API consistent with VideoFrame
message AllocAudioBufferRequest {
uint32 sample_rate = 1;
uint32 num_channels = 2;
uint32 samples_per_channel = 3;
}
message AllocAudioBufferResponse { AudioFrameBufferInfo buffer = 1; }
// Create a new AudioStream
// AudioStream is used to receive audio frames from a track
message NewAudioStreamRequest {
FFIHandleId room_handle = 1;
string participant_sid = 2;
string track_sid = 3;
AudioStreamType type = 4;
}
message NewAudioStreamResponse { AudioStreamInfo stream = 1; }
// Create a new AudioSource
message NewAudioSourceRequest { AudioSourceType type = 1; }
message NewAudioSourceResponse { AudioSourceInfo source = 1; }
// Push a frame to an AudioSource
message CaptureAudioFrameRequest {
FFIHandleId source_handle = 1;
FFIHandleId buffer_handle = 2;
}
message CaptureAudioFrameResponse {}
///
/// AudioFrame buffer ///
///
message AudioFrameBufferInfo {
FFIHandleId handle = 1;
uint64 data_ptr = 2; // *const i16
uint32 num_channels = 3;
uint32 sample_rate = 4;
uint32 samples_per_channel = 5;
}
///
/// AudioStream ///
///
enum AudioStreamType {
AUDIO_STREAM_NATIVE = 0;
AUDIO_STREAM_HTML = 1;
}
message AudioStreamInfo {
FFIHandleId handle = 1;
AudioStreamType type = 2;
string track_sid = 3;
}
message AudioStreamEvent {
FFIHandleId handle = 1;
oneof message { AudioFrameReceived frame_received = 2; }
}
message AudioFrameReceived {
AudioFrameBufferInfo frame = 1;
}
///
/// AudioSource ///
///
enum AudioSourceType {
AUDIO_SOURCE_NATIVE = 0;
}
message AudioSourceInfo {
FFIHandleId handle = 1;
AudioSourceType type = 2;
}
livekit-ffi/protocol/ffi.proto
+77 -315
View File
@@ -3,347 +3,109 @@ syntax = "proto3";
package livekit;
option csharp_namespace = "LiveKit.Proto";
/// IPC
/// # Safety
/// The foreign language is responsable for disposing an handle
/// Forgetting to dispose the handle may lead to memory leaks
/// Messages in this file can contain an FFIHandle
message FFIHandleId { uint64 id = 1; }
import "handle.proto";
import "track.proto";
import "room.proto";
import "participant.proto";
import "video_frame.proto";
import "audio_frame.proto";
/// This is the input of livekit_ffi_request function
/// We always expect a response (FFIResponse)
message FFIRequest {
oneof message {
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;
// Room
ConnectRequest connect = 3;
DisconnectRequest disconnect = 4;
PublishTrackRequest publish_track = 5;
UnpublishTrackRequest unpublish_track = 6;
// Track
CreateVideoTrackRequest create_video_track = 7;
CreateAudioTrackRequest create_audio_track = 8;
// Video
AllocVideoBufferRequest alloc_video_buffer = 9;
NewVideoStreamRequest new_video_stream = 10;
NewVideoSourceRequest new_video_source = 11;
CaptureVideoFrameRequest capture_video_frame = 12;
ToI420Request to_i420 = 13;
ToARGBRequest to_argb = 14;
// Audio
AllocAudioBufferRequest alloc_audio_buffer = 15;
NewAudioStreamRequest new_audio_stream = 16;
NewAudioSourceRequest new_audio_source = 17;
CaptureAudioFrameRequest capture_audio_frame = 18;
}
}
/// 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; }
oneof message {
InitializeResponse initialize = 1;
DisposeResponse dispose = 2;
// Room
ConnectResponse connect = 3;
DisconnectResponse disconnect = 4;
PublishTrackResponse publish_track = 5;
UnpublishTrackResponse unpublish_track = 6;
// Track
CreateVideoTrackResponse create_video_track = 7;
CreateAudioTrackResponse create_audio_track = 8;
// Video
AllocVideoBufferResponse alloc_video_buffer = 9;
NewVideoStreamResponse new_video_stream = 10;
NewVideoSourceResponse new_video_source = 11;
CaptureVideoFrameResponse capture_video_frame = 12;
ToI420Response to_i420 = 13;
ToARGBResponse to_argb = 14;
// Audio
AllocAudioBufferResponse alloc_audio_buffer = 15;
NewAudioStreamResponse new_audio_stream = 16;
NewAudioSourceResponse new_audio_source = 17;
CaptureAudioFrameResponse capture_audio_frame = 18;
}
}
/// 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 {
ConnectEvent connect_event = 2;
RoomEvent room_event = 3;
TrackEvent track_event = 4;
ParticipantEvent participant_event = 5;
RoomEvent room_event = 1;
TrackEvent track_event = 2;
ParticipantEvent participant_event = 3;
VideoStreamEvent video_stream_event = 4;
AudioStreamEvent audio_stream_event = 5;
ConnectCallback connect = 6;
DisposeCallback dispose = 7;
PublishTrackCallback publish_track = 8;
}
}
// Setup the callback where the foreign language can receive events
// and responses to asynchronous requests
message InitializeRequest { uint64 event_callback_ptr = 1; }
message InitializeResponse {}
message DisposeRequest {}
message ConnectRequest {
string url = 1;
string token = 2;
RoomOptions options = 3;
// Stop all rooms synchronously (Do we need async here?).
// e.g: This is used for the Unity Editor after each assemblies reload.
message DisposeRequest {
bool async = 1;
}
message DisconnectRequest { string room_sid = 1; }
/// Convert a VideoFrameBuffer to a I420Buffer
message ToI420Request {
FFIHandleId buffer = 1; // NOTE: This buffer will be dropped!
message DisposeResponse {
optional FFIAsyncId async_id = 1; // None if sync
}
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 DisposeCallback {
FFIAsyncId async_id = 1;
}
message ConnectEvent {
bool success = 1;
optional RoomInfo room = 2;
}
// TODO(theomonnom): Debug messages (Print handles, forward logs).
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;
}
message RoomInfo {
string sid = 1;
string name = 2;
string metadata = 3;
ParticipantInfo local_participant = 4;
repeated ParticipantInfo participants = 5;
}
message ParticipantInfo {
string sid = 1;
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 {
KIND_UNKNOWN = 0;
KIND_AUDIO = 1;
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;
}
enum VideoRotation {
VIDEO_ROTATION_0 = 0;
VIDEO_ROTATION_90 = 1;
VIDEO_ROTATION_180 = 2;
VIDEO_ROTATION_270 = 3;
}
enum DataPacketKind {
KIND_UNRELIABLE = 0;
KIND_RELIABLE = 1;
}
enum VideoFormatType {
FORMAT_ARGB = 0;
FORMAT_BGRA = 1;
FORMAT_ABGR = 2;
FORMAT_RGBA = 3;
}
/// Room Events
message RoomEvent {
string room_sid = 1;
oneof message {
ParticipantConnected participant_connected = 2;
ParticipantDisconnected participant_disconnected = 3;
TrackPublished track_published = 4;
TrackUnpublished track_unpublished = 5;
TrackSubscribed track_subscribed = 6;
TrackUnsubscribed track_unsubscribed = 7;
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;
}
}
message ParticipantConnected { ParticipantInfo info = 1; }
message ParticipantDisconnected { ParticipantInfo info = 1; }
message TrackPublished {
string participant_sid = 1;
TrackPublicationInfo publication = 2;
}
message TrackUnpublished {
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 {
string participant_sid = 1;
TrackInfo track = 2;
VideoSinkInfo sink = 3;
}
message TrackUnsubscribed {
// 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;
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 {
string track_sid = 1;
oneof message { FrameReceived frame_received = 2; }
}
message FrameReceived {
VideoFrameInfo frame = 1;
VideoFrameBufferInfo buffer = 2;
}
message VideoFrameInfo {
int64 timestamp = 1;
VideoRotation rotation = 2;
}
message VideoFrameBufferInfo {
FFIHandleId handle = 1;
VideoFrameBufferType buffer_type = 2;
int32 width = 3;
int32 height = 4;
oneof buffer {
PlanarYuvBufferInfo yuv = 5;
BiplanarYuvBufferInfo bi_yuv = 6;
NativeBufferInfo native = 7;
}
}
message PlanarYuvBufferInfo {
int32 chroma_width = 1;
int32 chroma_height = 2;
int32 stride_y = 3;
int32 stride_u = 4;
int32 stride_v = 5;
int32 stride_a = 6;
// *const u8 or *const u16
uint64 data_y_ptr = 7;
uint64 data_u_ptr = 8;
uint64 data_v_ptr = 9;
uint64 data_a_ptr = 10; // nullptr = no alpha
}
message BiplanarYuvBufferInfo {
int32 chroma_width = 1;
int32 chroma_height = 2;
int32 stride_y = 3;
int32 stride_uv = 4;
uint64 data_y_ptr = 5;
uint64 data_uv_ptr = 6;
}
message NativeBufferInfo {
// TODO(theomonnom): Expose graphic context?
}
enum VideoFrameBufferType {
NATIVE = 0;
I420 = 1;
I420A = 2;
I422 = 3;
I444 = 4;
I010 = 5;
NV12 = 6;
WEBGL = 7;
}
/// Participant Events
message ParticipantEvent {
string participant_sid = 1;
oneof message { IsSpeakingChanged speaking_changed = 2; }
}
message IsSpeakingChanged { bool speaking = 1; }
livekit-ffi/protocol/handle.proto
+17
View File
@@ -0,0 +1,17 @@
syntax = "proto3";
package livekit;
option csharp_namespace = "LiveKit.Proto";
/// # Safety
/// The foreign language is responsable for disposing handles
/// Forgetting to dispose the handle may lead to memory leaks
/// Messages in this file can contain an FFIHandle
message FFIHandleId {
uint64 id = 1;
}
/// Link the request/response of an asynchronous call
message FFIAsyncId {
uint64 id = 1;
}
livekit-ffi/protocol/participant.proto
+22
View File
@@ -0,0 +1,22 @@
syntax = "proto3";
package livekit;
option csharp_namespace = "LiveKit.Proto";
//import "handle.proto";
import "track.proto";
message ParticipantInfo {
string sid = 1;
string name = 2;
string identity = 3;
string metadata = 4;
repeated TrackPublicationInfo publications = 5;
}
message ParticipantEvent {
string participant_sid = 1;
oneof message { IsSpeakingChanged speaking_changed = 2; }
}
message IsSpeakingChanged { bool speaking = 1; }
livekit-ffi/protocol/room.proto
+199
View File
@@ -0,0 +1,199 @@
syntax = "proto3";
package livekit;
option csharp_namespace = "LiveKit.Proto";
import "handle.proto";
import "participant.proto";
import "track.proto";
import "video_frame.proto";
// Connect to a new LiveKit room
message ConnectRequest {
string url = 1;
string token = 2;
RoomOptions options = 3;
}
message ConnectResponse {
FFIAsyncId async_id = 1;
}
message ConnectCallback {
FFIAsyncId async_id = 1;
optional string error = 2;
RoomInfo room = 3;
}
// Disconnect from the a room
message DisconnectRequest { FFIHandleId room_handle = 1; }
message DisconnectResponse { FFIAsyncId async_id = 1; }
message DisconnectCallback { }
// Publish a track to the room
message PublishTrackRequest {
FFIHandleId room_handle = 1;
FFIHandleId track_handle = 2;
TrackPublishOptions options = 3;
}
message PublishTrackResponse {
FFIAsyncId async_id = 1;
}
message PublishTrackCallback {
FFIAsyncId async_id = 1;
optional string error = 2;
TrackPublicationInfo publication = 3;
}
// Unpublish a track from the room
message UnpublishTrackRequest {
FFIHandleId room_handle = 1;
string track_sid = 2;
bool stop_on_unpublish = 3;
}
message UnpublishTrackResponse {
FFIAsyncId async_id = 1;
}
message UnpublishTrackCallback {
optional string error = 1;
}
///
/// Options
///
message VideoEncoding {
uint64 max_bitrate = 1;
double max_framerate = 2;
}
message AudioEncoding {
uint64 max_bitrate = 1;
}
message TrackPublishOptions {
// encodings are optional
VideoEncoding video_encoding = 1;
AudioEncoding audio_encoding = 2;
VideoCodec video_codec = 3;
bool dtx = 4;
bool red = 5;
bool simulcast = 6;
string name = 7;
TrackSource source = 8;
}
message RoomOptions {
bool auto_subscribe = 1;
bool adaptive_stream = 2;
}
///
/// Room
///
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 DataPacketKind {
KIND_UNRELIABLE = 0;
KIND_RELIABLE = 1;
}
message RoomEvent {
FFIHandleId room_handle = 1;
oneof message {
ParticipantConnected participant_connected = 2;
ParticipantDisconnected participant_disconnected = 3;
TrackPublished track_published = 4;
TrackUnpublished track_unpublished = 5;
TrackSubscribed track_subscribed = 6;
TrackUnsubscribed track_unsubscribed = 7;
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;
}
}
message RoomInfo {
FFIHandleId handle = 1;
string sid = 2;
string name = 3;
string metadata = 4;
ParticipantInfo local_participant = 5;
repeated ParticipantInfo participants = 6;
}
message DataReceived {
FFIHandleId handle = 1;
optional string participant_sid = 2;
uint64 data_ptr = 3;
uint64 data_size = 4;
DataPacketKind kind = 5;
}
// Publication isn't needed for subscription events on the FFI
// The FFI will retrieve the publication using the Track sid
message TrackSubscribed {
string participant_sid = 1;
TrackInfo track = 2;
}
message TrackUnsubscribed {
// 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 ParticipantConnected { ParticipantInfo info = 1; }
message ParticipantDisconnected { ParticipantInfo info = 1; }
message TrackPublished {
string participant_sid = 1;
TrackPublicationInfo publication = 2;
}
message TrackUnpublished {
string participant_sid = 1;
string publication_sid = 2;
}
message ActiveSpeakersChanged { repeated string participant_sids = 1; }
message ConnectionQualityChanged {
string participant_sid = 1;
ConnectionQuality quality = 2;
}
message ConnectionStateChanged { ConnectionState state = 1; }
message Connected {}
message Disconnected {}
message Reconnecting {}
message Reconnected {}
livekit-ffi/protocol/track.proto
+91
View File
@@ -0,0 +1,91 @@
syntax = "proto3";
package livekit;
option csharp_namespace = "LiveKit.Proto";
import "handle.proto";
import "video_frame.proto";
import "audio_frame.proto";
message VideoCaptureOptions {
VideoResolution resolution = 1;
}
message AudioCaptureOptions {
bool echo_cancellation = 1;
bool noise_suppression = 2;
bool auto_gain_control = 3;
}
// Create a new VideoTrack from a VideoSource
message CreateVideoTrackRequest {
string name = 1;
VideoCaptureOptions options = 2;
FFIHandleId source_handle = 3;
}
message CreateVideoTrackResponse {
TrackInfo track = 1;
}
// Create a new AudioTrack from a AudioSource
message CreateAudioTrackRequest {
string name = 1;
AudioCaptureOptions options = 2;
FFIHandleId source_handle = 3;
}
message CreateAudioTrackResponse {
TrackInfo track = 1;
}
///
/// Track
///
message TrackEvent {}
enum TrackKind {
KIND_UNKNOWN = 0;
KIND_AUDIO = 1;
KIND_VIDEO = 2;
}
enum TrackSource {
SOURCE_UNKNOWN = 0;
SOURCE_CAMERA = 1;
SOURCE_MICROPHONE = 2;
SOURCE_SCREENSHARE = 3;
SOURCE_SCREENSHARE_AUDIO = 4;
}
enum StreamState {
STATE_UNKNOWN = 0;
STATE_ACTIVE = 1;
STATE_PAUSED = 2;
}
// TODO(theomonnom): Should we have a separate message whether the track is local or remote?
message TrackPublicationInfo {
string sid = 1;
string name = 2;
TrackKind kind = 3;
TrackSource source = 4;
bool simulcasted = 5;
uint32 width = 6;
uint32 height = 7;
string mime_type = 8;
bool muted = 9;
bool remote = 10;
}
message TrackInfo {
// Tracks created/owned by the client will have a handle
FFIHandleId opt_handle = 1;
string sid = 2;
string name = 3;
TrackKind kind = 4;
StreamState stream_state = 5;
bool muted = 6;
bool remote = 7;
}
livekit-ffi/protocol/video_frame.proto
+198
View File
@@ -0,0 +1,198 @@
syntax = "proto3";
package livekit;
option csharp_namespace = "LiveKit.Proto";
import "handle.proto";
// Allocate a new VideoFrameBuffer
message AllocVideoBufferRequest {
VideoFrameBufferType type = 1; // Only I420 is supported atm
uint32 width = 2;
uint32 height = 3;
}
message AllocVideoBufferResponse { VideoFrameBufferInfo buffer = 1; }
// Create a new VideoStream
// VideoStream is used to receive video frames from a track
message NewVideoStreamRequest {
FFIHandleId room_handle = 1;
string participant_sid = 2;
string track_sid = 3;
VideoStreamType type = 4;
}
message NewVideoStreamResponse { VideoStreamInfo stream = 1; }
// Create a new VideoSource
// VideoSource is used to send video frame to a track
message NewVideoSourceRequest { VideoSourceType type = 1; }
message NewVideoSourceResponse { VideoSourceInfo source = 1; }
// Push a frame to a VideoSource
message CaptureVideoFrameRequest {
FFIHandleId source_handle = 1;
VideoFrameInfo frame = 2;
FFIHandleId buffer_handle = 3;
}
message CaptureVideoFrameResponse {}
// Convert a RGBA frame to a I420 YUV frame
// Or convert another YUV frame format to I420
message ToI420Request {
bool flip_y = 1;
oneof from {
ARGBBufferInfo argb = 2;
FFIHandleId buffer = 3;
}
}
message ToI420Response { VideoFrameBufferInfo buffer = 1; }
// Convert a YUV frame to a RGBA frame
// Only I420 is supported atm
message ToARGBRequest {
FFIHandleId buffer = 1;
uint64 dst_ptr = 2;
VideoFormatType dst_format = 3;
uint32 dst_stride = 4;
uint32 dst_width = 5;
uint32 dst_height = 6;
bool flip_y = 7;
}
message ToARGBResponse {}
///
/// VideoFrame buffers ///
///
message VideoResolution {
uint32 width = 1;
uint32 height = 2;
double frame_rate = 3;
}
enum VideoCodec {
VP8 = 0;
H264 = 1;
AV1 = 2;
}
enum VideoRotation {
VIDEO_ROTATION_0 = 0;
VIDEO_ROTATION_90 = 1;
VIDEO_ROTATION_180 = 2;
VIDEO_ROTATION_270 = 3;
}
enum VideoFormatType {
FORMAT_ARGB = 0;
FORMAT_BGRA = 1;
FORMAT_ABGR = 2;
FORMAT_RGBA = 3;
}
enum VideoFrameBufferType {
NATIVE = 0;
I420 = 1;
I420A = 2;
I422 = 3;
I444 = 4;
I010 = 5;
NV12 = 6;
WEBGL = 7;
}
message ARGBBufferInfo {
uint64 ptr = 1;
VideoFormatType format = 2;
uint32 stride = 3;
uint32 width = 4;
uint32 height = 5;
}
message VideoFrameInfo {
int64 timestamp = 1;
VideoRotation rotation = 2;
}
message VideoFrameBufferInfo {
FFIHandleId handle = 1;
VideoFrameBufferType buffer_type = 2;
uint32 width = 3;
uint32 height = 4;
oneof buffer {
PlanarYuvBufferInfo yuv = 5;
BiplanarYuvBufferInfo bi_yuv = 6;
NativeBufferInfo native = 7;
}
}
message PlanarYuvBufferInfo {
uint32 chroma_width = 1;
uint32 chroma_height = 2;
uint32 stride_y = 3;
uint32 stride_u = 4;
uint32 stride_v = 5;
uint32 stride_a = 6;
// *const u8 or *const u16
uint64 data_y_ptr = 7;
uint64 data_u_ptr = 8;
uint64 data_v_ptr = 9;
uint64 data_a_ptr = 10; // nullptr = no alpha
}
message BiplanarYuvBufferInfo {
uint32 chroma_width = 1;
uint32 chroma_height = 2;
uint32 stride_y = 3;
uint32 stride_uv = 4;
uint64 data_y_ptr = 5;
uint64 data_uv_ptr = 6;
}
message NativeBufferInfo {
// TODO(theomonnom): Expose graphic context?
}
///
/// VideoStream ///
///
enum VideoStreamType {
VIDEO_STREAM_NATIVE = 0;
VIDEO_STREAM_WEBGL = 1;
VIDEO_STREAM_HTML = 2;
}
message VideoStreamInfo {
FFIHandleId handle = 1;
VideoStreamType type = 2;
string track_sid = 3;
}
message VideoStreamEvent {
FFIHandleId handle = 1;
oneof message { VideoFrameReceived frame_received = 2; }
}
message VideoFrameReceived {
VideoFrameInfo frame = 1;
VideoFrameBufferInfo buffer = 2;
}
///
/// VideoSource ///
///
enum VideoSourceType {
VIDEO_SOURCE_NATIVE = 0;
}
message VideoSourceInfo {
// # SAFETY
// This handle must not be dropped if a track is currently using it
FFIHandleId handle = 1;
VideoSourceType type = 2;
}
livekit-ffi/src/conversion/audio_frame.rs
+38
View File
@@ -0,0 +1,38 @@
use crate::server::audio_frame::{FfiAudioSource, FfiAudioSream};
use crate::{proto, FfiHandleId};
use livekit::webrtc::prelude::*;
impl proto::AudioFrameBufferInfo {
pub fn from(handle_id: FfiHandleId, buffer: &AudioFrame) -> Self {
Self {
handle: Some(handle_id.into()),
data_ptr: buffer.data.as_ptr() as u64,
samples_per_channel: buffer.samples_per_channel,
sample_rate: buffer.sample_rate,
num_channels: buffer.num_channels,
}
}
}
impl From<&FfiAudioSream> for proto::AudioStreamInfo {
fn from(stream: &FfiAudioSream) -> Self {
Self {
handle: Some(proto::FfiHandleId {
id: stream.handle_id() as u64,
}),
track_sid: stream.track_sid().clone().into(),
r#type: stream.stream_type() as i32,
}
}
}
impl From<&FfiAudioSource> for proto::AudioSourceInfo {
fn from(source: &FfiAudioSource) -> Self {
Self {
handle: Some(proto::FfiHandleId {
id: source.handle_id() as u64,
}),
r#type: source.source_type() as i32,
}
}
}
livekit-ffi/src/conversion/mod.rs
+21
View File
@@ -0,0 +1,21 @@
use crate::proto;
use crate::{FfiAsyncId, FfiHandleId};
pub mod audio_frame;
pub mod participant;
pub mod publication;
pub mod room;
pub mod track;
pub mod video_frame;
impl From<FfiHandleId> for proto::FfiHandleId {
fn from(id: FfiHandleId) -> Self {
Self { id: id as u64 }
}
}
impl From<FfiAsyncId> for proto::FfiAsyncId {
fn from(id: FfiAsyncId) -> Self {
Self { id: id as u64 }
}
}
livekit-ffi/src/conversion/participant.rs
+22
View File
@@ -0,0 +1,22 @@
use crate::proto;
use livekit::prelude::*;
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(),
publications: p.tracks().iter().map(|(_, p)| p.into()).collect(),
}
}
}
};
}
impl_participant_into!(&LocalParticipant);
impl_participant_into!(&RemoteParticipant);
impl_participant_into!(&Participant);
livekit-ffi/src/server/conversion/publication.rs → livekit-ffi/src/conversion/publication.rs
View File
livekit-ffi/src/conversion/room.rs
+117
View File
@@ -0,0 +1,117 @@
use crate::{proto, FfiHandleId, INVALID_HANDLE};
use livekit::options::{AudioEncoding, TrackPublishOptions, VideoEncoding};
use livekit::prelude::*;
impl proto::RoomEvent {
pub fn from(room_handle: FfiHandleId, 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_sid: publication.sid().into(),
},
)),
RoomEvent::TrackSubscribed {
track,
publication: _,
participant,
} => Some(proto::room_event::Message::TrackSubscribed(
proto::TrackSubscribed {
participant_sid: participant.sid().to_string(),
track: Some(proto::TrackInfo::from_remote_track(INVALID_HANDLE, &track)),
},
)),
RoomEvent::TrackUnsubscribed {
track,
publication: _,
participant,
} => Some(proto::room_event::Message::TrackUnsubscribed(
proto::TrackUnsubscribed {
participant_sid: participant.sid().to_string(),
track_sid: track.sid().to_string(),
},
)),
_ => None,
};
message.map(|message| proto::RoomEvent {
room_handle: Some(room_handle.into()),
message: Some(message),
})
}
}
impl proto::RoomInfo {
pub fn from_session(handle_id: FfiHandleId, session: &RoomSession) -> Self {
Self {
handle: Some(handle_id.into()),
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(),
}
}
}
impl From<proto::TrackPublishOptions> for TrackPublishOptions {
fn from(opts: proto::TrackPublishOptions) -> Self {
Self {
video_encoding: opts.video_encoding.map(Into::into),
audio_encoding: opts.audio_encoding.map(Into::into),
video_codec: proto::VideoCodec::from_i32(opts.video_codec)
.unwrap()
.into(),
dtx: opts.dtx,
red: opts.red,
simulcast: opts.simulcast,
name: opts.name,
source: proto::TrackSource::from_i32(opts.source).unwrap().into(),
}
}
}
impl From<proto::VideoEncoding> for VideoEncoding {
fn from(opts: proto::VideoEncoding) -> Self {
Self {
max_bitrate: opts.max_bitrate,
max_framerate: opts.max_framerate,
}
}
}
impl From<proto::AudioEncoding> for AudioEncoding {
fn from(opts: proto::AudioEncoding) -> Self {
Self {
max_bitrate: opts.max_bitrate,
}
}
}
livekit-ffi/src/conversion/track.rs
+114
View File
@@ -0,0 +1,114 @@
use crate::{proto, FfiHandleId};
use livekit::options::{AudioCaptureOptions, VideoCaptureOptions};
use livekit::prelude::*;
impl From<proto::VideoCaptureOptions> for VideoCaptureOptions {
fn from(opts: proto::VideoCaptureOptions) -> Self {
Self {
resolution: opts.resolution.unwrap_or_default().into(),
}
}
}
impl From<proto::AudioCaptureOptions> for AudioCaptureOptions {
fn from(opts: proto::AudioCaptureOptions) -> Self {
Self {
echo_cancellation: opts.echo_cancellation,
auto_gain_control: opts.auto_gain_control,
noise_suppression: opts.noise_suppression,
}
}
}
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 {
fn from(p: $p) -> Self {
Self {
name: p.name(),
sid: p.sid().to_string(),
kind: proto::TrackKind::from(p.kind()).into(),
source: proto::TrackSource::from(p.source()).into(),
width: p.dimension().0,
height: p.dimension().1,
mime_type: p.mime_type(),
simulcasted: p.simulcasted(),
muted: p.is_muted(),
remote: p.is_remote(),
}
}
}
};
}
impl_publication_into!(&LocalTrackPublication);
impl_publication_into!(&RemoteTrackPublication);
impl_publication_into!(&TrackPublication);
macro_rules! impl_track_into {
($fnc:ident, $t:ty) => {
impl proto::TrackInfo {
pub fn $fnc(handle_id: FfiHandleId, track: $t) -> Self {
Self {
opt_handle: Some(handle_id.into()),
name: track.name(),
stream_state: proto::StreamState::from(track.stream_state()).into(),
sid: track.sid().to_string(),
kind: proto::TrackKind::from(track.kind()).into(),
muted: track.is_muted(),
remote: track.is_remote(),
}
}
}
};
}
impl_track_into!(from_local_audio_track, &LocalAudioTrack);
impl_track_into!(from_local_video_track, &LocalVideoTrack);
impl_track_into!(from_remote_audio_track, &RemoteAudioTrack);
impl_track_into!(from_remote_video_track, &RemoteVideoTrack);
impl_track_into!(from_track, &Track);
impl_track_into!(from_local_track, &LocalTrack);
impl_track_into!(from_remote_track, &RemoteTrack);
impl From<TrackKind> for proto::TrackKind {
fn from(kind: TrackKind) -> Self {
match kind {
TrackKind::Audio => proto::TrackKind::KindAudio,
TrackKind::Video => proto::TrackKind::KindVideo,
}
}
}
impl From<StreamState> for proto::StreamState {
fn from(state: StreamState) -> Self {
match state {
StreamState::Active => Self::StateActive,
StreamState::Paused => Self::StatePaused,
}
}
}
impl From<proto::TrackSource> for TrackSource {
fn from(source: proto::TrackSource) -> Self {
match source {
proto::TrackSource::SourceUnknown => TrackSource::Unknown,
proto::TrackSource::SourceCamera => TrackSource::Camera,
proto::TrackSource::SourceMicrophone => TrackSource::Microphone,
proto::TrackSource::SourceScreenshare => TrackSource::Screenshare,
proto::TrackSource::SourceScreenshareAudio => TrackSource::ScreenshareAudio,
}
}
}
livekit-ffi/src/server/conversion/video_frame.rs → livekit-ffi/src/conversion/video_frame.rs
+106 -24
View File
@@ -1,5 +1,7 @@
use crate::proto;
use crate::server::FFIHandleId;
use crate::server::video_frame::{FfiVideoSource, FfiVideoStream};
use crate::FfiHandleId;
use livekit::options::{VideoCodec, VideoResolution};
use livekit::webrtc::prelude::*;
use livekit::webrtc::video_frame;
@@ -8,9 +10,9 @@ macro_rules! impl_yuv_into {
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(),
stride_y: $buffer.strides().0,
stride_u: $buffer.strides().1,
stride_v: $buffer.strides().2,
data_y_ptr: $data_y.as_ptr() as u64,
data_u_ptr: $data_u.as_ptr() as u64,
data_v_ptr: $data_v.as_ptr() as u64,
@@ -22,7 +24,7 @@ macro_rules! impl_yuv_into {
fn from(buffer: $buffer) -> Self {
let (data_y, data_u, data_v, data_a) = buffer.data();
let mut proto = impl_yuv_into!(@fields, buffer, data_y, data_u, data_v);
proto.stride_a = buffer.stride_a();
proto.stride_a = buffer.strides().3;
proto.data_a_ptr = data_a.map(|data_a| data_a.as_ptr() as u64).unwrap_or(0);
proto
}
@@ -42,12 +44,13 @@ macro_rules! impl_biyuv_into {
($b:ty) => {
impl From<$b> for proto::BiplanarYuvBufferInfo {
fn from(buffer: $b) -> Self {
let (stride_y, stride_uv) = buffer.strides();
let (data_y, data_uv) = buffer.data();
Self {
chroma_width: buffer.chroma_width(),
chroma_height: buffer.chroma_height(),
stride_y: buffer.stride_y(),
stride_uv: buffer.stride_uv(),
stride_y: stride_y,
stride_uv: stride_uv,
data_y_ptr: data_y.as_ptr() as u64,
data_uv_ptr: data_uv.as_ptr() as u64,
}
@@ -66,7 +69,7 @@ impl_biyuv_into!(&NV12Buffer);
impl proto::VideoFrameInfo {
pub fn from<T>(frame: &VideoFrame<T>) -> Self
where
T: VideoFrameBuffer,
T: AsRef<dyn VideoFrameBuffer>,
{
Self {
timestamp: frame.timestamp,
@@ -76,22 +79,36 @@ impl proto::VideoFrameInfo {
}
impl proto::VideoFrameBufferInfo {
pub fn from(handle: FFIHandleId, buffer: &dyn VideoFrameBuffer) -> Self {
match &buffer.buffer_type() {
pub fn from(handle: FfiHandleId, buffer: impl AsRef<dyn VideoFrameBuffer>) -> Self {
match &buffer.as_ref().buffer_type() {
#[cfg(not(target_arch = "wasm32"))]
VideoFrameBufferType::Native => Self::from_native(handle, buffer.as_native().unwrap()),
VideoFrameBufferType::I420 => Self::from_i420(handle, buffer.as_i420().unwrap()),
VideoFrameBufferType::I420A => Self::from_i420a(handle, buffer.as_i420a().unwrap()),
VideoFrameBufferType::I422 => Self::from_i422(handle, buffer.as_i422().unwrap()),
VideoFrameBufferType::I444 => Self::from_i444(handle, buffer.as_i444().unwrap()),
VideoFrameBufferType::I010 => Self::from_i010(handle, buffer.as_i010().unwrap()),
VideoFrameBufferType::NV12 => Self::from_nv12(handle, buffer.as_nv12().unwrap()),
VideoFrameBufferType::Native => {
Self::from_native(handle, buffer.as_ref().as_native().unwrap())
}
VideoFrameBufferType::I420 => {
Self::from_i420(handle, buffer.as_ref().as_i420().unwrap())
}
VideoFrameBufferType::I420A => {
Self::from_i420a(handle, buffer.as_ref().as_i420a().unwrap())
}
VideoFrameBufferType::I422 => {
Self::from_i422(handle, buffer.as_ref().as_i422().unwrap())
}
VideoFrameBufferType::I444 => {
Self::from_i444(handle, buffer.as_ref().as_i444().unwrap())
}
VideoFrameBufferType::I010 => {
Self::from_i010(handle, buffer.as_ref().as_i010().unwrap())
}
VideoFrameBufferType::NV12 => {
Self::from_nv12(handle, buffer.as_ref().as_nv12().unwrap())
}
_ => panic!("unsupported buffer type on this platform"),
}
}
#[cfg(not(target_arch = "wasm32"))]
pub fn from_native(handle_id: FFIHandleId, buffer: &video_frame::native::NativeBuffer) -> Self {
pub fn from_native(handle_id: FfiHandleId, buffer: &video_frame::native::NativeBuffer) -> Self {
Self {
handle: Some(handle_id.into()),
buffer_type: proto::VideoFrameBufferType::Native.into(),
@@ -103,7 +120,7 @@ impl proto::VideoFrameBufferInfo {
}
}
pub fn from_i420(handle_id: FFIHandleId, buffer: &I420Buffer) -> Self {
pub fn from_i420(handle_id: FfiHandleId, buffer: &I420Buffer) -> Self {
Self {
handle: Some(handle_id.into()),
buffer_type: proto::VideoFrameBufferType::I420.into(),
@@ -113,7 +130,7 @@ impl proto::VideoFrameBufferInfo {
}
}
pub fn from_i420a(handle_id: FFIHandleId, buffer: &I420ABuffer) -> Self {
pub fn from_i420a(handle_id: FfiHandleId, buffer: &I420ABuffer) -> Self {
Self {
handle: Some(handle_id.into()),
buffer_type: proto::VideoFrameBufferType::I420a.into(),
@@ -123,7 +140,7 @@ impl proto::VideoFrameBufferInfo {
}
}
pub fn from_i422(handle_id: FFIHandleId, buffer: &I422Buffer) -> Self {
pub fn from_i422(handle_id: FfiHandleId, buffer: &I422Buffer) -> Self {
Self {
handle: Some(handle_id.into()),
buffer_type: proto::VideoFrameBufferType::I422.into(),
@@ -133,7 +150,7 @@ impl proto::VideoFrameBufferInfo {
}
}
pub fn from_i444(handle_id: FFIHandleId, buffer: &I444Buffer) -> Self {
pub fn from_i444(handle_id: FfiHandleId, buffer: &I444Buffer) -> Self {
Self {
handle: Some(handle_id.into()),
buffer_type: proto::VideoFrameBufferType::I444.into(),
@@ -143,7 +160,7 @@ impl proto::VideoFrameBufferInfo {
}
}
pub fn from_i010(handle_id: FFIHandleId, buffer: &I010Buffer) -> Self {
pub fn from_i010(handle_id: FfiHandleId, buffer: &I010Buffer) -> Self {
Self {
handle: Some(handle_id.into()),
buffer_type: proto::VideoFrameBufferType::I010.into(),
@@ -153,7 +170,7 @@ impl proto::VideoFrameBufferInfo {
}
}
pub fn from_nv12(handle_id: FFIHandleId, buffer: &NV12Buffer) -> Self {
pub fn from_nv12(handle_id: FfiHandleId, buffer: &NV12Buffer) -> Self {
Self {
handle: Some(handle_id.into()),
buffer_type: proto::VideoFrameBufferType::Nv12.into(),
@@ -186,6 +203,17 @@ impl From<VideoRotation> for proto::VideoRotation {
}
}
impl From<proto::VideoRotation> for VideoRotation {
fn from(rotation: proto::VideoRotation) -> VideoRotation {
match rotation {
proto::VideoRotation::VideoRotation0 => Self::VideoRotation0,
proto::VideoRotation::VideoRotation90 => Self::VideoRotation90,
proto::VideoRotation::VideoRotation180 => Self::VideoRotation180,
proto::VideoRotation::VideoRotation270 => Self::VideoRotation270,
}
}
}
impl From<VideoFrameBufferType> for proto::VideoFrameBufferType {
fn from(buffer_type: VideoFrameBufferType) -> Self {
match buffer_type {
@@ -201,3 +229,57 @@ impl From<VideoFrameBufferType> for proto::VideoFrameBufferType {
}
}
}
impl From<&FfiVideoStream> for proto::VideoStreamInfo {
fn from(stream: &FfiVideoStream) -> Self {
Self {
handle: Some(proto::FfiHandleId {
id: stream.handle_id() as u64,
}),
track_sid: stream.track_sid().clone().into(),
r#type: stream.stream_type() as i32,
}
}
}
impl From<&FfiVideoSource> for proto::VideoSourceInfo {
fn from(source: &FfiVideoSource) -> Self {
Self {
handle: Some(proto::FfiHandleId {
id: source.handle_id() as u64,
}),
r#type: source.source_type() as i32,
}
}
}
impl From<VideoResolution> for proto::VideoResolution {
fn from(resolution: VideoResolution) -> Self {
Self {
width: resolution.width,
height: resolution.height,
frame_rate: resolution.frame_rate,
}
}
}
impl From<proto::VideoResolution> for VideoResolution {
fn from(resolution: proto::VideoResolution) -> Self {
Self {
width: resolution.width,
height: resolution.height,
frame_rate: resolution.frame_rate,
aspect_ratio: resolution.width as f32 / resolution.height as f32,
}
}
}
impl From<proto::VideoCodec> for VideoCodec {
fn from(codec: proto::VideoCodec) -> Self {
match codec {
proto::VideoCodec::Vp8 => Self::VP8,
proto::VideoCodec::H264 => Self::H264,
proto::VideoCodec::Av1 => Self::AV1,
}
}
}
livekit-ffi/src/lib.rs
+74 -1
View File
@@ -1,5 +1,78 @@
use livekit::prelude::*;
use prost::Message;
use std::any::Any;
use thiserror::Error;
mod proto {
include!(concat!(env!("OUT_DIR"), "/livekit.rs"));
}
mod conversion;
mod server;
#[derive(Error, Debug)]
pub enum FfiError {
#[error("the server is not configured")]
NotConfigured,
#[error("the server is already initialized")]
AlreadyInitialized,
#[error("room error {0}")]
Room(#[from] RoomError),
#[error("invalid request: {0}")]
InvalidRequest(&'static str),
}
/// # SAFTEY: The "C" callback must be threadsafe and not block
pub type FfiCallbackFn = unsafe extern "C" fn(*const u8, usize);
pub type FfiResult<T> = Result<T, FfiError>;
pub type FfiAsyncId = usize;
pub type FfiHandleId = usize;
pub type FfiHandle = Box<dyn Any + Send + Sync>;
pub const INVALID_HANDLE: FfiHandleId = 0;
#[no_mangle]
pub(crate) extern "C" fn livekit_ffi_request(
data: *const u8,
len: usize,
res_ptr: *mut *const u8,
res_len: *mut usize,
) -> FfiHandleId {
let data = unsafe { std::slice::from_raw_parts(data, len) };
let res = match proto::FfiRequest::decode(data) {
Ok(res) => res,
Err(err) => {
eprintln!("failed to decode request: {}", err);
return INVALID_HANDLE;
}
};
let res = match server::FFI_SERVER.handle_request(res) {
Ok(res) => res,
Err(err) => {
eprintln!("failed to handle request: {}", err);
return INVALID_HANDLE;
}
}
.encode_to_vec();
unsafe {
*res_ptr = res.as_ptr();
*res_len = res.len();
}
let handle_id = server::FFI_SERVER.next_id();
server::FFI_SERVER
.ffi_handles()
.insert(handle_id, Box::new(res));
handle_id
}
#[no_mangle]
pub(crate) extern "C" fn livekit_ffi_drop_handle(handle_id: FfiHandleId) -> bool {
// Free the memory
server::FFI_SERVER
.ffi_handles()
.remove(&handle_id)
.is_some()
}
livekit-ffi/src/server/audio_frame.rs
+217
View File
@@ -0,0 +1,217 @@
use crate::{proto, server, FfiError, FfiHandleId, FfiResult};
use futures_util::StreamExt;
use livekit::prelude::*;
use livekit::webrtc::audio_frame::AudioFrame;
use livekit::webrtc::audio_source::native::NativeAudioSource;
use livekit::webrtc::audio_stream::native::NativeAudioStream;
use livekit::webrtc::media_stream::MediaStreamTrack;
use log::warn;
use server::utils;
use tokio::sync::oneshot;
// ===== FFIAudioStream =====
pub struct FfiAudioSream {
handle_id: FfiHandleId,
stream_type: proto::AudioStreamType,
track_sid: TrackSid,
#[allow(dead_code)]
close_tx: oneshot::Sender<()>, // Close the stream on drop
}
impl FfiAudioSream {
/// Setup a new AudioStream and forward the audio data to the client/the foreign
/// language.
///
/// When FFIAudioStream is dropped (When the corresponding handle_id is dropped), the task
/// is being closed.
///
/// It is possible that the client receives an AudioFrame after the task is closed. The client
/// musts ignore it.
pub fn setup(
server: &'static server::FfiServer,
new_stream: proto::NewAudioStreamRequest,
) -> FfiResult<proto::AudioStreamInfo> {
let (close_tx, close_rx) = oneshot::channel();
let stream_type = proto::AudioStreamType::from_i32(new_stream.r#type).unwrap();
let track_sid: TrackSid = new_stream.track_sid.into();
let room_handle = new_stream
.room_handle
.ok_or(FfiError::InvalidRequest("room_handle is empty"))?
.id as FfiHandleId;
let track = utils::find_remote_track(
server,
&track_sid,
&new_stream.participant_sid.into(),
room_handle,
)?
.rtc_track();
let MediaStreamTrack::Audio(track) = track else {
return Err(FfiError::InvalidRequest("not an audio track"));
};
let audio_stream = match stream_type {
proto::AudioStreamType::AudioStreamNative => {
let audio_stream = Self {
handle_id: server.next_id(),
stream_type,
close_tx,
track_sid,
};
tokio::spawn(Self::native_audio_stream_task(
server,
audio_stream.handle_id,
NativeAudioStream::new(track),
close_rx,
));
Ok::<FfiAudioSream, FfiError>(audio_stream)
}
// TODO(theomonnom): Support other stream types
_ => return Err(FfiError::InvalidRequest("unsupported audio stream type")),
}?;
// Store the new audio stream and return the info
let info = proto::AudioStreamInfo::from(&audio_stream);
server
.ffi_handles()
.insert(audio_stream.handle_id, Box::new(audio_stream));
Ok(info)
}
pub fn handle_id(&self) -> FfiHandleId {
self.handle_id
}
pub fn stream_type(&self) -> proto::AudioStreamType {
self.stream_type
}
pub fn track_sid(&self) -> &TrackSid {
&self.track_sid
}
async fn native_audio_stream_task(
server: &'static server::FfiServer,
stream_handle_id: FfiHandleId,
mut native_stream: NativeAudioStream,
mut close_rx: oneshot::Receiver<()>,
) {
loop {
tokio::select! {
_ = &mut close_rx => {
break;
}
frame = native_stream.next() => {
let Some(frame) = frame else {
break;
};
let handle_id = server.next_id();
let buffer_info = proto::AudioFrameBufferInfo::from(handle_id, &frame);
server.ffi_handles().insert(handle_id, Box::new(frame));
if let Err(err) = server.send_event(proto::ffi_event::Message::AudioStreamEvent(
proto::AudioStreamEvent {
handle: Some(stream_handle_id.into()),
message: Some(proto::audio_stream_event::Message::FrameReceived(
proto::AudioFrameReceived {
frame: Some(buffer_info),
},
)),
},
)) {
warn!("failed to send audio frame: {}", err);
}
}
}
}
}
}
// ===== FFIAudioSource =====
pub struct FfiAudioSource {
handle_id: FfiHandleId,
source_type: proto::AudioSourceType,
source: AudioSource,
}
#[derive(Clone)]
pub enum AudioSource {
Native(NativeAudioSource),
}
impl FfiAudioSource {
pub fn setup(
server: &'static server::FfiServer,
new_source: proto::NewAudioSourceRequest,
) -> FfiResult<proto::AudioSourceInfo> {
let source_type = proto::AudioSourceType::from_i32(new_source.r#type).unwrap();
let source_inner = match source_type {
proto::AudioSourceType::AudioSourceNative => {
let audio_source = NativeAudioSource::default();
Ok::<AudioSource, FfiError>(AudioSource::Native(audio_source))
}
_ => return Err(FfiError::InvalidRequest("unsupported audio source type")),
}?;
let audio_source = Self {
handle_id: server.next_id(),
source_type,
source: source_inner,
};
let source_info = proto::AudioSourceInfo::from(&audio_source);
server
.ffi_handles()
.insert(audio_source.handle_id, Box::new(audio_source));
Ok(source_info)
}
pub fn capture_frame(
&self,
server: &'static server::FfiServer,
capture: proto::CaptureAudioFrameRequest,
) -> FfiResult<()> {
match self.source {
AudioSource::Native(ref source) => {
let buffer_handle = capture
.buffer_handle
.ok_or(FfiError::InvalidRequest("buffer_handle is empty"))?
.id as FfiHandleId;
let frame = server
.ffi_handles()
.get(&buffer_handle)
.ok_or(FfiError::InvalidRequest("handle not found"))?;
let frame = frame
.downcast_ref::<AudioFrame>()
.ok_or(FfiError::InvalidRequest("handle is not an audio frame"))?;
source.capture_frame(frame);
}
}
Ok(())
}
pub fn handle_id(&self) -> FfiHandleId {
self.handle_id
}
pub fn source_type(&self) -> proto::AudioSourceType {
self.source_type
}
pub fn inner_source(&self) -> &AudioSource {
&self.source
}
}
livekit-ffi/src/server/conversion/mod.rs
-196
View File
@@ -1,196 +0,0 @@
use crate::server::FFIHandleId;
use livekit::prelude::*;
use crate::proto;
pub mod participant;
pub mod publication;
pub mod room;
pub mod video_frame;
impl From<FFIHandleId> for proto::FfiHandleId {
fn from(id: FFIHandleId) -> Self {
Self { id: id as u64 }
}
}
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(),
publications: p.tracks().iter().map(|(_, p)| p.into()).collect(),
}
}
}
};
}
impl_participant_into!(&LocalParticipant);
impl_participant_into!(&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 {
fn from(p: $p) -> Self {
Self {
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(),
}
}
}
};
}
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(),
stream_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!(&Track);
impl_track_into!(&LocalTrack);
impl_track_into!(&RemoteTrack);
impl From<TrackKind> for proto::TrackKind {
fn from(kind: TrackKind) -> Self {
match kind {
TrackKind::Audio => proto::TrackKind::KindAudio,
TrackKind::Video => proto::TrackKind::KindVideo,
}
}
}
impl From<StreamState> for proto::StreamState {
fn from(state: StreamState) -> Self {
match state {
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_sid: publication.sid().into(),
},
)),
RoomEvent::TrackSubscribed {
track,
publication: _,
participant,
} => Some(proto::room_event::Message::TrackSubscribed(
proto::TrackSubscribed {
participant_sid: participant.sid().to_string(),
track: Some((&track).into()),
sink: Some(proto::VideoSinkInfo {
track_sid: track.sid().to_string(),
}),
},
)),
RoomEvent::TrackUnsubscribed {
track,
publication: _,
participant,
} => Some(proto::room_event::Message::TrackUnsubscribed(
proto::TrackUnsubscribed {
participant_sid: participant.sid().to_string(),
track_sid: track.sid().to_string(),
},
)),
_ => None,
};
message.map(|message| proto::RoomEvent {
room_sid: room_sid.into(),
message: Some(message),
})
}
}
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(),
}
}
}
livekit-ffi/src/server/conversion/participant.rs
View File
livekit-ffi/src/server/conversion/room.rs
View File
livekit-ffi/src/server/mod.rs
+626 -221
View File
@@ -1,272 +1,677 @@
use crate::{proto, FfiCallbackFn};
use crate::{FfiAsyncId, FfiError, FfiHandle, FfiHandleId, FfiResult};
use dashmap::DashMap;
use lazy_static::lazy_static;
use livekit::prelude::*;
use livekit::webrtc::video_frame::{native::VideoFrameBufferExt, BoxVideoFrame, VideoFrameBuffer};
use crate::proto;
use parking_lot::{Mutex, RwLock};
use livekit::webrtc::native::yuv_helper;
use livekit::webrtc::prelude::*;
use livekit::webrtc::video_frame::{native::I420BufferExt, BoxVideoFrameBuffer, I420Buffer};
use parking_lot::Mutex;
use prost::Message;
use std::any::Any;
use std::collections::HashMap;
use std::panic;
use std::slice;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::{AtomicBool, Ordering};
use thiserror::Error;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use std::sync::atomic::{AtomicUsize, Ordering};
mod conversion;
mod room;
pub mod audio_frame;
pub mod room;
pub mod utils;
pub mod video_frame;
#[derive(Error, Debug)]
pub enum FFIError {
#[error("the FFIServer isn't configured")]
NotConfigured,
#[error("failed to execute the FFICallback")]
CallbackFailed,
}
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
#[cfg(test)]
mod tests;
lazy_static! {
static ref FFI_SERVER: FFIServer = FFIServer::default();
pub static ref FFI_SERVER: FfiServer = FfiServer::default();
}
pub struct FFIConfig {
callback_fn: CallbackFn,
pub struct FfiConfig {
callback_fn: FfiCallbackFn,
}
/// 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
//
// NOTE: For VideoBuffers, we always store the enum VideoFrameBuffer
ffi_owned: RwLock<HashMap<FFIHandleId, FFIHandle>>,
next_handle_id: AtomicU64, // FFIHandleId
next_async_id: AtomicU64,
rooms: RwLock<HashMap<RoomSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
pub struct FfiServer {
rooms: Mutex<HashMap<RoomSid, FfiHandleId>>,
/// Store all FFI handles inside an HashMap, if this isn't efficient enough
/// We can still use Box::into_raw & Box::from_raw in the future (but keep it safe for now)
ffi_handles: DashMap<FfiHandleId, FfiHandle>,
next_id: AtomicUsize,
async_runtime: tokio::runtime::Runtime,
initialized: AtomicBool,
config: Mutex<Option<FFIConfig>>,
config: Mutex<Option<FfiConfig>>,
}
impl Default for FFIServer {
impl Default for FfiServer {
fn default() -> Self {
Self {
ffi_owned: RwLock::new(HashMap::new()),
next_handle_id: AtomicU64::new(1), // 0 is considered invalid
next_async_id: AtomicU64::new(1),
rooms: RwLock::new(HashMap::new()),
rooms: Default::default(),
ffi_handles: Default::default(),
next_id: AtomicUsize::new(1), // 0 is invalid
async_runtime: tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap(),
initialized: Default::default(),
config: Default::default(),
}
}
}
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) {
// Using &'static self inside the implementation, not sure if this is really idiomatic
// It simplifies the code a lot tho. In most cases the server is used until the end of the process
impl FfiServer {
pub async fn dispose(&'static self) {
// Close all rooms
for (_, (handle, shutdown_tx)) in self.rooms.write().drain() {
let _ = shutdown_tx.send(());
let _ = handle.await;
for (_, room_handle) in self.rooms.lock().drain() {
let room = self.ffi_handles.remove(&room_handle);
if let Some(room) = room {
let ffi_room = room.1.downcast::<room::FfiRoom>().unwrap();
ffi_room.close().await;
}
}
// Drop all handles
self.ffi_handles.clear();
// Invalidate the config
*self.config.lock() = None;
}
pub fn add_room(&self, sid: RoomSid, handle: (JoinHandle<()>, oneshot::Sender<()>)) {
self.rooms.write().insert(sid, handle);
pub fn next_id(&'static self) -> usize {
self.next_id.fetch_add(1, Ordering::Relaxed)
}
pub fn initialized(&self) -> bool {
self.initialized.load(Ordering::SeqCst)
pub fn ffi_handles(&'static self) -> &DashMap<FfiHandleId, FfiHandle> {
&self.ffi_handles
}
pub fn next_handle_id(&self) -> FFIHandleId {
self.next_handle_id.fetch_add(1, Ordering::SeqCst) as FFIHandleId
pub fn rooms(&'static self) -> &Mutex<HashMap<RoomSid, FfiHandleId>> {
&self.rooms
}
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) {
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_event(
&self,
message: proto::ffi_event::Message,
async_id: Option<u64>,
) -> Result<(), FFIError> {
let config = self.config.lock();
if !self.initialized() {
Err(FFIError::NotConfigured)?
}
pub fn send_event(&'static self, message: proto::ffi_event::Message) -> FfiResult<()> {
let callback_fn = self
.config
.lock()
.as_ref()
.map_or_else(|| Err(FfiError::NotConfigured), |c| Ok(c.callback_fn))?;
let message = proto::FfiEvent {
async_id,
message: Some(message),
}
.encode_to_vec();
let config = config.as_ref().unwrap();
if let Err(err) = panic::catch_unwind(|| unsafe {
(config.callback_fn)(message.as_ptr(), message.len());
}) {
eprintln!("panic when sending ffi event: {:?}", err);
Err(FFIError::CallbackFailed)?
unsafe {
callback_fn(message.as_ptr(), message.len());
}
Ok(())
}
}
pub fn handle_request(&self, message: proto::ffi_request::Message) -> proto::FfiResponse {
match message {
proto::ffi_request::Message::AsyncConnect(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);
if let Some(buffer) = buffer {
if let Ok(buffer) = buffer.downcast::<Box<dyn VideoFrameBuffer>>() {
let handle_id = self.next_handle_id();
let buffer = buffer.to_i420();
buffer_info = Some(proto::VideoFrameBufferInfo::from(handle_id, &buffer));
self.insert_handle(handle_id, Box::new(buffer));
}
}
return proto::FfiResponse {
message: Some(proto::ffi_response::Message::ToI420(
proto::ToI420Response {
new_buffer: buffer_info,
},
)),
..Default::default()
};
}
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::<Box<dyn 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);
}
}
}
}
_ => {}
impl FfiServer {
fn on_initialize(
&'static self,
init: proto::InitializeRequest,
) -> FfiResult<proto::InitializeResponse> {
if self.config.lock().is_some() {
return Err(FfiError::AlreadyInitialized);
}
proto::FfiResponse::default()
// # SAFETY: The foreign language is responsible for ensuring that the callback function is valid
unsafe {
*self.config.lock() = Some(FfiConfig {
callback_fn: std::mem::transmute(init.event_callback_ptr),
});
}
Ok(proto::InitializeResponse::default())
}
fn on_dispose(
&'static self,
dispose: proto::DisposeRequest,
) -> FfiResult<proto::DisposeResponse> {
*self.config.lock() = None;
let close = self.dispose();
if !dispose.r#async {
self.async_runtime.block_on(close);
Ok(proto::DisposeResponse::default())
} else {
let async_id = self.next_id();
self.async_runtime.spawn(async move {
close.await;
});
Ok(proto::DisposeResponse {
async_id: Some(proto::FfiAsyncId {
id: async_id as u64,
}),
})
}
}
// Room
fn on_connect(
&'static self,
connect: proto::ConnectRequest,
) -> FfiResult<proto::ConnectResponse> {
let async_id = self.next_id();
self.async_runtime.spawn(async move {
// Try to connect to the Room
let res = room::FfiRoom::connect(&self, connect).await;
// match res
match res {
Ok(room_info) => {
let _ = self.send_event(proto::ffi_event::Message::Connect(
proto::ConnectCallback {
async_id: Some(async_id.into()),
error: None,
room: Some(room_info),
},
));
}
Err(err) => {
let _ = self.send_event(proto::ffi_event::Message::Connect(
proto::ConnectCallback {
async_id: Some(async_id.into()),
error: Some(err.to_string()),
room: None,
},
));
}
}
});
Ok(proto::ConnectResponse {
async_id: Some(proto::FfiAsyncId {
id: async_id as u64,
}),
})
}
fn on_disconnect(
&'static self,
_disconnect: proto::DisconnectRequest,
) -> FfiResult<proto::DisconnectResponse> {
Ok(proto::DisconnectResponse::default())
}
fn on_publish_track(
&'static self,
publish: proto::PublishTrackRequest,
) -> FfiResult<proto::PublishTrackResponse> {
let async_id = self.next_id() as FfiAsyncId;
tokio::spawn(async move {
let res = async {
let room_handle = publish
.room_handle
.as_ref()
.ok_or(FfiError::InvalidRequest("room_handle is empty"))?
.id as FfiHandleId;
let room = self
.ffi_handles
.get(&room_handle)
.ok_or(FfiError::InvalidRequest("room not found"))?;
let room = room
.downcast_ref::<room::FfiRoom>()
.ok_or(FfiError::InvalidRequest("room is not a FfiRoom"))?;
let track_handle = publish
.track_handle
.as_ref()
.ok_or(FfiError::InvalidRequest("track_handle is empty"))?
.id as FfiHandleId;
let track = self
.ffi_handles
.get(&track_handle)
.ok_or(FfiError::InvalidRequest("track not found"))?;
let track = track
.downcast_ref::<LocalTrack>()
.ok_or(FfiError::InvalidRequest("track is not a LocalTrack"))?;
let publication = room
.session()
.local_participant()
.publish_track(
track.clone(),
publish.options.map(Into::into).unwrap_or_default(),
)
.await?;
Ok::<LocalTrackPublication, FfiError>(publication)
}
.await;
if let Err(err) = self.send_event(proto::ffi_event::Message::PublishTrack(
proto::PublishTrackCallback {
async_id: Some(async_id.into()),
error: res.as_ref().err().map(|e| e.to_string()),
publication: res.as_ref().ok().map(Into::into),
},
)) {
log::warn!("error sending PublishTrack callback: {}", err);
}
});
Ok(proto::PublishTrackResponse {
async_id: Some(async_id.into()),
})
}
fn on_unpublish_track(
&'static self,
_unpublish: proto::UnpublishTrackRequest,
) -> FfiResult<proto::UnpublishTrackResponse> {
Ok(proto::UnpublishTrackResponse::default())
}
// Track
fn on_create_video_track(
&'static self,
create: proto::CreateVideoTrackRequest,
) -> FfiResult<proto::CreateVideoTrackResponse> {
let handle_id = create
.source_handle
.as_ref()
.ok_or(FfiError::InvalidRequest("source_handle is empty"))?
.id as FfiHandleId;
let source = self
.ffi_handles
.get(&handle_id)
.ok_or(FfiError::InvalidRequest("source not found"))?;
let source = source
.downcast_ref::<video_frame::FfiVideoSource>()
.ok_or(FfiError::InvalidRequest("handle is not a video source"))?;
let source = source.inner_source().clone();
let video_track = match source {
video_frame::VideoSource::Native(native_source) => LocalVideoTrack::create_video_track(
&create.name,
create.options.unwrap_or_default().into(),
native_source,
),
};
let handle_id = self.next_id() as FfiHandleId;
let track_info = proto::TrackInfo::from_local_video_track(handle_id, &video_track);
self.ffi_handles
.insert(handle_id, Box::new(LocalTrack::Video(video_track)));
Ok(proto::CreateVideoTrackResponse {
track: Some(track_info),
})
}
fn on_create_audio_track(
&'static self,
create: proto::CreateAudioTrackRequest,
) -> FfiResult<proto::CreateAudioTrackResponse> {
let handle_id = create
.source_handle
.as_ref()
.ok_or(FfiError::InvalidRequest("source_handle is empty"))?
.id as FfiHandleId;
let source = self
.ffi_handles
.get(&handle_id)
.ok_or(FfiError::InvalidRequest("source not found"))?;
let source = source
.downcast_ref::<audio_frame::FfiAudioSource>()
.ok_or(FfiError::InvalidRequest("handle is not an audio source"))?;
let source = source.inner_source().clone();
let audio_track = match source {
audio_frame::AudioSource::Native(native_source) => LocalAudioTrack::create_audio_track(
&create.name,
create.options.unwrap_or_default().into(),
native_source,
),
};
let handle_id = self.next_id() as FfiHandleId;
let track_info = proto::TrackInfo::from_local_audio_track(handle_id, &audio_track);
self.ffi_handles
.insert(handle_id, Box::new(LocalTrack::Audio(audio_track)));
Ok(proto::CreateAudioTrackResponse {
track: Some(track_info),
})
}
// Video
fn on_alloc_video_buffer(
&'static self,
alloc: proto::AllocVideoBufferRequest,
) -> FfiResult<proto::AllocVideoBufferResponse> {
let frame_type = proto::VideoFrameBufferType::from_i32(alloc.r#type).unwrap();
let buffer: BoxVideoFrameBuffer = match frame_type {
proto::VideoFrameBufferType::I420 => {
Box::new(I420Buffer::new(alloc.width, alloc.height))
}
_ => return Err(FfiError::InvalidRequest("frame type is not supported")),
};
let handle_id = self.next_id();
let buffer_info = proto::VideoFrameBufferInfo::from(handle_id, &buffer);
self.ffi_handles.insert(handle_id, Box::new(buffer));
Ok(proto::AllocVideoBufferResponse {
buffer: Some(buffer_info),
})
}
fn on_new_video_stream(
&'static self,
new_stream: proto::NewVideoStreamRequest,
) -> FfiResult<proto::NewVideoStreamResponse> {
let stream_info = video_frame::FfiVideoStream::setup(&self, new_stream)?;
Ok(proto::NewVideoStreamResponse {
stream: Some(stream_info),
})
}
fn on_new_video_source(
&'static self,
new_source: proto::NewVideoSourceRequest,
) -> FfiResult<proto::NewVideoSourceResponse> {
let source_info = video_frame::FfiVideoSource::setup(&self, new_source)?;
Ok(proto::NewVideoSourceResponse {
source: Some(source_info),
})
}
fn on_capture_video_frame(
&'static self,
push: proto::CaptureVideoFrameRequest,
) -> FfiResult<proto::CaptureVideoFrameResponse> {
let handle_id = push
.source_handle
.as_ref()
.ok_or(FfiError::InvalidRequest("source_handle is empty"))?
.id as FfiHandleId;
let video_source = self
.ffi_handles
.get(&handle_id)
.ok_or(FfiError::InvalidRequest("source not found"))?;
let video_source = video_source
.downcast_ref::<video_frame::FfiVideoSource>()
.ok_or(FfiError::InvalidRequest("handle is not a video source"))?;
video_source.capture_frame(self, push)?;
Ok(proto::CaptureVideoFrameResponse::default())
}
fn on_to_i420(
&'static self,
to_i420: proto::ToI420Request,
) -> FfiResult<proto::ToI420Response> {
let from = to_i420
.from
.ok_or(FfiError::InvalidRequest("from is empty"))?;
let i420 = match from {
proto::to_i420_request::From::Argb(argb_info) => {
let mut i420 = I420Buffer::new(argb_info.width, argb_info.height);
let argb_format = proto::VideoFormatType::from_i32(argb_info.format).unwrap();
let argb_ptr = argb_info.ptr as *const u8;
let argb_len = (argb_info.stride * argb_info.height) as usize;
let argb = unsafe { slice::from_raw_parts(argb_ptr, argb_len) };
let argb_stride = argb_info.stride;
let (stride_y, stride_u, stride_v) = i420.strides();
let (data_y, data_u, data_v) = i420.data_mut();
let width = argb_info.width as i32;
let mut height = argb_info.height as i32;
if to_i420.flip_y {
height = -height;
}
match argb_format {
proto::VideoFormatType::FormatArgb => {
yuv_helper::argb_to_i420(
argb,
argb_stride,
data_y,
stride_y,
data_u,
stride_u,
data_v,
stride_v,
width,
height,
)
.unwrap();
}
proto::VideoFormatType::FormatAbgr => {
yuv_helper::abgr_to_i420(
argb,
argb_stride,
data_y,
stride_y,
data_u,
stride_u,
data_v,
stride_v,
width,
height,
)
.unwrap();
}
_ => return Err(FfiError::InvalidRequest("the format is not supported")),
}
i420
}
proto::to_i420_request::From::Buffer(handle) => {
let handle_id = handle.id as FfiHandleId;
let buffer = self
.ffi_handles
.get(&handle_id)
.ok_or(FfiError::InvalidRequest("handle not found"))?;
let i420 = buffer
.downcast_ref::<BoxVideoFrameBuffer>()
.ok_or(FfiError::InvalidRequest("handle is not a video buffer"))?
.to_i420();
i420
}
};
let i420: BoxVideoFrameBuffer = Box::new(i420);
let handle_id = self.next_id() as FfiHandleId;
let buffer_info = proto::VideoFrameBufferInfo::from(handle_id, &i420);
self.ffi_handles.insert(handle_id, Box::new(i420));
Ok(proto::ToI420Response {
buffer: Some(buffer_info),
})
}
fn on_to_argb(
&'static self,
to_argb: proto::ToArgbRequest,
) -> FfiResult<proto::ToArgbResponse> {
let handle_id = to_argb
.buffer
.ok_or(FfiError::InvalidRequest("buffer is empty"))?
.id as FfiHandleId;
let buffer = self
.ffi_handles
.get(&handle_id)
.ok_or(FfiError::InvalidRequest("buffer is not found"))?;
let buffer = buffer
.downcast_ref::<BoxVideoFrameBuffer>()
.ok_or(FfiError::InvalidRequest("handle is not a video buffer"))?;
let flip_y = to_argb.flip_y;
let dst_format = proto::VideoFormatType::from_i32(to_argb.dst_format).unwrap();
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,
)
};
let dst_stride = to_argb.dst_stride;
let dst_width = to_argb.dst_width as i32;
let mut dst_height = to_argb.dst_height as i32;
if flip_y {
dst_height = -dst_height;
}
buffer
.to_argb(
dst_format.into(),
dst_buf,
dst_stride,
dst_width,
dst_height,
)
.unwrap();
Ok(proto::ToArgbResponse::default())
}
// Audio
fn on_alloc_audio_buffer(
&'static self,
alloc: proto::AllocAudioBufferRequest,
) -> FfiResult<proto::AllocAudioBufferResponse> {
let frame = AudioFrame::new(
alloc.sample_rate,
alloc.num_channels,
alloc.samples_per_channel,
);
let handle_id = self.next_id() as FfiHandleId;
let frame_info = proto::AudioFrameBufferInfo::from(handle_id, &frame);
self.ffi_handles.insert(handle_id, Box::new(frame));
Ok(proto::AllocAudioBufferResponse {
buffer: Some(frame_info),
})
}
fn on_new_audio_stream(
&'static self,
new_stream: proto::NewAudioStreamRequest,
) -> FfiResult<proto::NewAudioStreamResponse> {
let stream_info = audio_frame::FfiAudioSream::setup(self, new_stream)?;
Ok(proto::NewAudioStreamResponse {
stream: Some(stream_info),
})
}
fn on_new_audio_source(
&'static self,
new_source: proto::NewAudioSourceRequest,
) -> FfiResult<proto::NewAudioSourceResponse> {
let source_info = audio_frame::FfiAudioSource::setup(self, new_source)?;
Ok(proto::NewAudioSourceResponse {
source: Some(source_info),
})
}
fn on_capture_audio_frame(
&'static self,
push: proto::CaptureAudioFrameRequest,
) -> FfiResult<proto::CaptureAudioFrameResponse> {
let handle_id = push
.source_handle
.as_ref()
.ok_or(FfiError::InvalidRequest("handle is empty"))?
.id as FfiHandleId;
let audio_source = self
.ffi_handles
.get(&handle_id)
.ok_or(FfiError::InvalidRequest("audio_source not found"))?;
let audio_source = audio_source
.downcast_ref::<audio_frame::FfiAudioSource>()
.ok_or(FfiError::InvalidRequest("handle is not a video source"))?;
audio_source.capture_frame(self, push)?;
Ok(proto::CaptureAudioFrameResponse::default())
}
pub fn handle_request(
&'static self,
request: proto::FfiRequest,
) -> FfiResult<proto::FfiResponse> {
let request = request
.message
.ok_or(FfiError::InvalidRequest("message is empty"))?;
let mut res = proto::FfiResponse::default();
res.message = Some(match request {
proto::ffi_request::Message::Initialize(init) => {
proto::ffi_response::Message::Initialize(self.on_initialize(init)?)
}
proto::ffi_request::Message::Dispose(dispose) => {
proto::ffi_response::Message::Dispose(self.on_dispose(dispose)?)
}
proto::ffi_request::Message::Connect(connect) => {
proto::ffi_response::Message::Connect(self.on_connect(connect)?)
}
proto::ffi_request::Message::Disconnect(disconnect) => {
proto::ffi_response::Message::Disconnect(self.on_disconnect(disconnect)?)
}
proto::ffi_request::Message::PublishTrack(publish) => {
proto::ffi_response::Message::PublishTrack(self.on_publish_track(publish)?)
}
proto::ffi_request::Message::UnpublishTrack(unpublish) => {
proto::ffi_response::Message::UnpublishTrack(self.on_unpublish_track(unpublish)?)
}
proto::ffi_request::Message::CreateVideoTrack(create) => {
proto::ffi_response::Message::CreateVideoTrack(self.on_create_video_track(create)?)
}
proto::ffi_request::Message::CreateAudioTrack(create) => {
proto::ffi_response::Message::CreateAudioTrack(self.on_create_audio_track(create)?)
}
proto::ffi_request::Message::AllocVideoBuffer(alloc) => {
proto::ffi_response::Message::AllocVideoBuffer(self.on_alloc_video_buffer(alloc)?)
}
proto::ffi_request::Message::NewVideoStream(new_stream) => {
proto::ffi_response::Message::NewVideoStream(self.on_new_video_stream(new_stream)?)
}
proto::ffi_request::Message::NewVideoSource(new_source) => {
proto::ffi_response::Message::NewVideoSource(self.on_new_video_source(new_source)?)
}
proto::ffi_request::Message::CaptureVideoFrame(push) => {
proto::ffi_response::Message::CaptureVideoFrame(self.on_capture_video_frame(push)?)
}
proto::ffi_request::Message::ToI420(to_i420) => {
proto::ffi_response::Message::ToI420(self.on_to_i420(to_i420)?)
}
proto::ffi_request::Message::ToArgb(to_argb) => {
proto::ffi_response::Message::ToArgb(self.on_to_argb(to_argb)?)
}
proto::ffi_request::Message::AllocAudioBuffer(alloc) => {
proto::ffi_response::Message::AllocAudioBuffer(self.on_alloc_audio_buffer(alloc)?)
}
proto::ffi_request::Message::NewAudioStream(new_stream) => {
proto::ffi_response::Message::NewAudioStream(self.on_new_audio_stream(new_stream)?)
}
proto::ffi_request::Message::NewAudioSource(new_source) => {
proto::ffi_response::Message::NewAudioSource(self.on_new_audio_source(new_source)?)
}
proto::ffi_request::Message::CaptureAudioFrame(push) => {
proto::ffi_response::Message::CaptureAudioFrame(self.on_capture_audio_frame(push)?)
}
});
Ok(res)
}
}
/// 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
}
#[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
}
livekit-ffi/src/server/room.rs
+53 -81
View File
@@ -1,54 +1,65 @@
use crate::server::FFIServer;
use futures_util::stream::StreamExt;
use crate::server::FfiServer;
use crate::{proto, FfiHandleId, FfiResult};
use livekit::prelude::*;
use livekit::webrtc::video_stream::native::NativeVideoStream;
use tokio::sync::{mpsc, oneshot};
use crate::proto;
use tokio::task::JoinHandle;
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;
pub struct FfiRoom {
room: Room,
handle_id: FfiHandleId,
handle: JoinHandle<()>,
close_tx: oneshot::Sender<()>,
}
impl FfiRoom {
pub async fn connect(
server: &'static FfiServer,
connect: proto::ConnectRequest,
) -> FfiResult<proto::RoomInfo> {
let (room, events) = Room::connect(&connect.url, &connect.token).await?;
let (close_tx, close_rx) = oneshot::channel();
let session = room.session();
let next_id = server.next_id() as FfiHandleId;
let handle = tokio::spawn(room_task(
server,
session.clone(),
next_id,
events,
close_rx,
));
let room_info = proto::RoomInfo::from_session(next_id, &session);
let ffi_room = Self {
handle_id: next_id,
room,
handle,
close_tx,
};
server.ffi_handles().insert(next_id, Box::new(ffi_room));
server.rooms().lock().insert(session.sid(), next_id);
Ok(room_info)
}
let (room, events) = res.unwrap();
let session = room.session();
pub async fn close(self) {
self.room.close().await;
let _ = self.close_tx.send(());
let _ = self.handle.await;
}
// 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));
pub fn session(&self) -> RoomSession {
self.room.session()
}
}
async fn room_task(
server: &'static FFIServer,
room: Room,
server: &'static FfiServer,
session: RoomSession,
room_handle: FfiHandleId,
mut events: mpsc::UnboundedReceiver<livekit::RoomEvent>,
mut close_rx: oneshot::Receiver<()>,
) {
let session = room.session();
tokio::spawn(participant_task(Participant::Local(
session.local_participant(),
)));
@@ -56,24 +67,14 @@ async fn room_task(
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);
if let Some(event) = proto::RoomEvent::from(room_handle, event.clone()) {
let _ = server.send_event(proto::ffi_event::Message::RoomEvent(event));
}
match event {
RoomEvent::ParticipantConnected(p) => {
tokio::spawn(participant_task(Participant::Remote(p)));
}
RoomEvent::TrackSubscribed {
track,
publication: _,
participant: _,
} => {
if let RemoteTrack::Video(video_track) = track {
let video_stream = NativeVideoStream::new(video_track.rtc_track());
tokio::spawn(video_frame_task(server, video_track.sid(), video_stream));
}
}
_ => {}
}
},
@@ -82,40 +83,11 @@ async fn room_task(
}
};
}
room.close().await;
}
async fn participant_task(participant: Participant) {
let mut participant_events = participant.register_observer();
while let Some(event) = participant_events.recv().await {
while let Some(_event) = participant_events.recv().await {
// TODO(theomonnom): convert event to proto
}
}
async fn video_frame_task(
server: &'static FFIServer,
track_sid: TrackSid,
mut stream: NativeVideoStream,
) {
while let Some(frame) = stream.next().await {
let handle_id = server.next_handle_id();
let frame_info = proto::VideoFrameInfo::from(&frame);
let buffer_info = proto::VideoFrameBufferInfo::from(handle_id, &frame.buffer);
server.insert_handle(handle_id, Box::new(frame.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_info),
buffer: Some(buffer_info),
},
)),
}),
None,
);
}
}
livekit-ffi/src/server/tests.rs
+334
View File
@@ -0,0 +1,334 @@
use std::time::Duration;
use crate::FfiHandleId;
use crate::{proto, server};
use livekit_api::access_token::{AccessToken, VideoGrants};
// Small FfiClient implementation used for testing
// This can be used as an example for a real implementation
mod client {
use crate::{
livekit_ffi_drop_handle, livekit_ffi_request, proto, FfiCallbackFn, FfiHandleId,
INVALID_HANDLE,
};
use lazy_static::lazy_static;
use prost::Message;
use std::sync::Mutex;
use tokio::sync::mpsc;
lazy_static! {
static ref EVENT_TX: Mutex<Option<mpsc::UnboundedSender<proto::ffi_event::Message>>> =
Default::default();
pub static ref FFI_CLIENT: Mutex<FfiClient> = Default::default();
}
pub struct FfiHandle(pub FfiHandleId);
pub struct FfiClient {
event_rx: mpsc::UnboundedReceiver<proto::ffi_event::Message>,
}
impl Default for FfiClient {
fn default() -> Self {
let (event_tx, event_rx) = mpsc::unbounded_channel();
*EVENT_TX.lock().unwrap() = Some(event_tx);
Self { event_rx }
}
}
impl FfiClient {
pub async fn recv_event(&mut self) -> proto::ffi_event::Message {
self.event_rx.recv().await.unwrap()
}
pub fn initialize(&self) {
self.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::Initialize(
proto::InitializeRequest {
event_callback_ptr: test_events_callback as FfiCallbackFn as u64,
},
)),
});
}
pub fn send_request(&self, request: proto::FfiRequest) -> proto::FfiResponse {
let data = request.encode_to_vec();
let mut res_ptr: Box<*const u8> = Box::new(std::ptr::null());
let mut res_len: Box<usize> = Box::new(0);
let handle = livekit_ffi_request(
data.as_ptr(),
data.len(),
res_ptr.as_mut(),
res_len.as_mut(),
);
let handle = FfiHandle(handle); // drop at end of scope
let res = unsafe {
assert_ne!(handle.0, INVALID_HANDLE);
assert_ne!(*res_ptr, std::ptr::null());
assert_ne!(*res_len, 0);
std::slice::from_raw_parts(*res_ptr, *res_len)
};
proto::FfiResponse::decode(res).unwrap()
}
}
impl Drop for FfiHandle {
fn drop(&mut self) {
assert!(livekit_ffi_drop_handle(self.0));
}
}
#[no_mangle]
unsafe extern "C" fn test_events_callback(data_ptr: *const u8, len: usize) {
let data = unsafe { std::slice::from_raw_parts(data_ptr, len) };
let event = proto::FfiEvent::decode(data).unwrap();
EVENT_TX
.lock()
.unwrap()
.as_ref()
.unwrap()
.send(event.message.unwrap())
.unwrap();
}
}
struct TestScope {}
impl TestScope {
fn new() -> (Self, std::sync::MutexGuard<'static, client::FfiClient>) {
// Run one test at a time
let client = client::FFI_CLIENT.lock().unwrap();
(TestScope {}, client)
}
}
impl Drop for TestScope {
fn drop(&mut self) {
// At the end of a test, no more handle should exist
assert!(server::FFI_SERVER.ffi_handles().is_empty());
}
}
fn test_env() -> (String, String, String) {
let lk_url = std::env::var("LK_TEST_URL").expect("LK_TEST_URL isn't set");
let lk_api_key = std::env::var("LK_TEST_API_KEY").expect("LK_TEST_API_KEY isn't set");
let lk_api_secret = std::env::var("LK_TEST_API_SECRET").expect("LK_TEST_API_SECRET isn't set");
(lk_url, lk_api_key, lk_api_secret)
}
macro_rules! wait_for_event {
($client:ident, $variant:ident, $timeout:expr) => {
tokio::time::timeout(Duration::from_secs($timeout), async {
loop {
let event = $client.recv_event().await;
if let proto::ffi_event::Message::$variant(event) = event {
return event;
}
}
})
};
}
#[test]
fn create_i420_buffer() {
let (_test, client) = TestScope::new();
// Create a new I420Buffer
let res = client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::AllocVideoBuffer(
proto::AllocVideoBufferRequest {
r#type: proto::VideoFrameBufferType::I420 as i32,
width: 640,
height: 480,
},
)),
});
let proto::ffi_response::Message::AllocVideoBuffer(alloc) = res.message.unwrap() else {
panic!("unexpected response");
};
// Convert to I420 (copy/no-op)
let i420_handle = client::FfiHandle(alloc.buffer.unwrap().handle.unwrap().id as FfiHandleId);
let res = client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::ToI420(proto::ToI420Request {
flip_y: false,
from: Some(proto::to_i420_request::From::Buffer(proto::FfiHandleId {
id: i420_handle.0 as u64,
})),
})),
});
let proto::ffi_response::Message::ToI420(to_i420) = res.message.unwrap() else {
panic!("unexpected response");
};
// Make sure to drop the handles
client::FfiHandle(to_i420.buffer.unwrap().handle.unwrap().id as FfiHandleId);
}
#[test]
#[ignore] // Ignore for now ( need to setup GHA )
fn publish_video_track() {
let (test, mut client) = TestScope::new();
let (lk_url, lk_api_key, lk_api_secret) = test_env();
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
.block_on(async {
client.initialize();
let token = AccessToken::with_api_key(&lk_api_key, &lk_api_secret)
.with_grants(VideoGrants {
room: "livekit-ffi-test".to_string(),
..Default::default()
})
.with_identity("video_test")
.to_jwt()
.unwrap();
// Connect to the room
client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::Connect(
proto::ConnectRequest {
url: lk_url.clone(),
token,
..Default::default()
},
)),
});
let connect = wait_for_event!(client, Connect, 5).await.unwrap();
assert!(connect.error.is_none());
let room_handle =
client::FfiHandle(connect.room.unwrap().handle.unwrap().id as FfiHandleId);
// Create a new VideoSource
let res = client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::NewVideoSource(
proto::NewVideoSourceRequest {
r#type: proto::VideoSourceType::VideoSourceNative as i32,
},
)),
});
const VIDEO_WIDTH: u32 = 640;
const VIDEO_HEIGHT: u32 = 480;
const VIDEO_FPS: f64 = 8.0;
let proto::ffi_response::Message::NewVideoSource(new_video_source) =
res.message.unwrap() else {
panic!("unexpected response");
};
let source_handle = client::FfiHandle(
new_video_source.source.unwrap().handle.unwrap().id as FfiHandleId,
);
// Create a new VideoTrack
let res = client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::CreateVideoTrack(
proto::CreateVideoTrackRequest {
name: "video_test".to_string(),
source_handle: Some(proto::FfiHandleId {
id: source_handle.0 as u64,
}),
options: Some(proto::VideoCaptureOptions {
resolution: Some(proto::VideoResolution {
width: VIDEO_WIDTH,
height: VIDEO_HEIGHT,
frame_rate: VIDEO_FPS,
}),
}),
},
)),
});
let proto::ffi_response::Message::CreateVideoTrack(create_video_track) =
res.message.unwrap() else {
panic!("unexpected response");
};
let track_handle = client::FfiHandle(
create_video_track.track.unwrap().opt_handle.unwrap().id as FfiHandleId,
);
let publish_options = proto::TrackPublishOptions {
name: "video_test".to_string(),
video_codec: proto::VideoCodec::H264 as i32,
source: proto::TrackSource::SourceCamera as i32,
..Default::default()
};
// Publish the VideoTrack
client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::PublishTrack(
proto::PublishTrackRequest {
room_handle: Some(proto::FfiHandleId {
id: room_handle.0 as u64,
}),
track_handle: Some(proto::FfiHandleId {
id: track_handle.0 as u64,
}),
options: Some(publish_options),
},
)),
});
let publish_track = wait_for_event!(client, PublishTrack, 5).await.unwrap();
assert!(publish_track.error.is_none());
// Send red frames
let rgba: Vec<u32> = vec![0xff0000ff; (VIDEO_WIDTH * VIDEO_HEIGHT) as usize];
let res = client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::ToI420(proto::ToI420Request {
flip_y: false,
from: Some(proto::to_i420_request::From::Argb(proto::ArgbBufferInfo {
ptr: rgba.as_ptr() as u64,
format: proto::VideoFormatType::FormatAbgr as i32,
width: VIDEO_WIDTH,
height: VIDEO_HEIGHT,
stride: VIDEO_WIDTH * 4,
})),
})),
});
let proto::ffi_response::Message::ToI420(to_i420) = res.message.unwrap() else {
panic!("unexpected response");
};
let buffer_handle =
client::FfiHandle(to_i420.buffer.unwrap().handle.unwrap().id as FfiHandleId);
// 2 seconds
for _ in 0..16 {
client.send_request(proto::FfiRequest {
message: Some(proto::ffi_request::Message::CaptureVideoFrame(
proto::CaptureVideoFrameRequest {
source_handle: Some(proto::FfiHandleId {
id: source_handle.0 as u64,
}),
buffer_handle: Some(proto::FfiHandleId {
id: buffer_handle.0 as u64,
}),
frame: Some(proto::VideoFrameInfo {
timestamp: 0, // TODO
rotation: proto::VideoRotation::VideoRotation0 as i32,
}),
},
)),
});
tokio::time::sleep(std::time::Duration::from_millis(1000 / VIDEO_FPS as u64)).await;
}
})
}
livekit-ffi/src/server/utils.rs
+32
View File
@@ -0,0 +1,32 @@
use crate::{server, FfiError, FfiHandleId, FfiResult};
use livekit::prelude::*;
pub fn find_remote_track(
server: &'static server::FfiServer,
track_sid: &TrackSid,
participant_sid: &ParticipantSid,
room_handle: FfiHandleId,
) -> FfiResult<RemoteTrack> {
let room = server
.ffi_handles()
.get(&room_handle)
.ok_or(FfiError::InvalidRequest("room not found"))?;
let room = room
.downcast_ref::<server::room::FfiRoom>()
.ok_or(FfiError::InvalidRequest("room is not ffi room"))?;
let session = room.session();
let participants = session.participants();
let participant = participants
.get(participant_sid)
.ok_or(FfiError::InvalidRequest("participant not found"))?;
let track = participant
.get_track_publication(track_sid)
.ok_or(FfiError::InvalidRequest("publication not found"))?
.track()
.ok_or(FfiError::InvalidRequest("track not found/subscribed"))?;
Ok(track)
}
livekit-ffi/src/server/video_frame.rs
+231
View File
@@ -0,0 +1,231 @@
use crate::{proto, server, FfiError, FfiHandleId, FfiResult};
use futures_util::StreamExt;
use livekit::prelude::*;
use livekit::webrtc::media_stream::MediaStreamTrack;
use livekit::webrtc::video_frame::{BoxVideoFrameBuffer, VideoFrame};
use livekit::webrtc::video_source::native::NativeVideoSource;
use livekit::webrtc::video_stream::native::NativeVideoStream;
use log::warn;
use server::utils;
use tokio::sync::oneshot;
// ===== FFIVideoStream =====
pub struct FfiVideoStream {
handle_id: FfiHandleId,
stream_type: proto::VideoStreamType,
track_sid: TrackSid,
#[allow(dead_code)]
close_tx: oneshot::Sender<()>, // Close the stream on drop
}
impl FfiVideoStream {
/// Setup a new VideoStream and forward the frame data to the client/the foreign
/// language.
///
/// When FFIVideoStream is dropped (When the corresponding handle_id is dropped), the task
/// is being closed.
///
/// It is possible that the client receives a VideoFrame after the task is closed. The client
/// musts ignore it.
pub fn setup(
server: &'static server::FfiServer,
new_stream: proto::NewVideoStreamRequest,
) -> FfiResult<proto::VideoStreamInfo> {
let (close_tx, close_rx) = oneshot::channel();
let stream_type = proto::VideoStreamType::from_i32(new_stream.r#type).unwrap();
let track_sid: TrackSid = new_stream.track_sid.into();
let room_handle = new_stream
.room_handle
.ok_or(FfiError::InvalidRequest("room_handle is empty"))?
.id as FfiHandleId;
let track = utils::find_remote_track(
server,
&track_sid,
&new_stream.participant_sid.into(),
room_handle,
)?
.rtc_track();
let MediaStreamTrack::Video(track) = track else {
return Err(FfiError::InvalidRequest("not a video track"));
};
let stream = match stream_type {
proto::VideoStreamType::VideoStreamNative => {
let video_stream = Self {
handle_id: server.next_id(),
close_tx,
stream_type,
track_sid,
};
tokio::spawn(Self::native_video_stream_task(
server,
video_stream.handle_id,
NativeVideoStream::new(track),
close_rx,
));
Ok::<FfiVideoStream, FfiError>(video_stream)
}
// TODO(theomonnom): Support other stream types
_ => return Err(FfiError::InvalidRequest("unsupported video stream type")),
}?;
// Store the new video stream and return the info
let info = proto::VideoStreamInfo::from(&stream);
server
.ffi_handles()
.insert(stream.handle_id, Box::new(stream));
Ok(info)
}
pub fn handle_id(&self) -> FfiHandleId {
self.handle_id
}
pub fn stream_type(&self) -> proto::VideoStreamType {
self.stream_type
}
pub fn track_sid(&self) -> &TrackSid {
&self.track_sid
}
async fn native_video_stream_task(
server: &'static server::FfiServer,
stream_handle_id: FfiHandleId,
mut native_stream: NativeVideoStream,
mut close_rx: oneshot::Receiver<()>,
) {
loop {
tokio::select! {
_ = &mut close_rx => {
break;
}
frame = native_stream.next() => {
let Some(frame) = frame else {
break;
};
let handle_id = server.next_id();
let frame_info = proto::VideoFrameInfo::from(&frame);
let buffer_info = proto::VideoFrameBufferInfo::from(handle_id, &frame.buffer);
server
.ffi_handles()
.insert(handle_id, Box::new(frame.buffer));
if let Err(err) = server.send_event(proto::ffi_event::Message::VideoStreamEvent(
proto::VideoStreamEvent {
handle: Some(stream_handle_id.into()),
message: Some(proto::video_stream_event::Message::FrameReceived(
proto::VideoFrameReceived {
frame: Some(frame_info),
buffer: Some(buffer_info),
}
)),
}
)) {
warn!("failed to send video frame: {}", err);
}
}
}
}
}
}
// ===== FFIVideoSource =====
pub struct FfiVideoSource {
handle_id: FfiHandleId,
source_type: proto::VideoSourceType,
source: VideoSource,
}
#[derive(Clone)]
pub enum VideoSource {
Native(NativeVideoSource),
}
impl FfiVideoSource {
pub fn setup(
server: &'static server::FfiServer,
new_source: proto::NewVideoSourceRequest,
) -> FfiResult<proto::VideoSourceInfo> {
let source_type = proto::VideoSourceType::from_i32(new_source.r#type).unwrap();
let source_inner = match source_type {
proto::VideoSourceType::VideoSourceNative => {
let video_source = NativeVideoSource::default();
Ok(VideoSource::Native(video_source))
}
_ => Err(FfiError::InvalidRequest("unsupported video source type")),
}?;
let video_source = Self {
handle_id: server.next_id(),
source_type,
source: source_inner,
};
let source_info = proto::VideoSourceInfo::from(&video_source);
server
.ffi_handles()
.insert(video_source.handle_id, Box::new(video_source));
Ok(source_info)
}
pub fn capture_frame(
&self,
server: &'static server::FfiServer,
capture: proto::CaptureVideoFrameRequest,
) -> FfiResult<()> {
match self.source {
VideoSource::Native(ref source) => {
let frame_info = capture
.frame
.ok_or(FfiError::InvalidRequest("frame is empty"))?;
let buffer_handle = capture
.buffer_handle
.ok_or(FfiError::InvalidRequest("buffer_handle is none"))?
.id as FfiHandleId;
let buffer = server
.ffi_handles()
.get(&buffer_handle)
.ok_or(FfiError::InvalidRequest("handle not found"))?;
let buffer = buffer
.downcast_ref::<BoxVideoFrameBuffer>()
.ok_or(FfiError::InvalidRequest("handle is not video frame"))?;
let rotation = proto::VideoRotation::from_i32(frame_info.rotation).unwrap();
let frame = VideoFrame {
rotation: rotation.into(),
timestamp: frame_info.timestamp,
buffer,
};
source.capture_frame(&frame);
}
}
Ok(())
}
pub fn handle_id(&self) -> FfiHandleId {
self.handle_id
}
pub fn source_type(&self) -> proto::VideoSourceType {
self.source_type
}
pub fn inner_source(&self) -> &VideoSource {
&self.source
}
}