From 7e96c0bca1c0e990e218ba8f87ef8dc9dec69e84 Mon Sep 17 00:00:00 2001 From: James Hugman Date: Fri, 11 Sep 2026 13:23:17 +0100 Subject: [PATCH 1/6] Anchor the SoxResampler behaviour with migration tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exercise SoxResampler both of the ways the migration has to keep working — directly, and through encoded FFI requests — and compare the two against each other, so the same tests can be re-run after it. Nothing outside the tests changes here. --- livekit-ffi/src/lib.rs | 2 + livekit-ffi/src/migration_tests.rs | 192 ++++++++++++++++++++++++++++ livekit-ffi/src/server/resampler.rs | 91 +++++++++++++ 3 files changed, 285 insertions(+) create mode 100644 livekit-ffi/src/migration_tests.rs diff --git a/livekit-ffi/src/lib.rs b/livekit-ffi/src/lib.rs index 1514b3612..722eab085 100644 --- a/livekit-ffi/src/lib.rs +++ b/livekit-ffi/src/lib.rs @@ -19,6 +19,8 @@ use livekit::prelude::*; use thiserror::Error; mod conversion; +#[cfg(test)] +mod migration_tests; pub mod build_info; pub mod cabi; diff --git a/livekit-ffi/src/migration_tests.rs b/livekit-ffi/src/migration_tests.rs new file mode 100644 index 000000000..6e34ac856 --- /dev/null +++ b/livekit-ffi/src/migration_tests.rs @@ -0,0 +1,192 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Protobuf-driven tests for the types being migrated to uniffi. +//! +//! The FFI request surface is the contract that must not move while the types +//! behind it grow uniffi annotations, so these tests drive it the way the +//! bindings do — encoded requests in, handles and raw pointers out — and check +//! the result against the same work done through the Rust API directly. The +//! direct-call half of each pair lives next to the code it exercises, e.g. +//! `server::resampler::migration_tests`. +//! +//! They are written to compile and pass either side of the migration, so they +//! can be replayed on the commit before it. + +use std::{mem::size_of, slice}; + +use crate::{proto, server::requests::handle_request, FfiHandleId, FFI_SERVER}; + +fn request(message: impl Into) -> proto::ffi_response::Message { + let response = handle_request(&FFI_SERVER, proto::FfiRequest { message: Some(message.into()) }) + .expect("request failed"); + response.message.expect("response carried no message") +} + +fn new_sox_resampler( + input_rate: f64, + output_rate: f64, + num_channels: u32, + quality: proto::SoxQualityRecipe, +) -> FfiHandleId { + let response = match request(proto::NewSoxResamplerRequest { + input_rate, + output_rate, + num_channels, + input_data_type: proto::SoxResamplerDataType::SoxrDatatypeInt16i as i32, + output_data_type: proto::SoxResamplerDataType::SoxrDatatypeInt16i as i32, + quality_recipe: quality as i32, + flags: None, + }) { + proto::ffi_response::Message::NewSoxResampler(response) => response, + _ => panic!("expected a NewSoxResampler response"), + }; + + match response.message.expect("response carried no message") { + proto::new_sox_resampler_response::Message::Resampler(resampler) => resampler.handle.id, + proto::new_sox_resampler_response::Message::Error(error) => panic!("{error}"), + } +} + +fn push_sox_resampler(handle: FfiHandleId, input: &[i16]) -> Vec { + let response = match request(proto::PushSoxResamplerRequest { + resampler_handle: handle, + data_ptr: input.as_ptr() as u64, + size: (input.len() * size_of::()) as u32, + }) { + proto::ffi_response::Message::PushSoxResampler(response) => response, + _ => panic!("expected a PushSoxResampler response"), + }; + + assert_eq!(response.error, None, "push failed"); + read_output(response.output_ptr, response.size) +} + +fn flush_sox_resampler(handle: FfiHandleId) -> Vec { + let response = match request(proto::FlushSoxResamplerRequest { resampler_handle: handle }) { + proto::ffi_response::Message::FlushSoxResampler(response) => response, + _ => panic!("expected a FlushSoxResampler response"), + }; + + assert_eq!(response.error, None, "flush failed"); + read_output(response.output_ptr, response.size) +} + +/// The pointer/size pair the bindings dereference: `size` counts bytes, the +/// samples are `i16`, a run that produced nothing is a null pointer, and the +/// samples stay readable until the next call on the same resampler. +fn read_output(output_ptr: u64, size: u32) -> Vec { + assert_eq!(size as usize % size_of::(), 0, "size {size} is not whole i16 samples"); + if output_ptr == 0 || size == 0 { + return Vec::new(); + } + unsafe { slice::from_raw_parts(output_ptr as *const i16, size as usize / size_of::()) } + .to_vec() +} + +/// soxr dithers its int16 output, so two instances fed identical input agree to +/// within a LSB or so rather than bit for bit. +fn assert_samples_match(actual: &[i16], expected: &[i16], context: &str) { + assert_eq!(actual.len(), expected.len(), "{context}: output length"); + for (i, (actual, expected)) in actual.iter().zip(expected).enumerate() { + assert!( + (*actual as i32 - *expected as i32).abs() <= 2, + "{context}: sample {i} is {actual}, expected ~{expected}" + ); + } +} + +/// Every knob of `NewSoxResamplerRequest` reaches the resampler: the same input +/// pushed through the FFI request surface and through the Rust API must come +/// back as the same audio, for each rate, channel count and quality recipe. +#[test] +fn proto_push_matches_direct_push() { + let cases = [ + (48000.0, 16000.0, 1, proto::SoxQualityRecipe::SoxrQualityQuick), + (48000.0, 16000.0, 1, proto::SoxQualityRecipe::SoxrQualityVeryhigh), + (16000.0, 48000.0, 1, proto::SoxQualityRecipe::SoxrQualityMedium), + (48000.0, 24000.0, 2, proto::SoxQualityRecipe::SoxrQualityHigh), + ]; + + for (input_rate, output_rate, num_channels, quality) in cases { + let input: Vec = (0..960 * num_channels as i16) + .map(|i| (i % 97) * 300 - 14000) // a repeating sawtooth, loud enough to compare + .collect(); + + let handle = new_sox_resampler(input_rate, output_rate, num_channels, quality); + #[allow(unused_mut)] + let mut direct = crate::sox_resampler!(input_rate, output_rate, num_channels, quality); + + let context = format!("{input_rate} -> {output_rate}, {num_channels}ch, {quality:?}"); + for _ in 0..3 { + assert_samples_match( + &push_sox_resampler(handle, &input), + &direct.push(&input).unwrap(), + &format!("push at {context}"), + ); + } + assert_samples_match( + &flush_sox_resampler(handle), + &direct.flush().unwrap(), + &format!("flush at {context}"), + ); + + FFI_SERVER.drop_handle(handle); + } +} + +/// 30ms in at 48kHz is 30ms out at 16kHz, counted in bytes off the wire. +#[test] +fn proto_push_then_flush_conserves_duration() { + let handle = new_sox_resampler(48000.0, 16000.0, 1, proto::SoxQualityRecipe::SoxrQualityQuick); + let frame = vec![1000i16; 480]; // 10ms @ 48kHz + + let mut frames = 0; + for _ in 0..3 { + frames += push_sox_resampler(handle, &frame).len(); + } + frames += flush_sox_resampler(handle).len(); + + assert!((frames as i64 - 480).abs() <= 8, "got {frames} frames, expected ~480"); + + FFI_SERVER.drop_handle(handle); +} + +/// The FFI handle map is a *second* owner of a migrated object, and it only +/// takes that ownership when `ffi_handle_id()` publishes an id. A resampler +/// that never crosses into the FFI path must die with its last `Arc`. +#[test] +fn the_handle_map_only_owns_published_resamplers() { + use crate::{server::resampler::SoxResampler, sox_resampler}; + use std::sync::Arc; + + let resampler = sox_resampler!(48000.0, 16000.0, 1, proto::SoxQualityRecipe::SoxrQualityQuick); + assert_eq!(Arc::strong_count(&resampler), 1, "an unpublished resampler has no other owner"); + + let handle = resampler.clone().ffi_handle_id(); + assert_eq!( + Arc::strong_count(&resampler), + 2, + "publishing an id makes the FFI server a co-owner" + ); + + drop(resampler); + assert!( + FFI_SERVER.retrieve_handle::>(handle).is_ok(), + "the FFI server holds the resampler until its handle is dropped" + ); + + FFI_SERVER.drop_handle(handle); + assert!(FFI_SERVER.retrieve_handle::>(handle).is_err()); +} diff --git a/livekit-ffi/src/server/resampler.rs b/livekit-ffi/src/server/resampler.rs index 32ea49e54..aedc8e0a1 100644 --- a/livekit-ffi/src/server/resampler.rs +++ b/livekit-ffi/src/server/resampler.rs @@ -173,3 +173,94 @@ fn to_soxr_datatype(datatype: proto::SoxResamplerDataType) -> soxr_sys::soxr_dat proto::SoxResamplerDataType::SoxrDatatypeInt16s => soxr_sys::soxr_datatype_t_SOXR_INT16_S, } } + +/// Direct-call tests for [`SoxResampler`], paired with the protobuf-driven tests +/// in `crate::migration_tests`. Both sets are written to compile and pass on +/// either side of the uniffi migration, so they can be replayed across it. +/// +/// `sox_resampler!` is the only migration seam: the migration renames and +/// re-types the constructor that takes the spec structs, so the macro body is the +/// one thing that differs between commits. The tests bind the resampler to a +/// `mut` local, which suits both a plain `SoxResampler` and an `Arc`. +#[cfg(test)] +// `mut` is load-bearing before the migration (`push(&mut self)`) and redundant +// after it (`Arc`, `push(&self)`). +#[allow(unused_mut)] +mod migration_tests { + use crate::proto; + + /// Builds a resampler from the same spec structs the FFI request handler uses. + /// Proto enums are passed through `.into()` so this reads identically before + /// the migration (identity conversion) and after it (proto -> uniffi enum). + #[macro_export] + macro_rules! sox_resampler { + ($input_rate:expr, $output_rate:expr, $num_channels:expr, $quality:expr) => { + $crate::server::resampler::SoxResampler::new( + $input_rate, + $output_rate, + $num_channels, + $crate::server::resampler::IOSpec { + input_type: $crate::proto::SoxResamplerDataType::SoxrDatatypeInt16i.into(), + output_type: $crate::proto::SoxResamplerDataType::SoxrDatatypeInt16i.into(), + }, + $crate::server::resampler::QualitySpec { quality: $quality.into(), flags: 0 }, + $crate::server::resampler::RuntimeSpec { num_threads: 1 }, + ) + .unwrap() + }; + } + + /// 30ms of 48kHz mono in, 30ms of 16kHz mono out, once the filter has been + /// drained by `flush`. + #[test] + fn push_then_flush_conserves_duration() { + let mut resampler = + sox_resampler!(48000.0, 16000.0, 1, proto::SoxQualityRecipe::SoxrQualityQuick); + let frame = vec![1000i16; 480]; // 10ms @ 48kHz + + let mut frames = 0; + for _ in 0..3 { + frames += resampler.push(&frame).unwrap().len(); + } + frames += resampler.flush().unwrap().len(); + + // 3 x 10ms @ 48kHz == 480 frames @ 16kHz, give or take the filter delay. + assert!((frames as i64 - 480).abs() <= 8, "got {frames} frames, expected ~480"); + } + + /// A steady level comes out at the same level, i.e. the sample data really is + /// being resampled rather than reinterpreted or truncated. + #[test] + fn steady_level_survives_resampling() { + let mut resampler = + sox_resampler!(48000.0, 16000.0, 1, proto::SoxQualityRecipe::SoxrQualityVeryhigh); + let frame = vec![8000i16; 4800]; // 100ms @ 48kHz + + resampler.push(&frame).unwrap(); // warm up past the filter ramp + let output = resampler.push(&frame).unwrap(); + + assert!(output.len() > 1000, "expected ~1600 frames, got {}", output.len()); + for (i, sample) in output.iter().enumerate() { + assert!((*sample as i32 - 8000).abs() < 80, "sample {i} is {sample}, expected ~8000"); + } + } + + /// Interleaved channels are resampled independently: a stereo frame of + /// (+8000, -8000) must not average out into silence. + #[test] + fn interleaved_channels_stay_separate() { + let mut resampler = + sox_resampler!(48000.0, 24000.0, 2, proto::SoxQualityRecipe::SoxrQualityQuick); + let frame: Vec = std::iter::repeat([8000, -8000]).take(4800).flatten().collect(); + + resampler.push(&frame).unwrap(); // warm up past the filter ramp + let output = resampler.push(&frame).unwrap(); + + assert_eq!(output.len() % 2, 0, "output is not a whole number of stereo frames"); + assert!(output.len() > 1000, "expected ~4800 samples, got {}", output.len()); + for (i, out_frame) in output.chunks(2).enumerate() { + assert!(out_frame[0] > 7000, "frame {i} left is {}, expected ~8000", out_frame[0]); + assert!(out_frame[1] < -7000, "frame {i} right is {}, expected ~-8000", out_frame[1]); + } + } +} From dd095c0b62b0290074b43184a3db9c850f459085 Mon Sep 17 00:00:00 2001 From: James Hugman Date: Wed, 9 Sep 2026 17:59:32 +0100 Subject: [PATCH 2/6] Add infrastructure to ease the migration between the FFI and uniffi. --- livekit-ffi/src/lib.rs | 1 + livekit-ffi/src/migration.rs | 100 +++++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+) create mode 100644 livekit-ffi/src/migration.rs diff --git a/livekit-ffi/src/lib.rs b/livekit-ffi/src/lib.rs index 722eab085..68b400aa5 100644 --- a/livekit-ffi/src/lib.rs +++ b/livekit-ffi/src/lib.rs @@ -19,6 +19,7 @@ use livekit::prelude::*; use thiserror::Error; mod conversion; +mod migration; #[cfg(test)] mod migration_tests; diff --git a/livekit-ffi/src/migration.rs b/livekit-ffi/src/migration.rs new file mode 100644 index 000000000..f50e12187 --- /dev/null +++ b/livekit-ffi/src/migration.rs @@ -0,0 +1,100 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/// This macro implements a bridge between the FFI handle system and the uniffi system. +/// +/// It depends on migrating structs carrying a `handle_id: OnceLock` field. +/// Migrating structs then expose methods via `#[uniffi::export]`. +/// +/// This macro generates foreign language methods: +/// 1) to get the `handle_id` so it can be passed via the old FFI system +/// 2) to construct an `Arc` from a `handle_id`, so we can migrate from the old FFI system +/// to the new uniffi system. +/// 3) to take ownership of a `&self` from the old FFI system, so the FFI side can release its +/// ownership of the handle but allow the uniffi side to continue use the struct. +/// +/// The FFI server is a second owner: it holds an `Arc` from the first `ffi_handle_id()` call +/// until `livekit_ffi_drop_handle` or `take_ffi_handle_id` releases it. A struct that never publishes +/// a handle is owned by the uniffi side alone and drops with its last `Arc`. +/// +/// The FFI system and uniffi system can be interchanged, however, once removed (via [take_ffi_handle_id] or +/// [livekit_ffi_drop_handle]), the FFI system cannot easily be used again for that handle id. +#[macro_export] +macro_rules! migrate_from_ffi { + ($T:ty) => { + impl crate::server::FfiHandle for ::std::sync::Arc<$T> {} + + #[::uniffi::export] + impl $T { + /// Publishes this struct to the FFI handle map and returns the id the FFI side + /// knows it by. + /// + /// The first call hands the FFI side co-ownership; later calls return the same id + /// without republishing. An id whose handle the FFI side has already released is + /// stale, and passing it back over the FFI fails with "handle not found". + pub fn ffi_handle_id(self: ::std::sync::Arc) -> crate::FfiHandleId { + let handle_id = *self.handle_id.get_or_init(|| { + let handle_id = crate::FFI_SERVER.next_id(); + // the FFI side co-owns from here; released by livekit_ffi_drop_handle + crate::FFI_SERVER.store_handle(handle_id, ::std::sync::Arc::clone(&self)); + handle_id + }); + handle_id + } + + #[::uniffi::constructor] + pub fn from_ffi_handle_id( + handle_id: crate::FfiHandleId, + ) -> ::std::result::Result<::std::sync::Arc, crate::migration::MigrationError> + { + match crate::FFI_SERVER.retrieve_handle::<::std::sync::Arc>(handle_id) { + Ok(arc) => Ok(arc.clone()), + Err(s) => Err(crate::migration::MigrationError::FromFfiHandleIdError(format!( + "MigrationError for handle_id {handle_id}: {s:?}" + ))), + } + } + + /// Takes the FFI side's ownership of this struct, leaving the uniffi side as the + /// only owner. The FFI side calls this when it is done with the handle. + /// + /// The release is final: the handle id is not reusable, a second call errors, and + /// so does a call on a struct that was never published. + pub fn take_ffi_handle_id( + &self, + ) -> ::std::result::Result<::std::sync::Arc, crate::migration::MigrationError> + { + let handle_id = self.handle_id.get().ok_or_else(|| { + crate::migration::MigrationError::TakeFfiHandleIdError( + "MigrationError for handle_id: handle_id not set yet".to_owned(), + ) + })?; + match crate::FFI_SERVER.take_handle::<::std::sync::Arc>(*handle_id) { + Ok(arc) => Ok(arc.clone()), + Err(s) => Err(crate::migration::MigrationError::TakeFfiHandleIdError(format!( + "MigrationError for handle_id {handle_id}: {s:?}" + ))), + } + } + } + }; +} + +#[derive(Debug, thiserror::Error, uniffi::Error)] +pub enum MigrationError { + #[error("Invalid handle id: {0}")] + FromFfiHandleIdError(String), + #[error("Failed to take handle id: {0}")] + TakeFfiHandleIdError(String), +} From 30e17f45bf46e02675ee686851c0f7dd45f99899 Mon Sep 17 00:00:00 2001 From: James Hugman Date: Wed, 9 Sep 2026 18:42:41 +0100 Subject: [PATCH 3/6] Mechanical transformation to add uniffi annotations to the Sox Resampler. --- livekit-ffi/src/conversion/resampler.rs | 47 +++++++- livekit-ffi/src/server/requests.rs | 30 ++--- livekit-ffi/src/server/resampler.rs | 145 +++++++++++++++++++++--- 3 files changed, 184 insertions(+), 38 deletions(-) diff --git a/livekit-ffi/src/conversion/resampler.rs b/livekit-ffi/src/conversion/resampler.rs index e0fbfd119..295b9831f 100644 --- a/livekit-ffi/src/conversion/resampler.rs +++ b/livekit-ffi/src/conversion/resampler.rs @@ -1,4 +1,4 @@ -// Copyright 2025 LiveKit, Inc. +// Copyright 2026 LiveKit, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -11,3 +11,48 @@ // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. + +use crate::{proto, server::resampler}; + +pub(crate) struct IOSpec { + pub input_type: proto::SoxResamplerDataType, + pub output_type: proto::SoxResamplerDataType, +} + +impl From for resampler::IOSpec { + fn from(value: IOSpec) -> Self { + Self { input_type: value.input_type.into(), output_type: value.output_type.into() } + } +} + +impl From for resampler::SoxResamplerDataType { + fn from(value: proto::SoxResamplerDataType) -> Self { + match value { + proto::SoxResamplerDataType::SoxrDatatypeInt16i => Self::Interleaved, + proto::SoxResamplerDataType::SoxrDatatypeInt16s => Self::Split, + } + } +} + +pub(crate) struct QualitySpec { + pub quality: proto::SoxQualityRecipe, + pub flags: u32, // proto::SoxQualityFlags +} + +impl From for resampler::QualitySpec { + fn from(value: QualitySpec) -> Self { + Self { quality: value.quality.into(), flags: value.flags } + } +} + +impl From for resampler::SoxQualityRecipe { + fn from(value: proto::SoxQualityRecipe) -> Self { + match value { + proto::SoxQualityRecipe::SoxrQualityQuick => Self::Quick, + proto::SoxQualityRecipe::SoxrQualityLow => Self::Low, + proto::SoxQualityRecipe::SoxrQualityMedium => Self::Medium, + proto::SoxQualityRecipe::SoxrQualityHigh => Self::High, + proto::SoxQualityRecipe::SoxrQualityVeryhigh => Self::VeryHigh, + } + } +} diff --git a/livekit-ffi/src/server/requests.rs b/livekit-ffi/src/server/requests.rs index 8ee569885..e83e6c51a 100644 --- a/livekit-ffi/src/server/requests.rs +++ b/livekit-ffi/src/server/requests.rs @@ -30,7 +30,7 @@ use super::{ room::{self, FfiPublication, FfiTrack}, video_source, video_stream, FfiError, FfiResult, FfiServer, }; -use crate::proto; +use crate::{conversion, proto}; /// Dispose the server, close all rooms and clean up all handles /// It is not mandatory to call this function. @@ -821,15 +821,15 @@ fn on_get_session_stats( } fn on_new_sox_resampler( - server: &'static FfiServer, + _server: &'static FfiServer, new_soxr: proto::NewSoxResamplerRequest, ) -> FfiResult { - let io_spec = resampler::IOSpec { + let io_spec = conversion::resampler::IOSpec { input_type: new_soxr.input_data_type(), output_type: new_soxr.output_data_type(), }; - let quality_spec = resampler::QualitySpec { + let quality_spec = conversion::resampler::QualitySpec { quality: new_soxr.quality_recipe(), flags: new_soxr.flags.unwrap_or(0), }; @@ -840,16 +840,12 @@ fn on_new_sox_resampler( new_soxr.input_rate, new_soxr.output_rate, new_soxr.num_channels, - io_spec, - quality_spec, + io_spec.into(), + quality_spec.into(), runtime_spec, ) { Ok(resampler) => { - let resampler = Arc::new(Mutex::new(resampler)); - - let handle_id = server.next_id(); - server.store_handle(handle_id, resampler); - + let handle_id = resampler.ffi_handle_id(); Ok(proto::NewSoxResamplerResponse { message: Some(proto::new_sox_resampler_response::Message::Resampler( proto::OwnedSoxResampler { @@ -869,9 +865,8 @@ fn on_push_sox_resampler( server: &'static FfiServer, push: proto::PushSoxResamplerRequest, ) -> FfiResult { - let resampler = server - .retrieve_handle::>>(push.resampler_handle)? - .clone(); + let resampler = + server.retrieve_handle::>(push.resampler_handle)?.clone(); let data_ptr = push.data_ptr; let data_size = push.size; @@ -883,7 +878,6 @@ fn on_push_sox_resampler( ) }; - let mut resampler = resampler.lock(); match resampler.push(data) { Ok(output) => { if output.is_empty() { @@ -910,11 +904,9 @@ fn on_flush_sox_resampler( server: &'static FfiServer, flush: proto::FlushSoxResamplerRequest, ) -> FfiResult { - let resampler = server - .retrieve_handle::>>(flush.resampler_handle)? - .clone(); + let resampler = + server.retrieve_handle::>(flush.resampler_handle)?.clone(); - let mut resampler = resampler.lock(); match resampler.flush() { Ok(output) => Ok(proto::FlushSoxResamplerResponse { output_ptr: output.as_ptr() as u64, diff --git a/livekit-ffi/src/server/resampler.rs b/livekit-ffi/src/server/resampler.rs index aedc8e0a1..b4496d350 100644 --- a/livekit-ffi/src/server/resampler.rs +++ b/livekit-ffi/src/server/resampler.rs @@ -15,27 +15,82 @@ use std::{ ffi::c_char, os::raw::{c_ulong, c_void}, + sync::Arc, }; +use parking_lot::Mutex; use soxr_sys; -use crate::proto; - +#[derive(uniffi::Record)] pub struct IOSpec { - pub input_type: proto::SoxResamplerDataType, - pub output_type: proto::SoxResamplerDataType, + pub input_type: SoxResamplerDataType, + pub output_type: SoxResamplerDataType, } +#[derive(uniffi::Record)] pub struct QualitySpec { - pub quality: proto::SoxQualityRecipe, - pub flags: u32, // proto::SoxQualityFlags + pub quality: SoxQualityRecipe, + pub flags: u32, } +#[derive(uniffi::Record)] pub struct RuntimeSpec { pub num_threads: u32, } +#[derive(uniffi::Object)] +/// New resampler using SoX (much better quality) pub struct SoxResampler { + inner: Mutex, + // After the migration is complete, this field can be removed. + handle_id: ::std::sync::OnceLock, +} +// After the migration is complete, this call can be removed. +crate::migrate_from_ffi!(SoxResampler); + +unsafe impl Send for SoxResampler {} +unsafe impl Sync for SoxResampler {} + +#[uniffi::export] +impl SoxResampler { + #[uniffi::constructor] + pub fn new( + input_rate: f64, + output_rate: f64, + num_channels: u32, + io_spec: IOSpec, + quality_spec: QualitySpec, + runtime_spec: RuntimeSpec, + ) -> Result, SoxResamplerError> { + let inner: SoxResamplerInner = SoxResamplerInner::new( + input_rate, + output_rate, + num_channels, + io_spec, + quality_spec, + runtime_spec, + ) + .map_err(|s| SoxResamplerError::NewError(s))?; + + let obj = Self { inner: Mutex::new(inner), handle_id: ::std::sync::OnceLock::new() }; + + Ok(Arc::new(obj)) + } + + pub fn push(&self, input: &[i16]) -> Result, SoxResamplerError> { + let mut inner = self.inner.lock(); + let output_slice = inner.push(input).map_err(|s| SoxResamplerError::PushError(s))?; + Ok(output_slice.to_vec()) + } + + pub fn flush(&self) -> Result, SoxResamplerError> { + let mut inner = self.inner.lock(); + let output_slice = inner.flush().map_err(|s| SoxResamplerError::FlushError(s))?; + Ok(output_slice.to_vec()) + } +} + +struct SoxResamplerInner { soxr_ptr: soxr_sys::soxr_t, out_buf: Vec, input_rate: f64, @@ -43,9 +98,7 @@ pub struct SoxResampler { num_channels: u32, } -unsafe impl Send for SoxResampler {} - -impl SoxResampler { +impl SoxResamplerInner { pub fn new( input_rate: f64, output_rate: f64, @@ -57,10 +110,8 @@ impl SoxResampler { let error: *mut *const c_char = std::ptr::null_mut(); let soxr_ptr = unsafe { - let io_spec = soxr_sys::soxr_io_spec( - to_soxr_datatype(io_spec.input_type), - to_soxr_datatype(io_spec.output_type), - ); + let io_spec = + soxr_sys::soxr_io_spec(io_spec.input_type.into(), io_spec.output_type.into()); let quality_spec = soxr_sys::soxr_quality_spec( quality_spec.quality as c_ulong, @@ -159,7 +210,7 @@ impl SoxResampler { } } -impl Drop for SoxResampler { +impl Drop for SoxResamplerInner { fn drop(&mut self) { unsafe { soxr_sys::soxr_delete(self.soxr_ptr); @@ -167,13 +218,71 @@ impl Drop for SoxResampler { } } -fn to_soxr_datatype(datatype: proto::SoxResamplerDataType) -> soxr_sys::soxr_datatype_t { - match datatype { - proto::SoxResamplerDataType::SoxrDatatypeInt16i => soxr_sys::soxr_datatype_t_SOXR_INT16_I, - proto::SoxResamplerDataType::SoxrDatatypeInt16s => soxr_sys::soxr_datatype_t_SOXR_INT16_S, +#[derive(uniffi::Enum)] +// TODO(theomonnom): support other datatypes (shouldn't really be needed) +pub enum SoxResamplerDataType { + Interleaved, + Split, +} + +impl From for soxr_sys::soxr_datatype_t { + fn from(value: SoxResamplerDataType) -> Self { + match value { + SoxResamplerDataType::Interleaved => soxr_sys::soxr_datatype_t_SOXR_INT16_I, + SoxResamplerDataType::Split => soxr_sys::soxr_datatype_t_SOXR_INT16_S, + } } } +#[derive(uniffi::Enum)] +pub enum SoxQualityRecipe { + Quick, + Low, + Medium, + High, + VeryHigh, +} + +/// SoX recipe numbers are not contiguous, so they are mapped here rather than +/// carried as enum discriminants. `SOXR_HQ` and `SOXR_VHQ` are aliases that +/// bindgen does not emit, so their targets are named directly. +impl From for c_ulong { + fn from(value: SoxQualityRecipe) -> Self { + let recipe = match value { + SoxQualityRecipe::Quick => soxr_sys::SOXR_QQ, + SoxQualityRecipe::Low => soxr_sys::SOXR_LQ, + SoxQualityRecipe::Medium => soxr_sys::SOXR_MQ, + SoxQualityRecipe::High => soxr_sys::SOXR_20_BITQ, + SoxQualityRecipe::VeryHigh => soxr_sys::SOXR_28_BITQ, + }; + recipe as c_ulong + } +} + +#[cfg(test)] +mod quality_recipe_tests { + use super::*; + + #[test] + fn recipes_map_to_sox_constants() { + assert_eq!(c_ulong::from(SoxQualityRecipe::Quick), 0); + assert_eq!(c_ulong::from(SoxQualityRecipe::Low), 1); + assert_eq!(c_ulong::from(SoxQualityRecipe::Medium), 2); + assert_eq!(c_ulong::from(SoxQualityRecipe::High), 4); + assert_eq!(c_ulong::from(SoxQualityRecipe::VeryHigh), 6); + } +} + +#[derive(Debug, thiserror::Error, uniffi::Error)] +pub enum SoxResamplerError { + #[error("{0}")] + NewError(String), + #[error("{0}")] + PushError(String), + #[error("{0}")] + FlushError(String), +} + /// Direct-call tests for [`SoxResampler`], paired with the protobuf-driven tests /// in `crate::migration_tests`. Both sets are written to compile and pass on /// either side of the uniffi migration, so they can be replayed across it. From 88368bfa1560a04c911b1223c2f4806be57e6f6e Mon Sep 17 00:00:00 2001 From: James Hugman Date: Sat, 12 Sep 2026 15:01:02 +0100 Subject: [PATCH 4/6] Read the sox resampler's FFI output under the lock, with push_ffi/flush_ffi The exported push and flush return their output by value, so the pointer the FFI response hands the client would aim at a buffer freed while the response is built. Give the request surface its own pair of methods that return the lock along with a slice of out_buf, which is what it read before: the samples stay put until the client's next call. They go with the rest of the FFI handle plumbing once the migration is done. Caught by the protobuf migration tests, which abort on the dangling pointer. --- livekit-ffi/src/server/requests.rs | 6 ++++-- livekit-ffi/src/server/resampler.rs | 20 +++++++++++++++++++- 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/livekit-ffi/src/server/requests.rs b/livekit-ffi/src/server/requests.rs index e83e6c51a..66bd2df81 100644 --- a/livekit-ffi/src/server/requests.rs +++ b/livekit-ffi/src/server/requests.rs @@ -878,7 +878,8 @@ fn on_push_sox_resampler( ) }; - match resampler.push(data) { + let output = resampler.push_ffi(data); + match output { Ok(output) => { if output.is_empty() { return Ok(proto::PushSoxResamplerResponse { @@ -907,7 +908,8 @@ fn on_flush_sox_resampler( let resampler = server.retrieve_handle::>(flush.resampler_handle)?.clone(); - match resampler.flush() { + let output = resampler.flush_ffi(); + match output { Ok(output) => Ok(proto::FlushSoxResamplerResponse { output_ptr: output.as_ptr() as u64, size: (output.len() * std::mem::size_of::()) as u32, diff --git a/livekit-ffi/src/server/resampler.rs b/livekit-ffi/src/server/resampler.rs index b4496d350..4e8413621 100644 --- a/livekit-ffi/src/server/resampler.rs +++ b/livekit-ffi/src/server/resampler.rs @@ -18,7 +18,7 @@ use std::{ sync::Arc, }; -use parking_lot::Mutex; +use parking_lot::{MappedMutexGuard, Mutex, MutexGuard}; use soxr_sys; #[derive(uniffi::Record)] @@ -90,6 +90,24 @@ impl SoxResampler { } } +// After the migration is complete, this impl can be removed. +impl SoxResampler { + pub(crate) fn push_ffi( + &self, + input: &[i16], + ) -> Result, SoxResamplerError> { + let mut inner = self.inner.lock(); + let output_len = inner.push(input).map_err(|s| SoxResamplerError::PushError(s))?.len(); + Ok(MutexGuard::map(inner, |inner| &mut inner.out_buf[..output_len])) + } + + pub(crate) fn flush_ffi(&self) -> Result, SoxResamplerError> { + let mut inner = self.inner.lock(); + let output_len = inner.flush().map_err(|s| SoxResamplerError::FlushError(s))?.len(); + Ok(MutexGuard::map(inner, |inner| &mut inner.out_buf[..output_len])) + } +} + struct SoxResamplerInner { soxr_ptr: soxr_sys::soxr_t, out_buf: Vec, From e1fac0bc49d51818fd1bdc83e3dc325e1b962f7a Mon Sep 17 00:00:00 2001 From: James Hugman Date: Thu, 10 Sep 2026 17:43:52 +0100 Subject: [PATCH 5/6] Refine the transformation to match the proto NewSoxResamplerRequest --- livekit-ffi/src/conversion/resampler.rs | 22 ---------------------- livekit-ffi/src/server/requests.rs | 21 +++++---------------- livekit-ffi/src/server/resampler.rs | 24 ++++++++++++++++++++++-- 3 files changed, 27 insertions(+), 40 deletions(-) diff --git a/livekit-ffi/src/conversion/resampler.rs b/livekit-ffi/src/conversion/resampler.rs index 295b9831f..6dea5e979 100644 --- a/livekit-ffi/src/conversion/resampler.rs +++ b/livekit-ffi/src/conversion/resampler.rs @@ -14,17 +14,6 @@ use crate::{proto, server::resampler}; -pub(crate) struct IOSpec { - pub input_type: proto::SoxResamplerDataType, - pub output_type: proto::SoxResamplerDataType, -} - -impl From for resampler::IOSpec { - fn from(value: IOSpec) -> Self { - Self { input_type: value.input_type.into(), output_type: value.output_type.into() } - } -} - impl From for resampler::SoxResamplerDataType { fn from(value: proto::SoxResamplerDataType) -> Self { match value { @@ -34,17 +23,6 @@ impl From for resampler::SoxResamplerDataType { } } -pub(crate) struct QualitySpec { - pub quality: proto::SoxQualityRecipe, - pub flags: u32, // proto::SoxQualityFlags -} - -impl From for resampler::QualitySpec { - fn from(value: QualitySpec) -> Self { - Self { quality: value.quality.into(), flags: value.flags } - } -} - impl From for resampler::SoxQualityRecipe { fn from(value: proto::SoxQualityRecipe) -> Self { match value { diff --git a/livekit-ffi/src/server/requests.rs b/livekit-ffi/src/server/requests.rs index 66bd2df81..ce2748ae6 100644 --- a/livekit-ffi/src/server/requests.rs +++ b/livekit-ffi/src/server/requests.rs @@ -30,7 +30,7 @@ use super::{ room::{self, FfiPublication, FfiTrack}, video_source, video_stream, FfiError, FfiResult, FfiServer, }; -use crate::{conversion, proto}; +use crate::proto; /// Dispose the server, close all rooms and clean up all handles /// It is not mandatory to call this function. @@ -824,25 +824,14 @@ fn on_new_sox_resampler( _server: &'static FfiServer, new_soxr: proto::NewSoxResamplerRequest, ) -> FfiResult { - let io_spec = conversion::resampler::IOSpec { - input_type: new_soxr.input_data_type(), - output_type: new_soxr.output_data_type(), - }; - - let quality_spec = conversion::resampler::QualitySpec { - quality: new_soxr.quality_recipe(), - flags: new_soxr.flags.unwrap_or(0), - }; - - let runtime_spec = resampler::RuntimeSpec { num_threads: 1 }; - match resampler::SoxResampler::new( new_soxr.input_rate, new_soxr.output_rate, new_soxr.num_channels, - io_spec.into(), - quality_spec.into(), - runtime_spec, + new_soxr.input_data_type().into(), + new_soxr.output_data_type().into(), + new_soxr.quality_recipe().into(), + new_soxr.flags.unwrap_or(0), ) { Ok(resampler) => { let handle_id = resampler.ffi_handle_id(); diff --git a/livekit-ffi/src/server/resampler.rs b/livekit-ffi/src/server/resampler.rs index 4e8413621..ab51acbf2 100644 --- a/livekit-ffi/src/server/resampler.rs +++ b/livekit-ffi/src/server/resampler.rs @@ -54,7 +54,7 @@ unsafe impl Sync for SoxResampler {} #[uniffi::export] impl SoxResampler { #[uniffi::constructor] - pub fn new( + pub fn new_with_options( input_rate: f64, output_rate: f64, num_channels: u32, @@ -77,6 +77,26 @@ impl SoxResampler { Ok(Arc::new(obj)) } + #[uniffi::constructor(default(flags = 0))] + pub fn new( + input_rate: f64, + output_rate: f64, + num_channels: u32, + input_data_type: SoxResamplerDataType, + output_data_type: SoxResamplerDataType, + quality_recipe: SoxQualityRecipe, + flags: u32, + ) -> Result, SoxResamplerError> { + Self::new_with_options( + input_rate, + output_rate, + num_channels, + IOSpec { input_type: input_data_type, output_type: output_data_type }, + QualitySpec { quality: quality_recipe, flags }, + RuntimeSpec { num_threads: 1 }, + ) + } + pub fn push(&self, input: &[i16]) -> Result, SoxResamplerError> { let mut inner = self.inner.lock(); let output_slice = inner.push(input).map_err(|s| SoxResamplerError::PushError(s))?; @@ -322,7 +342,7 @@ mod migration_tests { #[macro_export] macro_rules! sox_resampler { ($input_rate:expr, $output_rate:expr, $num_channels:expr, $quality:expr) => { - $crate::server::resampler::SoxResampler::new( + $crate::server::resampler::SoxResampler::new_with_options( $input_rate, $output_rate, $num_channels, From ff67917e9658c8fee30a769c93907c86370ed034 Mon Sep 17 00:00:00 2001 From: James Hugman Date: Sat, 12 Sep 2026 15:09:28 +0100 Subject: [PATCH 6/6] Add a changeset for the SoX resampler UniFFI migration --- .changeset/sox_resampler_uniffi.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .changeset/sox_resampler_uniffi.md diff --git a/.changeset/sox_resampler_uniffi.md b/.changeset/sox_resampler_uniffi.md new file mode 100644 index 000000000..d43bff850 --- /dev/null +++ b/.changeset/sox_resampler_uniffi.md @@ -0,0 +1,13 @@ +--- +livekit-ffi: patch +--- + +# Export SoxResampler through UniFFI + +`SoxResampler` is now a UniFFI object, so foreign hosts can construct one and +drive `push` / `flush` directly, receiving the resampled samples by value. + +The FFI request surface is unchanged: `NewSoxResampler`, `PushSoxResampler` and +`FlushSoxResampler` still hand back a pointer into the resampler's own buffer, +readable until the next call on that resampler. Both paths are covered by new +tests that drive them with the same input and compare the results.