fix(flutter): read the live DPR and clamp the painted cursor like the native one

CanvasModel caches devicePixelRatio and only refreshes it when the view
style changes, so after the window moves to a monitor with a different
DPR the unzoomed cursor kept the previous monitor's scale. Read it from
MediaQuery instead, which also rebuilds the cursor when it changes.

The native path clamps the scaled bitmap to kMinCursorSize; apply the
same clamp to the painted cursor so a small cursor does not change size
when the peer moves the mouse.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K8HSHTEHo27mDcJXVyVfSP

Cargo.lock

@@ -985,9 +985,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
 [[package]]
 name = "bytes"
-version = "1.11.1"
+version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
+checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a"
 dependencies = [
  "serde 1.0.228",
 ]
@@ -1791,9 +1791,9 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-epoch"
-version = "0.9.20"
+version = "0.9.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
+checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
 dependencies = [
  "crossbeam-utils",
 ]
@@ -6496,9 +6496,9 @@ dependencies = [
 
 [[package]]
 name = "quinn-proto"
-version = "0.11.15"
+version = "0.11.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e"
+checksum = "f1906b49b0c3bc04b5fe5d86a77925ae6524a19b816ae38ce1e426255f1d8a31"
 dependencies = [
  "bytes",
  "getrandom 0.3.2",
@@ -7139,7 +7139,6 @@ dependencies = [
  "lazy_static",
  "libpulse-binding",
  "libpulse-simple-binding",
- "libsamplerate-sys",
  "libxdo-sys",
  "mac_address",
  "magnum-opus",

Cargo.toml

@@ -22,7 +22,7 @@ path = "src/service.rs"
 
 [features]
 inline = []
-use_samplerate = ["samplerate", "libsamplerate-sys"]
+use_samplerate = ["samplerate"]
 use_rubato = ["rubato"]
 use_dasp = ["dasp"]
 flutter = ["flutter_rust_bridge"]
@@ -67,7 +67,6 @@ magnum-opus = { git = "https://github.com/rustdesk-org/magnum-opus" }
 dasp = { version = "0.11", features = ["signal", "interpolate-linear", "interpolate"], optional = true }
 rubato = { version = "0.12", optional = true }
 samplerate = { version = "0.2", optional = true }
-libsamplerate-sys = { version = "0.1.12", optional = true }
 uuid = { version = "1.3", features = ["v4"] }
 num_cpus = "1.15"
 bytes = { version = "1.4", features = ["serde"] }

flutter/lib/desktop/pages/remote_page.dart

@@ -1101,6 +1101,9 @@ class _ImagePaintState extends State<ImagePaint> {
     final m = Provider.of<ImageModel>(context);
     var c = Provider.of<CanvasModel>(context);
     final s = c.scale;
+    // CanvasModel caches the DPR and only refreshes it when the view style
+    // changes, so read it live to follow the window across monitors.
+    final dpr = MediaQuery.devicePixelRatioOf(context);
 
     bool isViewAdaptive() => c.viewStyle.style == kRemoteViewStyleAdaptive;
     bool isViewOriginal() => c.viewStyle.style == kRemoteViewStyleOriginal;
@@ -1117,6 +1120,12 @@ class _ImagePaintState extends State<ImagePaint> {
             } else {
               if (zoomCursor.value || isViewOriginal()) {
                 cursorScale = s;
+              } else {
+                // NSCursor and GdkCursor treat the bitmap size as logical
+                // pixels, so an unzoomed cursor must be shrunk by the DPR to
+                // keep 1 remote px == 1 physical px, the size Original view
+                // already renders it at.
+                cursorScale = 1.0 / dpr;
               }
             }
             return cursorScale;
@@ -1404,14 +1413,29 @@ class CursorPaint extends StatelessWidget {
       }
     }
 
-    double x = (m.x - hotx) * c.scale + cx;
+    double x = m.x * c.scale + cx - hotx;
-    double y = (m.y - hoty) * c.scale + cy;
+    double y = m.y * c.scale + cy - hoty;
     double scale = 1.0;
     final isViewOriginal = c.viewStyle.style == kRemoteViewStyleOriginal;
     if (zoomCursor.value || isViewOriginal) {
       x = m.x - hotx + cx / c.scale;
       y = m.y - hoty + cy / c.scale;
       scale = c.scale;
+    } else if (!isWindows) {
+      // Keep the painted cursor the same physical size as the native one
+      // built by getCursorScale() above, including its min-size clamp.
+      scale = 1.0 / MediaQuery.devicePixelRatioOf(context);
+      final image = m.image ?? preDefaultCursor.image;
+      if (scale != 1.0 &&
+          image != null &&
+          ((image.width * scale).toInt() < kMinCursorSize ||
+              (image.height * scale).toInt() < kMinCursorSize)) {
+        final sw = kMinCursorSize / image.width;
+        final sh = kMinCursorSize / image.height;
+        scale = sw < sh ? sh : sw;
+      }
+      x = (m.x * c.scale + cx) / scale - hotx;
+      y = (m.y * c.scale + cy) / scale - hoty;
     }
 
     return CustomPaint(

flutter/lib/models/model.dart

@@ -2884,7 +2884,7 @@ class CursorData {
     if (scale != 1.0) {
       // Update data if scale changed.
       final tgtWidth = (width * scale).toInt();
-      final tgtHeight = (width * scale).toInt();
+      final tgtHeight = (height * scale).toInt();
       if (tgtWidth < kMinCursorSize || tgtHeight < kMinCursorSize) {
         double sw = kMinCursorSize.toDouble() / width;
         double sh = kMinCursorSize.toDouble() / height;

src/audio_resampler.rs

@@ -1,242 +0,0 @@
-use hbb_common::thiserror;
-
-#[cfg(test)]
-pub(crate) mod allocation_tests;
-
-#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
-mod sinc;
-
-const INTERPOLATION_MARGIN_FRAMES: usize = 2;
-const PENDING_PACKET_CAPACITY: usize = 2;
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(crate) struct AudioResamplerConfig {
-    pub input_rate: u32,
-    pub output_rate: u32,
-    pub channels: u16,
-}
-
-#[derive(Debug, thiserror::Error, PartialEq, Eq)]
-pub(crate) enum AudioResamplerError {
-    #[error(
-        "invalid audio resampler configuration: input_rate={}, output_rate={}, channels={}",
-        .0.input_rate, .0.output_rate, .0.channels
-    )]
-    InvalidConfig(AudioResamplerConfig),
-    #[error("invalid resampler output frame size: {output_frames}")]
-    InvalidOutputFrameSize { output_frames: usize },
-    #[error("audio resampler input length {samples} is not divisible by channel count {channels}")]
-    IncompleteFrame { samples: usize, channels: usize },
-    #[error("audio resampler output capacity overflow")]
-    CapacityOverflow,
-    #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
-    #[error("audio resampler backend failed: {0}")]
-    Backend(String),
-}
-
-pub(crate) struct FixedFrameAudioResampler {
-    resampler: AudioResampler,
-    output_samples: usize,
-    pending_samples: Vec<f32>,
-}
-
-#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
-// SAFETY: libsamplerate's src_new state owns heap data and has no thread affinity.
-// This wrapper never exposes or shares that state; processing requires &mut self.
-unsafe impl Send for FixedFrameAudioResampler {}
-
-impl FixedFrameAudioResampler {
-    pub(crate) fn new(
-        config: AudioResamplerConfig,
-        output_frames: usize,
-    ) -> Result<Self, AudioResamplerError> {
-        if output_frames == 0 {
-            return Err(AudioResamplerError::InvalidOutputFrameSize { output_frames });
-        }
-        let channels = validate_config(config)?;
-        let output_samples = output_frames
-            .checked_mul(channels)
-            .ok_or(AudioResamplerError::CapacityOverflow)?;
-        let input_frames = output_frames
-            .checked_mul(config.input_rate as usize)
-            .ok_or(AudioResamplerError::CapacityOverflow)?
-            .div_ceil(config.output_rate as usize);
-        let capacity = output_samples
-            .checked_mul(PENDING_PACKET_CAPACITY)
-            .and_then(|samples| samples.checked_add(channels * INTERPOLATION_MARGIN_FRAMES))
-            .ok_or(AudioResamplerError::CapacityOverflow)?;
-        let mut resampler = AudioResampler::new(config)?;
-        resampler.reserve_input(input_frames)?;
-        Ok(Self {
-            resampler,
-            output_samples,
-            pending_samples: Vec::with_capacity(capacity),
-        })
-    }
-
-    pub(crate) fn process_with(
-        &mut self,
-        input: &[f32],
-        mut on_packet: impl FnMut(&[f32]),
-    ) -> Result<(), AudioResamplerError> {
-        self.resampler
-            .process_into(input, &mut self.pending_samples)?;
-        let complete_samples =
-            self.pending_samples.len() / self.output_samples * self.output_samples;
-        for packet in self.pending_samples[..complete_samples].chunks_exact(self.output_samples) {
-            on_packet(packet);
-        }
-        self.pending_samples.drain(..complete_samples);
-        Ok(())
-    }
-
-    #[cfg(test)]
-    pub(crate) fn process(&mut self, input: &[f32]) -> Result<Vec<Vec<f32>>, AudioResamplerError> {
-        let mut packets = Vec::new();
-        self.process_with(input, |packet| packets.push(packet.to_owned()))?;
-        Ok(packets)
-    }
-}
-
-pub(crate) struct AudioResampler {
-    #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
-    backend: StreamingLinearAudioResampler,
-    #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
-    backend: sinc::SincAudioResampler,
-}
-
-impl AudioResampler {
-    pub(crate) fn new(config: AudioResamplerConfig) -> Result<Self, AudioResamplerError> {
-        Ok(Self {
-            #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
-            backend: sinc::SincAudioResampler::new(config)?,
-            #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
-            backend: StreamingLinearAudioResampler::new(config)?,
-        })
-    }
-
-    pub(crate) fn process(&mut self, input: &[f32]) -> Result<Vec<f32>, AudioResamplerError> {
-        let mut output = Vec::new();
-        self.process_into(input, &mut output)?;
-        Ok(output)
-    }
-
-    // Append samples so capture can retain an incomplete output packet in the same buffer.
-    fn process_into(
-        &mut self,
-        input: &[f32],
-        output: &mut Vec<f32>,
-    ) -> Result<(), AudioResamplerError> {
-        self.backend.process_into(input, output)
-    }
-
-    fn reserve_input(&mut self, _frames: usize) -> Result<(), AudioResamplerError> {
-        #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
-        {
-            let capacity = _frames
-                .checked_add(INTERPOLATION_MARGIN_FRAMES)
-                .and_then(|frames| frames.checked_mul(self.backend.channels))
-                .ok_or(AudioResamplerError::CapacityOverflow)?;
-            self.backend.buffered_samples.reserve(capacity);
-        }
-        Ok(())
-    }
-}
-
-#[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
-struct StreamingLinearAudioResampler {
-    config: AudioResamplerConfig,
-    channels: usize,
-    buffered_samples: Vec<f32>,
-    next_position: u64,
-}
-
-#[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
-impl StreamingLinearAudioResampler {
-    fn new(config: AudioResamplerConfig) -> Result<Self, AudioResamplerError> {
-        Ok(Self {
-            config,
-            channels: validate_config(config)?,
-            buffered_samples: Vec::new(),
-            next_position: 0,
-        })
-    }
-
-    fn process_into(
-        &mut self,
-        input: &[f32],
-        output: &mut Vec<f32>,
-    ) -> Result<(), AudioResamplerError> {
-        validate_input(input, self.channels)?;
-        let capacity = self.output_capacity(input.len())?;
-        output.reserve(capacity);
-        self.buffered_samples.extend_from_slice(input);
-        while self.write_next_frame(output) {
-            self.next_position += self.config.input_rate as u64;
-        }
-        self.discard_consumed_frames();
-        Ok(())
-    }
-
-    fn output_capacity(&self, input_samples: usize) -> Result<usize, AudioResamplerError> {
-        let input_frames = input_samples / self.channels;
-        let scaled_frames = input_frames
-            .checked_mul(self.config.output_rate as usize)
-            .ok_or(AudioResamplerError::CapacityOverflow)?
-            / self.config.input_rate as usize;
-        scaled_frames
-            .checked_add(INTERPOLATION_MARGIN_FRAMES)
-            .and_then(|frames| frames.checked_mul(self.channels))
-            .ok_or(AudioResamplerError::CapacityOverflow)
-    }
-
-    fn write_next_frame(&self, output: &mut Vec<f32>) -> bool {
-        let output_rate = self.config.output_rate as u64;
-        let frame_count = self.buffered_samples.len() / self.channels;
-        let frame = (self.next_position / output_rate) as usize;
-        let fraction = self.next_position % output_rate;
-        if frame >= frame_count || (fraction != 0 && frame + 1 >= frame_count) {
-            return false;
-        }
-        let weight = fraction as f32 / output_rate as f32;
-        for channel in 0..self.channels {
-            let current = self.buffered_samples[frame * self.channels + channel];
-            let next_frame = frame + usize::from(fraction != 0);
-            let next = self.buffered_samples[next_frame * self.channels + channel];
-            output.push(current + (next - current) * weight);
-        }
-        true
-    }
-
-    fn discard_consumed_frames(&mut self) {
-        let output_rate = self.config.output_rate as u64;
-        let available_frames = self.buffered_samples.len() / self.channels;
-        let consumed_frames = ((self.next_position / output_rate) as usize).min(available_frames);
-        self.buffered_samples
-            .drain(0..consumed_frames * self.channels);
-        self.next_position -= consumed_frames as u64 * output_rate;
-    }
-}
-
-fn validate_config(config: AudioResamplerConfig) -> Result<usize, AudioResamplerError> {
-    if config.input_rate == 0 || config.output_rate == 0 || config.channels == 0 {
-        return Err(AudioResamplerError::InvalidConfig(config));
-    }
-    Ok(config.channels as usize)
-}
-
-fn validate_input(input: &[f32], channels: usize) -> Result<(), AudioResamplerError> {
-    if input.len() % channels != 0 {
-        return Err(AudioResamplerError::IncompleteFrame {
-            samples: input.len(),
-            channels,
-        });
-    }
-    Ok(())
-}
-
-#[cfg(all(test, not(all(feature = "use_samplerate", not(feature = "use_dasp")))))]
-mod tests;
-
-#[cfg(all(test, feature = "use_samplerate", not(feature = "use_dasp")))]
-mod samplerate_tests;

src/audio_resampler/allocation_tests.rs

@@ -1,142 +0,0 @@
-use super::{AudioResamplerConfig, FixedFrameAudioResampler};
-use std::alloc::{GlobalAlloc, Layout, System};
-use std::cell::Cell;
-
-struct CountingAllocator;
-
-thread_local! {
-    static ALLOCATIONS: Cell<Option<usize>> = const { Cell::new(None) };
-}
-
-fn record_allocation() {
-    let _ = ALLOCATIONS.try_with(|count| {
-        if let Some(value) = count.get() {
-            count.set(Some(value + 1));
-        }
-    });
-}
-
-unsafe impl GlobalAlloc for CountingAllocator {
-    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
-        record_allocation();
-        unsafe { System.alloc(layout) }
-    }
-
-    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
-        record_allocation();
-        unsafe { System.alloc_zeroed(layout) }
-    }
-
-    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 {
-        record_allocation();
-        unsafe { System.realloc(ptr, layout, size) }
-    }
-
-    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
-        unsafe { System.dealloc(ptr, layout) }
-    }
-}
-
-#[global_allocator]
-static ALLOCATOR: CountingAllocator = CountingAllocator;
-
-pub(crate) fn assert_no_allocations(process: impl FnOnce()) {
-    struct ResetCounter;
-    impl Drop for ResetCounter {
-        fn drop(&mut self) {
-            ALLOCATIONS.with(|count| count.set(None));
-        }
-    }
-
-    ALLOCATIONS.with(|count| assert!(count.replace(Some(0)).is_none()));
-    let reset = ResetCounter;
-    process();
-    let allocations = ALLOCATIONS.with(|count| count.get().unwrap());
-    drop(reset);
-    assert_eq!(
-        allocations, 0,
-        "PCM processing allocated on the capture thread"
-    );
-}
-
-#[test]
-fn capture_resampling_reuses_buffers() {
-    const PACKETS_PER_SECOND: usize = 100;
-    const PACKET_COUNT: usize = 100;
-    const MAX_STARTUP_DELAY_PACKETS: usize = 1;
-    const SIGNAL_LEVEL: f32 = 0.25;
-    const RATE_PAIRS: [(u32, u32); 6] = [
-        (32_000, 24_000),
-        (44_100, 24_000),
-        (44_100, 48_000),
-        (48_000, 24_000),
-        (96_000, 48_000),
-        (192_000, 48_000),
-    ];
-
-    for (input_rate, output_rate) in RATE_PAIRS {
-        for channels in [1, 2, 4, 6, 8] {
-            let config = AudioResamplerConfig {
-                input_rate,
-                output_rate,
-                channels,
-            };
-            let input =
-                vec![SIGNAL_LEVEL; input_rate as usize / PACKETS_PER_SECOND * channels as usize];
-            let frames = output_rate as usize / PACKETS_PER_SECOND;
-            let mut resampler = FixedFrameAudioResampler::new(config, frames).unwrap();
-            let mut packets = 0;
-            let mut energy = 0.0;
-            assert_no_allocations(|| {
-                for _ in 0..PACKET_COUNT {
-                    resampler
-                        .process_with(&input, |packet| {
-                            assert_eq!(packet.len(), frames * channels as usize);
-                            energy += packet.iter().map(|sample| sample * sample).sum::<f32>();
-                            packets += 1;
-                        })
-                        .unwrap();
-                }
-            });
-            assert!((PACKET_COUNT - MAX_STARTUP_DELAY_PACKETS..=PACKET_COUNT).contains(&packets));
-            assert!(energy > SIGNAL_LEVEL);
-        }
-    }
-}
-
-#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
-#[test]
-fn sinc_output_matches_the_existing_backend() {
-    use super::AudioResampler;
-
-    const INPUT_FRAMES: usize = 2_048;
-    const CHUNK_FRAMES: usize = 73;
-    const SIGNAL_STEP: f32 = 0.07;
-    for (input_rate, output_rate) in [(44_100, 24_000), (44_100, 48_000), (96_000, 48_000)] {
-        for channels in [1, 2, 4, 6, 8] {
-            let config = AudioResamplerConfig {
-                input_rate,
-                output_rate,
-                channels,
-            };
-            let input: Vec<_> = (0..INPUT_FRAMES * channels as usize)
-                .map(|sample| (sample as f32 * SIGNAL_STEP).sin())
-                .collect();
-            let mut actual = AudioResampler::new(config).unwrap();
-            let expected = samplerate::Samplerate::new(
-                samplerate::ConverterType::SincBestQuality,
-                input_rate,
-                output_rate,
-                channels as usize,
-            )
-            .unwrap();
-            for chunk in input.chunks(CHUNK_FRAMES * channels as usize) {
-                assert_eq!(
-                    actual.process(chunk).unwrap(),
-                    expected.process(chunk).unwrap()
-                );
-                assert_eq!(actual.process(&[]).unwrap(), expected.process(&[]).unwrap());
-            }
-        }
-    }
-}

src/audio_resampler/samplerate_tests.rs

@@ -1,157 +0,0 @@
-use super::{AudioResampler, AudioResamplerConfig, AudioResamplerError, FixedFrameAudioResampler};
-
-const INPUT_RATE: u32 = 44_100;
-const OUTPUT_RATE: u32 = 48_000;
-const CHANNELS: u16 = 2;
-const INPUT_PACKET_FRAMES: usize = INPUT_RATE as usize / PACKETS_PER_SECOND;
-const OUTPUT_PACKET_FRAMES: usize = OUTPUT_RATE as usize / PACKETS_PER_SECOND;
-const PACKET_COUNT: usize = 20;
-const PACKETS_PER_SECOND: usize = 100;
-const MIN_CONTINUITY_PACKETS: usize = 2;
-const TONE_FREQUENCY_HZ: f32 = 997.0;
-const TONE_AMPLITUDE: f32 = 0.5;
-const MAX_BOUNDARY_RESIDUAL: f32 = 0.02;
-const INCOMPLETE_SAMPLE_COUNT: usize = 1;
-const DOWNSAMPLE_RATE: u32 = 24_000;
-const REJECTED_TONE_HZ: f64 = 18_000.0;
-const MAX_ALIAS_RMS: f64 = 0.01;
-const MIN_PASSBAND_RMS: f64 = 0.3;
-
-fn stereo_tone(frames: usize) -> Vec<f32> {
-    (0..frames)
-        .flat_map(|frame| {
-            let phase =
-                std::f32::consts::TAU * TONE_FREQUENCY_HZ * frame as f32 / INPUT_RATE as f32;
-            let sample = TONE_AMPLITUDE * phase.sin();
-            [sample, sample]
-        })
-        .collect()
-}
-
-fn maximum_boundary_residual(packets: &[Vec<f32>]) -> f32 {
-    packets.windows(2).fold(0.0, |maximum, pair| {
-        let previous = &pair[0];
-        let current = &pair[1];
-        let last = previous.len() - CHANNELS as usize;
-        let penultimate = last - CHANNELS as usize;
-        (0..CHANNELS as usize).fold(maximum, |maximum, channel| {
-            let predicted = previous[last + channel]
-                + (previous[last + channel] - previous[penultimate + channel]);
-            maximum.max((current[channel] - predicted).abs())
-        })
-    })
-}
-
-fn stereo_config() -> AudioResamplerConfig {
-    AudioResamplerConfig {
-        input_rate: INPUT_RATE,
-        output_rate: OUTPUT_RATE,
-        channels: CHANNELS,
-    }
-}
-
-#[test]
-fn moving_capture_resampler_preserves_pending_audio() {
-    let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT);
-    let packet_samples = INPUT_PACKET_FRAMES * CHANNELS as usize;
-    let mut expected_resampler =
-        FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap();
-    let expected: Vec<_> = input
-        .chunks(packet_samples)
-        .flat_map(|packet| expected_resampler.process(packet).unwrap())
-        .collect();
-    let mut moved_resampler =
-        FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap();
-    let mut output = moved_resampler.process(&input[..packet_samples]).unwrap();
-    let remaining = std::thread::spawn(move || {
-        input[packet_samples..]
-            .chunks(packet_samples)
-            .flat_map(|packet| moved_resampler.process(packet).unwrap())
-            .collect::<Vec<_>>()
-    })
-    .join()
-    .unwrap();
-    output.extend(remaining);
-
-    assert!(output.len() >= MIN_CONTINUITY_PACKETS);
-    assert!(output
-        .iter()
-        .all(|packet| packet.len() == OUTPUT_PACKET_FRAMES * CHANNELS as usize));
-    assert!(maximum_boundary_residual(&output) <= MAX_BOUNDARY_RESIDUAL);
-    assert_eq!(output, expected);
-}
-
-fn downsampled_rms(input: &[f32]) -> f64 {
-    let config = AudioResamplerConfig {
-        output_rate: DOWNSAMPLE_RATE,
-        ..stereo_config()
-    };
-    let output_frames = DOWNSAMPLE_RATE as usize / PACKETS_PER_SECOND;
-    let mut resampler = FixedFrameAudioResampler::new(config, output_frames).unwrap();
-    let output: Vec<f32> = input
-        .chunks(INPUT_PACKET_FRAMES * CHANNELS as usize)
-        .flat_map(|packet| resampler.process(packet).unwrap().into_iter().flatten())
-        .collect();
-
-    assert!(output.len() >= output_frames * CHANNELS as usize * MIN_CONTINUITY_PACKETS);
-    let mean_square = output
-        .iter()
-        .map(|sample| f64::from(*sample).powi(2))
-        .sum::<f64>()
-        / output.len() as f64;
-    mean_square.sqrt()
-}
-
-#[test]
-fn capture_downsampling_filters_out_of_band_audio() {
-    let input: Vec<_> = (0..INPUT_PACKET_FRAMES * PACKET_COUNT)
-        .flat_map(|frame| {
-            let phase =
-                std::f64::consts::TAU * REJECTED_TONE_HZ * frame as f64 / f64::from(INPUT_RATE);
-            let sample = (f64::from(TONE_AMPLITUDE) * phase.sin()) as f32;
-            [sample, sample]
-        })
-        .collect();
-    let rms = downsampled_rms(&input);
-    assert!(
-        rms < MAX_ALIAS_RMS,
-        "out-of-band output RMS {rms} exceeded {MAX_ALIAS_RMS}"
-    );
-
-    let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT);
-    let rms = downsampled_rms(&input);
-    assert!(
-        rms > MIN_PASSBAND_RMS,
-        "in-band output RMS {rms} fell below {MIN_PASSBAND_RMS}"
-    );
-}
-
-#[test]
-fn samplerate_backend_preserves_streaming_continuity() {
-    let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT);
-    let mut resampler = AudioResampler::new(stereo_config()).unwrap();
-    let packets: Vec<_> = input
-        .chunks(INPUT_PACKET_FRAMES * CHANNELS as usize)
-        .map(|packet| resampler.process(packet).unwrap())
-        .filter(|packet| !packet.is_empty())
-        .collect();
-
-    assert!(packets.len() >= MIN_CONTINUITY_PACKETS);
-    assert!(packets
-        .iter()
-        .all(|packet| packet.len() % CHANNELS as usize == 0));
-    assert!(maximum_boundary_residual(&packets) <= MAX_BOUNDARY_RESIDUAL);
-}
-
-#[test]
-fn samplerate_backend_reports_incomplete_frame_context() {
-    let mut resampler = AudioResampler::new(stereo_config()).unwrap();
-
-    assert_eq!(
-        resampler.process(&[0.0]).unwrap_err(),
-        AudioResamplerError::IncompleteFrame {
-            samples: INCOMPLETE_SAMPLE_COUNT,
-            channels: CHANNELS as usize,
-        }
-    );
-}

