diff --git a/src-tauri/src/commands/audio.rs b/src-tauri/src/commands/audio.rs index 9f5d73f..e663822 100644 --- a/src-tauri/src/commands/audio.rs +++ b/src-tauri/src/commands/audio.rs @@ -1,9 +1,12 @@ use std::{ + collections::VecDeque, net::UdpSocket, sync::{ atomic::{AtomicBool, Ordering}, Arc, Mutex, }, + thread, + time::Duration, }; use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; @@ -11,6 +14,76 @@ use tauri::State; use crate::commands::config::ConfigState; +// ============================================================ +// AUDIO CONSTANTS +// ============================================================ + +/// Network audio format. +/// +/// Everything sent over UDP is: +/// 44100 Hz +/// mono +/// signed i16 +/// big endian +/// +/// 20 ms @ 44100 Hz = 882 samples. +/// 882 * 2 = 1764 bytes. +const NETWORK_SAMPLE_RATE: u32 = 44_100; +const PACKET_DURATION_MS: usize = 20; +const NETWORK_PACKET_SAMPLES: usize = NETWORK_SAMPLE_RATE as usize * PACKET_DURATION_MS / 1000; + +/// Amount of audio that must be queued before playback begins. +const PREBUFFER_MS: usize = 40; + +/// If the playback queue gets below this amount, we allow it to +/// continue normally but the callback will output silence if it +/// actually runs dry. +const LOW_WATERMARK_MS: usize = 20; + +/// Maximum amount of queued audio. +/// +/// If this is exceeded, OLD audio is discarded. This is intentional: +/// for voice chat, dropping old audio is much better than accumulating +/// hundreds of milliseconds/seconds of latency. +const MAX_BUFFER_MS: usize = 100; + +/// Maximum UDP packet size we expect. +const MAX_UDP_PACKET_SIZE: usize = 4096; + +// ============================================================ +// STATE +// ============================================================ + +pub struct VoiceState { + pub session: Mutex>, +} + +pub struct VoiceSession { + pub input_stream: cpal::Stream, + + // Wrapped so the watcher thread can replace the stream after + // the output device changes configuration. + pub output_stream: Arc>, + + pub socket: Arc, + + pub pin: u64, + + pub shutdown: Arc, +} + +impl Default for VoiceState { + fn default() -> Self { + Self { + session: Mutex::new(None), + } + } +} + +// ============================================================ +// DEVICE ENUMERATION +// ============================================================ + #[tauri::command] pub fn list_input_devices() -> Result, String> { let host = cpal::default_host(); @@ -20,7 +93,7 @@ pub fn list_input_devices() -> Result, String> { .map_err(|e| format!("Failed to enumerate input devices: {e}"))?; Ok(devices - .filter_map(|d| Some(d.description().ok()?.name().to_string())) + .filter_map(|device| Some(device.description().ok()?.name().to_string())) .collect()) } @@ -33,42 +106,32 @@ pub fn list_output_devices() -> Result, String> { .map_err(|e| format!("Failed to enumerate output devices: {e}"))?; Ok(devices - .filter_map(|d| Some(d.description().ok()?.name().to_string())) + .filter_map(|device| Some(device.description().ok()?.name().to_string())) .collect()) } -pub struct VoiceState { - pub session: Mutex>, -} - -pub struct VoiceSession { - pub input_stream: cpal::Stream, - // Wrapped so the watcher thread can swap in a freshly rebuilt stream. - pub output_stream: Arc>, - pub socket: Arc, - pub pin: u64, - // Set to true when the watcher thread should stop (on disconnect). - pub shutdown: Arc, -} - -impl Default for VoiceState { - fn default() -> Self { - Self { - session: Mutex::new(None), - } - } -} +// ============================================================ +// DISCONNECT +// ============================================================ #[tauri::command] pub fn disconnect_from_vc(voice_state: State) -> Result<(), String> { if let Some(session) = voice_state.session.lock().unwrap().take() { - // Tell the watcher thread to stop before dropping the streams. session.shutdown.store(true, Ordering::SeqCst); + + // Streams are dropped here. + // + // The UDP receiver has a 100ms read timeout, so it will notice + // shutdown shortly instead of remaining blocked forever. } - // Dropping the session stops both cpal streams. + Ok(()) } +// ============================================================ +// CONNECT +// ============================================================ + #[tauri::command] pub fn connect_to_vc( hostname: String, @@ -80,68 +143,134 @@ pub fn connect_to_vc( let config = config_state.0.lock().unwrap().clone(); - let socket = UdpSocket::bind("0.0.0.0:0").map_err(|e| e.to_string())?; - socket.connect(&hostname).map_err(|e| e.to_string())?; + // ======================================================== + // UDP SOCKET + // ======================================================== + + let socket = + UdpSocket::bind("0.0.0.0:0").map_err(|e| format!("Failed to bind UDP socket: {e}"))?; + + socket + .connect(&hostname) + .map_err(|e| format!("Failed to connect UDP socket: {e}"))?; + + // IMPORTANT: + // + // Without a timeout, recv() can remain blocked forever and the + // receiver thread can survive after disconnect. + socket + .set_read_timeout(Some(Duration::from_millis(100))) + .map_err(|e| format!("Failed to configure UDP timeout: {e}"))?; + let socket = Arc::new(socket); - // Send the pin as the very first packet — authenticates this UDP session + // PIN is always the first packet. socket .send(&pin.to_be_bytes()) .map_err(|e| format!("Failed to send pincode: {e}"))?; + // ======================================================== + // AUDIO DEVICES + // ======================================================== + let host = cpal::default_host(); let input_device = match &config.input_device_name { Some(name) => host .input_devices() .map_err(|e| e.to_string())? - .find(|d| { - d.description() + .find(|device| { + device + .description() .ok() - .map(|d| d.name() == name) + .map(|description| description.name() == name) .unwrap_or(false) }) - .ok_or("Input device not found")?, + .ok_or_else(|| "Input device not found".to_string())?, + None => host .default_input_device() - .ok_or("No default input device")?, + .ok_or_else(|| "No default input device".to_string())?, }; let output_device = match &config.output_device_name { Some(name) => host .output_devices() .map_err(|e| e.to_string())? - .find(|d| { - d.description() + .find(|device| { + device + .description() .ok() - .map(|d| d.name() == name) + .map(|description| description.name() == name) .unwrap_or(false) }) - .ok_or("Output device not found")?, + .ok_or_else(|| "Output device not found".to_string())?, + None => host .default_output_device() - .ok_or("No default output device")?, + .ok_or_else(|| "No default output device".to_string())?, }; - // ---------- INPUT (mic -> UDP) ---------- + // ======================================================== + // SHARED PLAYBACK STATE + // ======================================================== + + // + // IMPORTANT: + // + // This queue contains samples already converted to the + // CURRENT output device's channel layout. + // + // Therefore: + // + // stereo device: + // L R L R L R ... + // + // mono device: + // M M M M ... + // + let playback_buffer = Arc::new(Mutex::new(VecDeque::::new())); + + let needs_output_rebuild = Arc::new(AtomicBool::new(false)); + + let shutdown = Arc::new(AtomicBool::new(false)); + + let playback_started = Arc::new(AtomicBool::new(false)); + + // ======================================================== + // INPUT: MICROPHONE -> UDP + // ======================================================== let input_socket = socket.clone(); + let mut input_config: cpal::StreamConfig = input_device .default_input_config() - .map_err(|e| e.to_string())? + .map_err(|e| format!("Failed to get input config: {e}"))? .into(); + // We explicitly capture mono. input_config.channels = 1; + let input_sample_rate = input_config.sample_rate; + eprintln!( + "[vc] input: {} Hz -> {} Hz", + input_sample_rate, NETWORK_SAMPLE_RATE + ); + let mut input_resampler = resampler::ResamplerFft::new( 1, - input_sample_rate.try_into().map_err(|e| format!("{e:?}"))?, + input_sample_rate + .try_into() + .map_err(|e| format!("Invalid input sample rate: {e:?}"))?, resampler::SampleRate::Hz44100, ); let mut input_buffer = Vec::::new(); + // Resampled samples waiting to form a network packet. + let mut packet_buffer = Vec::::with_capacity(NETWORK_PACKET_SAMPLES * 2); + let input_stream = input_device .build_input_stream( input_config, @@ -153,190 +282,390 @@ pub fn connect_to_vc( while input_buffer.len() >= frame_size { let input: Vec = input_buffer.drain(..frame_size).collect(); - let mut output = vec![0.0f32; input_resampler.chunk_size_output()]; + let output_size = input_resampler.chunk_size_output(); + + let mut output = vec![0.0f32; output_size]; if let Err(e) = input_resampler.resample(&input, &mut output) { eprintln!("[vc] failed to resample input: {e}"); continue; } - // f32 -> i16 PCM - let mut packet = Vec::with_capacity(output.len() * 2); + packet_buffer.extend(output.into_iter().map(|sample| sample.clamp(-1.0, 1.0))); - for sample in output { - let sample = sample.clamp(-1.0, 1.0); - let pcm = (sample * i16::MAX as f32) as i16; + // ------------------------------------------------ + // Form exact 20ms network packets. + // ------------------------------------------------ - packet.extend_from_slice(&pcm.to_be_bytes()); - } + while packet_buffer.len() >= NETWORK_PACKET_SAMPLES { + let packet_samples: Vec = + packet_buffer.drain(..NETWORK_PACKET_SAMPLES).collect(); - eprintln!( - "[vc] sending audio: input={} samples, packet={} bytes", - frame_size, - packet.len() - ); + let mut packet = Vec::with_capacity(NETWORK_PACKET_SAMPLES * 2); - match input_socket.send(&packet) { - Ok(n) => eprintln!("[vc] UDP sent {n} bytes"), - Err(e) => eprintln!("[vc] UDP send failed: {e}"), + for sample in packet_samples { + let pcm = (sample * i16::MAX as f32) as i16; + + packet.extend_from_slice(&pcm.to_be_bytes()); + } + + match input_socket.send(&packet) { + Ok(_) => {} + + Err(e) => { + eprintln!("[vc] UDP send failed: {e}"); + } + } } } }, - |err| eprintln!("[vc] input stream error: {err}"), + |err| { + eprintln!("[vc] input stream error: {err}"); + }, None, ) .map_err(|e| e.to_string())?; - // ---------- OUTPUT (UDP -> speakers), rebuildable on device change ---------- + // ======================================================== + // OUTPUT CONFIG + // ======================================================== - let playback_buffer = Arc::new(Mutex::new(Vec::::new())); - let needs_output_rebuild = Arc::new(AtomicBool::new(false)); - let shutdown = Arc::new(AtomicBool::new(false)); - - // Output config is tracked in a shared cell so the UDP-receiving thread - // (below) always resamples toward whatever the *current* live stream expects, - // even after a rebuild changes the device's rate. let output_config = build_output_config(&output_device)?; + + eprintln!( + "[vc] output: {} Hz / {} channels", + output_config.sample_rate, output_config.channels + ); + let output_config_cell = Arc::new(Mutex::new(output_config.clone())); + // ======================================================== + // OUTPUT STREAM + // ======================================================== + let initial_output_stream = build_output_stream( &output_device, &output_config, playback_buffer.clone(), needs_output_rebuild.clone(), + playback_started.clone(), )?; - initial_output_stream.play().map_err(|e| e.to_string())?; + + initial_output_stream + .play() + .map_err(|e| format!("Failed to start output stream: {e}"))?; let output_stream = Arc::new(Mutex::new(initial_output_stream)); - // Watcher thread: rebuilds the output stream whenever the device signals - // its config has changed (e.g. "Device sample rate changed"), since cpal - // streams can't be reconfigured in place — only rebuilt from scratch. + // ======================================================== + // OUTPUT DEVICE WATCHER + // ======================================================== + { let output_device = output_device.clone(); + let needs_output_rebuild = needs_output_rebuild.clone(); + let output_stream = output_stream.clone(); + let output_config_cell = output_config_cell.clone(); + let shutdown = shutdown.clone(); + let playback_buffer = playback_buffer.clone(); - std::thread::spawn(move || loop { - if shutdown.load(Ordering::SeqCst) { - return; - } + let playback_started = playback_started.clone(); - std::thread::sleep(std::time::Duration::from_millis(200)); + thread::spawn(move || { + while !shutdown.load(Ordering::SeqCst) { + thread::sleep(Duration::from_millis(100)); - if needs_output_rebuild.swap(false, Ordering::SeqCst) { - match build_output_config(&output_device).and_then(|new_config| { + if shutdown.load(Ordering::SeqCst) { + return; + } + + if !needs_output_rebuild.swap(false, Ordering::SeqCst) { + continue; + } + + eprintln!("[vc] rebuilding output stream..."); + + let result = build_output_config(&output_device).and_then(|new_config| { build_output_stream( &output_device, &new_config, playback_buffer.clone(), needs_output_rebuild.clone(), + playback_started.clone(), ) .map(|stream| (new_config, stream)) - }) { + }); + + match result { Ok((new_config, new_stream)) => { if let Err(e) = new_stream.play() { - eprintln!("[vc] failed to start rebuilt output stream: {e}"); + eprintln!( + "[vc] failed to start rebuilt \ + output stream: {e}" + ); + + needs_output_rebuild.store(true, Ordering::SeqCst); + continue; } - *output_config_cell.lock().unwrap() = new_config; + + // Clear stale audio because it was generated + // for the old output timing/channel layout. + playback_buffer.lock().unwrap().clear(); + + playback_started.store(false, Ordering::SeqCst); + + *output_config_cell.lock().unwrap() = new_config.clone(); + *output_stream.lock().unwrap() = new_stream; - eprintln!("[vc] output stream rebuilt after device change"); + + eprintln!( + "[vc] output rebuilt: {} Hz / {} channels", + new_config.sample_rate, new_config.channels + ); + } + + Err(e) => { + eprintln!("[vc] failed to rebuild output: {e}"); + + needs_output_rebuild.store(true, Ordering::SeqCst); } - Err(e) => eprintln!("[vc] failed to rebuild output stream: {e}"), } } }); } - // ---------- UDP receive thread: incoming audio -> resample -> playback channel ---------- + // ======================================================== + // UDP RECEIVE -> RESAMPLE -> PLAYBACK QUEUE + // ======================================================== - let recv_socket = socket.clone(); { + let recv_socket = socket.clone(); + let output_config_cell = output_config_cell.clone(); + + let playback_buffer = playback_buffer.clone(); + + let playback_started = playback_started.clone(); + let shutdown = shutdown.clone(); - std::thread::spawn(move || { - let mut buf = [0u8; 4096]; - let mut output_buffer = Vec::::new(); + thread::spawn(move || { + let mut buf = [0u8; MAX_UDP_PACKET_SIZE]; - let mut output_resampler = resampler::ResamplerFft::new( - 1, - resampler::SampleRate::Hz44100, - output_config_cell - .lock() - .unwrap() - .sample_rate - .try_into() - .unwrap(), - ); + let initial_config = output_config_cell.lock().unwrap().clone(); - let mut last_rate = output_config_cell.lock().unwrap().sample_rate; + let mut last_sample_rate = initial_config.sample_rate; + + let mut output_resampler = match create_output_resampler(initial_config.sample_rate) { + Ok(resampler) => resampler, + + Err(e) => { + eprintln!( + "[vc] failed to create output \ + resampler: {e}" + ); + + return; + } + }; + + // Audio waiting to be fed into the resampler. + let mut resample_input = Vec::::new(); loop { if shutdown.load(Ordering::SeqCst) { return; } - let Ok(len) = recv_socket.recv(&mut buf) else { - continue; - }; + // ---------------------------------------------------- + // Receive UDP packet. + // ---------------------------------------------------- - // UDP contains BIG-ENDIAN i16 PCM. - let pcm = buf[..len] - .chunks_exact(2) - .map(|b| { - let sample = i16::from_be_bytes([b[0], b[1]]); - sample as f32 / i16::MAX as f32 - }) - .collect::>(); + let len = match recv_socket.recv(&mut buf) { + Ok(len) => len, - output_buffer.extend_from_slice(&pcm); - - let current_config = output_config_cell.lock().unwrap().clone(); - - if current_config.sample_rate != last_rate { - output_resampler = resampler::ResamplerFft::new( - 1, - resampler::SampleRate::Hz44100, - match current_config.sample_rate.try_into() { - Ok(rate) => rate, - Err(_) => { - eprintln!("[vc] unsupported output sample rate"); - last_rate = current_config.sample_rate; - continue; - } - }, - ); - - last_rate = current_config.sample_rate; - } - - let frame_size = output_resampler.chunk_size_input(); - - while output_buffer.len() >= frame_size { - let input: Vec = output_buffer.drain(..frame_size).collect(); - - let mut pcm_out = vec![0.0f32; output_resampler.chunk_size_output()]; - - if let Err(e) = output_resampler.resample(&input, &mut pcm_out) { - eprintln!("[vc] failed to resample output: {e}"); + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => + { continue; } - // Mono network audio -> device channel layout. - let output = mono_to_output_channels(&pcm_out, current_config.channels); + Err(e) => { + if !shutdown.load(Ordering::SeqCst) { + eprintln!("[vc] UDP receive failed: {e}"); + } - playback_buffer.lock().unwrap().extend(output); + continue; + } + }; + + // ---------------------------------------------------- + // Ignore malformed packets. + // + // Every audio packet must contain complete i16 + // samples. + // ---------------------------------------------------- + + if len < 2 { + continue; + } + + let usable_len = len - (len % 2); + + // ---------------------------------------------------- + // BIG-ENDIAN i16 -> f32 + // ---------------------------------------------------- + + for chunk in buf[..usable_len].chunks_exact(2) { + let pcm = i16::from_be_bytes([chunk[0], chunk[1]]); + + resample_input.push(pcm as f32 / i16::MAX as f32); + } + + // ---------------------------------------------------- + // Check current output device configuration. + // ---------------------------------------------------- + + let current_config = output_config_cell.lock().unwrap().clone(); + + // ---------------------------------------------------- + // Output sample rate changed. + // + // Recreate the resampler and discard samples from the + // old timing domain. + // ---------------------------------------------------- + + if current_config.sample_rate != last_sample_rate { + eprintln!( + "[vc] output rate changed: {} -> {}", + last_sample_rate, current_config.sample_rate + ); + + match create_output_resampler(current_config.sample_rate) { + Ok(new_resampler) => { + output_resampler = new_resampler; + + last_sample_rate = current_config.sample_rate; + + resample_input.clear(); + + playback_buffer.lock().unwrap().clear(); + + playback_started.store(false, Ordering::SeqCst); + } + + Err(e) => { + eprintln!( + "[vc] unsupported output \ + sample rate {}: {e}", + current_config.sample_rate + ); + + resample_input.clear(); + + continue; + } + } + } + + // ---------------------------------------------------- + // Resample incoming 44.1kHz mono audio into the + // output device's sample rate. + // ---------------------------------------------------- + + let frame_size = output_resampler.chunk_size_input(); + + while resample_input.len() >= frame_size { + let input: Vec = resample_input.drain(..frame_size).collect(); + + let output_size = output_resampler.chunk_size_output(); + + let mut resampled = vec![0.0f32; output_size]; + + if let Err(e) = output_resampler.resample(&input, &mut resampled) { + eprintln!( + "[vc] failed to resample \ + output: {e}" + ); + + continue; + } + + // ------------------------------------------------ + // Mono -> device channels. + // ------------------------------------------------ + + let output = mono_to_output_channels(&resampled, current_config.channels); + + // ------------------------------------------------ + // Push into bounded playback queue. + // ------------------------------------------------ + + let mut queue = playback_buffer.lock().unwrap(); + + queue.extend(output); + + let channels = current_config.channels.max(1) as usize; + + let max_frames = current_config.sample_rate as usize * MAX_BUFFER_MS / 1000; + + let max_samples = max_frames * channels; + + // ------------------------------------------------ + // If we have accumulated too much audio, discard + // OLD audio. + // + // This is critical for voice chat latency. + // ------------------------------------------------ + + while queue.len() > max_samples { + queue.pop_front(); + } + + // ------------------------------------------------ + // Start playback only after we have a small + // amount of audio buffered. + // ------------------------------------------------ + + if !playback_started.load(Ordering::SeqCst) { + let prebuffer_frames = + current_config.sample_rate as usize * PREBUFFER_MS / 1000; + + let prebuffer_samples = prebuffer_frames * channels; + + if queue.len() >= prebuffer_samples { + playback_started.store(true, Ordering::SeqCst); + + eprintln!( + "[vc] playback started \ + with ~{}ms buffered", + PREBUFFER_MS + ); + } + } } } }); } - input_stream.play().map_err(|e| e.to_string())?; + // ======================================================== + // START INPUT + // ======================================================== + + input_stream + .play() + .map_err(|e| format!("Failed to start input stream: {e}"))?; + + // ======================================================== + // STORE SESSION + // ======================================================== *voice_state.session.lock().unwrap() = Some(VoiceSession { input_stream, @@ -349,42 +678,91 @@ pub fn connect_to_vc( Ok(()) } +// ============================================================ +// OUTPUT CONFIG +// ============================================================ + fn build_output_config(output_device: &cpal::Device) -> Result { - Ok(output_device + output_device .default_output_config() - .map_err(|e| e.to_string())? - .into()) + .map_err(|e| e.to_string()) + .map(Into::into) } +// ============================================================ +// OUTPUT RESAMPLER +// ============================================================ + +fn create_output_resampler( + output_sample_rate: cpal::SampleRate, +) -> Result { + let output_rate = output_sample_rate + .try_into() + .map_err(|e| format!("Invalid output sample rate: {e:?}"))?; + + Ok(resampler::ResamplerFft::new( + 1, + resampler::SampleRate::Hz44100, + output_rate, + )) +} + +// ============================================================ +// BUILD OUTPUT STREAM +// ============================================================ + fn build_output_stream( output_device: &cpal::Device, output_config: &cpal::StreamConfig, - playback_buffer: Arc>>, + playback_buffer: Arc>>, needs_rebuild: Arc, + playback_started: Arc, ) -> Result { output_device .build_output_stream( output_config.clone(), + // ==================================================== + // CPAL OUTPUT CALLBACK + // ==================================================== move |data: &mut [f32], _| { - let mut buffer = playback_buffer.lock().unwrap(); + let started = playback_started.load(Ordering::Acquire); - for out in data.iter_mut() { - if buffer.is_empty() { - *out = 0.0; - } else { - *out = buffer.remove(0); - } + if !started { + // Do NOT consume audio before the prebuffer is + // ready. Just output silence. + data.fill(0.0); + return; + } + + let mut queue = playback_buffer.lock().unwrap(); + + // VecDeque::pop_front() is O(1). + // + // This is massively better than: + // + // Vec::remove(0) + // + // which shifts the entire vector every sample. + + for sample in data.iter_mut() { + *sample = queue.pop_front().unwrap_or(0.0); } }, + // ==================================================== + // OUTPUT ERROR CALLBACK + // ==================================================== { let needs_rebuild = needs_rebuild.clone(); move |err| { eprintln!("[vc] output stream error: {err}"); - let msg = err.to_string(); + let message = err.to_string(); - if msg.contains("sample rate changed") || msg.contains("DeviceNotAvailable") { + if message.contains("sample rate changed") + || message.contains("DeviceNotAvailable") + || message.contains("device not available") + { needs_rebuild.store(true, Ordering::SeqCst); } } @@ -394,26 +772,17 @@ fn build_output_stream( .map_err(|e| e.to_string()) } -fn stereo_to_mono(stereo_data: &[f32]) -> Vec { - let mut mono_data = Vec::with_capacity(stereo_data.len() / 2); - - for chunk in stereo_data.chunks_exact(2) { - let left = chunk[0]; - let right = chunk[1]; - let mono_sample = (left + right) / 2.0; - mono_data.push(mono_sample); - } - - mono_data -} +// ============================================================ +// MONO -> OUTPUT CHANNELS +// ============================================================ fn mono_to_output_channels(mono: &[f32], channels: u16) -> Vec { - if channels <= 1 { + let channels = channels.max(1) as usize; + + if channels == 1 { return mono.to_vec(); } - let channels = channels as usize; - let mut output = Vec::with_capacity(mono.len() * channels); for &sample in mono { @@ -424,3 +793,17 @@ fn mono_to_output_channels(mono: &[f32], channels: u16) -> Vec { output } + +// ============================================================ +// OPTIONAL UTILITY +// ============================================================ + +fn stereo_to_mono(stereo_data: &[f32]) -> Vec { + let mut mono = Vec::with_capacity(stereo_data.len() / 2); + + for chunk in stereo_data.chunks_exact(2) { + mono.push((chunk[0] + chunk[1]) * 0.5); + } + + mono +}