add error callback

livekit · Dec 20, 2024 · 94b6e75 · 94b6e75
1 parent f858454
commit 94b6e75
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 2 deletions.
diff --git a/livekit-ffi/protocol/ffi.proto b/livekit-ffi/protocol/ffi.proto
@@ -199,6 +199,8 @@ message FfiEvent {
     SendChatMessageCallback chat_message = 22;
     PerformRpcCallback perform_rpc = 23;
     RpcMethodInvocationEvent rpc_method_invocation = 24;
+    SendStreamHeaderCallback send_stream_header = 25;
+    SendStreamChunkCallback send_stream_chunk = 26;
   }
 }
 

diff --git a/livekit-ffi/protocol/room.proto b/livekit-ffi/protocol/room.proto
@@ -611,3 +611,13 @@ message SendStreamHeaderResponse {
 message SendStreamChunkResponse {
   required uint64 async_id = 1;
 }
+
+message SendStreamHeaderCallback {
+  required uint64 async_id = 1;
+  optional string error = 2;
+}
+
+message SendStreamChunkCallback {
+  required uint64 async_id = 1;
+  optional string error = 2;
+}
diff --git a/livekit-ffi/src/livekit.proto.rs b/livekit-ffi/src/livekit.proto.rs
@@ -3158,6 +3158,22 @@ pub struct SendStreamChunkResponse {
     #[prost(uint64, required, tag="1")]
     pub async_id: u64,
 }
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SendStreamHeaderCallback {
+    #[prost(uint64, required, tag="1")]
+    pub async_id: u64,
+    #[prost(string, optional, tag="2")]
+    pub error: ::core::option::Option<::prost::alloc::string::String>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SendStreamChunkCallback {
+    #[prost(uint64, required, tag="1")]
+    pub async_id: u64,
+    #[prost(string, optional, tag="2")]
+    pub error: ::core::option::Option<::prost::alloc::string::String>,
+}
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
 #[repr(i32)]
 pub enum IceTransportType {
@@ -4138,7 +4154,7 @@ pub mod ffi_response {
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FfiEvent {
-    #[prost(oneof="ffi_event::Message", tags="1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24")]
+    #[prost(oneof="ffi_event::Message", tags="1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26")]
     pub message: ::core::option::Option<ffi_event::Message>,
 }
 /// Nested message and enum types in `FfiEvent`.
@@ -4192,6 +4208,10 @@ pub mod ffi_event {
         PerformRpc(super::PerformRpcCallback),
         #[prost(message, tag="24")]
         RpcMethodInvocation(super::RpcMethodInvocationEvent),
+        #[prost(message, tag="25")]
+        SendStreamHeader(super::SendStreamHeaderCallback),
+        #[prost(message, tag="26")]
+        SendStreamChunk(super::SendStreamChunkCallback),
     }
 }
 /// Stop all rooms synchronously (Do we need async here?).

diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs
@@ -695,6 +695,11 @@ impl RoomInner {
         let inner = self.clone();
         let handle = server.async_runtime.spawn(async move {
             let res = inner.room.local_participant().publish_raw_data(packet, true).await;
+            let cb = proto::SendStreamHeaderCallback {
+                async_id,
+                error: res.err().map(|e| e.to_string()),
+            };
+            let _ = server.send_event(proto::ffi_event::Message::SendStreamHeader(cb));
         });
         server.watch_panic(handle);
         proto::SendStreamHeaderResponse { async_id }
@@ -717,7 +722,13 @@ impl RoomInner {
         let async_id = server.next_id();
         let inner = self.clone();
         let handle = server.async_runtime.spawn(async move {
-            let res = inner.room.local_participant().publish_raw_data(packet, true).await;
+            let res: Result<(), RoomError> =
+                inner.room.local_participant().publish_raw_data(packet, true).await;
+            let cb = proto::SendStreamChunkCallback {
+                async_id,
+                error: res.err().map(|e| e.to_string()),
+            };
+            let _ = server.send_event(proto::ffi_event::Message::SendStreamChunk(cb));
         });
         server.watch_panic(handle);
         proto::SendStreamChunkResponse { async_id }