src/audio_resampler/sinc.rs

@@ -1,112 +0,0 @@
-use super::{AudioResamplerConfig, AudioResamplerError};
-use libsamplerate_sys as sys;
-use std::ptr::NonNull;
-
-const OUTPUT_MARGIN_FRAMES: usize = 1;
-
-pub(super) struct SincAudioResampler {
-    state: NonNull<sys::SRC_STATE>,
-    config: AudioResamplerConfig,
-}
-
-impl SincAudioResampler {
-    pub(super) fn new(config: AudioResamplerConfig) -> Result<Self, AudioResamplerError> {
-        super::validate_config(config)?;
-        let ratio = f64::from(config.output_rate) / f64::from(config.input_rate);
-        if unsafe { sys::src_is_valid_ratio(ratio) } == 0 {
-            return Err(backend_error(
-                config,
-                samplerate::ErrorCode::BadSrcRatio as _,
-            ));
-        }
-        let mut error = 0;
-        // SAFETY: src_new allocates independent state; this owner releases it in Drop.
-        let state = unsafe {
-            sys::src_new(
-                sys::SRC_SINC_BEST_QUALITY as _,
-                config.channels.into(),
-                &mut error,
-            )
-        };
-        let state = NonNull::new(state).ok_or_else(|| backend_error(config, error))?;
-        Ok(Self { state, config })
-    }
-
-    pub(super) fn process_into(
-        &mut self,
-        input: &[f32],
-        output: &mut Vec<f32>,
-    ) -> Result<(), AudioResamplerError> {
-        super::validate_input(input, self.config.channels as usize)?;
-        let mut consumed = 0;
-        loop {
-            let (used, generated) = self.process_block(&input[consumed..], output)?;
-            consumed += used;
-            if consumed == input.len() {
-                return Ok(());
-            }
-            if used == 0 && generated == 0 {
-                return Err(AudioResamplerError::Backend(
-                    "libsamplerate made no progress while input remained".to_owned(),
-                ));
-            }
-        }
-    }
-
-    fn process_block(
-        &mut self,
-        input: &[f32],
-        output: &mut Vec<f32>,
-    ) -> Result<(usize, usize), AudioResamplerError> {
-        let channels = self.config.channels as usize;
-        let input_frames = input.len() / channels;
-        let output_frames = input_frames
-            .checked_mul(self.config.output_rate as usize)
-            .map(|frames| frames / self.config.input_rate as usize)
-            .and_then(|frames| frames.checked_add(OUTPUT_MARGIN_FRAMES))
-            .ok_or(AudioResamplerError::CapacityOverflow)?;
-        let start = output.len();
-        let end = output_frames
-            .checked_mul(channels)
-            .and_then(|samples| start.checked_add(samples))
-            .ok_or(AudioResamplerError::CapacityOverflow)?;
-        let mut data = sys::SRC_DATA {
-            data_in: input.as_ptr(),
-            input_frames: input_frames
-                .try_into()
-                .map_err(|_| AudioResamplerError::CapacityOverflow)?,
-            output_frames: output_frames
-                .try_into()
-                .map_err(|_| AudioResamplerError::CapacityOverflow)?,
-            src_ratio: f64::from(self.config.output_rate) / f64::from(self.config.input_rate),
-            ..Default::default()
-        };
-        output.resize(end, 0.0);
-        data.data_out = output[start..].as_mut_ptr();
-        // SAFETY: state is exclusively owned; disjoint slices cover the declared frame counts.
-        let error = unsafe { sys::src_process(self.state.as_ptr(), &mut data) };
-        let generated = data.output_frames_gen as usize * channels;
-        output.truncate(start + generated);
-        if error != 0 {
-            return Err(backend_error(self.config, error));
-        }
-        Ok((data.input_frames_used as usize * channels, generated))
-    }
-}
-
-impl Drop for SincAudioResampler {
-    fn drop(&mut self) {
-        // SAFETY: this owner holds the only handle returned by src_new.
-        unsafe { sys::src_delete(self.state.as_ptr()) };
-    }
-}
-
-fn backend_error(config: AudioResamplerConfig, code: i32) -> AudioResamplerError {
-    AudioResamplerError::Backend(format!(
-        "input_rate={}, output_rate={}, channels={}: {:?}",
-        config.input_rate,
-        config.output_rate,
-        config.channels,
-        samplerate::Error::from_int(code)
-    ))
-}

src/audio_resampler/tests.rs

@@ -1,178 +0,0 @@
-use super::{AudioResampler, AudioResamplerConfig, FixedFrameAudioResampler};
-
-const INPUT_RATE: u32 = 24_000;
-const OUTPUT_RATE: u32 = 48_000;
-const CHANNELS: u16 = 2;
-const CHUNK_FRAMES: usize = 240;
-const CHUNK_COUNT: usize = 4;
-const TONE_FREQUENCY_HZ: f32 = 997.0;
-const TONE_AMPLITUDE: f32 = 0.5;
-const MAX_BOUNDARY_RESIDUAL: f32 = 0.02;
-const LOOK_AHEAD_OUTPUT_FRAMES: usize = 1;
-const UNEVEN_CHUNK_FRAMES: usize = 73;
-const MONO_CHANNELS: u16 = 1;
-const UNIT_RATE: u32 = 1;
-const DOUBLE_RATE: u32 = 2;
-const FIRST_DOWNSAMPLE_PACKET: [f32; 3] = [0.0, 1.0, 2.0];
-const SECOND_DOWNSAMPLE_PACKET: [f32; 4] = [3.0, 4.0, 5.0, 6.0];
-const EXPECTED_DOWNSAMPLED_OUTPUT: [f32; 4] = [0.0, 2.0, 4.0, 6.0];
-const PACKETS_PER_SECOND: usize = 100;
-const OUTPUT_PACKET_FRAMES: usize = OUTPUT_RATE as usize / PACKETS_PER_SECOND;
-const RATE_44_1_KHZ: u32 = 44_100;
-const FLOAT_TOLERANCE: f32 = 0.000_001;
-const MIN_CONTINUITY_PACKETS: usize = 2;
-
-fn stereo_tone_at_rate(frames: usize, sample_rate: u32) -> Vec<f32> {
-    (0..frames)
-        .flat_map(|frame| {
-            let phase =
-                std::f32::consts::TAU * TONE_FREQUENCY_HZ * frame as f32 / sample_rate as f32;
-            let sample = TONE_AMPLITUDE * phase.sin();
-            [sample, sample]
-        })
-        .collect()
-}
-
-fn stereo_tone(frames: usize) -> Vec<f32> {
-    stereo_tone_at_rate(frames, INPUT_RATE)
-}
-
-fn maximum_tone_prediction_residual(sample_rate: u32) -> f32 {
-    let half_step = std::f32::consts::PI * TONE_FREQUENCY_HZ / sample_rate as f32;
-    4.0 * TONE_AMPLITUDE * half_step.sin().powi(2)
-}
-
-fn maximum_boundary_residual(chunks: &[Vec<f32>]) -> f32 {
-    chunks.windows(2).fold(0.0, |maximum, pair| {
-        let previous = &pair[0];
-        let current = &pair[1];
-        let last = previous.len() - CHANNELS as usize;
-        let penultimate = last - CHANNELS as usize;
-        (0..CHANNELS as usize).fold(maximum, |maximum, channel| {
-            let predicted = previous[last + channel]
-                + (previous[last + channel] - previous[penultimate + channel]);
-            maximum.max((current[channel] - predicted).abs())
-        })
-    })
-}
-
-fn stereo_config() -> AudioResamplerConfig {
-    AudioResamplerConfig {
-        input_rate: INPUT_RATE,
-        output_rate: OUTPUT_RATE,
-        channels: CHANNELS,
-    }
-}
-
-#[test]
-fn preserves_decoded_packet_continuity_and_output_ratio() {
-    let input = stereo_tone(CHUNK_FRAMES * CHUNK_COUNT);
-    let mut whole_resampler = AudioResampler::new(stereo_config()).unwrap();
-    let whole_output = whole_resampler.process(&input).unwrap();
-    let expected_frames = CHUNK_FRAMES * CHUNK_COUNT * OUTPUT_RATE as usize / INPUT_RATE as usize
-        - LOOK_AHEAD_OUTPUT_FRAMES;
-
-    for chunk_frames in [CHUNK_FRAMES, UNEVEN_CHUNK_FRAMES] {
-        let mut resampler = AudioResampler::new(stereo_config()).unwrap();
-        let output: Vec<_> = input
-            .chunks(chunk_frames * CHANNELS as usize)
-            .map(|chunk| resampler.process(chunk).unwrap())
-            .collect();
-        let residual = maximum_boundary_residual(&output);
-        assert!(
-            residual <= MAX_BOUNDARY_RESIDUAL,
-            "packet boundary residual {residual} exceeded {MAX_BOUNDARY_RESIDUAL}, chunk_frames={chunk_frames}"
-        );
-        let output_frames = output.iter().map(Vec::len).sum::<usize>() / CHANNELS as usize;
-        assert_eq!(output_frames, expected_frames);
-        assert_eq!(output.concat(), whole_output, "chunk_frames={chunk_frames}");
-    }
-}
-
-#[test]
-fn rejects_incomplete_interleaved_frames() {
-    let mut resampler = AudioResampler::new(stereo_config()).unwrap();
-
-    assert!(resampler.process(&[TONE_AMPLITUDE]).is_err());
-}
-
-#[test]
-fn interpolates_mono_samples() {
-    let config = AudioResamplerConfig {
-        input_rate: UNIT_RATE,
-        output_rate: DOUBLE_RATE,
-        channels: MONO_CHANNELS,
-    };
-    let mut resampler = AudioResampler::new(config).unwrap();
-
-    assert_eq!(
-        resampler.process(&[0.0, 1.0, 2.0]).unwrap(),
-        [0.0, 0.5, 1.0, 1.5, 2.0]
-    );
-}
-
-#[test]
-fn rejects_zero_rate_configuration() {
-    let config = AudioResamplerConfig {
-        input_rate: 0,
-        output_rate: OUTPUT_RATE,
-        channels: CHANNELS,
-    };
-
-    assert!(AudioResampler::new(config).is_err());
-}
-
-#[test]
-fn downsamples_across_packet_boundaries() {
-    let config = AudioResamplerConfig {
-        input_rate: DOUBLE_RATE,
-        output_rate: UNIT_RATE,
-        channels: MONO_CHANNELS,
-    };
-    let mut resampler = AudioResampler::new(config).unwrap();
-    let mut output = resampler.process(&FIRST_DOWNSAMPLE_PACKET).unwrap();
-    output.extend(resampler.process(&SECOND_DOWNSAMPLE_PACKET).unwrap());
-
-    assert_eq!(output, EXPECTED_DOWNSAMPLED_OUTPUT);
-}
-
-#[test]
-fn sender_resampler_emits_only_complete_continuous_frames() {
-    let input = stereo_tone(CHUNK_FRAMES * CHUNK_COUNT);
-    let mut resampler =
-        FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap();
-    let output: Vec<_> = input
-        .chunks(CHUNK_FRAMES * CHANNELS as usize)
-        .flat_map(|chunk| resampler.process(chunk).unwrap())
-        .collect();
-
-    assert!(output.len() >= MIN_CONTINUITY_PACKETS);
-    assert!(output
-        .iter()
-        .all(|packet| packet.len() == OUTPUT_PACKET_FRAMES * CHANNELS as usize));
-    assert!(maximum_boundary_residual(&output) <= MAX_BOUNDARY_RESIDUAL);
-}
-
-#[test]
-fn sender_downsampling_preserves_packet_continuity() {
-    let input_packet_frames = RATE_44_1_KHZ as usize / PACKETS_PER_SECOND;
-    let input = stereo_tone_at_rate(input_packet_frames * CHUNK_COUNT, RATE_44_1_KHZ);
-    let config = AudioResamplerConfig {
-        input_rate: RATE_44_1_KHZ,
-        output_rate: INPUT_RATE,
-        channels: CHANNELS,
-    };
-    let mut resampler =
-        FixedFrameAudioResampler::new(config, INPUT_RATE as usize / PACKETS_PER_SECOND).unwrap();
-    let packets: Vec<_> = input
-        .chunks(input_packet_frames * CHANNELS as usize)
-        .flat_map(|packet| resampler.process(packet).unwrap())
-        .collect();
-    let residual = maximum_boundary_residual(&packets);
-
-    assert_eq!(packets.len(), CHUNK_COUNT);
-    assert!(
-        residual <= maximum_tone_prediction_residual(INPUT_RATE) + FLOAT_TOLERANCE,
-        "sender packet boundary residual {residual} exceeded the tone curvature"
-    );
-}

