diff --git a/src-tauri/src/audio/commands.rs b/src-tauri/src/audio/commands.rs index 71c9e03f..62f7614b 100644 --- a/src-tauri/src/audio/commands.rs +++ b/src-tauri/src/audio/commands.rs @@ -1,6 +1,8 @@ -//! Tauri commands: play/pause/seek, gapless chain, radio, EQ, devices, normalization. +//! Tauri commands: audio_play / chain_preload / preload + the shared +//! spawn_progress_task helper. Transport (pause/resume/stop/seek), device, +//! radio, mix-mode and AutoEQ commands live in sibling modules. use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex, TryLockError}; +use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use futures_util::StreamExt; @@ -18,8 +20,8 @@ use super::ipc::{maybe_emit_normalization_state, NormalizationStatePayload}; use super::preview::preview_clear_for_new_main_playback; use super::state::{ChainedInfo, PreloadedTrack}; use super::stream::{ - radio_download_task, ranged_download_task, track_download_task, AudioStreamReader, - LocalFileSource, RangedHttpSource, RADIO_BUF_CAPACITY, RADIO_READ_TIMEOUT_SECS, + ranged_download_task, track_download_task, AudioStreamReader, + LocalFileSource, RangedHttpSource, RADIO_READ_TIMEOUT_SECS, TRACK_STREAM_MAX_BUF_CAPACITY, TRACK_STREAM_MIN_BUF_CAPACITY, LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES, }; @@ -1179,214 +1181,6 @@ pub(super) fn spawn_progress_task( }); } -#[tauri::command] -pub fn audio_pause(state: State<'_, AudioEngine>) { - let mut cur = state.current.lock().unwrap(); - if let Some(sink) = &cur.sink { - if !sink.is_paused() { - let pos = cur.position(); - sink.pause(); - cur.paused_at = Some(pos); - cur.play_started = None; - } - } - // Notify the download task so it can start measuring the hard-pause stall timer. - if let Some(rs) = state.radio_state.lock().unwrap().as_ref() { - rs.flags.is_paused.store(true, Ordering::Release); - } -} - -/// Resume playback. -/// -/// **Warm resume** (`is_hard_paused = false`): download task is still running, -/// buffer has buffered audio. `sink.play()` suffices. -/// -/// **Cold resume** (`is_hard_paused = true`): TCP was dropped. A fresh 4 MB -/// ring buffer is created, its consumer is sent to `AudioStreamReader` (which -/// swaps it in on the next `read()`), and a new download task is spawned. -#[tauri::command] -pub async fn audio_resume(state: State<'_, AudioEngine>, app: AppHandle) -> Result<(), String> { - // If a preview is running, cancel it first — otherwise sink.play() on the - // main sink would mix on top of the preview sink. - preview_clear_for_new_main_playback(&state, &app); - - // Detect radio hard-disconnect. - let reconnect_info = { - let guard = state.radio_state.lock().unwrap(); - guard - .as_ref() - .filter(|rs| rs.flags.is_hard_paused.load(Ordering::Acquire)) - .map(|rs| (rs.url.clone(), rs.gen, rs.flags.clone())) - }; - - if let Some((url, gen, flags)) = reconnect_info { - let rb = HeapRb::::new(RADIO_BUF_CAPACITY); - let (new_prod, new_cons) = rb.split(); - - // Send new consumer to AudioStreamReader (non-blocking; unbounded channel). - let ok = flags.new_cons_tx.lock().unwrap().send(new_cons).is_ok(); - - if ok { - let new_task = tokio::spawn(radio_download_task( - gen, - state.generation.clone(), - None, // task performs its own fresh GET - audio_http_client(&state), - url, - new_prod, - flags.clone(), - app, - )); - if let Some(rs) = state.radio_state.lock().unwrap().as_mut() { - let old = std::mem::replace(&mut rs.task, new_task); - old.abort(); // ensure any lingering old task is gone - rs.flags.is_hard_paused.store(false, Ordering::Release); - rs.flags.is_paused.store(false, Ordering::Release); - } - } else { - crate::app_eprintln!("[radio] resume: AudioStreamReader gone — skipping reconnect"); - } - } - - // Resume the rodio Sink (works for both warm and cold resume). - { - let mut cur = state.current.lock().unwrap(); - if let Some(sink) = &cur.sink { - if sink.is_paused() { - let pos = cur.paused_at.unwrap_or(cur.seek_offset); - sink.play(); - cur.seek_offset = pos; - cur.play_started = Some(Instant::now()); - cur.paused_at = None; - } - } - } - if let Some(rs) = state.radio_state.lock().unwrap().as_ref() { - rs.flags.is_paused.store(false, Ordering::Release); - } - Ok(()) -} - -#[tauri::command] -pub fn audio_stop(state: State<'_, AudioEngine>, app: AppHandle) { - preview_clear_for_new_main_playback(&state, &app); - state.generation.fetch_add(1, Ordering::SeqCst); - *state.current_playback_url.lock().unwrap() = None; - *state.current_analysis_track_id.lock().unwrap() = None; - *state.chained_info.lock().unwrap() = None; - // Keep `stream_completed_cache`: natural track end often calls `audio_stop` when the - // queue is exhausted; clearing here dropped the full ranged buffer and forced a - // re-download on replay. The slot is only consumed on `take`/overwrite for another URL. - // Drop RadioLiveState → triggers Drop → task.abort() → TCP released. - drop(state.radio_state.lock().unwrap().take()); - let mut cur = state.current.lock().unwrap(); - if let Some(sink) = cur.sink.take() { sink.stop(); } - cur.duration_secs = 0.0; - cur.seek_offset = 0.0; - cur.play_started = None; - cur.paused_at = None; -} - -#[tauri::command] -pub fn audio_seek(seconds: f64, state: State<'_, AudioEngine>) -> Result<(), String> { - const AUDIO_SEEK_TIMEOUT_MS: u64 = 700; - const AUDIO_SEEK_LOCK_TIMEOUT_MS: u64 = 40; - // Ghost-command guard: reject seeks within 500 ms of a gapless auto-advance. - { - let switch_ms = state.gapless_switch_at.load(Ordering::SeqCst); - if switch_ms > 0 { - let now_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as u64; - if now_ms.saturating_sub(switch_ms) < 500 { - return Ok(()); - } - } - } - - // Reject seek up-front for non-seekable streaming sources so the frontend's - // restart-fallback engages instead of rolling the dice on the format reader - // (which can consume the ring buffer to EOF for forward seeks → next song). - if !state.current_is_seekable.load(Ordering::SeqCst) { - crate::app_deprintln!("[seek] rejected → not-seekable source (legacy stream)"); - return Err("source is not seekable".into()); - } - crate::app_deprintln!("[seek] target={:.2}s", seconds); - - let lock_current_with_timeout = |timeout_ms: u64| { - let deadline = Instant::now() + Duration::from_millis(timeout_ms); - loop { - match state.current.try_lock() { - Ok(guard) => break Ok(guard), - Err(TryLockError::WouldBlock) => { - if Instant::now() >= deadline { - break Err("audio seek busy".to_string()); - } - std::thread::sleep(Duration::from_millis(2)); - } - Err(TryLockError::Poisoned(_)) => { - break Err("audio state lock poisoned".to_string()); - } - } - } - }; - - // Seeking back invalidates any pending gapless chain. - let cur_pos = { - let cur = lock_current_with_timeout(AUDIO_SEEK_LOCK_TIMEOUT_MS)?; - cur.position() - }; - if seconds < cur_pos - 1.0 { - *state.chained_info.lock().unwrap() = None; - } - - let seek_seconds = seconds.max(0.0); - let seek_duration = Duration::from_secs_f64(seek_seconds); - let seek_generation = state.generation.load(Ordering::SeqCst); - let sink = { - let cur = lock_current_with_timeout(AUDIO_SEEK_LOCK_TIMEOUT_MS)?; - match cur.sink.as_ref() { - Some(sink) => Arc::clone(sink), - None => return Ok(()), - } - }; - - let (tx, rx) = std::sync::mpsc::channel::>(); - std::thread::spawn(move || { - let result = sink.try_seek(seek_duration).map_err(|e| e.to_string()); - let _ = tx.send(result); - }); - - match rx.recv_timeout(Duration::from_millis(AUDIO_SEEK_TIMEOUT_MS)) { - Ok(Ok(())) => {} - Ok(Err(e)) => return Err(e), - Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { - return Err("audio seek timeout".into()); - } - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - return Err("audio seek worker disconnected".into()); - } - } - - // If playback switched while seek was in flight, skip timestamp updates. - if state.generation.load(Ordering::SeqCst) != seek_generation { - return Ok(()); - } - - let mut cur = lock_current_with_timeout(AUDIO_SEEK_LOCK_TIMEOUT_MS)?; - if cur.sink.is_none() { return Ok(()); } - - if cur.paused_at.is_some() { - cur.paused_at = Some(seek_seconds); - } else { - cur.seek_offset = seek_seconds; - cur.play_started = Some(Instant::now()); - } - Ok(()) -} - - #[tauri::command] diff --git a/src-tauri/src/audio/mod.rs b/src-tauri/src/audio/mod.rs index bc80ddb7..23196bdf 100644 --- a/src-tauri/src/audio/mod.rs +++ b/src-tauri/src/audio/mod.rs @@ -11,6 +11,7 @@ mod dev_io; pub mod device_commands; pub mod mix_commands; pub mod radio_commands; +pub mod transport_commands; mod device_watcher; mod engine; mod power_resume; diff --git a/src-tauri/src/audio/transport_commands.rs b/src-tauri/src/audio/transport_commands.rs new file mode 100644 index 00000000..a56a26bc --- /dev/null +++ b/src-tauri/src/audio/transport_commands.rs @@ -0,0 +1,222 @@ +//! Transport-control Tauri commands: pause / resume / stop / seek. +//! These don't drive playback startup — they mutate state on an already-running +//! sink (or coordinate radio reconnect for cold-resume). + +use std::sync::atomic::Ordering; +use std::sync::{Arc, TryLockError}; +use std::time::{Duration, Instant}; + +use ringbuf::traits::Split; +use ringbuf::HeapRb; +use tauri::{AppHandle, State}; + +use super::engine::{audio_http_client, AudioEngine}; +use super::preview::preview_clear_for_new_main_playback; +use super::stream::{radio_download_task, RADIO_BUF_CAPACITY}; + +#[tauri::command] +pub fn audio_pause(state: State<'_, AudioEngine>) { + let mut cur = state.current.lock().unwrap(); + if let Some(sink) = &cur.sink { + if !sink.is_paused() { + let pos = cur.position(); + sink.pause(); + cur.paused_at = Some(pos); + cur.play_started = None; + } + } + // Notify the download task so it can start measuring the hard-pause stall timer. + if let Some(rs) = state.radio_state.lock().unwrap().as_ref() { + rs.flags.is_paused.store(true, Ordering::Release); + } +} + +/// Resume playback. +/// +/// **Warm resume** (`is_hard_paused = false`): download task is still running, +/// buffer has buffered audio. `sink.play()` suffices. +/// +/// **Cold resume** (`is_hard_paused = true`): TCP was dropped. A fresh 4 MB +/// ring buffer is created, its consumer is sent to `AudioStreamReader` (which +/// swaps it in on the next `read()`), and a new download task is spawned. +#[tauri::command] +pub async fn audio_resume(state: State<'_, AudioEngine>, app: AppHandle) -> Result<(), String> { + // If a preview is running, cancel it first — otherwise sink.play() on the + // main sink would mix on top of the preview sink. + preview_clear_for_new_main_playback(&state, &app); + + // Detect radio hard-disconnect. + let reconnect_info = { + let guard = state.radio_state.lock().unwrap(); + guard + .as_ref() + .filter(|rs| rs.flags.is_hard_paused.load(Ordering::Acquire)) + .map(|rs| (rs.url.clone(), rs.gen, rs.flags.clone())) + }; + + if let Some((url, gen, flags)) = reconnect_info { + let rb = HeapRb::::new(RADIO_BUF_CAPACITY); + let (new_prod, new_cons) = rb.split(); + + // Send new consumer to AudioStreamReader (non-blocking; unbounded channel). + let ok = flags.new_cons_tx.lock().unwrap().send(new_cons).is_ok(); + + if ok { + let new_task = tokio::spawn(radio_download_task( + gen, + state.generation.clone(), + None, // task performs its own fresh GET + audio_http_client(&state), + url, + new_prod, + flags.clone(), + app, + )); + if let Some(rs) = state.radio_state.lock().unwrap().as_mut() { + let old = std::mem::replace(&mut rs.task, new_task); + old.abort(); // ensure any lingering old task is gone + rs.flags.is_hard_paused.store(false, Ordering::Release); + rs.flags.is_paused.store(false, Ordering::Release); + } + } else { + crate::app_eprintln!("[radio] resume: AudioStreamReader gone — skipping reconnect"); + } + } + + // Resume the rodio Sink (works for both warm and cold resume). + { + let mut cur = state.current.lock().unwrap(); + if let Some(sink) = &cur.sink { + if sink.is_paused() { + let pos = cur.paused_at.unwrap_or(cur.seek_offset); + sink.play(); + cur.seek_offset = pos; + cur.play_started = Some(Instant::now()); + cur.paused_at = None; + } + } + } + if let Some(rs) = state.radio_state.lock().unwrap().as_ref() { + rs.flags.is_paused.store(false, Ordering::Release); + } + Ok(()) +} + +#[tauri::command] +pub fn audio_stop(state: State<'_, AudioEngine>, app: AppHandle) { + preview_clear_for_new_main_playback(&state, &app); + state.generation.fetch_add(1, Ordering::SeqCst); + *state.current_playback_url.lock().unwrap() = None; + *state.current_analysis_track_id.lock().unwrap() = None; + *state.chained_info.lock().unwrap() = None; + // Keep `stream_completed_cache`: natural track end often calls `audio_stop` when the + // queue is exhausted; clearing here dropped the full ranged buffer and forced a + // re-download on replay. The slot is only consumed on `take`/overwrite for another URL. + // Drop RadioLiveState → triggers Drop → task.abort() → TCP released. + drop(state.radio_state.lock().unwrap().take()); + let mut cur = state.current.lock().unwrap(); + if let Some(sink) = cur.sink.take() { sink.stop(); } + cur.duration_secs = 0.0; + cur.seek_offset = 0.0; + cur.play_started = None; + cur.paused_at = None; +} + +#[tauri::command] +pub fn audio_seek(seconds: f64, state: State<'_, AudioEngine>) -> Result<(), String> { + const AUDIO_SEEK_TIMEOUT_MS: u64 = 700; + const AUDIO_SEEK_LOCK_TIMEOUT_MS: u64 = 40; + // Ghost-command guard: reject seeks within 500 ms of a gapless auto-advance. + { + let switch_ms = state.gapless_switch_at.load(Ordering::SeqCst); + if switch_ms > 0 { + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64; + if now_ms.saturating_sub(switch_ms) < 500 { + return Ok(()); + } + } + } + + // Reject seek up-front for non-seekable streaming sources so the frontend's + // restart-fallback engages instead of rolling the dice on the format reader + // (which can consume the ring buffer to EOF for forward seeks → next song). + if !state.current_is_seekable.load(Ordering::SeqCst) { + crate::app_deprintln!("[seek] rejected → not-seekable source (legacy stream)"); + return Err("source is not seekable".into()); + } + crate::app_deprintln!("[seek] target={:.2}s", seconds); + + let lock_current_with_timeout = |timeout_ms: u64| { + let deadline = Instant::now() + Duration::from_millis(timeout_ms); + loop { + match state.current.try_lock() { + Ok(guard) => break Ok(guard), + Err(TryLockError::WouldBlock) => { + if Instant::now() >= deadline { + break Err("audio seek busy".to_string()); + } + std::thread::sleep(Duration::from_millis(2)); + } + Err(TryLockError::Poisoned(_)) => { + break Err("audio state lock poisoned".to_string()); + } + } + } + }; + + // Seeking back invalidates any pending gapless chain. + let cur_pos = { + let cur = lock_current_with_timeout(AUDIO_SEEK_LOCK_TIMEOUT_MS)?; + cur.position() + }; + if seconds < cur_pos - 1.0 { + *state.chained_info.lock().unwrap() = None; + } + + let seek_seconds = seconds.max(0.0); + let seek_duration = Duration::from_secs_f64(seek_seconds); + let seek_generation = state.generation.load(Ordering::SeqCst); + let sink = { + let cur = lock_current_with_timeout(AUDIO_SEEK_LOCK_TIMEOUT_MS)?; + match cur.sink.as_ref() { + Some(sink) => Arc::clone(sink), + None => return Ok(()), + } + }; + + let (tx, rx) = std::sync::mpsc::channel::>(); + std::thread::spawn(move || { + let result = sink.try_seek(seek_duration).map_err(|e| e.to_string()); + let _ = tx.send(result); + }); + + match rx.recv_timeout(Duration::from_millis(AUDIO_SEEK_TIMEOUT_MS)) { + Ok(Ok(())) => {} + Ok(Err(e)) => return Err(e), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + return Err("audio seek timeout".into()); + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + return Err("audio seek worker disconnected".into()); + } + } + + // If playback switched while seek was in flight, skip timestamp updates. + if state.generation.load(Ordering::SeqCst) != seek_generation { + return Ok(()); + } + + let mut cur = lock_current_with_timeout(AUDIO_SEEK_LOCK_TIMEOUT_MS)?; + if cur.sink.is_none() { return Ok(()); } + + if cur.paused_at.is_some() { + cur.paused_at = Some(seek_seconds); + } else { + cur.seek_offset = seek_seconds; + cur.play_started = Some(Instant::now()); + } + Ok(()) +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index f45663be..6060d2b7 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -388,10 +388,10 @@ pub fn run() { mpris_set_metadata, mpris_set_playback, audio::commands::audio_play, - audio::commands::audio_pause, - audio::commands::audio_resume, - audio::commands::audio_stop, - audio::commands::audio_seek, + audio::transport_commands::audio_pause, + audio::transport_commands::audio_resume, + audio::transport_commands::audio_stop, + audio::transport_commands::audio_seek, audio::mix_commands::audio_set_volume, audio::mix_commands::audio_update_replay_gain, audio::mix_commands::audio_set_eq,