From 47c58047cec96676dbe05123b399c89212766b17 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Thu, 18 Jun 2026 17:15:56 -0400 Subject: [PATCH] Windows port Phase 1: real cpal/WASAPI audio backend Replace the Phase 0 no-op CpalBackend stub with a working cpal backend (WASAPI on Windows), preserving the exact PipeWire AudioBackend contract so the mixer/encoder/jitter pipeline is unchanged. - Capture: input stream -> downmix to mono -> 960-sample (20ms) i16 frames -> tx, matching the encoder/jitter frame size. - Playback: 200ms stereo ring prefilled to PLAYBACK_TARGET_SAMPLES; the output callback drains it (silence on underrun) while the owning thread feeds it from rx. ring_fill is the exact delta-maintained occupancy counter (fetch_add on push, fetch_sub on pop), preserving the clock-paced production design (not ringbuf's stale occupied_len). - cpal::Stream is !Send, but AudioBackend is Send+Sync and shared via Arc, so each stream lives on its own owning thread (built/played/dropped there); the struct holds only the running flag + JoinHandle. stop() flips the flag and joins. - Generic over F32/I16/U16 sample formats; device selected by name else default; requires a native 48kHz config (clear error otherwise, no resampling yet). Mirrors the PipeWire drain_loop and playout-health line. - Cargo.toml: add cpal 0.15 under cfg(windows). Verified by temporarily compiling cpal_impl against real cpal on Linux/ALSA: build + clippy clean, 6/6 cpal_impl unit tests pass. Reverted to windows-only gating; shipped Linux state green (316/316). Runtime/WASAPI end-to-end is unverified and pending a Windows host (plan M2). Co-Authored-By: Claude Opus 4.8 --- Cargo.lock | 292 ++++++++++++++++++-- Cargo.toml | 6 +- src/audio/cpal_impl.rs | 599 ++++++++++++++++++++++++++++++++++++++--- 3 files changed, 838 insertions(+), 59 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a177314..c7e880b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -105,6 +105,28 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "alsa" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed7572b7ba83a31e20d1b48970ee402d2e3e0537dcfe0a3ff4d6eb7508617d43" +dependencies = [ + "alsa-sys", + "bitflags 2.11.1", + "cfg-if", + "libc", +] + +[[package]] +name = "alsa-sys" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db8fee663d06c4e303404ef5f40488a53e062f89ba8bfed81f42325aafad1527" +dependencies = [ + "libc", + "pkg-config", +] + [[package]] name = "android-activity" version = "0.6.1" @@ -114,12 +136,12 @@ dependencies = [ "android-properties", "bitflags 2.11.1", "cc", - "jni", + "jni 0.22.4", "libc", "log", - "ndk", + "ndk 0.9.0", "ndk-context", - "ndk-sys", + "ndk-sys 0.6.0+11769913", "num_enum", "thiserror 2.0.18", ] @@ -712,6 +734,12 @@ dependencies = [ "shlex", ] +[[package]] +name = "cesu8" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d43a04d8753f35258c91f8ec639f792891f748a1edbd759cf1dcea3382ad83c" + [[package]] name = "cexpr" version = "0.6.0" @@ -1002,6 +1030,26 @@ dependencies = [ "libm", ] +[[package]] +name = "coreaudio-rs" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "321077172d79c662f64f5071a03120748d5bb652f5231570141be24cfcd2bace" +dependencies = [ + "bitflags 1.3.2", + "core-foundation-sys", + "coreaudio-sys", +] + +[[package]] +name = "coreaudio-sys" +version = "0.2.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9b4739a805a62757a83e5654fa3faabec0442666b263bb2287d5a8185bfd953" +dependencies = [ + "bindgen", +] + [[package]] name = "cosmic-text" version = "0.15.0" @@ -1026,6 +1074,29 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "cpal" +version = "0.15.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "873dab07c8f743075e57f524c583985fbaf745602acbe916a01539364369a779" +dependencies = [ + "alsa", + "core-foundation-sys", + "coreaudio-rs", + "dasp_sample", + "jni 0.21.1", + "js-sys", + "libc", + "mach2", + "ndk 0.8.0", + "ndk-context", + "oboe", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "windows 0.54.0", +] + [[package]] name = "cpufeatures" version = "0.2.17" @@ -1228,6 +1299,12 @@ dependencies = [ "syn", ] +[[package]] +name = "dasp_sample" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c87e182de0887fd5361989c677c4e8f5000cd9491d6d563161a8f3a5519fc7f" + [[package]] name = "data-encoding" version = "2.11.0" @@ -2227,7 +2304,7 @@ dependencies = [ "http", "idna", "ipnet", - "jni", + "jni 0.22.4", "rand 0.10.1", "rustls", "thiserror 2.0.18", @@ -2247,7 +2324,7 @@ dependencies = [ "data-encoding", "idna", "ipnet", - "jni", + "jni 0.22.4", "once_cell", "prefix-trie", "rand 0.10.1", @@ -2270,7 +2347,7 @@ dependencies = [ "hickory-proto", "ipconfig", "ipnet", - "jni", + "jni 0.22.4", "moka", "ndk-context", "once_cell", @@ -3102,6 +3179,22 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jni" +version = "0.21.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a87aa2bb7d2af34197c04845522473242e1aa17c12f4935d5856491a7fb8c97" +dependencies = [ + "cesu8", + "cfg-if", + "combine", + "jni-sys 0.3.1", + "log", + "thiserror 1.0.69", + "walkdir", + "windows-sys 0.45.0", +] + [[package]] name = "jni" version = "0.22.4" @@ -3463,6 +3556,15 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3d25b0e0b648a86960ac23b7ad4abb9717601dec6f66c165f5b037f3f03065f" +[[package]] +name = "mach2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d640282b302c0bb0a2a8e0233ead9035e3bed871f0b7e81fe4a1ec829765db44" +dependencies = [ + "libc", +] + [[package]] name = "malloc_buf" version = "0.0.6" @@ -3596,7 +3698,7 @@ dependencies = [ "dispatch", "futures-channel", "futures-lite", - "jni", + "jni 0.22.4", "ndk-context", "objc2 0.6.4", "objc2-app-kit 0.3.2", @@ -3695,6 +3797,20 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "ndk" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2076a31b7010b17a38c01907c45b945e8f11495ee4dd588309718901b1f7a5b7" +dependencies = [ + "bitflags 2.11.1", + "jni-sys 0.3.1", + "log", + "ndk-sys 0.5.0+25.2.9519653", + "num_enum", + "thiserror 1.0.69", +] + [[package]] name = "ndk" version = "0.9.0" @@ -3704,7 +3820,7 @@ dependencies = [ "bitflags 2.11.1", "jni-sys 0.3.1", "log", - "ndk-sys", + "ndk-sys 0.6.0+11769913", "num_enum", "raw-window-handle", "thiserror 1.0.69", @@ -3716,6 +3832,15 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b" +[[package]] +name = "ndk-sys" +version = "0.5.0+25.2.9519653" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c196769dd60fd4f363e11d948139556a344e79d451aeb2fa2fd040738ef7691" +dependencies = [ + "jni-sys 0.3.1", +] + [[package]] name = "ndk-sys" version = "0.6.0+11769913" @@ -4466,6 +4591,29 @@ dependencies = [ "objc2-foundation 0.2.2", ] +[[package]] +name = "oboe" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8b61bebd49e5d43f5f8cc7ee2891c16e0f41ec7954d36bcb6c14c5e0de867fb" +dependencies = [ + "jni 0.21.1", + "ndk 0.8.0", + "ndk-context", + "num-derive", + "num-traits", + "oboe-sys", +] + +[[package]] +name = "oboe-sys" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c8bb09a4a2b1d668170cfe0a7d5bc103f8999fb316c98099b6a9939c9f2e79d" +dependencies = [ + "cc", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -4600,6 +4748,7 @@ dependencies = [ "async-trait", "base64", "bytes", + "cpal", "dirs", "iced", "image", @@ -5436,7 +5585,7 @@ checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0" dependencies = [ "core-foundation 0.10.1", "core-foundation-sys", - "jni", + "jni 0.22.4", "log", "once_cell", "rustls", @@ -5877,7 +6026,7 @@ dependencies = [ "fastrand", "js-sys", "memmap2", - "ndk", + "ndk 0.9.0", "objc2 0.6.4", "objc2-core-foundation", "objc2-core-graphics", @@ -7162,7 +7311,7 @@ dependencies = [ "log", "metal", "naga", - "ndk-sys", + "ndk-sys 0.6.0+11769913", "objc", "once_cell", "ordered-float", @@ -7247,6 +7396,16 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "windows" +version = "0.54.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9252e5725dbed82865af151df558e754e4a3c2c30818359eb17465f1346a1b49" +dependencies = [ + "windows-core 0.54.0", + "windows-targets 0.52.6", +] + [[package]] name = "windows" version = "0.58.0" @@ -7254,7 +7413,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd04d41d93c4992d421894c18c8b43496aa748dd4c081bac0dc93eb0489272b6" dependencies = [ "windows-core 0.58.0", - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -7278,6 +7437,16 @@ dependencies = [ "windows-core 0.62.2", ] +[[package]] +name = "windows-core" +version = "0.54.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12661b9c89351d684a50a8a643ce5f608e20243b9fb84687800163429f161d65" +dependencies = [ + "windows-result 0.1.2", + "windows-targets 0.52.6", +] + [[package]] name = "windows-core" version = "0.58.0" @@ -7288,7 +7457,7 @@ dependencies = [ "windows-interface 0.58.0", "windows-result 0.2.0", "windows-strings 0.1.0", - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -7386,13 +7555,22 @@ dependencies = [ "windows-strings 0.5.1", ] +[[package]] +name = "windows-result" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e383302e8ec8515204254685643de10811af0ed97ea37210dc26fb0032647f8" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-result" version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d1043d8214f791817bab27572aaa8af63732e11bf84aa21a45a78d6c317ae0e" dependencies = [ - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -7411,7 +7589,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4cd9b125c486025df0eabcb585e62173c6c9eddcec5d117d3b6e8c30e2ee4d10" dependencies = [ "windows-result 0.2.0", - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -7423,13 +7601,22 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.45.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0" +dependencies = [ + "windows-targets 0.42.2", +] + [[package]] name = "windows-sys" version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -7441,20 +7628,35 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071" +dependencies = [ + "windows_aarch64_gnullvm 0.42.2", + "windows_aarch64_msvc 0.42.2", + "windows_i686_gnu 0.42.2", + "windows_i686_msvc 0.42.2", + "windows_x86_64_gnu 0.42.2", + "windows_x86_64_gnullvm 0.42.2", + "windows_x86_64_msvc 0.42.2", +] + [[package]] name = "windows-targets" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", + "windows_aarch64_gnullvm 0.52.6", + "windows_aarch64_msvc 0.52.6", + "windows_i686_gnu 0.52.6", "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_i686_msvc 0.52.6", + "windows_x86_64_gnu 0.52.6", + "windows_x86_64_gnullvm 0.52.6", + "windows_x86_64_msvc 0.52.6", ] [[package]] @@ -7466,18 +7668,36 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8" + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_i686_gnu" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -7490,24 +7710,48 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060" + [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_x86_64_gnu" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" @@ -7536,7 +7780,7 @@ dependencies = [ "js-sys", "libc", "memmap2", - "ndk", + "ndk 0.9.0", "objc2 0.5.2", "objc2-app-kit 0.2.2", "objc2-foundation 0.2.2", diff --git a/Cargo.toml b/Cargo.toml index 54421d2..8ed7269 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,6 +61,6 @@ rfd = { version = "0.17", default-features = false, features = ["xdg-portal"] } [target.'cfg(windows)'.dependencies] # Native file picker using the built-in Win32 dialog backend on Windows. rfd = { version = "0.17", default-features = false } -# NOTE: the Windows audio backend (cpal/WASAPI) lands in Phase 1. Until then the -# Windows build uses the no-op `CpalBackend` stub in `src/audio/cpal_impl.rs`, -# which needs no extra dependency. +# Windows audio backend: cpal drives WASAPI for capture/playback behind the +# AudioBackend trait (src/audio/cpal_impl.rs). The Linux counterpart is pipewire. +cpal = "0.15" diff --git a/src/audio/cpal_impl.rs b/src/audio/cpal_impl.rs index 783a814..4555a97 100644 --- a/src/audio/cpal_impl.rs +++ b/src/audio/cpal_impl.rs @@ -1,31 +1,80 @@ -//! Windows audio backend (cpal/WASAPI) — **Phase 0 stub**. +//! Windows audio backend — cpal / WASAPI (Phase 1). //! -//! This is a compile-and-run placeholder so the Windows build links and the app -//! starts up (networking, UI, and text chat all functional) while the real -//! capture/playback implementation lands in Phase 1. Every method satisfies the -//! [`AudioBackend`] contract as a no-op: no microphone is captured and nothing is -//! played. It deliberately pulls in no extra dependency — `cpal` is added only -//! when the real implementation arrives. +//! Implements [`AudioBackend`] on top of [`cpal`], which wraps WASAPI on Windows. +//! It is the Windows counterpart to `pipewire_impl.rs` and deliberately preserves +//! the exact same contract so the rest of the app (mixer, encoder, jitter buffer) +//! is unchanged: //! -//! Phase 1 will replace this with cpal streams on the WASAPI host, mapping: -//! - `start_capture` → input stream, f32→i16, mono 48 kHz, into `tx`; -//! - `start_playback` → output stream draining a `ringbuf`, keeping `ring_fill` -//! updated so the existing hardware-clock pacing in the mixer keeps working; -//! - `stop` → drop the streams. +//! - **Capture**: mono, 48 kHz, S16 PCM, emitted as `Vec` frames of +//! [`CAPTURE_FRAME`] (960 = 20 ms) samples — matching the encoder/jitter frame. +//! - **Playback**: stereo interleaved ([`PLAYBACK_CHANNELS`]) S16 PCM at 48 kHz, +//! drained from a ring buffer that is paced to the device's hardware clock via +//! `ring_fill` exactly as the PipeWire backend does. +//! +//! ## Threading and the `!Send` stream +//! +//! `cpal::Stream` is `!Send` (some backends require it to be created and dropped +//! on the same thread), but [`AudioBackend`] is `Send + Sync` and the backend is +//! shared through an `Arc`. So the stream never lives in the struct: each of +//! `start_capture`/`start_playback` spawns one owning thread that builds the +//! stream, plays it, and keeps it alive until the per-worker `running` flag flips +//! (set by `stop`). The struct holds only `Send` handles (the flag + the join +//! handle). The stream's RT callback does the actual audio work; the owning +//! thread additionally feeds the playback ring from the network mixer. +//! +//! ## Sample rate +//! +//! The whole pipeline assumes 48 kHz (Opus + the 960-sample frame). Phase 1 only +//! selects a native-48 kHz device config; if the device can't do 48 kHz we return +//! a clear error rather than silently producing pitch-shifted audio. Arbitrary +//! sample-rate support (resampling) is a Phase 1.1 follow-up. -use std::sync::Arc; -use std::sync::atomic::AtomicUsize; -use std::sync::mpsc::{Receiver, Sender}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender}; +use std::sync::{Arc, Mutex}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; -use super::{AudioBackend, AudioError}; +use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; +use cpal::{Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig}; +use ringbuf::{ + traits::{Consumer, Producer, Split}, + HeapRb, +}; -/// No-op Windows audio backend (Phase 0). See module docs. -pub struct CpalBackend; +use super::{AudioBackend, AudioError, PLAYBACK_CHANNELS, PLAYBACK_TARGET_SAMPLES}; + +/// The one sample rate the pipeline supports (Opus + the 20 ms frame). +const SAMPLE_RATE: u32 = 48_000; +/// Mono capture frame: 960 samples = 20 ms @ 48 kHz. Matches the PipeWire backend +/// and `core::jitter::FRAME_SAMPLES`. +const CAPTURE_FRAME: usize = 960; +/// Playback ring capacity in interleaved samples: 200 ms of stereo @ 48 kHz. +/// Comfortably above [`PLAYBACK_TARGET_SAMPLES`] so the clock-paced producer has +/// headroom and never has to drop frames in steady state. +const RING_CAPACITY: usize = 9600 * PLAYBACK_CHANNELS; +/// How often a blocked playback worker re-checks its `running` flag, bounding how +/// long `stop()` can take to join it (mirrors the PipeWire backend's `WORKER_POLL`). +const WORKER_POLL: Duration = Duration::from_millis(100); + +/// Windows audio backend. See module docs. +pub struct CpalBackend { + capture: Mutex>, + playback: Mutex>, +} + +/// A spawned owning thread plus the flag that tells it to drop its stream and exit. +struct StreamWorker { + running: Arc, + thread: JoinHandle<()>, +} impl CpalBackend { pub fn new() -> Self { - crate::log_msg("CpalBackend: Phase 0 stub active (no audio I/O yet)"); - CpalBackend + Self { + capture: Mutex::new(None), + playback: Mutex::new(None), + } } } @@ -38,30 +87,516 @@ impl Default for CpalBackend { impl AudioBackend for CpalBackend { fn start_capture( &self, - _tx: Sender>, - _target_node: Option, + tx: Sender>, + target_node: Option, ) -> Result<(), AudioError> { - // No capture stream yet: dropping `_tx` simply means no samples are ever - // produced (silent mic), which is the intended Phase 0 behaviour. - crate::log_msg("CpalBackend::start_capture: not yet implemented (Phase 1) — capturing silence"); + let mut guard = self.capture.lock().unwrap(); + if guard.is_some() { + return Err(AudioError::Stream("Capture already started".to_string())); + } + let running = Arc::new(AtomicBool::new(true)); + let running_thread = running.clone(); + let thread = thread::Builder::new() + .name("peerspeak-cpal-capture".to_string()) + .spawn(move || { + if let Err(e) = run_capture(tx, target_node, running_thread) { + crate::log_msg(&format!("cpal capture error: {e}")); + } + }) + .map_err(|e| AudioError::Init(e.to_string()))?; + *guard = Some(StreamWorker { running, thread }); Ok(()) } fn start_playback( &self, rx: Receiver>, - _target_node: Option, - _ring_fill: Arc, + target_node: Option, + ring_fill: Arc, ) -> Result<(), AudioError> { - // Drain and discard incoming audio on a detached thread so the mixer's - // producer never blocks or sees a closed channel. This keeps the rest of - // the pipeline running normally while output is silent. - std::thread::spawn(move || while rx.recv().is_ok() {}); - crate::log_msg("CpalBackend::start_playback: not yet implemented (Phase 1) — discarding output"); + let mut guard = self.playback.lock().unwrap(); + if guard.is_some() { + return Err(AudioError::Stream("Playback already started".to_string())); + } + let running = Arc::new(AtomicBool::new(true)); + let running_thread = running.clone(); + let thread = thread::Builder::new() + .name("peerspeak-cpal-playback".to_string()) + .spawn(move || { + if let Err(e) = run_playback(rx, target_node, ring_fill, running_thread) { + crate::log_msg(&format!("cpal playback error: {e}")); + } + }) + .map_err(|e| AudioError::Init(e.to_string()))?; + *guard = Some(StreamWorker { running, thread }); Ok(()) } fn stop(&self) -> Result<(), AudioError> { + for slot in [&self.capture, &self.playback] { + if let Some(worker) = slot.lock().unwrap().take() { + worker.running.store(false, Ordering::Relaxed); + let _ = worker.thread.join(); + } + } Ok(()) } } + +// --------------------------------------------------------------------------- +// Device / config selection +// --------------------------------------------------------------------------- + +/// Resolve a device (by `target` name, else the system default) and a stream +/// config running natively at [`SAMPLE_RATE`]. +/// +/// For output we require [`PLAYBACK_CHANNELS`] (stereo) so the interleaved ring +/// maps 1:1 to the device buffer; for input we prefer mono but accept any channel +/// count and downmix. A device with no 48 kHz config is a hard error (no +/// resampling yet — see module docs). +fn resolve( + output: bool, + target: Option, +) -> Result<(Device, StreamConfig, SampleFormat), AudioError> { + let host = cpal::default_host(); + + let default = || { + if output { + host.default_output_device() + } else { + host.default_input_device() + } + }; + let device = match target { + Some(name) => find_device_by_name(&host, output, &name).or_else(default), + None => default(), + } + .ok_or_else(|| AudioError::Device("no audio device available".to_string()))?; + + let supported = choose_config(&device, output)?; + let sample_format = supported.sample_format(); + let config = supported.config(); + Ok((device, config, sample_format)) +} + +fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option { + let devices = if output { + host.output_devices().ok()? + } else { + host.input_devices().ok()? + }; + devices.into_iter().find(|d| d.name().is_ok_and(|n| n == name)) +} + +/// Pick a supported config at exactly [`SAMPLE_RATE`]. Output must be stereo; +/// input prefers mono, then any channel count (downmixed later). +fn choose_config( + device: &Device, + output: bool, +) -> Result { + let ranges: Vec = if output { + device + .supported_output_configs() + .map_err(|e| AudioError::Device(e.to_string()))? + .collect() + } else { + device + .supported_input_configs() + .map_err(|e| AudioError::Device(e.to_string()))? + .collect() + }; + + // A range covers a sample-rate span and a fixed channel count. + let supports_48k = |r: &cpal::SupportedStreamConfigRange| { + r.min_sample_rate().0 <= SAMPLE_RATE && SAMPLE_RATE <= r.max_sample_rate().0 + }; + let pick = |channels: Option| { + ranges + .iter() + .find(|r| supports_48k(r) && channels.is_none_or(|c| r.channels() == c)) + .cloned() + }; + + let chosen = if output { + pick(Some(PLAYBACK_CHANNELS as u16)) + } else { + pick(Some(1)).or_else(|| pick(None)) + }; + + chosen + .map(|r| r.with_sample_rate(SampleRate(SAMPLE_RATE))) + .ok_or_else(|| { + AudioError::Device(format!( + "device '{}' has no {SAMPLE_RATE} Hz {} config; resampling not yet implemented (Phase 1.1)", + device.name().unwrap_or_else(|_| "".to_string()), + if output { "stereo output" } else { "input" }, + )) + }) +} + +// --------------------------------------------------------------------------- +// Capture +// --------------------------------------------------------------------------- + +fn run_capture( + tx: Sender>, + target: Option, + running: Arc, +) -> Result<(), AudioError> { + let (device, config, sample_format) = resolve(false, target)?; + let channels = config.channels as usize; + + let stream = match sample_format { + SampleFormat::F32 => build_input::(&device, &config, tx, channels), + SampleFormat::I16 => build_input::(&device, &config, tx, channels), + SampleFormat::U16 => build_input::(&device, &config, tx, channels), + other => Err(AudioError::Stream(format!( + "unsupported capture sample format: {other:?}" + ))), + }?; + + stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + + // The RT callback does the work; this thread just keeps `stream` alive until + // `stop()` flips the flag, at which point the stream is dropped (= stopped). + while running.load(Ordering::Relaxed) { + thread::sleep(WORKER_POLL); + } + Ok(()) +} + +fn build_input( + device: &Device, + config: &StreamConfig, + tx: Sender>, + channels: usize, +) -> Result +where + T: SizedSample + Send + 'static, + i16: FromSample, +{ + let mut acc = FrameAccumulator::new(CAPTURE_FRAME); + let err_fn = |e| crate::log_msg(&format!("cpal capture stream error: {e}")); + device + .build_input_stream::( + config, + move |data: &[T], _| { + for frame in data.chunks_exact(channels) { + let mono = downmix_to_mono(frame); + if let Some(full) = acc.push(mono) { + // Consumer gone (call ended) → stop feeding; the owning + // thread will drop the stream on `stop()`. + if tx.send(full).is_err() { + return; + } + } + } + }, + err_fn, + None, + ) + .map_err(|e| AudioError::Stream(e.to_string())) +} + +/// Average a device frame's channels down to a single mono i16. For a 1-channel +/// device this is just the converted sample. +fn downmix_to_mono(frame: &[T]) -> i16 +where + T: Copy, + i16: FromSample, +{ + if frame.is_empty() { + return 0; + } + let sum: i32 = frame.iter().map(|&s| i16::from_sample(s) as i32).sum(); + (sum / frame.len() as i32) as i16 +} + +/// Accumulates mono samples into fixed-size [`CAPTURE_FRAME`] frames. Pulled out +/// of the RT callback so the framing is unit-testable. +struct FrameAccumulator { + buf: Vec, + frame_len: usize, +} + +impl FrameAccumulator { + fn new(frame_len: usize) -> Self { + Self { + buf: Vec::with_capacity(frame_len), + frame_len, + } + } + + /// Push one sample; returns a completed frame when the buffer fills. + fn push(&mut self, sample: i16) -> Option> { + self.buf.push(sample); + if self.buf.len() == self.frame_len { + Some(std::mem::replace( + &mut self.buf, + Vec::with_capacity(self.frame_len), + )) + } else { + None + } + } +} + +// --------------------------------------------------------------------------- +// Playback +// --------------------------------------------------------------------------- + +fn run_playback( + rx: Receiver>, + target: Option, + ring_fill: Arc, + running: Arc, +) -> Result<(), AudioError> { + let (device, config, sample_format) = resolve(true, target)?; + + let rb = HeapRb::::new(RING_CAPACITY); + let (mut producer, consumer) = rb.split(); + + // Prefill to the steady-state depth so playout starts at target. `ring_fill` + // is an EXACT occupancy counter maintained by deltas (worker fetch_add on + // push, RT callback fetch_sub on pop) — not ringbuf's cached `occupied_len`, + // which is stale across the split halves and would lie high and starve the + // ring. See pipewire_impl.rs for the full rationale. + for _ in 0..PLAYBACK_TARGET_SAMPLES { + let _ = producer.try_push(0); + } + ring_fill.store(PLAYBACK_TARGET_SAMPLES, Ordering::Relaxed); + + // Diagnostics (mirrors the PipeWire backend's playout-health line). + let underrun = Arc::new(AtomicU64::new(0)); + let dropped = Arc::new(AtomicU64::new(0)); + + let stream = match sample_format { + SampleFormat::F32 => { + build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) + } + SampleFormat::I16 => { + build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) + } + SampleFormat::U16 => { + build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) + } + other => Err(AudioError::Stream(format!( + "unsupported playback sample format: {other:?}" + ))), + }?; + + stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + + let logger = spawn_health_logger( + running.clone(), + ring_fill.clone(), + underrun.clone(), + dropped.clone(), + ); + + // Feed the ring from the network mixer until `stop()` flips `running` or the + // sender disconnects (call ended). Clock-paced production keeps the ring near + // target, so the drop path below should never fire in steady state. + drain_loop(&rx, &running, |frame| { + if ring_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY { + dropped.fetch_add(1, Ordering::Relaxed); + return; + } + for &sample in &frame { + let _ = producer.try_push(sample); + } + ring_fill.fetch_add(frame.len(), Ordering::Relaxed); + }); + + // We're shutting down (either stop() or disconnect). Ensure the logger sees it + // even on the disconnect path, then drop the stream. + running.store(false, Ordering::Relaxed); + let _ = logger.join(); + drop(stream); + Ok(()) +} + +fn build_output( + device: &Device, + config: &StreamConfig, + mut consumer: C, + ring_fill: Arc, + underrun: Arc, +) -> Result +where + T: SizedSample + FromSample + Send + 'static, + C: Consumer + Send + 'static, +{ + let err_fn = |e| crate::log_msg(&format!("cpal playback stream error: {e}")); + device + .build_output_stream::( + config, + move |data: &mut [T], _| { + let (popped, starved) = fill_output(&mut consumer, data); + if starved > 0 { + underrun.fetch_add(starved, Ordering::Relaxed); + } + if popped > 0 { + // Decrement the exact occupancy by what we actually pulled + // (underruns removed nothing) so the mixer paces against the + // true ring depth. + ring_fill.fetch_sub(popped, Ordering::Relaxed); + } + }, + err_fn, + None, + ) + .map_err(|e| AudioError::Stream(e.to_string())) +} + +/// Drain the ring into the device buffer, substituting silence on underrun. +/// Returns `(samples_popped, samples_starved)`. RT-safe (wait-free `try_pop`). +fn fill_output(consumer: &mut C, out: &mut [T]) -> (usize, u64) +where + T: Sample + FromSample, + C: Consumer, +{ + let mut popped = 0usize; + let mut starved = 0u64; + for slot in out.iter_mut() { + match consumer.try_pop() { + Some(v) => { + *slot = T::from_sample(v); + popped += 1; + } + None => { + *slot = T::from_sample(0i16); + starved += 1; + } + } + } + (popped, starved) +} + +/// Once-per-second playout-health line (mirrors the PipeWire backend). Quiet +/// unless a second actually glitched, or `PEERSPEAK_AUDIO_VERBOSE` is set. +fn spawn_health_logger( + running: Arc, + ring_fill: Arc, + underrun: Arc, + dropped: Arc, +) -> JoinHandle<()> { + let verbose = std::env::var_os("PEERSPEAK_AUDIO_VERBOSE").is_some(); + thread::spawn(move || { + let (mut last_u, mut last_d) = (0u64, 0u64); + while running.load(Ordering::Relaxed) { + thread::sleep(Duration::from_secs(1)); + let u = underrun.load(Ordering::Relaxed); + let d = dropped.load(Ordering::Relaxed); + let fill = ring_fill.load(Ordering::Relaxed); + let (du, dd) = (u - last_u, d - last_d); + last_u = u; + last_d = d; + if verbose || du > 0 || dd > 0 { + crate::log_msg(&format!( + "playout-health: fill={fill} samples (~{}ms) | underrun +{du} samples/s (total {u}) | dropped +{dd} frames/s (total {d})", + fill / (48 * PLAYBACK_CHANNELS), + )); + } + } + }) +} + +/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender +/// disconnects. The timed receive re-checks `running` at least every +/// [`WORKER_POLL`], so `stop()` can join the worker promptly instead of hanging +/// on a parked blocking `recv()` (same A7 fix as the PipeWire backend). Pure +/// w.r.t. its inputs, so it's unit-testable. +fn drain_loop(rx: &Receiver>, running: &AtomicBool, mut on_frame: impl FnMut(Vec)) { + while running.load(Ordering::Relaxed) { + match rx.recv_timeout(WORKER_POLL) { + Ok(frame) => on_frame(frame), + Err(RecvTimeoutError::Timeout) => continue, + Err(RecvTimeoutError::Disconnected) => return, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + + #[test] + fn downmix_averages_channels() { + assert_eq!(downmix_to_mono::(&[100, 100]), 100); + assert_eq!(downmix_to_mono::(&[100, -100]), 0); + assert_eq!(downmix_to_mono::(&[50]), 50); + assert_eq!(downmix_to_mono::(&[]), 0); + // 4-channel average rounds toward zero (integer division). + assert_eq!(downmix_to_mono::(&[10, 20, 30, 41]), 25); + } + + #[test] + fn frame_accumulator_emits_full_frames() { + let mut acc = FrameAccumulator::new(3); + assert_eq!(acc.push(1), None); + assert_eq!(acc.push(2), None); + assert_eq!(acc.push(3), Some(vec![1, 2, 3])); + // Resets for the next frame. + assert_eq!(acc.push(4), None); + assert_eq!(acc.push(5), None); + assert_eq!(acc.push(6), Some(vec![4, 5, 6])); + } + + #[test] + fn fill_output_pops_then_substitutes_silence() { + let rb = HeapRb::::new(8); + let (mut prod, mut cons) = rb.split(); + for v in [1, 2, 3] { + prod.try_push(v).unwrap(); + } + let mut out = [0i16; 5]; + let (popped, starved) = fill_output(&mut cons, &mut out); + assert_eq!(popped, 3); + assert_eq!(starved, 2); + assert_eq!(out, [1, 2, 3, 0, 0]); + } + + #[test] + fn drain_loop_exits_when_running_flips_even_with_sender_alive() { + let (tx, rx) = mpsc::channel::>(); + let running = Arc::new(AtomicBool::new(true)); + let r2 = running.clone(); + let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {})); + thread::sleep(Duration::from_millis(50)); + running.store(false, Ordering::Relaxed); + thread::sleep(WORKER_POLL + Duration::from_millis(150)); + assert!( + h.is_finished(), + "drain_loop must exit after running=false even while the sender is alive" + ); + drop(tx); + h.join().unwrap(); + } + + #[test] + fn drain_loop_returns_on_disconnect() { + let (tx, rx) = mpsc::channel::>(); + let running = Arc::new(AtomicBool::new(true)); + drop(tx); + drain_loop(&rx, &running, |_| panic!("no frame should arrive")); + } + + #[test] + fn drain_loop_delivers_frames() { + let (tx, rx) = mpsc::channel::>(); + let running = Arc::new(AtomicBool::new(true)); + let r2 = running.clone(); + let got = Arc::new(Mutex::new(Vec::new())); + let g2 = got.clone(); + let h = thread::spawn(move || drain_loop(&rx, &r2, |f| g2.lock().unwrap().push(f))); + tx.send(vec![1, 2, 3]).unwrap(); + tx.send(vec![4, 5]).unwrap(); + thread::sleep(Duration::from_millis(50)); + running.store(false, Ordering::Relaxed); + drop(tx); + h.join().unwrap(); + assert_eq!(*got.lock().unwrap(), vec![vec![1, 2, 3], vec![4, 5]]); + } +}