src/client.rs

@@ -93,10 +93,6 @@ use crate::ui_session_interface::SessionPermissionConfig;
 
 pub use super::lang::*;
 
-#[cfg(not(target_os = "linux"))]
-mod audio_playback;
-#[cfg(all(test, not(target_os = "linux")))]
-mod audio_state_tests;
 pub mod file_trait;
 pub mod helper;
 pub mod io_loop;
@@ -2057,8 +2053,6 @@ pub struct AudioHandler {
     simple: Option<psimple::Simple>,
     #[cfg(not(target_os = "linux"))]
     audio_buffer: AudioBuffer,
-    #[cfg(not(target_os = "linux"))]
-    audio_resampler: Option<crate::audio_resampler::AudioResampler>,
     sample_rate: (u32, u32),
     #[cfg(not(target_os = "linux"))]
     audio_stream: Option<Box<dyn StreamTrait>>,
@@ -2066,55 +2060,7 @@ pub struct AudioHandler {
     #[cfg(not(target_os = "linux"))]
     device_channel: u16,
     #[cfg(not(target_os = "linux"))]
-    playback_status: Arc<audio_playback::AudioPlaybackStatus>,
+    ready: Arc<std::sync::Mutex<bool>>,
-}
-
-#[cfg(not(target_os = "linux"))]
-#[derive(Clone, Copy)]
-struct DecodedAudioConfig {
-    sample_rate: u32,
-    input_channels: u16,
-    output_channels: u16,
-}
-
-#[cfg(not(target_os = "linux"))]
-fn create_audio_resampler(
-    input_rate: u32,
-    output_rate: u32,
-    channels: u16,
-) -> ResultType<Option<crate::audio_resampler::AudioResampler>> {
-    if input_rate == output_rate {
-        return Ok(None);
-    }
-    Ok(Some(crate::audio_resampler::AudioResampler::new(
-        crate::audio_resampler::AudioResamplerConfig {
-            input_rate,
-            output_rate,
-            channels,
-        },
-    )?))
-}
-
-#[cfg(not(target_os = "linux"))]
-fn prepare_decoded_audio(
-    input: &[f32],
-    resampler: Option<&mut crate::audio_resampler::AudioResampler>,
-    config: DecodedAudioConfig,
-) -> Result<Vec<f32>, crate::audio_resampler::AudioResamplerError> {
-    let mut output = match resampler {
-        Some(resampler) => resampler.process(input)?,
-        None => input.to_owned(),
-    };
-    if config.input_channels != config.output_channels {
-        output = crate::audio_rechannel(
-            output,
-            config.sample_rate,
-            config.sample_rate,
-            config.input_channels,
-            config.output_channels,
-        );
-    }
-    Ok(output)
 }
 
 #[cfg(not(target_os = "linux"))]
@@ -2122,7 +2068,6 @@ struct AudioBuffer(
     pub Arc<std::sync::Mutex<ringbuf::HeapRb<f32>>>,
     usize,
     [usize; 30],
-    Arc<std::sync::atomic::AtomicUsize>,
 );
 
 #[cfg(not(target_os = "linux"))]
@@ -2134,7 +2079,6 @@ impl Default for AudioBuffer {
             )),
             48000 * 2,
             [0; 30],
-            Arc::new(std::sync::atomic::AtomicUsize::new(0)),
         )
     }
 }
@@ -2209,36 +2153,27 @@ impl AudioBuffer {
         let skip = (cap * max / (30 * N) + 1) & (!1);
         if (having > skip * 3) && (skip > 0) {
             lock.skip(skip);
-            let generation = self.signal_discontinuity();
+            log::info!("skip {skip}, based {max} {zero}");
-            drop(lock);
-            log::info!("skip {skip}, based {max} {zero}, generation={generation}");
         }
     }
 
-    /// The caller must hold the PCM buffer lock while signaling the discard.
-    fn signal_discontinuity(&self) -> usize {
-        self.3
-            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
-            .wrapping_add(1)
-    }
-
     /// append pcm to audio buffer, if buffered data
     /// exceeds AUDIO_BUFFER_MS,  only AUDIO_BUFFER_MS
     /// will be kept.
     fn append_pcm2(&self, buffer: &[f32]) -> usize {
         let mut lock = self.0.lock().unwrap();
         let cap = lock.capacity();
-        let having = lock.occupied_len() + buffer.len();
+        if buffer.len() > cap {
-        lock.push_slice_overwrite(buffer);
+            lock.push_slice_overwrite(buffer);
-        let discard = (having > cap).then(|| (having - cap, self.signal_discontinuity()));
+            return cap;
-        let occupied = lock.occupied_len();
-        drop(lock);
-        if let Some((discarded, generation)) = discard {
-            log::debug!(
-                "Audio buffer capacity discard: samples={discarded}, generation={generation}"
-            );
         }
-        occupied
+
+        let having = lock.occupied_len() + buffer.len();
+        if having > cap {
+            lock.skip(having - cap);
+        }
+        lock.push_slice_overwrite(buffer);
+        lock.occupied_len()
     }
 
     /// append pcm to audio buffer, trying to drop data
@@ -2250,41 +2185,6 @@ impl AudioBuffer {
     }
 }
 
