Add a mechanism to update stream information for applied audio filters after AudioStream initialization (#596)

* add a way to update stream_info after AudioStream is initialized

* fmt

* add a changeset

Author: typester

.nanpa/candy-blimp-twice.kdl

@@ -0,0 +1 @@
+patch package="livekit-ffi" type="fixed" "Fixed metric report issue on audio filter where room_id is sometimes empty"

Cargo.lock

@@ -57,7 +57,7 @@ impl FfiAudioStream {
             return Err(FfiError::InvalidRequest("not an audio track".into()));
         };
 
-        let (audio_filter, stream_info) = match &new_stream.audio_filter_module_id {
+        let (audio_filter, info) = match &new_stream.audio_filter_module_id {
             Some(module_id) => {
                 let Some(room_handle) = ffi_track.room_handle else {
                     return Err(FfiError::InvalidRequest(
@@ -83,9 +83,9 @@ impl FfiAudioStream {
                     track_id: rtc_track.id(),
                 };
 
-                (Some(filter), stream_info)
+                (Some(filter), Some(AudioFilterInfo { stream_info, room_handle }))
             }
-            None => (None, AudioFilterStreamInfo::default()),
+            None => (None, None),
         };
 
         let stream_type = new_stream.r#type();
@@ -104,7 +104,7 @@ impl FfiAudioStream {
                     let session = audio_filter.clone().new_session(
                         sample_rate,
                         new_stream.audio_filter_options.unwrap_or("".into()),
-                        stream_info,
+                        info.as_ref().map(|i| i.stream_info.clone()).unwrap(),
                     );
 
                     match session {
@@ -134,6 +134,7 @@ impl FfiAudioStream {
                     self_dropped_rx,
                     server.watch_handle_dropped(new_stream.track_handle),
                     true,
+                    info,
                 ));
                 server.watch_panic(handle);
                 Ok::<FfiAudioStream, FfiError>(audio_stream)
@@ -208,7 +209,8 @@ impl FfiAudioStream {
         // track_tx is no longer held, so the track_rx will be closed when track_changed_trigger is done
 
         let url = ffi_participant.room.url();
-        let room_sid = ffi_participant.room.room.sid().await;
+        let room_sid =
+            ffi_participant.room.room.maybe_sid().map(|id| id.to_string()).unwrap_or("".into());
         let room_name = ffi_participant.room.room.name();
         let participant_identity = ffi_participant.participant.identity();
         let participant_id = ffi_participant.participant.sid();
@@ -247,7 +249,7 @@ impl FfiAudioStream {
                     }
                 });
 
-                let mut audio_filter_session = match &filter {
+                let (mut audio_filter_session, info) = match &filter {
                     Some(filter) => match &request.audio_filter_options {
                         Some(options) => {
                             let stream_info = AudioFilterStreamInfo {
@@ -259,19 +261,24 @@ impl FfiAudioStream {
                                 track_id: track.sid().into(),
                             };
 
+                            let info = AudioFilterInfo {
+                                stream_info,
+                                room_handle: ffi_participant.room.handle_id,
+                            };
+
                             let session = filter.clone().new_session(
                                 sample_rate as u32,
                                 &options,
-                                stream_info,
+                                info.stream_info.clone(),
                             );
                             if session.is_none() {
                                 log::error!("failed to initialize the audio filter. it will not be enabled for this session.");
                             }
-                            session
+                            (session, Some(info))
                         }
-                        None => None,
+                        None => (None, None),
                     },
-                    None => None,
+                    None => (None, None),
                 };
 
                 let native_stream = NativeAudioStream::new(rtc_track, sample_rate, num_channels);
@@ -297,6 +304,7 @@ impl FfiAudioStream {
                         c_rx,
                         handle_dropped_rx,
                         false,
+                        info,
                     )
                     .await;
                     let _ = done_tx.send(());
@@ -332,6 +340,7 @@ impl FfiAudioStream {
         mut self_dropped_rx: oneshot::Receiver<()>,
         mut handle_dropped_rx: oneshot::Receiver<()>,
         send_eos: bool,
+        mut filter_info: Option<AudioFilterInfo>,
     ) {
         loop {
             tokio::select! {
@@ -346,6 +355,21 @@ impl FfiAudioStream {
                         break;
                     };
 
+                    if let Some(ref mut info) = filter_info {
+                        if info.stream_info.room_id == "" {
+                            // check if room_id is updated
+                            if info.update_room_id(server) {
+                                if info.stream_info.room_id != "" {
+                                    if let AudioStreamKind::Filtered(ref mut filter) = native_stream {
+                                        filter.update_stream_info(info.stream_info.clone());
+                                    }
+                                    // room_id is updated, this check is no longer needed.
+                                    filter_info = None;
+                                }
+                            }
+                        }
+                    }
+
                     let handle_id = server.next_id();
                     let buffer_info = proto::AudioFrameBufferInfo::from(&frame);
                     server.store_handle(handle_id, frame);
@@ -383,3 +407,24 @@ impl FfiAudioStream {
         }
     }
 }
+
+// Used to update audio filter session when the stream info is changed. (Mainly room_id
+#[derive(Default)]
+struct AudioFilterInfo {
+    stream_info: AudioFilterStreamInfo,
+    room_handle: FfiHandleId,
+}
+
+impl AudioFilterInfo {
+    fn update_room_id(&mut self, server: &'static server::FfiServer) -> bool {
+        let Ok(room) = server.retrieve_handle::<FfiRoom>(self.room_handle) else {
+            return false;
+        };
+        let room_id = room.inner.room.maybe_sid().map(|id| id.to_string()).unwrap_or("".into());
+        if room_id != "" {
+            self.stream_info.room_id = room_id.into();
+            return true;
+        }
+        false
+    }
+}

livekit-ffi/src/server/audio_stream.rs

@@ -56,7 +56,7 @@ pub struct FfiRoom {
 
 pub struct RoomInner {
     pub room: Room,
-    handle_id: FfiHandleId,
+    pub(crate) handle_id: FfiHandleId,
     data_tx: mpsc::UnboundedSender<FfiDataPacket>,
     transcription_tx: mpsc::UnboundedSender<FfiTranscription>,
     dtmf_tx: mpsc::UnboundedSender<FfiSipDtmfPacket>,

livekit-ffi/src/server/room.rs

@@ -33,6 +33,7 @@ type CreateFn = unsafe extern "C" fn(
 type DestroyFn = unsafe extern "C" fn(*const c_void);
 type ProcessI16Fn = unsafe extern "C" fn(*const c_void, usize, *const i16, *mut i16);
 type ProcessF32Fn = unsafe extern "C" fn(*const c_void, usize, *const f32, *mut f32);
+type UpdateStreamInfoFn = unsafe extern "C" fn(*const c_void, *const c_char);
 
 static REGISTERED_PLUGINS: LazyLock<RwLock<HashMap<String, Arc<AudioFilterPlugin>>>> =
     LazyLock::new(|| RwLock::new(HashMap::new()));
@@ -57,6 +58,7 @@ pub struct AudioFilterPlugin {
     destroy_fn_ptr: *const c_void,
     process_i16_fn_ptr: *const c_void,
     process_f32_fn_ptr: *const c_void,
+    update_stream_info_fn_ptr: *const c_void,
 }
 
 impl AudioFilterPlugin {
@@ -116,6 +118,11 @@ impl AudioFilterPlugin {
                 .try_as_raw_ptr()
                 .unwrap()
         };
+        let update_stream_info_fn_ptr = unsafe {
+            lib.get::<Symbol<UpdateStreamInfoFn>>(b"audio_filter_update_stream_info")?
+                .try_as_raw_ptr()
+                .unwrap()
+        };
 
         Ok(Self {
             lib,
@@ -125,6 +132,7 @@ impl AudioFilterPlugin {
             destroy_fn_ptr,
             process_i16_fn_ptr,
             process_f32_fn_ptr,
+            update_stream_info_fn_ptr,
         })
     }
 
@@ -196,6 +204,17 @@ impl AudioFilterSession {
         let process: ProcessF32Fn = unsafe { std::mem::transmute(self.plugin.process_f32_fn_ptr) };
         unsafe { process(self.ptr, num_samples, input.as_ptr(), output.as_mut_ptr()) };
     }
+
+    pub fn update_stream_info(&self, info: AudioFilterStreamInfo) {
+        if self.plugin.update_stream_info_fn_ptr.is_null() {
+            return;
+        }
+        let update_stream_info_fn: UpdateStreamInfoFn =
+            unsafe { std::mem::transmute(self.plugin.update_stream_info_fn_ptr) };
+        let info_json = serde_json::to_string(&info).unwrap();
+        let info_json = CString::new(info_json).unwrap_or(CString::new("").unwrap());
+        unsafe { update_stream_info_fn(self.ptr, info_json.as_ptr()) }
+    }
 }
 
 impl Drop for AudioFilterSession {
@@ -234,6 +253,10 @@ impl AudioFilterAudioStream {
             frame_size,
         }
     }
+
+    pub fn update_stream_info(&mut self, info: AudioFilterStreamInfo) {
+        self.session.update_stream_info(info);
+    }
 }
 
 impl Stream for AudioFilterAudioStream {
@@ -267,7 +290,7 @@ impl Stream for AudioFilterAudioStream {
     }
 }
 
-#[derive(Debug, Serialize, Default)]
+#[derive(Debug, Serialize, Default, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct AudioFilterStreamInfo {
     pub url: String,