-#[cfg(all(test, not(target_os = "linux")))]
-mod audio_buffer_discontinuity_tests {
-    use super::AudioBuffer;
-    use std::sync::{
-        atomic::{AtomicUsize, Ordering},
-        Arc, Mutex,
-    };
-
-    const BUFFER_CAPACITY: usize = 4;
-    const BUFFER_LEVELS: usize = 30;
-    const FIRST_INPUT: [f32; 2] = [0.1, 0.2];
-    const OVERFLOWING_INPUT: [f32; 3] = [0.3, 0.4, 0.5];
-    const OVERSIZED_INPUT: [f32; 5] = [0.6, 0.7, 0.8, 0.9, 1.0];
-
-    #[test]
-    fn capacity_discards_signal_discontinuities() {
-        let audio_buffer = AudioBuffer(
-            Arc::new(Mutex::new(ringbuf::HeapRb::new(BUFFER_CAPACITY))),
-            BUFFER_CAPACITY,
-            [0; BUFFER_LEVELS],
-            Arc::new(AtomicUsize::new(0)),
-        );
-
-        assert_eq!(audio_buffer.append_pcm2(&FIRST_INPUT), FIRST_INPUT.len());
-        assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 0);
-        assert_eq!(
-            audio_buffer.append_pcm2(&OVERFLOWING_INPUT),
-            BUFFER_CAPACITY
-        );
-        assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 1);
-        assert_eq!(audio_buffer.append_pcm2(&OVERSIZED_INPUT), BUFFER_CAPACITY);
-        assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 2);
-    }
-}
-
 impl AudioHandler {
     #[cfg(target_os = "linux")]
     fn start_audio(&mut self, format0: AudioFormat) -> ResultType<()> {
@@ -2338,9 +2238,6 @@ impl AudioHandler {
         }
 
         self.sample_rate = (format0.sample_rate, config.sample_rate.0);
-        let audio_resampler = create_audio_resampler(
-            format0.sample_rate, config.sample_rate.0, format0.channels as _,
-        )?;
         let mut build_output_stream = |config: StreamConfig| match sample_format {
             cpal::SampleFormat::I8 => self.build_output_stream::<i8>(&config, &device),
             cpal::SampleFormat::I16 => self.build_output_stream::<i16>(&config, &device),
@@ -2365,7 +2262,6 @@ impl AudioHandler {
         } else {
             build_output_stream(config)?;
         }
-        self.audio_resampler = audio_resampler;
 
         Ok(())
     }
@@ -2378,17 +2274,10 @@ impl AudioHandler {
         }
         match AudioDecoder::new(f.sample_rate, if f.channels > 1 { Stereo } else { Mono }) {
             Ok(d) => {
-                #[cfg(target_os = "linux")]
-                let keep_existing_stream = self.simple.is_some()
-                    && self.sample_rate.0 == f.sample_rate
-                    && u32::from(self.channels) == f.channels;
-                #[cfg(not(target_os = "linux"))]
-                let keep_existing_stream = false;
                 let buffer = vec![0.; f.sample_rate as usize * f.channels as usize];
                 self.audio_decoder = Some((d, buffer));
                 self.channels = f.channels as _;
-                let result = self.start_audio(f);
+                allow_err!(self.start_audio(f));
-                self.handle_audio_start_result(result, keep_existing_stream);
             }
             Err(err) => {
                 log::error!("Failed to create audio decoder: {}", err);
@@ -2396,31 +2285,11 @@ impl AudioHandler {
         }
     }
 
-    fn handle_audio_start_result(&mut self, result: ResultType<()>, keep_existing_stream: bool) {
-        if let Err(error) = result {
-            if keep_existing_stream {
-                log::error!(
-                    "Failed to replace audio playback stream; keeping the existing compatible stream: {error:#}"
-                );
-            } else {
-                *self = Self::default();
-                log::error!("Failed to start audio playback: {error:#}");
-            }
-        }
-    }
-
     /// Handle audio frame and play it.
     #[inline]
     pub fn handle_frame(&mut self, frame: AudioFrame) {
         #[cfg(not(target_os = "linux"))]
-        self.playback_status.report_errors();
+        if self.audio_stream.is_none() || !self.ready.lock().unwrap().clone() {
-        #[cfg(not(target_os = "linux"))]
-        if self.audio_stream.is_none()
-            || !self
-                .playback_status
-                .ready
-                .load(std::sync::atomic::Ordering::Acquire)
-        {
             return;
         }
         #[cfg(target_os = "linux")]
@@ -2429,40 +2298,39 @@ impl AudioHandler {
             return;
         }
         self.audio_decoder.as_mut().map(|(d, buffer)| {
-            let decoded_frames = match d.decode_float(&frame.data, buffer, false) {
+            if let Ok(n) = d.decode_float(&frame.data, buffer, false) {
-                Ok(decoded_frames) => decoded_frames,
+                let channels = self.channels;
-                Err(error) => {
+                let n = n * (channels as usize);
-                    log::warn!("Failed to decode audio frame: {error:?}");
+                #[cfg(not(target_os = "linux"))]
-                    return;
+                {
-                }
+                    let sample_rate0 = self.sample_rate.0;
-            };
+                    let sample_rate = self.sample_rate.1;
-            let channels = self.channels;
+                    let mut buffer = buffer[0..n].to_owned();
-            let n = decoded_frames * channels as usize;
+                    if sample_rate != sample_rate0 {
-            #[cfg(not(target_os = "linux"))]
+                        buffer = crate::audio_resample(
-            {
+                            &buffer[0..n],
-                let config = DecodedAudioConfig {
+                            sample_rate0,
-                    sample_rate: self.sample_rate.1,
+                            sample_rate,
-                    input_channels: self.channels,
+                            channels,
-                    output_channels: self.device_channel,
+                        );
-                };
-                let buffer = match prepare_decoded_audio(
-                    &buffer[0..n],
-                    self.audio_resampler.as_mut(),
-                    config,
-                ) {
-                    Ok(output) => output,
-                    Err(error) => {
-                        log::error!("Failed to resample decoded audio: {error:#}");
-                        return;
                     }
-                };
+                    if self.channels != self.device_channel {
-                self.audio_buffer.append_pcm(&buffer);
+                        buffer = crate::audio_rechannel(
-            }
+                            buffer,
-            #[cfg(target_os = "linux")]
+                            sample_rate,
-            {
+                            sample_rate,
-                let data_u8 =
+                            self.channels,
-                    unsafe { std::slice::from_raw_parts::<u8>(buffer.as_ptr() as _, n * 4) };
+                            self.device_channel,
-                self.simple.as_mut().map(|x| x.write(data_u8));
+                        );
+                    }
+                    self.audio_buffer.append_pcm(&buffer);
+                }
+                #[cfg(target_os = "linux")]
+                {
+                    let data_u8 =
+                        unsafe { std::slice::from_raw_parts::<u8>(buffer.as_ptr() as _, n * 4) };
+                    self.simple.as_mut().map(|x| x.write(data_u8));
+                }
             }
         });
     }
@@ -2482,28 +2350,63 @@ impl AudioHandler {
         self.audio_buffer
             .resize(config.sample_rate.0 as _, config.channels as _);
         let audio_buffer = self.audio_buffer.0.clone();
-        let discontinuity_generation = self.audio_buffer.3.clone();
+        let ready = self.ready.clone();
-        let mut playback_writer = audio_playback::AudioPlaybackWriter::new(
-            audio_playback::AudioPlaybackConfig {
-                sample_rate: config.sample_rate.0,
-                channels: config.channels as usize,
-            },
-            audio_buffer,
-            discontinuity_generation,
-        )?;
-        let playback_status = playback_writer.status.clone();
         let timeout = None;
         let stream = device.build_output_stream(
             config,
-            move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
+            move |data: &mut [T], info: &cpal::OutputCallbackInfo| {
-                playback_writer.write_output(data);
+                if !*ready.lock().unwrap() {
+                    *ready.lock().unwrap() = true;
+                }
+
+                let mut n = data.len();
+                let mut lock = audio_buffer.lock().unwrap();
+                let mut having = lock.occupied_len();
+                // android two timestamps, one from zero, another not
+                #[cfg(not(target_os = "android"))]
+                if having < n {
+                    let tms = info.timestamp();
+                    let how_long = tms
+                        .playback
+                        .duration_since(&tms.callback)
+                        .unwrap_or(Duration::from_millis(0));
+
+                    // must long enough to fight back scheuler delay
+                    if how_long > Duration::from_millis(6) && how_long < Duration::from_millis(3000)
+                    {
+                        drop(lock);
+                        std::thread::sleep(how_long.div_f32(1.2));
+                        lock = audio_buffer.lock().unwrap();
+                        having = lock.occupied_len();
+                    }
+
+                    if having < n {
+                        n = having;
+                    }
+                }
+                #[cfg(target_os = "android")]
+                if having < n {
+                    n = having;
+                }
+                let mut elems = vec![0.0f32; n];
+                if n > 0 {
+                    lock.pop_slice(&mut elems);
+                }
+                drop(lock);
+
+                let mut input = elems.into_iter();
+                for sample in data.iter_mut() {
+                    *sample = match input.next() {
+                        Some(x) => T::from_sample(x),
+                        _ => T::from_sample(0.),
+                    };
+                }
             },
             err_fn,
             timeout,
         )?;
         stream.play()?;
         self.audio_stream = Some(Box::new(stream));
-        self.playback_status = playback_status;
         Ok(())
     }
 }
@@ -2523,27 +2426,6 @@ mod audio_format_tests {
         assert!(!is_supported_audio_channel_count(0));
         assert!(!is_supported_audio_channel_count(u32::MAX));
     }
-
-    #[test]
-    fn failed_audio_start_discards_format_state() {
-        use super::{anyhow, AudioDecoder, AudioHandler, Stereo};
-
-        const SAMPLE_RATE: u32 = 48_000;
-        const CHANNELS: u16 = 2;
-        let decoder = AudioDecoder::new(SAMPLE_RATE, Stereo).unwrap();
-        let mut handler = AudioHandler {
-            audio_decoder: Some((decoder, Vec::new())),
-            sample_rate: (SAMPLE_RATE, SAMPLE_RATE),
-            channels: CHANNELS,
-            ..Default::default()
-        };
-
-        handler.handle_audio_start_result(Err(anyhow!("Injected playback startup failure")), false);
-
-        assert!(handler.audio_decoder.is_none());
-        assert_eq!(handler.channels, 0);
-        assert_eq!(handler.sample_rate, (0, 0));
-    }
 }
 
 /// Video handler for the [`Client`].

src/client/audio_playback.rs

@@ -1,218 +0,0 @@
-use hbb_common::{log, thiserror};
-use ringbuf::{ring_buffer::RbBase, Rb};
-use std::sync::{
-    atomic::{AtomicBool, AtomicUsize, Ordering},
-    TryLockError,
-};
-
-pub(super) const UNDERRUN_DECLICK_MS: usize = 5;
-const MILLISECONDS_PER_SECOND: usize = 1_000;
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(super) struct AudioPlaybackConfig {
-    pub sample_rate: u32,
-    pub channels: usize,
-}
-
-#[derive(Debug, thiserror::Error, PartialEq, Eq)]
-pub(super) enum AudioPlaybackError {
-    #[error(
-        "invalid audio playback configuration: sample_rate={}, channels={}",
-        .0.sample_rate, .0.channels
-    )]
-    InvalidConfig(AudioPlaybackConfig),
-    #[error("audio playback frame has {samples} samples for {channels} channels")]
-    IncompleteFrame { samples: usize, channels: usize },
-    #[error("audio playback transition frame count overflow")]
-    FrameCountOverflow,
-}
-
-pub(super) struct AudioPlaybackRecovery {
-    channels: usize,
-    transition_frames: usize,
-    transition_frame: usize,
-    had_input: bool,
-    transition_start: Vec<f32>,
-    output_frame: Vec<f32>,
-}
-
-#[derive(Default)]
-pub(super) struct AudioPlaybackStatus {
-    pub(super) ready: AtomicBool,
-    contentions: AtomicUsize,
-    buffer_poisoned: AtomicBool,
-}
-
-impl AudioPlaybackStatus {
-    pub(super) fn report_errors(&self) {
-        let contentions = self.contentions.swap(0, Ordering::Relaxed);
-        if contentions != 0 {
-            log::debug!("Audio playback PCM buffer contention: callbacks={contentions}");
-        }
-        if self.buffer_poisoned.swap(false, Ordering::Relaxed) {
-            log::error!("Audio playback stopped reading a poisoned PCM buffer");
-        }
-    }
-}
-
-pub(super) struct AudioPlaybackWriter {
-    audio_buffer: std::sync::Arc<std::sync::Mutex<ringbuf::HeapRb<f32>>>,
-    discontinuity_generation: std::sync::Arc<AtomicUsize>,
-    observed_discontinuity_generation: usize,
-    buffered_input: Vec<f32>,
-    recovery: AudioPlaybackRecovery,
-    pub(super) status: std::sync::Arc<AudioPlaybackStatus>,
-    buffer_failed: bool,
-}
-
-impl AudioPlaybackWriter {
-    pub(super) fn new(
-        config: AudioPlaybackConfig,
-        audio_buffer: std::sync::Arc<std::sync::Mutex<ringbuf::HeapRb<f32>>>,
-        discontinuity_generation: std::sync::Arc<AtomicUsize>,
-    ) -> Result<Self, AudioPlaybackError> {
-        let recovery = AudioPlaybackRecovery::new(config)?;
-        let buffer_capacity = audio_buffer.lock().unwrap().capacity();
-        let observed_discontinuity_generation = discontinuity_generation.load(Ordering::Relaxed);
-        Ok(Self {
-            audio_buffer,
-            discontinuity_generation,
-            observed_discontinuity_generation,
-            buffered_input: vec![0.0; buffer_capacity],
-            recovery,
-            status: Default::default(),
-            buffer_failed: false,
-        })
-    }
-
-    fn read_buffer(&mut self, requested_samples: usize) -> usize {
-        if self.buffer_failed {
-            return 0;
-        }
-        let mut buffer = match self.audio_buffer.try_lock() {
-            Ok(buffer) => buffer,
-            Err(TryLockError::WouldBlock) => {
-                // Keep queued PCM and its generation for the next successful read.
-                self.status.contentions.fetch_add(1, Ordering::Relaxed);
-                return 0;
-            }
-            Err(TryLockError::Poisoned(_)) => {
-                self.buffer_failed = true;
-                self.status.ready.store(false, Ordering::Release);
-                self.status.buffer_poisoned.store(true, Ordering::Relaxed);
-                return 0;
-            }
-        };
-        let generation = self.discontinuity_generation.load(Ordering::Relaxed);
-        let channels = self.recovery.channels;
-        let samples = buffer.occupied_len().min(requested_samples) / channels * channels;
-        buffer.pop_slice(&mut self.buffered_input[..samples]);
-        drop(buffer);
-        if generation != self.observed_discontinuity_generation {
-            self.recovery.begin_discontinuity();
-            self.observed_discontinuity_generation = generation;
-        }
-        samples
-    }
-
-    pub(super) fn write_output<T>(&mut self, output: &mut [T])
-    where
-        T: cpal::Sample + cpal::FromSample<f32>,
-    {
-        self.status
-            .ready
-            .store(!self.buffer_failed, Ordering::Release);
-        let requested_samples = output.len().min(self.buffered_input.len());
-        let channel_count = self.recovery.channels;
-        let available_samples = self.read_buffer(requested_samples);
-        let available_frames = available_samples / channel_count;
-        for (frame_index, output_frame) in output.chunks_mut(channel_count).enumerate() {
-            let input = if frame_index < available_frames {
-                let start = frame_index * channel_count;
-                Some(&self.buffered_input[start..start + channel_count])
-            } else {
-                None
-            };
-            match self.recovery.process_frame(input) {
-                Ok(recovered) => {
-                    for (output, sample) in output_frame.iter_mut().zip(recovered) {
-                        *output = T::from_sample(*sample);
-                    }
-                }
-                Err(error) => {
-                    log::error!("Failed to recover audio underflow: {error}");
-                    output_frame.fill(T::from_sample(0.0));
-                }
-            }
-        }
-    }
-}
-
-impl AudioPlaybackRecovery {
-    pub(super) fn new(config: AudioPlaybackConfig) -> Result<Self, AudioPlaybackError> {
-        if config.sample_rate == 0 || config.channels == 0 {
-            return Err(AudioPlaybackError::InvalidConfig(config));
-        }
-        let transition_frames = (config.sample_rate as usize)
-            .checked_mul(UNDERRUN_DECLICK_MS)
-            .ok_or(AudioPlaybackError::FrameCountOverflow)?
-            / MILLISECONDS_PER_SECOND;
-        if transition_frames == 0 {
-            return Err(AudioPlaybackError::InvalidConfig(config));
-        }
-        Ok(Self {
-            channels: config.channels,
-            transition_frames,
-            transition_frame: transition_frames,
-            had_input: false,
-            transition_start: vec![0.0; config.channels],
-            output_frame: vec![0.0; config.channels],
-        })
-    }
-
-    pub(super) fn process_frame(
-        &mut self,
-        input: Option<&[f32]>,
-    ) -> Result<&[f32], AudioPlaybackError> {
-        if input.is_some_and(|frame| frame.len() != self.channels) {
-            return Err(AudioPlaybackError::IncompleteFrame {
-                samples: input.map_or(0, <[f32]>::len),
-                channels: self.channels,
-            });
-        }
-        self.begin_transition(input.is_some());
-        let target_weight = self.advance_transition();
-        for channel in 0..self.channels {
-            let target = input.map_or(0.0, |frame| frame[channel]);
-            self.output_frame[channel] =
-                self.transition_start[channel] * (1.0 - target_weight) + target * target_weight;
-        }
-        Ok(&self.output_frame)
-    }
-
-    pub(super) fn begin_discontinuity(&mut self) {
-        self.transition_start.copy_from_slice(&self.output_frame);
-        self.transition_frame = 0;
-    }
-
-    fn begin_transition(&mut self, has_input: bool) {
-        if has_input == self.had_input {
-            return;
-        }
-        self.transition_start.copy_from_slice(&self.output_frame);
-        self.transition_frame = 0;
-        self.had_input = has_input;
-    }
-
-    fn advance_transition(&mut self) -> f32 {
-        if self.transition_frame >= self.transition_frames {
-            return 1.0;
-        }
-        self.transition_frame += 1;
-        self.transition_frame as f32 / self.transition_frames as f32
-    }
-}
-
-#[cfg(test)]
-#[path = "audio_playback_tests.rs"]
-mod tests;

src/client/audio_playback_tests.rs

@@ -1,218 +0,0 @@
-use super::{AudioPlaybackConfig, AudioPlaybackError, AudioPlaybackRecovery, AudioPlaybackWriter};
-use ringbuf::{ring_buffer::RbBase, Rb};
-use std::{
-    sync::{atomic::Ordering, mpsc, Arc, Mutex},
-    time::Duration,
-};
-
-const SAMPLE_RATE: u32 = 48_000;
-const CHANNELS: usize = 2;
-const ACTIVE_FRAME: [f32; CHANNELS] = [0.8, -0.8];
-const OPPOSITE_ACTIVE_FRAME: [f32; CHANNELS] = [-0.8, 0.8];
-const ACTIVE_FRAMES: usize = 300;
-const SILENT_FRAMES: usize = 300;
-const TRANSITION_FRAMES: usize =
-    SAMPLE_RATE as usize * super::UNDERRUN_DECLICK_MS / super::MILLISECONDS_PER_SECOND;
-const MAX_SAMPLE_STEP: f32 = 0.01;
-
-#[test]
-fn writing_audio_observes_discard_and_releases_buffer_lock() {
-    const INPUT: [f32; 4] = [0.1, 0.2, 0.3, 0.4];
-    const GENERATION: usize = 7;
-    let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(INPUT.len())));
-    let generation = Arc::new(super::AtomicUsize::new(0));
-    let config = AudioPlaybackConfig {
-        sample_rate: SAMPLE_RATE,
-        channels: CHANNELS,
-    };
-    let mut writer = AudioPlaybackWriter::new(config, buffer.clone(), generation.clone()).unwrap();
-    {
-        let mut buffer = buffer.lock().unwrap();
-        buffer.push_slice(&INPUT);
-        generation.store(GENERATION, Ordering::Relaxed);
-    }
-    let mut output = [0.0_f32; INPUT.len()];
-
-    writer.write_output(&mut output);
-
-    assert_eq!(writer.buffered_input, INPUT);
-    assert_eq!(writer.observed_discontinuity_generation, GENERATION);
-    assert_eq!(buffer.try_lock().unwrap().occupied_len(), 0);
-}
-
-fn maximum_sample_step(samples: &[f32]) -> f32 {
-    samples
-        .windows(CHANNELS + 1)
-        .map(|window| (window[CHANNELS] - window[0]).abs())
-        .fold(0.0, f32::max)
-}
-
-#[test]
-fn smooths_underflow_and_explicit_audio_discontinuities() {
-    for explicit_discontinuity in [false, true] {
-        let config = AudioPlaybackConfig {
-            sample_rate: SAMPLE_RATE,
-            channels: CHANNELS,
-        };
-        let mut recovery = AudioPlaybackRecovery::new(config).unwrap();
-        let mut output = Vec::new();
-        for _ in 0..ACTIVE_FRAMES {
-            output.extend_from_slice(recovery.process_frame(Some(&ACTIVE_FRAME)).unwrap());
-        }
-        let transition_end = TRANSITION_FRAMES * CHANNELS;
-        assert_eq!(
-            &output[transition_end - CHANNELS..transition_end],
-            ACTIVE_FRAME.as_slice(),
-            "explicit_discontinuity={explicit_discontinuity}"
-        );
-        let resumed_frame = if explicit_discontinuity {
-            recovery.begin_discontinuity();
-            &OPPOSITE_ACTIVE_FRAME
-        } else {
-            for _ in 0..SILENT_FRAMES {
-                output.extend_from_slice(recovery.process_frame(None).unwrap());
-            }
-            &ACTIVE_FRAME
-        };
-        for _ in 0..ACTIVE_FRAMES {
-            output.extend_from_slice(recovery.process_frame(Some(resumed_frame)).unwrap());
-        }
-        let maximum = maximum_sample_step(&output);
-        assert!(
-            maximum <= MAX_SAMPLE_STEP,
-            "step {maximum} exceeded {MAX_SAMPLE_STEP}, explicit={explicit_discontinuity}"
-        );
-        assert_eq!(
-            &output[output.len() - CHANNELS..],
-            resumed_frame,
-            "explicit_discontinuity={explicit_discontinuity}"
-        );
-    }
-}
-
-#[test]
-fn validates_configuration_and_frame_size() {
-    let invalid_config = AudioPlaybackConfig {
-        sample_rate: 0,
-        channels: CHANNELS,
-    };
-    assert_eq!(
-        AudioPlaybackRecovery::new(invalid_config).err(),
-        Some(AudioPlaybackError::InvalidConfig(invalid_config))
-    );
-
-    let config = AudioPlaybackConfig {
-        sample_rate: SAMPLE_RATE,
-        channels: CHANNELS,
-    };
-    let mut recovery = AudioPlaybackRecovery::new(config).unwrap();
-    assert_eq!(
-        recovery.process_frame(Some(&[0.5])).err(),
-        Some(AudioPlaybackError::IncompleteFrame {
-            samples: 1,
-            channels: CHANNELS,
-        })
-    );
-}
-
-const CALLBACK_SAMPLES: usize = 64;
-const CALLBACK_TIMEOUT: Duration = Duration::from_secs(2);
-const DISCARD_GENERATION: usize = 1;
-
-fn write_while_buffer_is_locked(
-    mut writer: AudioPlaybackWriter,
-    buffer: &Arc<Mutex<ringbuf::HeapRb<f32>>>,
-    generation: &Arc<super::AtomicUsize>,
-) -> (AudioPlaybackWriter, [f32; CALLBACK_SAMPLES]) {
-    let mut guard = buffer.lock().unwrap();
-    let queued = OPPOSITE_ACTIVE_FRAME.repeat(ACTIVE_FRAMES);
-    guard.push_slice(&queued);
-    generation.store(DISCARD_GENERATION, Ordering::Relaxed);
-    let (completed_tx, completed_rx) = mpsc::channel();
-    let callback = std::thread::spawn(move || {
-        let mut output = [0.0; CALLBACK_SAMPLES];
-        crate::audio_resampler::allocation_tests::assert_no_allocations(|| {
-            writer.write_output(&mut output);
-        });
-        completed_tx.send((writer, output)).unwrap();
-    });
-    let completed = completed_rx.recv_timeout(CALLBACK_TIMEOUT);
-    let retained = guard.occupied_len();
-    drop(guard);
-    callback.join().unwrap();
-    let result = completed.expect("playback callback waited for the buffer owner");
-    assert_eq!(retained, queued.len());
-    result
-}
-
-#[test]
-fn playback_contention_preserves_queued_audio_and_recovers_after_release() {
-    let samples = ACTIVE_FRAMES * CHANNELS;
-    let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(samples)));
-    let generation = Arc::new(super::AtomicUsize::new(0));
-    let config = AudioPlaybackConfig {
-        sample_rate: SAMPLE_RATE,
-        channels: CHANNELS,
-    };
-    let mut writer = AudioPlaybackWriter::new(config, buffer.clone(), generation.clone()).unwrap();
-    buffer
-        .lock()
-        .unwrap()
-        .push_slice(&ACTIVE_FRAME.repeat(ACTIVE_FRAMES));
-    let mut output = vec![0.0; samples];
-    writer.write_output(&mut output);
-    assert_eq!(&output[samples - CHANNELS..], &ACTIVE_FRAME);
-
-    let (mut writer, gap) = write_while_buffer_is_locked(writer, &buffer, &generation);
-
-    assert_eq!(writer.status.contentions.load(Ordering::Relaxed), 1);
-    assert!(writer.status.ready.load(Ordering::Acquire));
-    assert_eq!(writer.observed_discontinuity_generation, 0);
-    assert!(maximum_sample_step(&gap) <= MAX_SAMPLE_STEP);
-    assert!(gap[0] > 0.0 && gap[0] < ACTIVE_FRAME[0]);
-    assert_eq!(gap[1], -gap[0]);
-    assert!(gap[CALLBACK_SAMPLES - CHANNELS] > 0.0);
-    writer.write_output(&mut output);
-    let mut transition = gap[gap.len() - CHANNELS..].to_vec();
-    transition.extend_from_slice(&output[..TRANSITION_FRAMES * CHANNELS]);
-    assert!(maximum_sample_step(&transition) <= MAX_SAMPLE_STEP);
-    assert_eq!(
-        writer.buffered_input,
-        OPPOSITE_ACTIVE_FRAME.repeat(ACTIVE_FRAMES)
-    );
-    assert_eq!(writer.observed_discontinuity_generation, DISCARD_GENERATION);
-    assert_eq!(&output[samples - CHANNELS..], &OPPOSITE_ACTIVE_FRAME);
-    assert_eq!(buffer.lock().unwrap().occupied_len(), 0);
-}
-
-#[test]
-fn poisoned_playback_buffer_reports_once_without_panicking_in_the_callback() {
-    let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(CALLBACK_SAMPLES)));
-    let config = AudioPlaybackConfig {
-        sample_rate: SAMPLE_RATE,
-        channels: CHANNELS,
-    };
-    let mut writer =
-        AudioPlaybackWriter::new(config, buffer.clone(), Arc::new(super::AtomicUsize::new(0)))
-            .unwrap();
-    assert!(std::thread::spawn(move || {
-        let _guard = buffer.lock().unwrap();
-        panic!("Injected PCM buffer failure");
-    })
-    .join()
-    .is_err());
-    let mut output = [ACTIVE_FRAME[0]; CALLBACK_SAMPLES];
-
-    crate::audio_resampler::allocation_tests::assert_no_allocations(|| {
-        writer.write_output(&mut output);
-    });
-
-    assert_eq!(output, [0.0; CALLBACK_SAMPLES]);
-    assert!(!writer.status.ready.load(Ordering::Acquire));
-    assert!(writer.status.buffer_poisoned.load(Ordering::Relaxed));
-    writer.status.report_errors();
-    writer.write_output(&mut output);
-    assert!(!writer.status.buffer_poisoned.load(Ordering::Relaxed));
-    assert_eq!(writer.status.contentions.load(Ordering::Relaxed), 0);
-    assert!(!writer.status.ready.load(Ordering::Acquire));
-}

src/client/audio_state_tests.rs

@@ -1,113 +0,0 @@
-use super::{create_audio_resampler, AudioDecoder, AudioFrame, AudioHandler, Stereo};
-use cpal::traits::StreamTrait;
-use hbb_common::anyhow::anyhow;
-use magnum_opus::{Application::LowDelay, Encoder};
-use ringbuf::{ring_buffer::RbBase, Rb};
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc,
-};
-
-const INPUT_RATE: u32 = 24_000;
-const OUTPUT_RATE: u32 = 48_000;
-const CHANNELS: u16 = 2;
-const PACKETS_PER_SECOND: usize = 100;
-const MAX_PACKET_BYTES: usize = 4_096;
-const SAMPLE_VALUE: f32 = 0.25;
-
-struct TrackedAudioStream(Arc<AtomicBool>);
-
-impl StreamTrait for TrackedAudioStream {
-    fn play(&self) -> Result<(), cpal::PlayStreamError> {
-        Ok(())
-    }
-
-    fn pause(&self) -> Result<(), cpal::PauseStreamError> {
-        Ok(())
-    }
-}
-
-impl Drop for TrackedAudioStream {
-    fn drop(&mut self) {
-        self.0.store(true, Ordering::SeqCst);
-    }
-}
-
-fn decoder(sample_rate: u32) -> (AudioDecoder, Vec<f32>) {
-    (
-        AudioDecoder::new(sample_rate, Stereo).unwrap(),
-        vec![0.0; sample_rate as usize * CHANNELS as usize],
-    )
-}
-
-fn active_handler(input_rate: u32) -> (AudioHandler, Arc<AtomicBool>) {
-    let dropped = Arc::new(AtomicBool::new(false));
-    let handler = AudioHandler {
-        audio_decoder: Some(decoder(input_rate)),
-        audio_resampler: create_audio_resampler(input_rate, OUTPUT_RATE, CHANNELS).unwrap(),
-        sample_rate: (input_rate, OUTPUT_RATE),
-        audio_stream: Some(Box::new(TrackedAudioStream(dropped.clone()))),
-        channels: CHANNELS,
-        device_channel: CHANNELS,
-        ..Default::default()
-    };
-    handler.playback_status.ready.store(true, Ordering::Release);
-    (handler, dropped)
-}
-
-fn audio_frame() -> AudioFrame {
-    let samples = OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize;
-    let mut encoder = Encoder::new(OUTPUT_RATE, Stereo, LowDelay).unwrap();
-    AudioFrame {
-        data: encoder
-            .encode_vec_float(&vec![SAMPLE_VALUE; samples], MAX_PACKET_BYTES)
-            .unwrap()
-            .into(),
-        ..Default::default()
-    }
-}
-
-#[test]
-fn failed_format_change_discards_old_playback_state() {
-    let (mut handler, dropped) = active_handler(INPUT_RATE);
-    handler
-        .audio_buffer
-        .0
-        .lock()
-        .unwrap()
-        .push_slice(&[SAMPLE_VALUE; CHANNELS as usize]);
-    handler.audio_decoder = Some(decoder(OUTPUT_RATE));
-    handler.sample_rate = (OUTPUT_RATE, OUTPUT_RATE);
-
-    handler.handle_audio_start_result(
-        Err(anyhow!("Injected output stream startup failure")),
-        false,
-    );
-
-    assert!(dropped.load(Ordering::SeqCst));
-    assert!(handler.audio_stream.is_none());
-    assert!(handler.audio_resampler.is_none());
-    assert!(handler.audio_decoder.is_none());
-    assert!(!handler.playback_status.ready.load(Ordering::Acquire));
-    handler.handle_frame(audio_frame());
-    assert_eq!(handler.audio_buffer.0.lock().unwrap().occupied_len(), 0);
-}
-
-#[test]
-fn successful_start_or_compatible_failure_preserves_audio_packet_duration() {
-    for result in [
-        Ok(()),
-        Err(anyhow!("Injected compatible stream replacement failure")),
-    ] {
-        let (mut handler, dropped) = active_handler(OUTPUT_RATE);
-
-        handler.handle_audio_start_result(result, true);
-        handler.handle_frame(audio_frame());
-
-        assert!(!dropped.load(Ordering::SeqCst));
-        assert_eq!(
-            handler.audio_buffer.0.lock().unwrap().occupied_len(),
-            OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize
-        );
-    }
-}

src/common.rs

@@ -408,11 +408,6 @@ pub fn resample_channels(
     }
 }
 
-#[cfg(all(feature = "use_dasp", feature = "use_samplerate"))]
-compile_error!(
-    "features `use_dasp` and `use_samplerate` are mutually exclusive; disable default features before selecting `use_samplerate`"
-);
-
 #[cfg(feature = "use_dasp")]
 pub fn audio_resample(
     data: &[f32],
@@ -449,7 +444,7 @@ pub fn audio_resample(
     }
 }
 
-#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
+#[cfg(feature = "use_samplerate")]
 pub fn audio_resample(
     data: &[f32],
     sample_rate0: u32,

src/lib.rs

@@ -1,5 +1,3 @@
-#[cfg(any(test, not(target_os = "linux")))]
-mod audio_resampler;
 mod keyboard;
 /// cbindgen:ignore
 pub mod platform;

src/server/audio_capture_error.rs

@@ -11,10 +11,7 @@ pub(super) struct CaptureErrorHandler {
 
 impl CaptureErrorHandler {
     pub(super) fn handle(&self, error: cpal::StreamError) {
-        if matches!(
+        if matches!(error, cpal::StreamError::StreamInterrupted { .. }) {
-            error,
-            cpal::StreamError::StreamInterrupted { .. } | cpal::StreamError::DeviceNotAvailable
-        ) {
             // ScreenCaptureKit can stop capture while the remote session stays open.
             // The observed -3821 error does not identify its underlying trigger.
             // https://developer.apple.com/documentation/screencapturekit/scstreamdelegate/stream(_:didstopwitherror:)
@@ -26,20 +23,6 @@ impl CaptureErrorHandler {
         }
     }
 
-    pub(super) fn process_frame(&self, process: impl FnOnce() -> hbb_common::ResultType<()>) {
-        // Defensive recovery: persistent processing failures with valid capture input
-        // have not been reproduced. Stop using a failed processor until stream replacement.
-        if self.needs_restart() {
-            return;
-        }
-        if let Err(error) = process() {
-            self.interrupted.store(true, Ordering::Relaxed);
-            hbb_common::log::error!(
-                "Failed to process captured audio frame; requesting stream restart: {error:#}"
-            );
-        }
-    }
-
     pub(super) fn needs_restart(&self) -> bool {
         self.interrupted.load(Ordering::Relaxed)
     }
@@ -86,44 +69,4 @@ mod tests {
         });
         assert!(!errors.needs_restart());
     }
-
-    #[test]
-    fn processing_failure_skips_remaining_frames_until_stream_replacement() {
-        const FRAME_SAMPLES: usize = 2;
-        const FRAMES_PER_CALLBACK: usize = 3;
-        const CALLBACK_COUNT: usize = 2;
-        const FAILURE_CALL: usize = 2;
-        let errors = CaptureErrorHandler::default();
-        let callback_errors = errors.clone();
-        let mut framer =
-            super::super::audio_capture::CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap();
-        let input = [0.0; FRAME_SAMPLES * FRAMES_PER_CALLBACK];
-        let mut processed = 0;
-        let mut failures = 0;
-
-        // Inject an error to test recovery; this is not a valid-input backend failure reproduction.
-        for _ in 0..CALLBACK_COUNT {
-            framer.process(input.iter().copied(), |_| {
-                callback_errors.process_frame(|| {
-                    processed += 1;
-                    if processed >= FAILURE_CALL {
-                        failures += 1;
-                        hbb_common::anyhow::bail!("Injected capture processing failure");
-                    }
-                    Ok(())
-                });
-            });
-        }
-
-        assert_eq!(processed, FAILURE_CALL);
-        assert_eq!(failures, 1);
-        assert!(errors.needs_restart());
-        let replacement = CaptureErrorHandler::default();
-        replacement.process_frame(|| {
-            processed += 1;
-            Ok(())
-        });
-        assert_eq!(processed, FAILURE_CALL + 1);
-        assert!(!replacement.needs_restart());
-    }
 }

src/server/audio_service.rs

@@ -15,9 +15,7 @@
 use super::*;
 #[cfg(not(any(target_os = "linux", target_os = "android")))]
 use hbb_common::anyhow::anyhow;
-#[cfg(any(target_os = "linux", target_os = "android"))]
+use magnum_opus::{Application::*, Channels::*, Encoder};
-use magnum_opus::Application::LowDelay;
-use magnum_opus::{Channels::*, Encoder};
 use std::sync::atomic::{AtomicBool, Ordering};
 
 pub const NAME: &'static str = "audio";
@@ -99,11 +97,10 @@ mod pa_impl {
         RESTARTING.store(false, Ordering::SeqCst);
         #[cfg(target_os = "linux")]
         let mut stream = crate::ipc::connect(1000, "_pa").await?;
-        let mut encoder = AudioEncoder::new(Encoder::new(
+        unsafe {
-            crate::platform::PA_SAMPLE_RATE,
+            AUDIO_ZERO_COUNT = 0;
-            Stereo,
+        }
-            LowDelay,
+        let mut encoder = Encoder::new(crate::platform::PA_SAMPLE_RATE, Stereo, LowDelay)?;
-        )?);
         #[cfg(target_os = "linux")]
         allow_err!(
             stream
@@ -175,11 +172,8 @@ pub fn is_screen_capture_kit_available() -> bool {
 }
 
 #[cfg(not(any(target_os = "linux", target_os = "android")))]
-mod audio_capture;
+#[path = "audio_capture_error.rs"]
-#[cfg(not(any(target_os = "linux", target_os = "android")))]
 mod audio_capture_error;
-#[cfg(not(any(target_os = "linux", target_os = "android")))]
-mod audio_capture_queue;
 
 #[cfg(not(any(target_os = "linux", target_os = "android")))]
 mod cpal_impl {
@@ -188,15 +182,14 @@ mod cpal_impl {
     use super::*;
     use cpal::{
         traits::{DeviceTrait, HostTrait, StreamTrait},
-        Device, Host, InputCallbackInfo, SupportedStreamConfig,
+        BufferSize, Device, Host, InputCallbackInfo, StreamConfig, SupportedStreamConfig,
     };
 
     lazy_static::lazy_static! {
         static ref HOST: Host = cpal::default_host();
+        static ref INPUT_BUFFER: Arc<Mutex<std::collections::VecDeque<f32>>> = Default::default();
     }
 
-    const AUDIO_PACKETS_PER_SECOND: usize = 100;
-
     #[cfg(feature = "screencapturekit")]
     lazy_static::lazy_static! {
         static ref HOST_SCREEN_CAPTURE_KIT: Result<Host, cpal::HostUnavailable> = cpal::host_from_id(cpal::HostId::ScreenCaptureKit);
@@ -204,20 +197,7 @@ mod cpal_impl {
 
     #[derive(Default)]
     pub struct State {
-        stream: Option<ActiveCaptureStream>,
+        stream: Option<(Box<dyn StreamTrait>, Arc<Message>, CaptureErrorHandler)>,
-    }
-
-    struct ActiveCaptureStream {
-        stream: Option<Box<dyn StreamTrait>>,
-        format: Arc<Message>,
-        _encoder_worker: audio_capture_queue::CaptureEncoderWorker,
-        errors: CaptureErrorHandler,
-    }
-
-    impl Drop for ActiveCaptureStream {
-        fn drop(&mut self) {
-            self.stream.take();
-        }
     }
 
     impl super::service::Reset for State {
@@ -235,8 +215,8 @@ mod cpal_impl {
             }
             _ => {}
         }
-        if let Some(stream) = &state.stream {
+        if let Some((_, format, _)) = &state.stream {
-            sp.send_shared(stream.format.clone());
+            sp.send_shared(format.clone());
             #[cfg(target_os = "macos")]
             log::info!("Audio capture stream recreated; replacement format sent");
         }
@@ -252,8 +232,8 @@ mod cpal_impl {
                 }
                 _ => {}
             }
-            if let Some(stream) = &state.stream {
+            if let Some((_, format, _)) = &state.stream {
-                sps.send_shared(stream.format.clone());
+                sps.send_shared(format.clone());
             }
             Ok(())
         })?;
@@ -261,10 +241,10 @@ mod cpal_impl {
     }
 
     pub fn run(sp: EmptyExtraFieldService, state: &mut State) -> ResultType<()> {
-        if let Some(stream) = &state.stream {
+        if let Some((_, _, errors)) = &state.stream {
-            if stream.errors.needs_restart() {
+            if errors.needs_restart() {
-                // Recreate on the service thread, outside the capture callbacks.
+                // Recreate on the service thread, outside the backend's error callback.
-                log::warn!("Recreating audio capture stream after an error");
+                log::warn!("Recreating interrupted audio capture stream");
                 super::restart();
             }
         }
@@ -275,89 +255,29 @@ mod cpal_impl {
         }
     }
 
-    #[derive(Clone, Copy)]
+    fn send(
-    struct CaptureFrameProcessorConfig {
+        data: Vec<f32>,
-        input_rate: u32,
+        sample_rate0: u32,
-        output_rate: u32,
+        sample_rate: u32,
         device_channel: u16,
         encode_channel: u16,
-    }
+        encoder: &mut Encoder,
-
+        sp: &GenericService,
-    struct CaptureFrameProcessor {
+    ) {
-        config: CaptureFrameProcessorConfig,
+        let mut data = data;
-        resampler: Option<crate::audio_resampler::FixedFrameAudioResampler>,
+        if sample_rate0 != sample_rate {
-        sender: audio_capture_queue::CapturePcmSender,
+            data = crate::common::audio_resample(&data, sample_rate0, sample_rate, device_channel);
-        rechannel_buffer: Vec<f32>,
-    }
-
-    struct CaptureStreamOutput {
-        sender: audio_capture_queue::CapturePcmSender,
-        sample_rate: u32,
-        encode_channel: magnum_opus::Channels,
-    }
-
-    impl CaptureFrameProcessor {
-        fn new(
-            config: CaptureFrameProcessorConfig,
-            sender: audio_capture_queue::CapturePcmSender,
-        ) -> ResultType<Self> {
-            let resampler = if config.input_rate == config.output_rate {
-                None
-            } else {
-                let output_frames = config.output_rate as usize / AUDIO_PACKETS_PER_SECOND;
-                Some(crate::audio_resampler::FixedFrameAudioResampler::new(
-                    crate::audio_resampler::AudioResamplerConfig {
-                        input_rate: config.input_rate,
-                        output_rate: config.output_rate,
-                        channels: config.device_channel,
-                    },
-                    output_frames,
-                )?)
-            };
-            Ok(Self {
-                config,
-                resampler,
-                sender,
-                rechannel_buffer: Vec::with_capacity(
-                    capture_packet_layout(config.output_rate, config.encode_channel)?.1,
-                ),
-            })
         }
-
+        if device_channel != encode_channel {
-        fn process(&mut self, data: &[f32]) -> ResultType<()> {
+            data = crate::common::audio_rechannel(
-            let config = self.config;
+                data,
-            let sender = &mut self.sender;
+                sample_rate,
-            let rechannel_buffer = &mut self.rechannel_buffer;
+                sample_rate,
-            let mut send_packet = |packet: &[f32]| {
+                device_channel,
-                let packet =
+                encode_channel,
-                    audio_capture::rechannel(packet, config.device_channel, rechannel_buffer);
-                sender.submit(packet);
-            };
-            if let Some(resampler) = self.resampler.as_mut() {
-                resampler.process_with(data, send_packet).with_context(|| {
-                    format!(
-                        "Failed to resample captured audio from {} Hz to {} Hz",
-                        config.input_rate, config.output_rate
-                    )
-                })?;
-            } else {
-                send_packet(data);
-            }
-            Ok(())
-        }
-    }
-
-    fn capture_packet_layout(sample_rate: u32, channels: u16) -> ResultType<(usize, usize)> {
-        if sample_rate < AUDIO_PACKETS_PER_SECOND as u32 || channels == 0 {
-            bail!("Invalid audio capture layout: sample_rate={sample_rate}, channels={channels}");
-        }
-        let frames = sample_rate as usize / AUDIO_PACKETS_PER_SECOND;
-        let samples = frames.checked_mul(channels as usize).with_context(|| {
-            format!(
-                "Audio capture frame size overflow: sample_rate={sample_rate}, channels={channels}"
             )
-        })?;
+        }
-        Ok((frames, samples))
+        send_f32(&data, encoder, sp);
     }
 
     #[cfg(feature = "screencapturekit")]
@@ -447,7 +367,9 @@ mod cpal_impl {
         Ok((device, format))
     }
 
-    fn play(sp: &GenericService) -> ResultType<ActiveCaptureStream> {
+    fn play(
+        sp: &GenericService,
+    ) -> ResultType<(Box<dyn StreamTrait>, Arc<Message>, CaptureErrorHandler)> {
         use cpal::SampleFormat::*;
         let (device, config) = get_device()?;
         let sp = sp.clone();
@@ -465,274 +387,109 @@ mod cpal_impl {
             48000
         };
         let ch = if config.channels() > 1 { Stereo } else { Mono };
-        let max_channels = config.channels().max(ch as u16);
-        let (_, max_packet_samples) = capture_packet_layout(sample_rate, max_channels)?;
-        let encoder_config = audio_capture_queue::CaptureEncoderConfig {
-            sample_rate,
-            encode_channel: ch,
-            max_packet_samples,
-        };
-        let (sender, encoder_worker) =
-            audio_capture_queue::start_capture_encoder(encoder_config, sp)?;
-        let output = CaptureStreamOutput {
-            sender,
-            sample_rate,
-            encode_channel: ch,
-        };
         let (stream, errors) = match config.sample_format() {
-            I8 => build_input_stream::<i8>(device, &config, output)?,
+            I8 => build_input_stream::<i8>(device, &config, sp, sample_rate, ch)?,
-            I16 => build_input_stream::<i16>(device, &config, output)?,
+            I16 => build_input_stream::<i16>(device, &config, sp, sample_rate, ch)?,
-            I32 => build_input_stream::<i32>(device, &config, output)?,
+            I32 => build_input_stream::<i32>(device, &config, sp, sample_rate, ch)?,
-            I64 => build_input_stream::<i64>(device, &config, output)?,
+            I64 => build_input_stream::<i64>(device, &config, sp, sample_rate, ch)?,
-            U8 => build_input_stream::<u8>(device, &config, output)?,
+            U8 => build_input_stream::<u8>(device, &config, sp, sample_rate, ch)?,
-            U16 => build_input_stream::<u16>(device, &config, output)?,
+            U16 => build_input_stream::<u16>(device, &config, sp, sample_rate, ch)?,
-            U32 => build_input_stream::<u32>(device, &config, output)?,
+            U32 => build_input_stream::<u32>(device, &config, sp, sample_rate, ch)?,
-            U64 => build_input_stream::<u64>(device, &config, output)?,
+            U64 => build_input_stream::<u64>(device, &config, sp, sample_rate, ch)?,
-            F32 => build_input_stream::<f32>(device, &config, output)?,
+            F32 => build_input_stream::<f32>(device, &config, sp, sample_rate, ch)?,
-            F64 => build_input_stream::<f64>(device, &config, output)?,
+            F64 => build_input_stream::<f64>(device, &config, sp, sample_rate, ch)?,
             f => bail!("unsupported audio format: {:?}", f),
         };
         stream.play()?;
         #[cfg(target_os = "macos")]
         log::info!("Audio capture start call succeeded");
-        Ok(ActiveCaptureStream {
+        Ok((
-            stream: Some(Box::new(stream)),
+            Box::new(stream),
-            format: Arc::new(create_format_msg(sample_rate, ch as _)),
+            Arc::new(create_format_msg(sample_rate, ch as _)),
-            _encoder_worker: encoder_worker,
             errors,
-        })
+        ))
-    }
-
-    fn convert_input_samples<T>(data: &[T]) -> impl Iterator<Item = f32> + '_
-    where
-        T: cpal::SizedSample,
-        f32: cpal::FromSample<T>,
-    {
-        data.iter()
-            .map(|sample| <f32 as cpal::FromSample<T>>::from_sample_(*sample))
-    }
-
-    #[cfg(target_os = "macos")]
-    fn log_capture_startup<T>(
-        data: &[T],
-        received_samples: bool,
-        received_signal: bool,
-    ) -> (bool, bool)
-    where
-        T: cpal::SizedSample,
-        f32: cpal::FromSample<T>,
-    {
-        // Starting capture does not guarantee sample delivery or audible data.
-        if !received_samples && !data.is_empty() {
-            log::info!(
-                "Audio capture received first PCM block: {} samples",
-                data.len()
-            );
-        }
-        let has_signal = received_signal
-            || convert_input_samples(data).any(|sample| sample.is_finite() && sample != 0.0);
-        if !received_signal && has_signal {
-            log::info!("Audio capture received first nonzero PCM");
-        }
-        (received_samples || !data.is_empty(), has_signal)
     }
 
     fn build_input_stream<T>(
         device: cpal::Device,
         config: &cpal::SupportedStreamConfig,
-        output: CaptureStreamOutput,
+        sp: GenericService,
+        sample_rate: u32,
+        encode_channel: magnum_opus::Channels,
     ) -> ResultType<(cpal::Stream, CaptureErrorHandler)>
     where
-        T: cpal::SizedSample,
+        T: cpal::SizedSample + dasp::sample::ToSample<f32>,
-        f32: cpal::FromSample<T>,
     {
         let errors = CaptureErrorHandler::default();
         let callback_errors = errors.clone();
         let err_fn = move |err| callback_errors.handle(err);
-        let processor_errors = errors.clone();
         #[cfg(target_os = "macos")]
         let (mut received_samples, mut received_signal) = (false, false);
         let sample_rate_0 = config.sample_rate().0;
-        log::debug!("Audio sample rate : {}", output.sample_rate);
+        log::debug!("Audio sample rate : {}", sample_rate);
+        unsafe {
+            AUDIO_ZERO_COUNT = 0;
+        }
         let device_channel = config.channels();
-        let (_, capture_frame_samples) = capture_packet_layout(sample_rate_0, device_channel)?;
+        let mut encoder = Encoder::new(sample_rate, encode_channel, LowDelay)?;
-        let mut frame = audio_capture::CaptureFrameBuffer::new(capture_frame_samples)?;
+        // https://www.opus-codec.org/docs/html_api/group__opusencoder.html#gace941e4ef26ed844879fde342ffbe546
-        let processor_config = CaptureFrameProcessorConfig {
+        // https://chromium.googlesource.com/chromium/deps/opus/+/1.1.1/include/opus.h
-            input_rate: sample_rate_0,
+        // Do not set `frame_size = sample_rate as usize / 100;`
-            output_rate: output.sample_rate,
+        // Because we find `sample_rate as usize / 100` will cause encoder error in `encoder.encode_vec_float()` sometimes.
-            device_channel,
+        // https://github.com/xiph/opus/blob/2554a89e02c7fc30a980b4f7e635ceae1ecba5d6/src/opus_encoder.c#L725
-            encode_channel: output.encode_channel as _,
+        let frame_size = sample_rate_0 as usize / 100; // 10 ms
-        };
+        let encode_len = frame_size * encode_channel as usize;
-        let mut processor = CaptureFrameProcessor::new(processor_config, output.sender)?;
+        let rechannel_len = encode_len * device_channel as usize / encode_channel as usize;
+        INPUT_BUFFER.lock().unwrap().clear();
         let timeout = None;
+        let stream_config = StreamConfig {
+            channels: device_channel,
+            sample_rate: config.sample_rate(),
+            buffer_size: BufferSize::Default,
+        };
         let stream = device.build_input_stream(
-            &config.config(),
+            &stream_config,
             move |data: &[T], _: &InputCallbackInfo| {
-                if processor_errors.needs_restart() {
+                let buffer: Vec<f32> = data.iter().map(|s| T::to_sample(*s)).collect();
-                    return;
-                }
                 #[cfg(target_os = "macos")]
                 {
-                    (received_samples, received_signal) =
+                    // Starting capture does not guarantee sample delivery or audible data.
-                        log_capture_startup(data, received_samples, received_signal);
+                    if !received_samples && !buffer.is_empty() {
+                        received_samples = true;
+                        log::info!(
+                            "Audio capture received first PCM block: {} samples",
+                            buffer.len()
+                        );
+                    }
+                    if !received_signal
+                        && buffer
+                            .iter()
+                            .any(|sample| sample.is_finite() && *sample != 0.0)
+                    {
+                        received_signal = true;
+                        log::info!("Audio capture received first nonzero PCM");
+                    }
+                }
+                let mut lock = INPUT_BUFFER.lock().unwrap();
+                lock.extend(buffer);
+                while lock.len() >= rechannel_len {
+                    let frame: Vec<f32> = lock.drain(0..rechannel_len).collect();
+                    send(
+                        frame,
+                        sample_rate_0,
+                        sample_rate,
+                        device_channel,
+                        encode_channel as _,
+                        &mut encoder,
+                        &sp,
+                    );
                 }
-                frame.process(convert_input_samples(data), |frame| {
-                    processor_errors.process_frame(|| processor.process(frame));
-                });
             },
             err_fn,
             timeout,
         )?;
         Ok((stream, errors))
     }
-
-    #[cfg(test)]
-    mod tests {
-        use super::super::audio_capture_queue::{
-            new_pcm_handoff, start_capture_encoder, CaptureEncoderConfig,
-        };
-        use super::{
-            capture_packet_layout, convert_input_samples, CaptureFrameProcessor,
-            CaptureFrameProcessorConfig,
-        };
-        use crate::audio_resampler::allocation_tests::assert_no_allocations;
-        use crate::server::EmptyExtraFieldService;
-        use magnum_opus::Channels::{Mono, Stereo};
-
-        const INVALID_CAPTURE_RATE: u32 = 99;
-        const RATE_24_KHZ: u32 = 24_000;
-        const RATE_44_1_KHZ: u32 = 44_100;
-        const RATE_48_KHZ: u32 = 48_000;
-        const MONO_CHANNELS: u16 = 1;
-        const NEGATIVE_FULL_SCALE_LIMIT: f32 = -0.99;
-        const POSITIVE_FULL_SCALE_LIMIT: f32 = 0.99;
-        const STEREO_CHANNELS: u16 = 2;
-        const SURROUND_CHANNELS: u16 = 6;
-        const ZERO_CHANNELS: u16 = 0;
-
-        #[test]
-        fn capture_sample_conversion_uses_cpal_traits() {
-            let input = [i16::MIN, 0, i16::MAX];
-            let output: Vec<_> = convert_input_samples(&input).collect();
-
-            assert_eq!(output.len(), input.len());
-            assert!(output[0] <= NEGATIVE_FULL_SCALE_LIMIT);
-            assert_eq!(output[1], 0.0);
-            assert!(output[2] >= POSITIVE_FULL_SCALE_LIMIT);
-        }
-
-        #[test]
-        fn capture_packet_layout_validates_rate_and_channels() {
-            let expected_frames = RATE_48_KHZ as usize / super::AUDIO_PACKETS_PER_SECOND;
-            assert_eq!(
-                capture_packet_layout(RATE_48_KHZ, STEREO_CHANNELS).unwrap(),
-                (expected_frames, expected_frames * STEREO_CHANNELS as usize)
-            );
-            assert!(capture_packet_layout(INVALID_CAPTURE_RATE, MONO_CHANNELS).is_err());
-            assert!(capture_packet_layout(RATE_48_KHZ, ZERO_CHANNELS).is_err());
-        }
-
-        #[test]
-        fn capture_callback_pipeline_does_not_allocate_after_warmup() {
-            for (input_rate, output_rate, device_channel, encode_channel) in [
-                (RATE_48_KHZ, RATE_48_KHZ, MONO_CHANNELS, MONO_CHANNELS),
-                (RATE_48_KHZ, RATE_48_KHZ, STEREO_CHANNELS, STEREO_CHANNELS),
-                (RATE_44_1_KHZ, RATE_24_KHZ, STEREO_CHANNELS, STEREO_CHANNELS),
-                (RATE_48_KHZ, RATE_48_KHZ, SURROUND_CHANNELS, STEREO_CHANNELS),
-            ] {
-                assert_capture_processor_does_not_allocate(CaptureFrameProcessorConfig {
-                    input_rate,
-                    output_rate,
-                    device_channel,
-                    encode_channel,
-                });
-            }
-        }
-
-        #[test]
-        fn capture_pcm_handoff_reuses_buffers_and_accounts_for_loss() {
-            const QUEUE_CAPACITY: usize = 2;
-            const PACKET_SAMPLES: usize = 4;
-            const FIRST: [f32; PACKET_SAMPLES] = [1.0; PACKET_SAMPLES];
-            const SECOND: [f32; PACKET_SAMPLES] = [2.0; PACKET_SAMPLES];
-            const THIRD: [f32; PACKET_SAMPLES] = [3.0; PACKET_SAMPLES];
-            const OVERSIZED_SAMPLES: usize = PACKET_SAMPLES + 1;
-            const OVERSIZED: [f32; OVERSIZED_SAMPLES] = [1.0; OVERSIZED_SAMPLES];
-
-            let (mut sender, receiver) = new_pcm_handoff(QUEUE_CAPACITY, PACKET_SAMPLES).unwrap();
-            sender.set_wake_thread(std::thread::current()).unwrap();
-            assert_no_allocations(|| {
-                sender.submit(&FIRST);
-                sender.submit(&SECOND);
-                sender.submit(&THIRD);
-            });
-
-            let loss = receiver.take_loss();
-            assert_eq!(loss.dropped, 1);
-            assert_eq!(loss.oversized, 0);
-            assert_eq!(loss.recycle_failures, 0);
-            let second = receiver.pop().unwrap();
-            let third = receiver.pop().unwrap();
-            assert_eq!(second, SECOND);
-            assert_eq!(third, THIRD);
-            receiver.recycle(second);
-            receiver.recycle(third);
-            assert!(receiver.is_empty());
-
-            assert_no_allocations(|| sender.submit(&OVERSIZED));
-            let loss = receiver.take_loss();
-            assert_eq!(loss.dropped, 0);
-            assert_eq!(loss.oversized, 1);
-            assert_eq!(loss.recycle_failures, 0);
-            assert!(receiver.is_empty());
-        }
-
-        #[test]
-        fn capture_pcm_handoff_rejects_invalid_layouts() {
-            assert!(new_pcm_handoff(0, 1).is_err());
-            assert!(new_pcm_handoff(1, 0).is_err());
-        }
-
-        fn assert_capture_processor_does_not_allocate(config: CaptureFrameProcessorConfig) {
-            const INPUT_LEVEL: f32 = 0.25;
-            const TEST_SERVICE_NAME: &str = "audio-allocation-test";
-
-            let service = EmptyExtraFieldService::new(TEST_SERVICE_NAME.to_owned(), true).sp;
-            let encode_channel = if config.encode_channel == MONO_CHANNELS {
-                Mono
-            } else {
-                Stereo
-            };
-            let encoder_config = CaptureEncoderConfig {
-                sample_rate: config.output_rate,
-                encode_channel,
-                max_packet_samples: config.output_rate as usize / super::AUDIO_PACKETS_PER_SECOND
-                    * config.device_channel.max(config.encode_channel) as usize,
-            };
-            let (sender, worker) = start_capture_encoder(encoder_config, service).unwrap();
-            let mut processor = CaptureFrameProcessor::new(config, sender).unwrap();
-            let errors = super::CaptureErrorHandler::default();
-            let input = vec![
-                INPUT_LEVEL;
-                config.input_rate as usize / super::AUDIO_PACKETS_PER_SECOND
-                    * config.device_channel as usize
-            ];
-            let mut frame_buffer =
-                super::audio_capture::CaptureFrameBuffer::new(input.len()).unwrap();
-
-            frame_buffer.process(convert_input_samples(&input), |frame| {
-                errors.process_frame(|| processor.process(frame));
-            });
-            assert_no_allocations(|| {
-                frame_buffer.process(convert_input_samples(&input), |frame| {
-                    errors.process_frame(|| processor.process(frame));
-                });
-            });
-            assert!(!errors.needs_restart());
-            drop(processor);
-            drop(worker);
-        }
-    }
 }
 
 fn create_format_msg(sample_rate: u32, channels: u16) -> Message {
@@ -748,43 +505,28 @@ fn create_format_msg(sample_rate: u32, channels: u16) -> Message {
     msg
 }
 
-// Use a per-encoder counter for the Noise(Zero) Gate Attack Time.
+// use AUDIO_ZERO_COUNT for the Noise(Zero) Gate Attack Time
 // every audio data length is set to 480
 // MAX_AUDIO_ZERO_COUNT=800 is similar as Gate Attack Time 3~5s(Linux) || 6~8s(Windows)
 const MAX_AUDIO_ZERO_COUNT: u16 = 800;
+static mut AUDIO_ZERO_COUNT: u16 = 0;
 
-struct AudioEncoder {
+fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
-    encoder: Encoder,
+    if data.iter().filter(|x| **x != 0.).next().is_some() {
-    zero_count: u16,
+        unsafe {
-}
+            AUDIO_ZERO_COUNT = 0;
-
-impl AudioEncoder {
-    fn new(encoder: Encoder) -> Self {
-        Self {
-            encoder,
-            zero_count: 0,
         }
-    }
+    } else {
-
+        unsafe {
-    fn should_encode(&mut self, data: &[f32]) -> bool {
+            if AUDIO_ZERO_COUNT > MAX_AUDIO_ZERO_COUNT {
-        if data.iter().filter(|x| **x != 0.).next().is_some() {
+                if AUDIO_ZERO_COUNT == MAX_AUDIO_ZERO_COUNT + 1 {
-            self.zero_count = 0;
+                    log::debug!("Audio Zero Gate Attack");
-        } else if self.zero_count > MAX_AUDIO_ZERO_COUNT {
+                    AUDIO_ZERO_COUNT += 1;
-            if self.zero_count == MAX_AUDIO_ZERO_COUNT + 1 {
+                }
-                log::debug!("Audio Zero Gate Attack");
+                return;
-                self.zero_count += 1;
             }
-            return false;
+            AUDIO_ZERO_COUNT += 1;
-        } else {
-            self.zero_count += 1;
         }
-        true
-    }
-}
-
-fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) {
-    if !encoder.should_encode(data) {
-        return;
     }
     #[cfg(target_os = "android")]
     {
@@ -797,7 +539,6 @@ fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) {
             let n = input_size / BATCH_SIZE;
             for i in 0..n {
                 match encoder
-                    .encoder
                     .encode_vec_float(&data[i * BATCH_SIZE..(i + 1) * BATCH_SIZE], BATCH_SIZE)
                 {
                     Ok(data) => {
@@ -808,7 +549,7 @@ fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) {
                         });
                         sp.send(msg_out);
                     }
-                    Err(error) => log::warn!("Failed to encode audio frame: {error:?}"),
+                    Err(_) => {}
                 }
             }
         } else {
@@ -818,7 +559,7 @@ fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) {
     }
 
     #[cfg(not(target_os = "android"))]
-    match encoder.encoder.encode_vec_float(data, data.len() * 6) {
+    match encoder.encode_vec_float(data, data.len() * 6) {
         Ok(data) => {
             let mut msg_out = Message::new();
             msg_out.set_audio_frame(AudioFrame {
@@ -827,6 +568,6 @@ fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) {
             });
             sp.send(msg_out);
         }
-        Err(error) => log::warn!("Failed to encode audio frame: {error:?}"),
+        Err(_) => {}
     }
 }

src/server/audio_service/audio_capture.rs

@@ -1,144 +0,0 @@
-use hbb_common::anyhow::{bail, Result};
-
-const STEREO_CHANNELS: usize = 2;
-
-pub(super) struct CaptureFrameBuffer {
-    samples: Vec<f32>,
-    filled: usize,
-}
-
-impl CaptureFrameBuffer {
-    pub(super) fn new(samples: usize) -> Result<Self> {
-        if samples == 0 {
-            bail!("Audio capture frame must contain at least one sample");
-        }
-        Ok(Self {
-            samples: vec![0.0; samples],
-            filled: 0,
-        })
-    }
-
-    pub(super) fn process(
-        &mut self,
-        input: impl Iterator<Item = f32>,
-        mut on_frame: impl FnMut(&[f32]),
-    ) {
-        for sample in input {
-            self.samples[self.filled] = sample;
-            self.filled += 1;
-            if self.filled == self.samples.len() {
-                self.filled = 0;
-                on_frame(&self.samples);
-            }
-        }
-    }
-}
-
-pub(super) fn rechannel<'a>(
-    input: &'a [f32],
-    channels: u16,
-    output: &'a mut Vec<f32>,
-) -> &'a [f32] {
-    let input = if channels > STEREO_CHANNELS as u16 {
-        &input[..input.len() / channels as usize * channels as usize]
-    } else {
-        input
-    };
-    output.clear();
-    match channels {
-        3 => rechannel_frame::<3>(input, output),
-        4 => rechannel_frame::<4>(input, output),
-        5 => rechannel_frame::<5>(input, output),
-        6 => rechannel_frame::<6>(input, output),
-        7 => rechannel_frame::<7>(input, output),
-        8 => rechannel_frame::<8>(input, output),
-        // Preserve the existing passthrough for mono/stereo and unsupported layouts.
-        _ => return input,
-    }
-    output
-}
-
-fn rechannel_frame<const CHANNELS: usize>(input: &[f32], output: &mut Vec<f32>) {
-    use fon::{
-        chan::{Ch32, Channel},
-        Frame,
-    };
-
-    for samples in input.chunks_exact(CHANNELS) {
-        let mut frame = Frame::<Ch32, CHANNELS>::default();
-        for (channel, sample) in frame.channels_mut().iter_mut().zip(samples) {
-            *channel = (*sample).into();
-        }
-        // Match the same-rate Stream::pipe conversion before fon's SinkTo conversion.
-        let stereo = frame.to::<Ch32, CHANNELS>().to::<Ch32, STEREO_CHANNELS>();
-        output.extend(stereo.channels().iter().map(|channel| channel.to_f32()));
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{rechannel, CaptureFrameBuffer};
-    use crate::audio_resampler::allocation_tests::assert_no_allocations;
-
-    #[test]
-    fn capture_channel_conversion_reuses_storage_and_preserves_mapping() {
-        const SAMPLE_RATE: u32 = 48_000;
-        const FRAMES: usize = 7;
-        const STEREO_CHANNELS: u16 = 2;
-        const MIN_SAMPLE: f32 = -1.25;
-        const SAMPLE_STEP: f32 = 0.2;
-
-        for channels in [1, 2, 3, 4, 5, 6, 7, 8, 16] {
-            let input: Vec<_> = (0..FRAMES * channels as usize + 1)
-                .map(|sample| MIN_SAMPLE + sample as f32 * SAMPLE_STEP)
-                .collect();
-            let expected = crate::common::audio_rechannel(
-                input.clone(),
-                SAMPLE_RATE,
-                SAMPLE_RATE,
-                channels,
-                channels.min(STEREO_CHANNELS),
-            );
-            let mut output = Vec::with_capacity(FRAMES * STEREO_CHANNELS as usize);
-            assert_no_allocations(|| {
-                let actual = rechannel(&input, channels, &mut output);
-                assert_eq!(
-                    actual, expected,
-                    "channel mapping changed for {channels} channels"
-                );
-            });
-        }
-    }
-
-    #[test]
-    fn capture_framing_retains_partial_input_without_allocating() {
-        const FRAME_SAMPLES: usize = 6;
-        const CALLBACK_SIZES: [usize; 7] = [0, 1, 17, 2, 257, 492, 5];
-        let input: Vec<_> = (0..CALLBACK_SIZES.iter().sum::<usize>())
-            .map(|sample| sample as f32)
-            .collect();
-        let mut buffer = CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap();
-        let mut input_position = 0;
-        let mut output_position = 0;
-
-        assert!(CaptureFrameBuffer::new(0).is_err());
-        assert_no_allocations(|| {
-            for samples in CALLBACK_SIZES {
-                let end = input_position + samples;
-                buffer.process(input[input_position..end].iter().copied(), |frame| {
-                    assert_eq!(
-                        frame,
-                        &input[output_position..output_position + FRAME_SAMPLES]
-                    );
-                    output_position += FRAME_SAMPLES;
-                });
-                input_position = end;
-                assert_eq!(
-                    output_position,
-                    input_position / FRAME_SAMPLES * FRAME_SAMPLES
-                );
-            }
-        });
-        assert_eq!(output_position, input.len());
-    }
-}

src/server/audio_service/audio_capture_encoder.rs

@@ -1,128 +0,0 @@
-use super::super::AudioEncoder;
-use super::{
-    send_f32, CaptureEncoderConfig, CaptureEncoderContext, CapturePcmReceiver, CapturePcmStats,
-    CAPTURE_PCM_QUEUE_PACKETS,
-};
-use hbb_common::log;
-use magnum_opus::Channels;
-use std::{
-    sync::atomic::Ordering,
-    time::{Duration, Instant},
-};
-
-const CAPTURE_DECLICK_MS: usize = 5;
-const CAPTURE_PACKET_MS: usize = 10;
-const MILLISECONDS_PER_SECOND: usize = 1_000;
-const MAX_ENCODE_CHANNELS: usize = Channels::Stereo as usize;
-const CAPTURE_STATS_LOG_INTERVAL: Duration = Duration::from_secs(5);
-
-struct CaptureEncoderState {
-    channels: usize,
-    expected_sequence: usize,
-    fade_frames: usize,
-    last_frame: [f32; MAX_ENCODE_CHANNELS],
-    reporter: CaptureStatsReporter,
-}
-
-impl CaptureEncoderState {
-    fn new(sample_rate: u32, channels: Channels) -> Self {
-        // The encoder has already validated the supported Opus rate and channel count.
-        let fade_frames = sample_rate as usize * CAPTURE_DECLICK_MS / MILLISECONDS_PER_SECOND;
-        Self {
-            channels: channels as usize,
-            expected_sequence: 0,
-            fade_frames,
-            last_frame: [0.0; MAX_ENCODE_CHANNELS],
-            reporter: CaptureStatsReporter::new(),
-        }
-    }
-
-    fn next_packet(&mut self, receiver: &CapturePcmReceiver) -> Option<Vec<f32>> {
-        self.reporter.pending.add(receiver.take_stats());
-        self.reporter.report(false);
-        let (sequence, mut packet) = receiver.pop_packet()?;
-        self.smooth_packet(sequence, &mut packet);
-        Some(packet)
-    }
-
-    fn smooth_packet(&mut self, sequence: usize, packet: &mut [f32]) {
-        if sequence != self.expected_sequence {
-            // Capture packets contain 10 ms of PCM, so the transition fits in this packet.
-            for (index, frame) in packet
-                .chunks_exact_mut(self.channels)
-                .take(self.fade_frames)
-                .enumerate()
-            {
-                let weight = (index + 1) as f32 / self.fade_frames as f32;
-                for (channel, sample) in frame.iter_mut().enumerate() {
-                    *sample = self.last_frame[channel] * (1.0 - weight) + *sample * weight;
-                }
-            }
-        }
-        self.expected_sequence = sequence.wrapping_add(1);
-        if let Some(frame) = packet.chunks_exact(self.channels).next_back() {
-            self.last_frame[..self.channels].copy_from_slice(frame);
-        }
-    }
-}
-
-struct CaptureStatsReporter {
-    pending: CapturePcmStats,
-    last_report: Instant,
-}
-
-impl CaptureStatsReporter {
-    fn new() -> Self {
-        Self {
-            pending: Default::default(),
-            last_report: Instant::now(),
-        }
-    }
-
-    fn report(&mut self, force: bool) {
-        if self.pending.is_empty() {
-            return;
-        }
-        if !force && self.last_report.elapsed() < CAPTURE_STATS_LOG_INTERVAL {
-            return;
-        }
-        let stats = std::mem::take(&mut self.pending);
-        log::debug!(
-            "Audio capture PCM handoff stats: observed_max_queued_packets={}, approx_queued_audio_ms={}, capacity_packets={}",
-            stats.max_queued_packets,
-            stats.max_queued_packets.saturating_mul(CAPTURE_PACKET_MS),
-            CAPTURE_PCM_QUEUE_PACKETS
-        );
-        if !stats.loss.is_empty() {
-            log::warn!(
-                "Audio capture PCM handoff loss: dropped={}, contention_dropped={}, oversized={}, recycle_failures={}",
-                stats.loss.dropped,
-                stats.loss.contention_dropped,
-                stats.loss.oversized,
-                stats.loss.recycle_failures
-            );
-        }
-        self.last_report = Instant::now();
-    }
-}
-
-pub(super) fn run_capture_encoder(context: CaptureEncoderContext, config: CaptureEncoderConfig) {
-    let mut encoder = AudioEncoder::new(context.encoder);
-    let mut state = CaptureEncoderState::new(config.sample_rate, config.encode_channel);
-    loop {
-        while let Some(packet) = state.next_packet(&context.receiver) {
-            send_f32(&packet, &mut encoder, &context.service);
-            context.receiver.recycle(packet);
-        }
-        if context.stop.load(Ordering::Acquire) && context.receiver.is_empty() {
-            state.reporter.pending.add(context.receiver.take_stats());
-            state.reporter.report(true);
-            return;
-        }
-        std::thread::park_timeout(CAPTURE_STATS_LOG_INTERVAL);
-    }
-}
-
-#[cfg(test)]
-#[path = "audio_capture_encoder_tests.rs"]
-mod tests;

src/server/audio_service/audio_capture_encoder_tests.rs

@@ -1,186 +0,0 @@
-use super::super::{new_pcm_handoff, CAPTURE_PCM_QUEUE_PACKETS};
-use super::*;
-use crate::audio_resampler::allocation_tests::assert_no_allocations;
-use magnum_opus::{Application::LowDelay, Decoder, Encoder};
-
-const SAMPLE_RATE: u32 = 48_000;
-const PACKETS_PER_SECOND: usize = 100;
-const PACKET_COUNT: usize = 120;
-const PAUSE_PACKET: usize = 50;
-const DROPPED_PACKETS: usize = 6;
-const SIGNAL_FREQUENCY: f64 = 97.0;
-const SIGNAL_AMPLITUDE: f64 = 0.5;
-const ACTIVE_LEVEL: f32 = 0.8;
-const SAMPLE_TOLERANCE: f32 = 0.000001;
-const MAX_ENCODE_BYTES_PER_SAMPLE: usize = 6;
-
-fn signal_packet(index: usize, channels: usize) -> Vec<f32> {
-    let frames = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
-    (0..frames)
-        .flat_map(|frame| {
-            let phase = std::f64::consts::TAU * SIGNAL_FREQUENCY * (index * frames + frame) as f64
-                / f64::from(SAMPLE_RATE);
-            (0..channels)
-                .map(move |channel| (SIGNAL_AMPLITUDE * (phase + channel as f64).sin()) as f32)
-        })
-        .collect()
-}
-
-fn encoded_audio(dropped: usize, channels: Channels) -> Vec<f32> {
-    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels as usize;
-    let (mut sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, samples).unwrap();
-    let mut state = CaptureEncoderState::new(SAMPLE_RATE, channels);
-    let mut encoder = Encoder::new(SAMPLE_RATE, channels, LowDelay).unwrap();
-    let mut decoder = Decoder::new(SAMPLE_RATE, channels).unwrap();
-    let mut decoded = vec![0.0; samples];
-    let mut output = Vec::new();
-    let resume_packet = PAUSE_PACKET + CAPTURE_PCM_QUEUE_PACKETS + dropped - 1;
-    for index in 0..PACKET_COUNT {
-        sender.submit(&signal_packet(index, channels as usize));
-        if (PAUSE_PACKET..resume_packet).contains(&index) {
-            continue;
-        }
-        if index == resume_packet {
-            let stats = receiver.take_stats();
-            assert_eq!(stats.max_queued_packets, CAPTURE_PCM_QUEUE_PACKETS);
-            assert_eq!(stats.loss.dropped, dropped);
-            assert_eq!(stats.loss.contention_dropped, 0);
-            assert_eq!(stats.loss.oversized, 0);
-            assert_eq!(stats.loss.recycle_failures, 0);
-        }
-        while let Some(packet) = state.next_packet(&receiver) {
-            let encoded = encoder
-                .encode_vec_float(&packet, samples * MAX_ENCODE_BYTES_PER_SAMPLE)
-                .unwrap();
-            let frames = decoder.decode_float(&encoded, &mut decoded, false).unwrap();
-            assert_eq!(frames * channels as usize, samples);
-            output.extend_from_slice(&decoded);
-            receiver.recycle(packet);
-        }
-    }
-    assert_eq!(output.len(), (PACKET_COUNT - dropped) * samples);
-    output
-}
-
-fn maximum_join_step(pcm: &[f32], channels: usize) -> f32 {
-    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels;
-    pcm[(PAUSE_PACKET - 1) * samples..(PAUSE_PACKET + 2) * samples]
-        .windows(channels + 1)
-        .map(|window| (window[channels] - window[0]).abs())
-        .fold(0.0, f32::max)
-}
-
-#[test]
-fn capture_overflow_is_smoothed_before_encoding() {
-    const MAX_STEP_RATIO: f32 = 2.0;
-    for channels in [Channels::Mono, Channels::Stereo] {
-        let clean = encoded_audio(0, channels);
-        let overflow = encoded_audio(DROPPED_PACKETS, channels);
-        let clean_step = maximum_join_step(&clean, channels as usize);
-        let overflow_step = maximum_join_step(&overflow, channels as usize);
-        assert!(clean_step > 0.0);
-        assert!(
-            overflow_step < clean_step * MAX_STEP_RATIO,
-            "capture discard introduced a sharp join: clean={clean_step}, overflow={overflow_step}"
-        );
-    }
-}
-
-#[test]
-fn rejected_packets_mark_the_gap_after_already_queued_audio() {
-    const CAPACITY: usize = 3;
-    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
-    let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap();
-    let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono);
-    let active = vec![ACTIVE_LEVEL; samples];
-    let opposite = vec![-ACTIVE_LEVEL; samples];
-    sender.submit(&active);
-    let packet = state.next_packet(&receiver).unwrap();
-    assert_eq!(packet, active);
-    receiver.recycle(packet);
-    sender.submit(&active);
-    sender.submit(&active);
-    sender.submit(&vec![ACTIVE_LEVEL; samples + 1]);
-    sender.submit(&opposite);
-    assert_eq!(receiver.take_loss().oversized, 1);
-    assert_no_allocations(|| {
-        for _ in 0..CAPACITY - 1 {
-            let packet = state.next_packet(&receiver).unwrap();
-            assert_eq!(packet, active);
-            receiver.recycle(packet);
-        }
-        let packet = state.next_packet(&receiver).unwrap();
-        let expected = ACTIVE_LEVEL * (1.0 - 2.0 / state.fade_frames as f32);
-        assert!((packet[0] - expected).abs() < SAMPLE_TOLERANCE);
-        assert_eq!(&packet[samples / 2..], &opposite[samples / 2..]);
-        receiver.recycle(packet);
-    });
-}
-
-#[test]
-fn gaps_and_sequence_wrap_preserve_channel_history() {
-    const RATE_8_KHZ: u32 = 8_000;
-    for (rate, channels) in [
-        (RATE_8_KHZ, Channels::Mono),
-        (RATE_8_KHZ, Channels::Stereo),
-        (SAMPLE_RATE, Channels::Mono),
-        (SAMPLE_RATE, Channels::Stereo),
-    ] {
-        let channels_count = channels as usize;
-        let mut state = CaptureEncoderState::new(rate, channels);
-        let samples = rate as usize / PACKETS_PER_SECOND * channels_count;
-        let active: Vec<_> = [ACTIVE_LEVEL, -ACTIVE_LEVEL][..channels_count]
-            .iter()
-            .copied()
-            .cycle()
-            .take(samples)
-            .collect();
-        let mut first = active.clone();
-        state.smooth_packet(0, &mut first);
-        assert_eq!(first, active);
-        let mut opposite: Vec<_> = active.iter().map(|sample| -sample).collect();
-        let mut resumed = active.clone();
-        assert_no_allocations(|| {
-            state.smooth_packet(2, &mut opposite);
-            state.smooth_packet(4, &mut resumed);
-        });
-        for channel in 0..channels_count {
-            let expected = active[channel] * (1.0 - 2.0 / state.fade_frames as f32);
-            assert!((opposite[channel] - expected).abs() < SAMPLE_TOLERANCE);
-            let previous = opposite[samples - channels_count + channel];
-            let expected = previous + (active[channel] - previous) / state.fade_frames as f32;
-            assert!((resumed[channel] - expected).abs() < SAMPLE_TOLERANCE);
-        }
-        assert_eq!(&resumed[samples / 2..], &active[samples / 2..]);
-        state.expected_sequence = usize::MAX;
-        let mut last = active.clone();
-        let mut wrapped: Vec<_> = active.iter().map(|sample| -sample).collect();
-        let expected = wrapped.clone();
-        assert_no_allocations(|| {
-            state.smooth_packet(usize::MAX, &mut last);
-            state.smooth_packet(0, &mut wrapped);
-        });
-        assert_eq!(last, active);
-        assert_eq!(wrapped, expected);
-    }
-}
-
-#[test]
-fn capture_loss_is_reported_while_packets_remain_queued() {
-    const CAPACITY: usize = 2;
-    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
-    let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap();
-    let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono);
-    let before_report = Instant::now() - CAPTURE_STATS_LOG_INTERVAL;
-    state.reporter.last_report = before_report;
-    let packet = vec![ACTIVE_LEVEL; samples];
-    for _ in 0..=CAPACITY {
-        sender.submit(&packet);
-    }
-    let packet = state.next_packet(&receiver).unwrap();
-    assert!(!receiver.is_empty());
-    assert!(state.reporter.last_report > before_report);
-    assert!(state.reporter.pending.is_empty());
-    assert!(receiver.take_loss().is_empty());
-    receiver.recycle(packet);
-}

src/server/audio_service/audio_capture_queue.rs

@@ -1,289 +0,0 @@
-use super::{send_f32, GenericService};
-use hbb_common::{
-    anyhow::{bail, Context, Result},
-    log,
-};
-use magnum_opus::{Application::LowDelay, Channels, Encoder};
-use std::{
-    collections::VecDeque,
-    sync::{
-        atomic::{AtomicBool, AtomicUsize, Ordering},
-        Arc, Mutex, OnceLock, TryLockError,
-    },
-    thread::{JoinHandle, Thread},
-};
-
-#[path = "audio_capture_encoder.rs"]
-mod encoder;
-
-const CAPTURE_PCM_QUEUE_PACKETS: usize = 10;
-const CAPTURE_ENCODER_THREAD_NAME: &str = "audio-encoder";
-
-#[derive(Debug, Default, PartialEq, Eq)]
-pub(super) struct CapturePcmLoss {
-    pub(super) dropped: usize,
-    pub(super) contention_dropped: usize,
-    pub(super) oversized: usize,
-    pub(super) recycle_failures: usize,
-}
-
-impl CapturePcmLoss {
-    pub(super) fn is_empty(&self) -> bool {
-        self.dropped == 0 && self.oversized == 0 && self.recycle_failures == 0
-    }
-
-    pub(super) fn add(&mut self, other: Self) {
-        self.dropped += other.dropped;
-        self.contention_dropped += other.contention_dropped;
-        self.oversized += other.oversized;
-        self.recycle_failures += other.recycle_failures;
-    }
-}
-
-#[derive(Debug, Default, PartialEq, Eq)]
-pub(super) struct CapturePcmStats {
-    pub(super) loss: CapturePcmLoss,
-    pub(super) max_queued_packets: usize,
-}
-
-impl CapturePcmStats {
-    pub(super) fn is_empty(&self) -> bool {
-        self.loss.is_empty() && self.max_queued_packets == 0
-    }
-
-    pub(super) fn add(&mut self, other: Self) {
-        self.loss.add(other.loss);
-        self.max_queued_packets = self.max_queued_packets.max(other.max_queued_packets);
-    }
-}
-
-struct CapturePcmHandoff {
-    buffers: Mutex<CapturePcmBuffers>,
-    wake_thread: OnceLock<Thread>,
-    other_dropped: AtomicUsize,
-    contention_dropped: AtomicUsize,
-    oversized: AtomicUsize,
-    recycle_failures: AtomicUsize,
-    max_queued_packets: AtomicUsize,
-    max_samples: usize,
-}
-
-struct CapturePcmBuffers {
-    available: Vec<Vec<f32>>,
-    ready: VecDeque<(usize, Vec<f32>)>,
-}
-
-pub(super) struct CapturePcmSender {
-    handoff: Arc<CapturePcmHandoff>,
-    sequence: usize,
-}
-
-pub(super) struct CapturePcmReceiver {
-    handoff: Arc<CapturePcmHandoff>,
-}
-
-pub(super) struct CaptureEncoderConfig {
-    pub(super) sample_rate: u32,
-    pub(super) encode_channel: Channels,
-    pub(super) max_packet_samples: usize,
-}
-
-pub(super) struct CaptureEncoderWorker {
-    stop: Arc<AtomicBool>,
-    handle: Option<JoinHandle<()>>,
-}
-
-impl Drop for CaptureEncoderWorker {
-    fn drop(&mut self) {
-        self.stop.store(true, Ordering::Release);
-        if let Some(handle) = self.handle.take() {
-            handle.thread().unpark();
-            if let Err(error) = handle.join() {
-                log::error!("Failed to join audio encoder thread: {error:?}");
-            }
-        }
-    }
-}
-
-struct CaptureEncoderContext {
-    receiver: CapturePcmReceiver,
-    encoder: Encoder,
-    service: GenericService,
-    stop: Arc<AtomicBool>,
-}
-
-pub(super) fn new_pcm_handoff(
-    capacity: usize,
-    max_samples: usize,
-) -> Result<(CapturePcmSender, CapturePcmReceiver)> {
-    if capacity == 0 || max_samples == 0 {
-        bail!("Audio capture PCM handoff requires nonzero capacity and packet size");
-    }
-    let handoff = Arc::new(CapturePcmHandoff {
-        buffers: Mutex::new(CapturePcmBuffers {
-            available: Vec::with_capacity(capacity),
-            ready: VecDeque::with_capacity(capacity),
-        }),
-        wake_thread: OnceLock::new(),
-        other_dropped: AtomicUsize::new(0),
-        contention_dropped: AtomicUsize::new(0),
-        oversized: AtomicUsize::new(0),
-        recycle_failures: AtomicUsize::new(0),
-        max_queued_packets: AtomicUsize::new(0),
-        max_samples,
-    });
-    {
-        // Initialize the mutex on this thread, including on platforms with lazy allocation.
-        let mut buffers = handoff.buffers.lock().unwrap();
-        for _ in 0..capacity {
-            buffers.available.push(Vec::with_capacity(max_samples));
-        }
-    }
-    Ok((
-        CapturePcmSender {
-            handoff: handoff.clone(),
-            sequence: 0,
-        },
-        CapturePcmReceiver { handoff },
-    ))
-}
-
-impl CapturePcmSender {
-    pub(super) fn set_wake_thread(&self, thread: Thread) -> Result<()> {
-        if self.handoff.wake_thread.set(thread).is_err() {
-            bail!("Audio capture PCM wake thread is already configured");
-        }
-        Ok(())
-    }
-
-    pub(super) fn submit(&mut self, input: &[f32]) {
-        let sequence = self.sequence;
-        self.sequence = self.sequence.wrapping_add(1);
-        if input.len() > self.handoff.max_samples {
-            self.handoff.oversized.fetch_add(1, Ordering::Relaxed);
-            self.wake();
-            return;
-        }
-        // Do not wait for a descheduled worker. Contention rejects the current
-        // packet even if buffers are available; this is separate from drop-oldest
-        // when the buffer pool is exhausted.
-        let mut buffers = match self.handoff.buffers.try_lock() {
-            Ok(buffers) => buffers,
-            Err(TryLockError::WouldBlock) => {
-                self.handoff
-                    .contention_dropped
-                    .fetch_add(1, Ordering::Relaxed);
-                self.wake();
-                return;
-            }
-            Err(TryLockError::Poisoned(error)) => {
-                self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
-                log::error!("Audio capture PCM handoff is poisoned: {error}");
-                self.wake();
-                return;
-            }
-        };
-        if let Some(mut buffer) = self.take_buffer(&mut buffers) {
-            buffer.clear();
-            buffer.extend_from_slice(input);
-            buffers.ready.push_back((sequence, buffer));
-            self.handoff
-                .max_queued_packets
-                .fetch_max(buffers.ready.len(), Ordering::Relaxed);
-        } else {
-            self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
-        }
-        drop(buffers);
-        self.wake();
-    }
-
-    fn take_buffer(&self, buffers: &mut CapturePcmBuffers) -> Option<Vec<f32>> {
-        buffers.available.pop().or_else(|| {
-            let buffer = buffers.ready.pop_front().map(|(_, buffer)| buffer);
-            if buffer.is_some() {
-                self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
-            }
-            buffer
-        })
-    }
-
-    fn wake(&self) {
-        if let Some(thread) = self.handoff.wake_thread.get() {
-            thread.unpark();
-        }
-    }
-}
-
-impl CapturePcmReceiver {
-    #[cfg(test)]
-    pub(super) fn pop(&self) -> Option<Vec<f32>> {
-        self.pop_packet().map(|(_, buffer)| buffer)
-    }
-
-    fn pop_packet(&self) -> Option<(usize, Vec<f32>)> {
-        self.handoff.buffers.lock().unwrap().ready.pop_front()
-    }
-
-    pub(super) fn recycle(&self, mut buffer: Vec<f32>) {
-        buffer.clear();
-        let mut buffers = self.handoff.buffers.lock().unwrap();
-        if buffers.available.len() == buffers.available.capacity() {
-            self.handoff
-                .recycle_failures
-                .fetch_add(1, Ordering::Relaxed);
-        } else {
-            buffers.available.push(buffer);
-        }
-    }
-
-    pub(super) fn is_empty(&self) -> bool {
-        self.handoff.buffers.lock().unwrap().ready.is_empty()
-    }
-
-    pub(super) fn take_loss(&self) -> CapturePcmLoss {
-        let contention_dropped = self.handoff.contention_dropped.swap(0, Ordering::Relaxed);
-        CapturePcmLoss {
-            dropped: self.handoff.other_dropped.swap(0, Ordering::Relaxed) + contention_dropped,
-            contention_dropped,
-            oversized: self.handoff.oversized.swap(0, Ordering::Relaxed),
-            recycle_failures: self.handoff.recycle_failures.swap(0, Ordering::Relaxed),
-        }
-    }
-
-    pub(super) fn take_stats(&self) -> CapturePcmStats {
-        CapturePcmStats {
-            loss: self.take_loss(),
-            max_queued_packets: self.handoff.max_queued_packets.swap(0, Ordering::Relaxed),
-        }
-    }
-}
-
-pub(super) fn start_capture_encoder(
-    config: CaptureEncoderConfig,
-    service: GenericService,
-) -> Result<(CapturePcmSender, CaptureEncoderWorker)> {
-    let (sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, config.max_packet_samples)?;
-    let encoder = Encoder::new(config.sample_rate, config.encode_channel, LowDelay)?;
-    let stop = Arc::new(AtomicBool::new(false));
-    let context = CaptureEncoderContext {
-        receiver,
-        encoder,
-        service,
-        stop: stop.clone(),
-    };
-    let handle = std::thread::Builder::new()
-        .name(CAPTURE_ENCODER_THREAD_NAME.to_owned())
-        .spawn(move || encoder::run_capture_encoder(context, config))
-        .with_context(|| "Failed to start audio encoder thread")?;
-    let wake_thread = handle.thread().clone();
-    let worker = CaptureEncoderWorker {
-        stop,
-        handle: Some(handle),
-    };
-    sender.set_wake_thread(wake_thread)?;
-    Ok((sender, worker))
-}
-
-#[cfg(test)]
-#[path = "audio_capture_queue_tests.rs"]
-mod tests;

src/server/audio_service/audio_capture_queue_tests.rs

@@ -1,169 +0,0 @@
-use super::*;
-use crate::audio_resampler::allocation_tests::assert_no_allocations;
-use std::{sync::mpsc, time::Duration};
-
-const PACKET_SAMPLES: usize = 4;
-const TEST_TIMEOUT: Duration = Duration::from_secs(2);
-const CONCURRENT_PACKETS: usize = 10_000;
-
-#[derive(Clone, Copy)]
-enum PausePoint {
-    Recycle,
-    Consume,
-}
-
-struct WorkerPause {
-    entered: mpsc::Sender<()>,
-    resume: mpsc::Receiver<()>,
-}
-
-fn paused_worker(
-    receiver: CapturePcmReceiver,
-    point: PausePoint,
-    pause: WorkerPause,
-) -> CapturePcmReceiver {
-    let recycled = match point {
-        PausePoint::Recycle => Some(receiver.pop_packet().unwrap().1),
-        PausePoint::Consume => None,
-    };
-    let consumed = {
-        let mut buffers = receiver.handoff.buffers.lock().unwrap();
-        let consumed = match recycled {
-            Some(mut packet) => {
-                packet.clear();
-                buffers.available.push(packet);
-                None
-            }
-            None => Some(buffers.ready.pop_front().unwrap().1),
-        };
-        pause.entered.send(()).unwrap();
-        pause.resume.recv().unwrap();
-        consumed
-    };
-    if let Some(packet) = consumed {
-        receiver.recycle(packet);
-    }
-    receiver
-}
-
-fn assert_callback_progress(point: PausePoint) {
-    let (mut sender, receiver) =
-        new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, PACKET_SAMPLES).unwrap();
-    for sequence in 0..CAPTURE_PCM_QUEUE_PACKETS {
-        sender.submit(&[sequence as f32; PACKET_SAMPLES]);
-    }
-    let (entered_tx, entered_rx) = mpsc::channel();
-    let (resume_tx, resume_rx) = mpsc::channel();
-    let pause = WorkerPause {
-        entered: entered_tx,
-        resume: resume_rx,
-    };
-    let worker = std::thread::spawn(move || paused_worker(receiver, point, pause));
-    sender.set_wake_thread(worker.thread().clone()).unwrap();
-    entered_rx.recv_timeout(TEST_TIMEOUT).unwrap();
-    let (completed_tx, completed_rx) = mpsc::channel();
-    let callback = std::thread::spawn(move || {
-        let packet = [CAPTURE_PCM_QUEUE_PACKETS as f32; PACKET_SAMPLES];
-        assert_no_allocations(|| sender.submit(&packet));
-        completed_tx.send(()).unwrap();
-        sender
-    });
-    let completed = completed_rx.recv_timeout(TEST_TIMEOUT);
-    // Release and join both threads before asserting progress, including on failure.
-    resume_tx.send(()).unwrap();
-    let receiver = worker.join().unwrap();
-    let mut sender = callback.join().unwrap();
-    assert!(completed.is_ok(), "callback waited for the encoder worker");
-    assert_eq!(
-        receiver.take_loss(),
-        CapturePcmLoss {
-            dropped: 1,
-            contention_dropped: 1,
-            ..Default::default()
-        }
-    );
-    assert_packets_after_contention(&mut sender, &receiver);
-}
-
-fn assert_packets_after_contention(sender: &mut CapturePcmSender, receiver: &CapturePcmReceiver) {
-    let next_sequence = CAPTURE_PCM_QUEUE_PACKETS + 1;
-    sender.submit(&[next_sequence as f32; PACKET_SAMPLES]);
-    for expected in (1..CAPTURE_PCM_QUEUE_PACKETS).chain(std::iter::once(next_sequence)) {
-        let (sequence, packet) = receiver.pop_packet().unwrap();
-        assert_eq!(sequence, expected);
-        assert_eq!(packet, [expected as f32; PACKET_SAMPLES]);
-        receiver.recycle(packet);
-    }
-    assert!(receiver.is_empty());
-    assert!(receiver.take_loss().is_empty());
-    assert_pool_restored(receiver, CAPTURE_PCM_QUEUE_PACKETS);
-}
-
-fn assert_pool_restored(receiver: &CapturePcmReceiver, capacity: usize) {
-    let buffers = receiver.handoff.buffers.lock().unwrap();
-    assert_eq!(buffers.available.len(), capacity);
-    assert!(buffers
-        .available
-        .iter()
-        .all(|buffer| buffer.capacity() >= PACKET_SAMPLES));
-}
-
-#[test]
-fn callback_finishes_while_worker_recycles_a_buffer() {
-    assert_callback_progress(PausePoint::Recycle);
-}
-
-#[test]
-fn callback_finishes_while_worker_releases_a_ready_packet() {
-    assert_callback_progress(PausePoint::Consume);
-}
-
-fn consume_concurrently(
-    receiver: CapturePcmReceiver,
-    finished: Arc<AtomicBool>,
-    first_sequence: usize,
-) -> (CapturePcmReceiver, usize) {
-    let mut received = 0;
-    let mut next_ordinal = 0;
-    loop {
-        if let Some((sequence, packet)) = receiver.pop_packet() {
-            let ordinal = sequence.wrapping_sub(first_sequence);
-            assert!(ordinal >= next_ordinal && ordinal < CONCURRENT_PACKETS);
-            assert_eq!(packet, [ordinal as f32; PACKET_SAMPLES]);
-            next_ordinal = ordinal + 1;
-            received += 1;
-            receiver.recycle(packet);
-        } else if finished.load(Ordering::Acquire) && receiver.is_empty() {
-            return (receiver, received);
-        } else {
-            std::thread::yield_now();
-        }
-    }
-}
-
-#[test]
-fn concurrent_handoff_preserves_order_buffers_and_loss_counts_across_sequence_wrap() {
-    const FIRST_SEQUENCE: usize = usize::MAX - CONCURRENT_PACKETS / 2;
-    for capacity in [1, 2, CAPTURE_PCM_QUEUE_PACKETS] {
-        let (mut sender, receiver) = new_pcm_handoff(capacity, PACKET_SAMPLES).unwrap();
-        sender.sequence = FIRST_SEQUENCE;
-        let finished = Arc::new(AtomicBool::new(false));
-        let worker_finished = finished.clone();
-        let worker = std::thread::spawn(move || {
-            consume_concurrently(receiver, worker_finished, FIRST_SEQUENCE)
-        });
-        assert_no_allocations(|| {
-            for ordinal in 0..CONCURRENT_PACKETS {
-                sender.submit(&[ordinal as f32; PACKET_SAMPLES]);
-            }
-        });
-        finished.store(true, Ordering::Release);
-        let (receiver, received) = worker.join().unwrap();
-        let stats = receiver.take_stats();
-        assert_eq!(received + stats.loss.dropped, CONCURRENT_PACKETS);
-        assert_eq!(stats.loss.oversized, 0);
-        assert_eq!(stats.loss.recycle_failures, 0);
-        assert!(stats.max_queued_packets <= capacity);
-        assert_pool_restored(&receiver, capacity);
-    }
-}