use std::io::{Cursor, Read, Seek, SeekFrom}; use std::sync::{Arc, Mutex, OnceLock, RwLock, TryLockError}; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering}; use std::time::{Duration, Instant}; #[cfg(unix)] use libc; use ringbuf::{HeapConsumer, HeapProducer, HeapRb}; use biquad::{Biquad, Coefficients, DirectForm2Transposed, ToHertz, Type as FilterType}; use rodio::{Sink, Source}; use rodio::source::UniformSourceIterator; use serde::Serialize; use symphonia::core::{ audio::{AudioBufferRef, SampleBuffer, SignalSpec}, codecs::{CodecRegistry, DecoderOptions, CODEC_TYPE_NULL}, formats::{FormatOptions, FormatReader, SeekMode, SeekTo}, io::{MediaSource, MediaSourceStream, MediaSourceStreamOptions}, meta::MetadataOptions, probe::Hint, units::{self, Time}, }; use futures_util::StreamExt; use tauri::{AppHandle, Emitter, Manager, State}; #[derive(Clone, Serialize)] #[serde(rename_all = "camelCase")] struct PartialLoudnessPayload { track_id: Option, gain_db: f32, target_lufs: f32, is_partial: bool, } #[derive(Clone, Serialize)] #[serde(rename_all = "camelCase")] struct WaveformUpdatedPayload { track_id: String, is_partial: bool, } #[derive(Clone, Serialize)] #[serde(rename_all = "camelCase")] struct NormalizationStatePayload { engine: String, current_gain_db: Option, target_lufs: f32, } // ─── 10-Band Graphic Equalizer ──────────────────────────────────────────────── const EQ_BANDS_HZ: [f32; 10] = [31.0, 62.0, 125.0, 250.0, 500.0, 1000.0, 2000.0, 4000.0, 8000.0, 16000.0]; const EQ_Q: f32 = 1.41; const EQ_CHECK_INTERVAL: usize = 1024; struct EqSource> { inner: S, sample_rate: u32, channels: u16, gains: Arc<[AtomicU32; 10]>, enabled: Arc, pre_gain: Arc, filters: [[DirectForm2Transposed; 2]; 10], current_gains: [f32; 10], sample_counter: usize, channel_idx: usize, } impl> EqSource { fn new(inner: S, gains: Arc<[AtomicU32; 10]>, enabled: Arc, pre_gain: Arc) -> Self { let sample_rate = inner.sample_rate(); let channels = inner.channels(); let filters = std::array::from_fn(|band| { let freq = EQ_BANDS_HZ[band].clamp(20.0, (sample_rate as f32 / 2.0) - 100.0); std::array::from_fn(|_| { let coeffs = Coefficients::::from_params( FilterType::PeakingEQ(0.0), (sample_rate as f32).hz(), freq.hz(), EQ_Q, ).unwrap_or_else(|_| Coefficients::::from_params( FilterType::PeakingEQ(0.0), (sample_rate as f32).hz(), 1000.0f32.hz(), EQ_Q, ).unwrap()); DirectForm2Transposed::::new(coeffs) }) }); Self { inner, sample_rate, channels, gains, enabled, pre_gain, filters, current_gains: [0.0; 10], sample_counter: 0, channel_idx: 0, } } fn refresh_if_needed(&mut self) { for band in 0..10 { let gain_db = f32::from_bits(self.gains[band].load(Ordering::Relaxed)); if (gain_db - self.current_gains[band]).abs() > 0.01 { self.current_gains[band] = gain_db; let freq = EQ_BANDS_HZ[band].clamp(20.0, (self.sample_rate as f32 / 2.0) - 100.0); if let Ok(coeffs) = Coefficients::::from_params( FilterType::PeakingEQ(gain_db), (self.sample_rate as f32).hz(), freq.hz(), EQ_Q, ) { for ch in 0..2 { self.filters[band][ch].update_coefficients(coeffs); } } } } } } impl> Iterator for EqSource { type Item = f32; fn next(&mut self) -> Option { let sample = self.inner.next()?; if self.sample_counter % EQ_CHECK_INTERVAL == 0 { self.refresh_if_needed(); } self.sample_counter = self.sample_counter.wrapping_add(1); if !self.enabled.load(Ordering::Relaxed) { self.channel_idx = (self.channel_idx + 1) % self.channels as usize; return Some(sample); } let ch = self.channel_idx.min(1); self.channel_idx = (self.channel_idx + 1) % self.channels as usize; let pre_gain_db = f32::from_bits(self.pre_gain.load(Ordering::Relaxed)); let pre_gain_factor = 10_f32.powf(pre_gain_db / 20.0); let mut s = sample * pre_gain_factor; for band in 0..10 { s = self.filters[band][ch].run(s); } Some(s.clamp(-1.0, 1.0)) } } impl> Source for EqSource { fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } fn channels(&self) -> u16 { self.channels } fn sample_rate(&self) -> u32 { self.sample_rate } fn total_duration(&self) -> Option { self.inner.total_duration() } fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> { // Reset biquad filter state to avoid glitches after seek. for band in 0..10 { let gain_db = f32::from_bits(self.gains[band].load(Ordering::Relaxed)); self.current_gains[band] = gain_db; let freq = EQ_BANDS_HZ[band].clamp(20.0, (self.sample_rate as f32 / 2.0) - 100.0); if let Ok(coeffs) = Coefficients::::from_params( FilterType::PeakingEQ(gain_db), (self.sample_rate as f32).hz(), freq.hz(), EQ_Q, ) { for ch in 0..2 { self.filters[band][ch] = DirectForm2Transposed::::new(coeffs); } } } self.channel_idx = 0; self.sample_counter = 0; self.inner.try_seek(pos) } } // ─── DynSource — type-erased Source wrapper ─────────────────────────────────── // // Allows chaining differently-typed sources (with trimming applied) into a // single concrete type accepted by EqSource>. struct DynSource { inner: Box + Send>, channels: u16, sample_rate: u32, } impl DynSource { fn new(src: impl Source + Send + 'static) -> Self { let channels = src.channels(); let sample_rate = src.sample_rate(); Self { inner: Box::new(src), channels, sample_rate } } } impl Iterator for DynSource { type Item = f32; fn next(&mut self) -> Option { self.inner.next() } } impl Source for DynSource { fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } fn channels(&self) -> u16 { self.channels } fn sample_rate(&self) -> u32 { self.sample_rate } fn total_duration(&self) -> Option { self.inner.total_duration() } fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> { self.inner.try_seek(pos) } } // ─── EqualPowerFadeIn — per-sample sin(t·π/2) fade-in envelope ─────────────── // // Applied to every new track: // • Crossfade: fade_dur = crossfade_secs → symmetric equal-power fade-in // • Hard cut: fade_dur = 5 ms → micro-fade eliminates DC-click // • Gapless: fade_dur = 0 → unity gain (no modification) // // gain(t) = sin(t · π/2), t ∈ [0, 1) // At t = 0 gain = 0, at t = 1 gain = 1. // Equal-power property: cos²+sin² = 1 → combined with cos fade-out on Track A // the total perceived loudness stays constant across the crossfade. struct EqualPowerFadeIn> { inner: S, sample_count: u64, fade_samples: u64, } impl> EqualPowerFadeIn { fn new(inner: S, fade_dur: Duration) -> Self { let sample_rate = inner.sample_rate(); let channels = inner.channels() as u64; let fade_samples = if fade_dur.is_zero() { 0 } else { (fade_dur.as_secs_f64() * sample_rate as f64 * channels as f64) as u64 }; Self { inner, sample_count: 0, fade_samples } } } impl> Iterator for EqualPowerFadeIn { type Item = f32; fn next(&mut self) -> Option { let sample = self.inner.next()?; let gain = if self.fade_samples == 0 || self.sample_count >= self.fade_samples { 1.0 } else { let t = self.sample_count as f32 / self.fade_samples as f32; (t * std::f32::consts::FRAC_PI_2).sin() }; self.sample_count += 1; Some((sample * gain).clamp(-1.0, 1.0)) } } impl> Source for EqualPowerFadeIn { fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } fn channels(&self) -> u16 { self.inner.channels() } fn sample_rate(&self) -> u32 { self.inner.sample_rate() } fn total_duration(&self) -> Option { self.inner.total_duration() } fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> { // For mid-track seeks: skip straight to unity gain so the new position // plays at full volume immediately — no audible fade-in glitch. // For seeks to the very start (< 100 ms): keep the micro-fade to // suppress any DC-offset click from the fresh decode. if pos.as_millis() < 100 { self.sample_count = 0; } else { self.sample_count = self.fade_samples; } self.inner.try_seek(pos) } } // ─── TriggeredFadeOut — sample-level cos(t·π/2) fade-out triggered externally ─ // // Every track source is wrapped with this. It passes through at unity gain // until `trigger` is set to true, at which point it reads `fade_total_samples` // and applies a cos(t·π/2) envelope: // gain(t) = cos(t · π/2), t ∈ [0, 1] // At t = 0 gain = 1, at t = 1 gain = 0. // After the fade completes, returns None to exhaust the source. // // Combined with EqualPowerFadeIn (sin curve) on Track B, this gives a // symmetric constant-power crossfade: sin²+cos² = 1. struct TriggeredFadeOut> { inner: S, trigger: Arc, fade_total_samples: Arc, fade_progress: u64, fading: bool, cached_total: u64, } impl> TriggeredFadeOut { fn new(inner: S, trigger: Arc, fade_total_samples: Arc) -> Self { Self { inner, trigger, fade_total_samples, fade_progress: 0, fading: false, cached_total: 0, } } } impl> Iterator for TriggeredFadeOut { type Item = f32; fn next(&mut self) -> Option { // Check trigger on first fade sample only (avoid atomic load per sample). if !self.fading && self.trigger.load(Ordering::Relaxed) { self.fading = true; self.cached_total = self.fade_total_samples.load(Ordering::Relaxed).max(1); self.fade_progress = 0; } if self.fading { if self.fade_progress >= self.cached_total { // Fade complete — exhaust the source. return None; } let sample = self.inner.next()?; let t = self.fade_progress as f32 / self.cached_total as f32; let gain = (t * std::f32::consts::FRAC_PI_2).cos(); self.fade_progress += 1; Some((sample * gain).clamp(-1.0, 1.0)) } else { self.inner.next() } } } impl> Source for TriggeredFadeOut { fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } fn channels(&self) -> u16 { self.inner.channels() } fn sample_rate(&self) -> u32 { self.inner.sample_rate() } fn total_duration(&self) -> Option { self.inner.total_duration() } fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> { // If we seek back during a fade, cancel the fade. if self.fading { self.fading = false; self.trigger.store(false, Ordering::Relaxed); } self.fade_progress = 0; self.inner.try_seek(pos) } } // ─── NotifyingSource — sets a flag when the inner iterator is exhausted ─────── // // This is the key mechanism for gapless: the progress task polls `done` to know // exactly when source N has finished inside the Sink, without relying on // wall-clock estimation or the unreliable `Sink::empty()`. struct NotifyingSource> { inner: S, done: Arc, signalled: bool, } impl> NotifyingSource { fn new(inner: S, done: Arc) -> Self { Self { inner, done, signalled: false } } } impl> Iterator for NotifyingSource { type Item = f32; fn next(&mut self) -> Option { let sample = self.inner.next(); if sample.is_none() && !self.signalled { self.signalled = true; self.done.store(true, Ordering::SeqCst); } sample } } impl> Source for NotifyingSource { fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } fn channels(&self) -> u16 { self.inner.channels() } fn sample_rate(&self) -> u32 { self.inner.sample_rate() } fn total_duration(&self) -> Option { self.inner.total_duration() } fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> { // If we seek backwards the source is no longer exhausted. self.signalled = false; self.done.store(false, Ordering::SeqCst); self.inner.try_seek(pos) } } // ─── CountingSource — atomic sample counter for drift-free position tracking ─ // // Wraps the outermost source and increments a shared AtomicU64 on every sample. // The progress task reads this counter and divides by (sample_rate * channels) // to get the exact playback position — no wall-clock drift. struct CountingSource> { inner: S, counter: Arc, } impl> CountingSource { fn new(inner: S, counter: Arc) -> Self { Self { inner, counter } } } impl> Iterator for CountingSource { type Item = f32; fn next(&mut self) -> Option { let sample = self.inner.next(); if sample.is_some() { self.counter.fetch_add(1, Ordering::Relaxed); } sample } } impl> Source for CountingSource { fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } fn channels(&self) -> u16 { self.inner.channels() } fn sample_rate(&self) -> u32 { self.inner.sample_rate() } fn total_duration(&self) -> Option { self.inner.total_duration() } fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> { // Reset counter only after confirming the inner seek succeeded. // If we reset first and the seek fails, the counter ends up at the // new position while the decoder is still at the old one — causing // a permanent desync between displayed time and actual audio. let result = self.inner.try_seek(pos); if result.is_ok() { let samples = (pos.as_secs_f64() * self.inner.sample_rate() as f64 * self.inner.channels() as f64) as u64; self.counter.store(samples, Ordering::Relaxed); } result } } // ─── Internet Radio v2 — Lock-Free SPSC + ICY Metadata + Hybrid Pause ──────── // // HTTP task (tokio) // └─[IcyInterceptor]─► HeapProducer // │ (4 MB HeapRb, lock-free) // HeapConsumer // │ // AudioStreamReader (Read + Seek + MediaSource) // │ // SizedDecoder (symphonia) // │ // rodio Sink // // Pause modes: // Logical pause — sink.pause(); download task keeps filling (time-shift). // Hard pause — buffer ≥ RADIO_HARD_PAUSE_THRESH full + paused ≥ 5 s // → TCP disconnect, is_hard_paused = true. // Resume (warm) — sink.play(); buffer drains seamlessly. // Resume (cold) — new HeapRb + new GET; consumer swapped in AudioStreamReader. // // New Tauri event: "radio:metadata" → String (ICY StreamTitle) /// 256 KB on the heap — ≈16 s at 128 kbps, ≈6 s at 320 kbps. /// Small enough that stale audio drains within a few seconds on reconnect; /// large enough to absorb brief network hiccups without stuttering. const RADIO_BUF_CAPACITY: usize = 256 * 1024; /// Minimum ring buffer for on-demand track streaming starts. const TRACK_STREAM_MIN_BUF_CAPACITY: usize = 1024 * 1024; /// Cap ring buffer growth when content-length is known. const TRACK_STREAM_MAX_BUF_CAPACITY: usize = 32 * 1024 * 1024; /// Max bytes kept in memory to promote a completed streamed track for fast replay/seek recovery. const TRACK_STREAM_PROMOTE_MAX_BYTES: usize = 64 * 1024 * 1024; /// Consecutive body-stream failures tolerated for track streaming before abort. const TRACK_STREAM_MAX_RECONNECTS: u32 = 3; /// Seconds at stall threshold while paused before hard-disconnect. const RADIO_HARD_PAUSE_SECS: u64 = 5; /// AudioStreamReader timeout: if no audio bytes arrive for this long → EOF. const RADIO_READ_TIMEOUT_SECS: u64 = 15; /// Sleep interval when ring buffer is empty (prevents CPU spin). const RADIO_YIELD_MS: u64 = 2; // ── ICY Metadata State Machine ──────────────────────────────────────────────── // // Shoutcast/Icecast embed metadata every `metaint` audio bytes: // // ┌──────────────────────┬───┬─────────────┐ // │ audio × metaint │ N │ meta × N×16 │ (repeating) // └──────────────────────┴───┴─────────────┘ // // N = 0 → no metadata this block. Metadata bytes are stripped so only // pure audio reaches the ring buffer and Symphonia never sees text bytes. enum IcyState { /// Forwarding audio bytes; `remaining` counts down to the next boundary. ReadingAudio { remaining: usize }, /// Next byte is the metadata length multiplier N. ReadingLengthByte, /// Accumulating N×16 metadata bytes. ReadingMetadata { remaining: usize, buf: Vec }, } struct IcyInterceptor { state: IcyState, metaint: usize, } impl IcyInterceptor { fn new(metaint: usize) -> Self { Self { metaint, state: IcyState::ReadingAudio { remaining: metaint } } } /// Feed a raw HTTP chunk. /// Appends only audio bytes to `audio_out`. /// Returns `Some(IcyMeta)` when a StreamTitle is extracted. fn process(&mut self, input: &[u8], audio_out: &mut Vec) -> Option { let mut extracted: Option = None; let mut i = 0; while i < input.len() { match &mut self.state { IcyState::ReadingAudio { remaining } => { let n = (input.len() - i).min(*remaining); audio_out.extend_from_slice(&input[i..i + n]); i += n; *remaining -= n; if *remaining == 0 { self.state = IcyState::ReadingLengthByte; } } IcyState::ReadingLengthByte => { let len_n = input[i] as usize; i += 1; self.state = if len_n == 0 { IcyState::ReadingAudio { remaining: self.metaint } } else { IcyState::ReadingMetadata { remaining: len_n * 16, buf: Vec::with_capacity(len_n * 16), } }; } IcyState::ReadingMetadata { remaining, buf } => { let n = (input.len() - i).min(*remaining); buf.extend_from_slice(&input[i..i + n]); i += n; *remaining -= n; if *remaining == 0 { let bytes = std::mem::take(buf); extracted = parse_icy_meta(&bytes); self.state = IcyState::ReadingAudio { remaining: self.metaint }; } } } } extracted } } /// ICY metadata parsed from a raw metadata block. #[derive(serde::Serialize, Clone)] pub(crate) struct IcyMeta { pub title: String, /// `true` when `StreamUrl='0'` — indicates a CDN-injected ad/promo. pub is_ad: bool, } /// Extract `StreamTitle` and `StreamUrl` from a raw ICY metadata block. /// Tolerates null padding and non-UTF-8 bytes (lossy conversion). fn parse_icy_meta(raw: &[u8]) -> Option { let s = String::from_utf8_lossy(raw); let s = s.trim_end_matches('\0'); const TITLE_TAG: &str = "StreamTitle='"; let title_start = s.find(TITLE_TAG)? + TITLE_TAG.len(); let title_rest = &s[title_start..]; // find (not rfind) — rfind would skip past StreamUrl and corrupt the title let title_end = title_rest.find("';")?; let title = title_rest[..title_end].trim().to_string(); if title.is_empty() { return None; } const URL_TAG: &str = "StreamUrl='"; let stream_url = s.find(URL_TAG).map(|pos| { let rest = &s[pos + URL_TAG.len()..]; let end = rest.find("';").unwrap_or(rest.len()); rest[..end].trim().to_string() }).unwrap_or_default(); Some(IcyMeta { title, is_ad: stream_url == "0" }) } // ── AudioStreamReader — SPSC consumer → std::io::Read ──────────────────────── // // Bridges HeapConsumer (non-blocking) into the synchronous Read interface // that Symphonia requires. Designed to run inside tokio::task::spawn_blocking. // // Empty buffer: sleeps RADIO_YIELD_MS ms, retries. Never busy-spins. // Timeout: after RADIO_READ_TIMEOUT_SECS with no data → TimedOut. // Generation: if gen_arc != self.gen → Ok(0) (EOF; new track started). // Reconnect: audio_resume sends a fresh HeapConsumer via new_cons_rx. // On the next read() we drain the channel (keep latest) and swap. struct AudioStreamReader { cons: HeapConsumer, /// Delivers fresh consumers on hard-pause reconnect (unbounded; drain to latest). /// Wrapped in Mutex so AudioStreamReader is Sync (required by symphonia::MediaSource). /// No real contention: only the audio thread ever calls read(). new_cons_rx: Mutex>>, deadline: std::time::Instant, gen_arc: Arc, gen: u64, /// Diagnostic tag for logs ("radio" or "track-stream"). source_tag: &'static str, /// Optional completion marker: when true and the ring buffer is empty, /// return EOF immediately (used by one-shot track streaming). eof_when_empty: Option>, /// Monotonic byte offset for SeekFrom::Current(0) "tell" (Symphonia probe). pos: u64, } impl Read for AudioStreamReader { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { // EOF guard: new track started. if self.gen_arc.load(Ordering::SeqCst) != self.gen { return Ok(0); } // Drain reconnect channel; keep only the most recently delivered consumer // so a double-tap of resume doesn't leave stale data in place. let mut newest: Option> = None; while let Ok(c) = self.new_cons_rx.lock().unwrap().try_recv() { newest = Some(c); } if let Some(c) = newest { self.cons = c; self.deadline = std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); } loop { if self.gen_arc.load(Ordering::SeqCst) != self.gen { return Ok(0); } let available = self.cons.len(); if available > 0 { let n = buf.len().min(available); let read = self.cons.pop_slice(&mut buf[..n]); self.pos += read as u64; // Reset deadline: data arrived, so connection is alive. self.deadline = std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); return Ok(read); } if self .eof_when_empty .as_ref() .is_some_and(|done| done.load(Ordering::SeqCst)) { return Ok(0); } if std::time::Instant::now() >= self.deadline { crate::app_eprintln!( "[{}] AudioStreamReader: {}s without data → EOF", self.source_tag, RADIO_READ_TIMEOUT_SECS ); return Err(std::io::Error::new( std::io::ErrorKind::TimedOut, format!("{}: no data received", self.source_tag), )); } std::thread::sleep(Duration::from_millis(RADIO_YIELD_MS)); } } } impl Seek for AudioStreamReader { fn seek(&mut self, pos: SeekFrom) -> std::io::Result { match pos { SeekFrom::Current(0) => Ok(self.pos), _ => Err(std::io::Error::new( std::io::ErrorKind::Unsupported, format!("{} stream is not seekable", self.source_tag), )), } } } impl MediaSource for AudioStreamReader { fn is_seekable(&self) -> bool { false } fn byte_len(&self) -> Option { None } } // ── RangedHttpSource — seekable HTTP-backed MediaSource ────────────────────── // // Pre-allocates a Vec of total track size. A background task fills it // linearly from offset 0 via streaming HTTP. Read blocks (with timeout) until // requested bytes are downloaded; Seek only updates the cursor. // // Reports is_seekable=true so Symphonia performs time-based seeks via the // format reader. Backward seeks: instant (data in buffer). Forward seeks // beyond downloaded_to: Read blocks until the linear download catches up. // // Requires server to have responded with both Content-Length and // `Accept-Ranges: bytes` so reconnects can resume via HTTP Range. struct RangedHttpSource { /// Pre-allocated buffer of total size. Filled linearly from offset 0. buf: Arc>>, /// Bytes contiguously downloaded from offset 0. downloaded_to: Arc, total_size: u64, pos: u64, /// Set when the download task terminates (success or hard error). done: Arc, gen_arc: Arc, gen: u64, } impl Read for RangedHttpSource { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { if self.gen_arc.load(Ordering::SeqCst) != self.gen { return Ok(0); } if self.pos >= self.total_size { return Ok(0); } let max_read = ((self.total_size - self.pos) as usize).min(buf.len()); if max_read == 0 { return Ok(0); } let target_end = self.pos + max_read as u64; let deadline = Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); loop { if self.gen_arc.load(Ordering::SeqCst) != self.gen { return Ok(0); } let dl = self.downloaded_to.load(Ordering::SeqCst) as u64; if dl >= target_end { break; } // Download finished but our cursor is past downloaded_to (e.g. seek // beyond a partial download that aborted). Return what we have. if self.done.load(Ordering::SeqCst) { if dl > self.pos { let avail = (dl - self.pos) as usize; let src = self.buf.lock().unwrap(); let start = self.pos as usize; buf[..avail].copy_from_slice(&src[start..start + avail]); drop(src); self.pos += avail as u64; return Ok(avail); } return Ok(0); } if Instant::now() >= deadline { return Err(std::io::Error::new( std::io::ErrorKind::TimedOut, "ranged-http: no data within timeout", )); } std::thread::sleep(Duration::from_millis(RADIO_YIELD_MS)); } let src = self.buf.lock().unwrap(); let start = self.pos as usize; let end = start + max_read; buf[..max_read].copy_from_slice(&src[start..end]); drop(src); self.pos += max_read as u64; Ok(max_read) } } impl Seek for RangedHttpSource { fn seek(&mut self, pos: SeekFrom) -> std::io::Result { let new_pos: i64 = match pos { SeekFrom::Start(p) => p as i64, SeekFrom::Current(p) => self.pos as i64 + p, SeekFrom::End(p) => self.total_size as i64 + p, }; if new_pos < 0 { return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, "ranged-http: seek before start", )); } self.pos = (new_pos as u64).min(self.total_size); Ok(self.pos) } } impl MediaSource for RangedHttpSource { fn is_seekable(&self) -> bool { true } fn byte_len(&self) -> Option { Some(self.total_size) } } // ── LocalFileSource — seekable file-backed MediaSource ─────────────────────── // // Wraps `std::fs::File` so the decoder reads on-demand from disk instead of // pre-loading the whole file into a Vec. Used for `psysonic-local://` URLs // (offline library + hot playback cache hits) — gives instant track-start // because Symphonia only needs to read ~64 KB during probe before playback // can begin, vs the previous behaviour of `tokio::fs::read` which blocked // until the entire file (often 100+ MB for hi-res FLAC) was in RAM. struct LocalFileSource { file: std::fs::File, len: u64, } impl Read for LocalFileSource { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { self.file.read(buf) } } impl Seek for LocalFileSource { fn seek(&mut self, pos: SeekFrom) -> std::io::Result { self.file.seek(pos) } } impl MediaSource for LocalFileSource { fn is_seekable(&self) -> bool { true } fn byte_len(&self) -> Option { Some(self.len) } } // ── Pause / Reconnect Coordination ─────────────────────────────────────────── pub(crate) struct RadioSharedFlags { /// Set by audio_pause; cleared by audio_resume. is_paused: AtomicBool, /// Set by download task on hard disconnect; cleared on resume-reconnect. is_hard_paused: AtomicBool, /// Delivers a fresh HeapConsumer to AudioStreamReader on reconnect. new_cons_tx: Mutex>>, } /// Live state for the current radio session, stored in AudioEngine. /// Dropping this struct aborts the HTTP download task immediately. pub(crate) struct RadioLiveState { pub url: String, pub gen: u64, pub task: tokio::task::JoinHandle<()>, pub flags: Arc, } impl Drop for RadioLiveState { fn drop(&mut self) { self.task.abort(); } } // ── HE-AAC / FDK-AAC Fallback ──────────────────────────────────────────────── // // Symphonia 0.5.x: AAC-LC only. HE-AAC (AAC+) and HE-AACv2 lack SBR/PS → // streams play at half speed with muffled audio. // // With Cargo feature "fdk-aac": FdkAacDecoder is tried first for CODEC_TYPE_AAC. // Enable in Cargo.toml: // symphonia-adapter-fdk-aac = { version = "0.1", optional = true } // [features] // fdk-aac = ["dep:symphonia-adapter-fdk-aac"] /// Symphonia’s default codec set for our enabled features, plus Opus via libopus. fn psysonic_codec_registry() -> &'static CodecRegistry { static REGISTRY: OnceLock = OnceLock::new(); REGISTRY.get_or_init(|| { let mut registry = CodecRegistry::new(); symphonia::default::register_enabled_codecs(&mut registry); registry.register_all::(); registry }) } fn try_make_radio_decoder( params: &symphonia::core::codecs::CodecParameters, opts: &DecoderOptions, ) -> Result, symphonia::core::errors::Error> { psysonic_codec_registry().make(params, opts) } // ── Async HTTP Download Task ────────────────────────────────────────────────── // // Lifecycle: // 'outer loop — reconnect on TCP drop (up to MAX_RECONNECTS) // 'inner loop — read HTTP chunks → ICY interceptor → push audio to ring buffer // // Hard-pause detection: if push_slice() returns 0 (buffer full) AND sink is // paused AND that condition persists for RADIO_HARD_PAUSE_SECS → disconnect. // Sets is_hard_paused = true so audio_resume knows it must reconnect. async fn radio_download_task( gen: u64, gen_arc: Arc, mut initial_response: Option, http_client: reqwest::Client, url: String, mut prod: HeapProducer, flags: Arc, app: AppHandle, ) { let mut bytes_total: u64 = 0; // Counts consecutive failures (reset on each successful chunk). // laut.fm and similar CDNs force-reconnect every ~700 KB; this is normal. let mut reconnect_count: u32 = 0; const MAX_CONSECUTIVE_FAILURES: u32 = 5; let mut audio_scratch: Vec = Vec::with_capacity(65_536); 'outer: loop { if gen_arc.load(Ordering::SeqCst) != gen { return; } // ── Obtain response (initial or reconnect) ──────────────────────────── let response = match initial_response.take() { Some(r) => r, None => { if reconnect_count >= MAX_CONSECUTIVE_FAILURES { crate::app_eprintln!("[radio] {MAX_CONSECUTIVE_FAILURES} consecutive failures — giving up"); break 'outer; } tokio::time::sleep(Duration::from_millis(500)).await; if gen_arc.load(Ordering::SeqCst) != gen { return; } match http_client .get(&url) .header("Icy-MetaData", "1") .send() .await { Ok(r) if r.status().is_success() => { crate::app_eprintln!("[radio] reconnected ({bytes_total} B so far)"); r } Ok(r) => { crate::app_eprintln!("[radio] reconnect: HTTP {} — giving up", r.status()); break 'outer; } Err(e) => { crate::app_eprintln!("[radio] reconnect error: {e} — giving up"); break 'outer; } } } }; // Parse ICY metaint from each response (consistent across reconnects). let metaint: Option = response .headers() .get("icy-metaint") .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse().ok()); let mut icy = metaint.map(IcyInterceptor::new); let mut byte_stream = response.bytes_stream(); // Stall timer: tracks how long push_slice() returns 0 while paused. let mut stall_since: Option = None; 'inner: loop { if gen_arc.load(Ordering::SeqCst) != gen { return; } // ── Back-pressure + hard-pause detection ────────────────────────── if prod.is_full() { if flags.is_paused.load(Ordering::Relaxed) { let since = stall_since.get_or_insert(std::time::Instant::now()); if since.elapsed() >= Duration::from_secs(RADIO_HARD_PAUSE_SECS) { let fill_pct = ((1.0 - prod.free_len() as f32 / RADIO_BUF_CAPACITY as f32) * 100.0) as u32; crate::app_eprintln!( "[radio] hard pause: {fill_pct}% full, \ paused >{RADIO_HARD_PAUSE_SECS}s → disconnecting" ); flags.is_hard_paused.store(true, Ordering::Release); return; // Drop HeapProducer → TCP connection released. } } else { stall_since = None; } tokio::time::sleep(Duration::from_millis(50)).await; continue 'inner; } stall_since = None; // ── Read HTTP chunk ─────────────────────────────────────────────── match byte_stream.next().await { Some(Ok(chunk)) => { bytes_total += chunk.len() as u64; // Successful data → reset consecutive-failure counter. reconnect_count = 0; audio_scratch.clear(); if let Some(ref mut interceptor) = icy { if let Some(meta) = interceptor.process(&chunk, &mut audio_scratch) { let label = if meta.is_ad { "[Ad]" } else { "" }; crate::app_eprintln!("[radio] ICY StreamTitle: {}{}", label, meta.title); let _ = app.emit("radio:metadata", &meta); } } else { audio_scratch.extend_from_slice(&chunk); } // Push with per-chunk back-pressure: yield 5 ms if full mid-chunk. let mut offset = 0; while offset < audio_scratch.len() { if gen_arc.load(Ordering::SeqCst) != gen { return; } let pushed = prod.push_slice(&audio_scratch[offset..]); if pushed == 0 { tokio::time::sleep(Duration::from_millis(5)).await; } else { offset += pushed; } } } Some(Err(e)) => { reconnect_count += 1; crate::app_eprintln!("[radio] stream error: {e} → reconnecting (consecutive #{reconnect_count})"); break 'inner; } None => { reconnect_count += 1; crate::app_eprintln!("[radio] stream ended cleanly → reconnecting (consecutive #{reconnect_count})"); break 'inner; } } } // 'inner // Do NOT swap the ring buffer here. The remaining bytes in the buffer // are still valid audio and will drain naturally during reconnect. // Clearing it would cause an immediate underrun/glitch. // The buffer is kept small (RADIO_BUF_CAPACITY) so stale audio drains // within a few seconds rather than minutes. } // 'outer crate::app_eprintln!("[radio] download task done ({bytes_total} B total)"); } /// One-shot HTTP downloader for track streaming starts. /// /// Pushes response chunks into an SPSC ring buffer consumed by `AudioStreamReader`. /// Terminates when: /// - generation changes (track superseded), /// - response stream ends, or /// - response emits an error. async fn track_download_task( gen: u64, gen_arc: Arc, http_client: reqwest::Client, app: AppHandle, url: String, initial_response: reqwest::Response, mut prod: HeapProducer, done: Arc, promote_cache_slot: Arc>>, normalization_target_lufs: Arc, loudness_pre_analysis_attenuation_db: Arc, cache_track_id: Option, ) { let mut downloaded: u64 = 0; let mut reconnects: u32 = 0; let mut next_response: Option = Some(initial_response); let mut capture: Vec = Vec::new(); let mut capture_over_limit = false; let mut last_partial_loudness_emit = Instant::now() - Duration::from_secs(5); 'outer: loop { let response = if let Some(r) = next_response.take() { r } else { let mut req = http_client.get(&url); if downloaded > 0 { req = req.header(reqwest::header::RANGE, format!("bytes={downloaded}-")); } match req.send().await { Ok(r) => r, Err(err) => { if reconnects >= TRACK_STREAM_MAX_RECONNECTS { crate::app_eprintln!( "[audio] streaming reconnect failed after {} attempts: {}", reconnects, err ); done.store(true, Ordering::SeqCst); return; } reconnects += 1; tokio::time::sleep(Duration::from_millis(200)).await; continue 'outer; } } }; if downloaded > 0 && response.status() != reqwest::StatusCode::PARTIAL_CONTENT { crate::app_eprintln!( "[audio] streaming reconnect returned {}, expected 206 for range resume", response.status() ); done.store(true, Ordering::SeqCst); return; } if downloaded == 0 && !response.status().is_success() { crate::app_eprintln!("[audio] streaming HTTP {}", response.status()); done.store(true, Ordering::SeqCst); return; } let mut byte_stream = response.bytes_stream(); while let Some(chunk) = byte_stream.next().await { if gen_arc.load(Ordering::SeqCst) != gen { done.store(true, Ordering::SeqCst); return; } let chunk = match chunk { Ok(c) => c, Err(e) => { if reconnects >= TRACK_STREAM_MAX_RECONNECTS { crate::app_eprintln!( "[audio] streaming download error after {} reconnects: {}", reconnects, e ); done.store(true, Ordering::SeqCst); return; } reconnects += 1; crate::app_eprintln!( "[audio] streaming download error (attempt {}/{}): {} — reconnecting", reconnects, TRACK_STREAM_MAX_RECONNECTS, e ); next_response = None; continue 'outer; } }; reconnects = 0; let mut offset = 0; while offset < chunk.len() { if gen_arc.load(Ordering::SeqCst) != gen { done.store(true, Ordering::SeqCst); return; } let pushed = prod.push_slice(&chunk[offset..]); if pushed == 0 { tokio::time::sleep(Duration::from_millis(5)).await; } else { if !capture_over_limit { if capture.len().saturating_add(pushed) <= TRACK_STREAM_PROMOTE_MAX_BYTES { let from = offset; let to = offset + pushed; capture.extend_from_slice(&chunk[from..to]); } else { capture.clear(); capture_over_limit = true; } } if !capture_over_limit && last_partial_loudness_emit.elapsed() >= Duration::from_millis(PARTIAL_LOUDNESS_EMIT_INTERVAL_MS) { let target_lufs = f32::from_bits(normalization_target_lufs.load(Ordering::Relaxed)); let pre_db = f32::from_bits( loudness_pre_analysis_attenuation_db.load(Ordering::Relaxed), ) .clamp(-24.0, 0.0); emit_partial_loudness_from_bytes(&app, &url, &capture, target_lufs, pre_db); last_partial_loudness_emit = Instant::now(); } offset += pushed; downloaded += pushed as u64; } } } if !capture_over_limit && !capture.is_empty() { if let Some(track_id) = cache_track_id { match crate::analysis_cache::seed_from_bytes(&app, &track_id, &capture) { Err(e) => crate::app_eprintln!("[analysis] track seed failed for {}: {}", track_id, e), Ok(crate::analysis_cache::SeedFromBytesOutcome::Upserted) => { let _ = app.emit( "analysis:waveform-updated", WaveformUpdatedPayload { track_id: track_id.clone(), is_partial: false, }, ); } Ok(_) => {} } } *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url: url.clone(), data: capture, }); } done.store(true, Ordering::SeqCst); return; } } /// Linear downloader for `RangedHttpSource`: fills the pre-allocated buffer /// from offset 0 to total_size. Reconnects via HTTP Range from the current /// `downloaded` offset on transient errors. On completion (full track) the /// data is promoted to `stream_completed_cache` for fast replay. async fn ranged_download_task( gen: u64, gen_arc: Arc, http_client: reqwest::Client, app: AppHandle, _duration_hint: f64, url: String, initial_response: reqwest::Response, buf: Arc>>, downloaded_to: Arc, done: Arc, promote_cache_slot: Arc>>, normalization_target_lufs: Arc, loudness_pre_analysis_attenuation_db: Arc, cache_track_id: Option, ) { let total_size = buf.lock().unwrap().len(); let mut downloaded: usize = 0; let mut reconnects: u32 = 0; let mut next_response: Option = Some(initial_response); let dl_started = Instant::now(); let mut next_progress_mb: usize = 1; let mut last_partial_loudness_emit = Instant::now() - Duration::from_secs(5); 'outer: loop { let response = if let Some(r) = next_response.take() { r } else { let mut req = http_client.get(&url); if downloaded > 0 { req = req.header(reqwest::header::RANGE, format!("bytes={downloaded}-")); } match req.send().await { Ok(r) => r, Err(err) => { if reconnects >= TRACK_STREAM_MAX_RECONNECTS { crate::app_eprintln!( "[audio] ranged reconnect failed after {} attempts: {}", reconnects, err ); break 'outer; } reconnects += 1; tokio::time::sleep(Duration::from_millis(200)).await; continue 'outer; } } }; if downloaded > 0 && response.status() != reqwest::StatusCode::PARTIAL_CONTENT { crate::app_eprintln!( "[audio] ranged reconnect returned {}, expected 206", response.status() ); break 'outer; } if downloaded == 0 && !response.status().is_success() { crate::app_eprintln!("[audio] ranged HTTP {}", response.status()); break 'outer; } let mut byte_stream = response.bytes_stream(); while let Some(chunk) = byte_stream.next().await { if gen_arc.load(Ordering::SeqCst) != gen { done.store(true, Ordering::SeqCst); return; } let chunk = match chunk { Ok(c) => c, Err(e) => { if reconnects >= TRACK_STREAM_MAX_RECONNECTS { crate::app_eprintln!( "[audio] ranged dl error after {} reconnects: {}", reconnects, e ); break 'outer; } reconnects += 1; crate::app_eprintln!( "[audio] ranged dl error (attempt {}/{}): {} — reconnecting", reconnects, TRACK_STREAM_MAX_RECONNECTS, e ); next_response = None; continue 'outer; } }; reconnects = 0; let writable = total_size.saturating_sub(downloaded); if writable == 0 { break; } let n = chunk.len().min(writable); { let mut b = buf.lock().unwrap(); b[downloaded..downloaded + n].copy_from_slice(&chunk[..n]); } downloaded += n; downloaded_to.store(downloaded, Ordering::SeqCst); if downloaded >= PARTIAL_LOUDNESS_MIN_BYTES && total_size > 0 && last_partial_loudness_emit.elapsed() >= Duration::from_millis(PARTIAL_LOUDNESS_EMIT_INTERVAL_MS) { let target_lufs = f32::from_bits(normalization_target_lufs.load(Ordering::Relaxed)); let start_db = f32::from_bits(loudness_pre_analysis_attenuation_db.load(Ordering::Relaxed)) .clamp(-24.0, 0.0); if let Some(provisional_db) = provisional_loudness_gain_from_progress(downloaded, total_size, target_lufs, start_db) { let _ = app.emit( "analysis:loudness-partial", PartialLoudnessPayload { track_id: playback_identity(&url), gain_db: provisional_db, target_lufs, is_partial: true, }, ); } last_partial_loudness_emit = Instant::now(); } let mb = downloaded / (1024 * 1024); if mb >= next_progress_mb { let pct = (downloaded as f64 / total_size as f64 * 100.0) as u32; crate::app_deprintln!( "[stream] dl progress: {} MB / {} MB ({}%)", mb, total_size / (1024 * 1024), pct ); next_progress_mb = mb + 1; } if downloaded >= total_size { break; } } // Stream ended cleanly (or hit total_size). break 'outer; } done.store(true, Ordering::SeqCst); crate::app_deprintln!( "[stream] dl done: {} / {} bytes in {:.2}s ({} reconnects)", downloaded, total_size, dl_started.elapsed().as_secs_f64(), reconnects ); if downloaded == total_size && total_size > 0 && total_size <= TRACK_STREAM_PROMOTE_MAX_BYTES { let data = buf.lock().unwrap().clone(); if let Some(track_id) = cache_track_id { match crate::analysis_cache::seed_from_bytes(&app, &track_id, &data) { Err(e) => crate::app_eprintln!("[analysis] ranged seed failed for {}: {}", track_id, e), Ok(crate::analysis_cache::SeedFromBytesOutcome::Upserted) => { let _ = app.emit( "analysis:waveform-updated", WaveformUpdatedPayload { track_id: track_id.clone(), is_partial: false, }, ); } Ok(_) => {} } } *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url, data }); crate::app_deprintln!("[stream] promoted to stream_completed_cache for replay"); } } fn emit_partial_loudness_from_bytes( app: &AppHandle, url: &str, bytes: &[u8], target_lufs: f32, pre_analysis_attenuation_db: f32, ) { if bytes.len() < PARTIAL_LOUDNESS_MIN_BYTES { crate::app_deprintln!( "[normalization] partial-loudness skip reason=insufficient-bytes bytes={} min_bytes={}", bytes.len(), PARTIAL_LOUDNESS_MIN_BYTES ); return; } // Lightweight fallback based on buffered bytes count to keep CPU low. let mb = bytes.len() as f32 / (1024.0 * 1024.0); let pre_floor = pre_analysis_attenuation_db.clamp(-24.0, 0.0); // Target-derived hint (e.g. -12 LUFS → -1 dB). Old `(hint).clamp(pre, 0)` left // the hint when it lay inside [pre, 0] — e.g. -1 with pre=-6, so AAC/M4A // streaming often sat at -1 dB until full analysis. Combine with user trim: // stricter (more negative) pre wins; milder pre still caps vs the hint. let heuristic_floor = (target_lufs + 11.0).clamp(-6.0, 0.0); let floor_db = if pre_floor < heuristic_floor { pre_floor } else { pre_floor.max(heuristic_floor) }; let gain_db = (-(mb * 0.7)).max(floor_db).min(0.0); crate::app_deprintln!( "[normalization] partial-loudness emit bytes={} gain_db={:.2} target_lufs={:.2} track_id={:?}", bytes.len(), gain_db, target_lufs, playback_identity(url) ); let _ = app.emit( "analysis:loudness-partial", PartialLoudnessPayload { track_id: playback_identity(url), gain_db: gain_db as f32, target_lufs, is_partial: true, }, ); } fn provisional_loudness_gain_from_progress( downloaded: usize, total_size: usize, target_lufs: f32, start_db_in: f32, ) -> Option { if total_size == 0 || downloaded == 0 { return None; } let progress = (downloaded as f32 / total_size as f32).clamp(0.0, 1.0); // Move from startup attenuation toward a more realistic late-stream level. // This avoids staying near -2 dB and then jumping hard when final LUFS lands. let start_db = start_db_in.clamp(-24.0, 0.0).min(0.0); let end_db = (target_lufs + 6.0).clamp(-10.0, -3.0).min(0.0); let shaped = progress.powf(0.75); Some(start_db + (end_db - start_db) * shaped) } fn content_type_to_hint(ct: &str) -> Option { let ct = ct.to_ascii_lowercase(); if ct.contains("mpeg") || ct.contains("mp3") { Some("mp3".into()) } else if ct.contains("aac") || ct.contains("aacp") { Some("aac".into()) } else if ct.contains("ogg") { Some("ogg".into()) } else if ct.contains("flac") { Some("flac".into()) } else if ct.contains("wav") || ct.contains("wave") { Some("wav".into()) } else if ct.contains("opus") { Some("opus".into()) } else { None } } // ─── SizedCursorSource — correct byte_len for seekable in-memory sources ────── // // rodio's internal ReadSeekSource wraps Cursor> but hardcodes // byte_len() → None. This tells symphonia "stream length unknown", which // prevents the FLAC demuxer from seeking (it validates seek offsets against // the total stream length from byte_len). MP3 is unaffected because its // demuxer uses Xing/LAME headers instead. // // This wrapper provides the actual byte length, fixing seek for all formats. struct SizedCursorSource { inner: Cursor>, len: u64, } impl Read for SizedCursorSource { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { self.inner.read(buf) } } impl Seek for SizedCursorSource { fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result { self.inner.seek(pos) } } impl MediaSource for SizedCursorSource { fn is_seekable(&self) -> bool { true } fn byte_len(&self) -> Option { Some(self.len) } } // ─── SizedDecoder — symphonia decoder with correct byte_len ─────────────────── // // Replaces rodio::Decoder::new() which wraps the source in ReadSeekSource // (byte_len = None). This constructs the symphonia pipeline directly, // providing the correct byte_len via SizedCursorSource. // // Implements Iterator + Source — identical interface to // rodio::Decoder, so the rest of the source chain is unchanged. /// Debug logging: codec parameters in human-readable form to verify whether /// playback is genuinely lossless. fn log_codec_resolution( tag: &str, params: &symphonia::core::codecs::CodecParameters, container_hint: Option<&str>, ) { let codec_name = symphonia::default::get_codecs() .get_codec(params.codec) .map(|d| d.short_name) .unwrap_or("?"); let rate = params.sample_rate.map(|r| format!("{} Hz", r)).unwrap_or_else(|| "? Hz".into()); let bits = params.bits_per_sample .or(params.bits_per_coded_sample) .map(|b| format!("{}-bit", b)) .unwrap_or_else(|| "?-bit".into()); let ch = params.channels .map(|c| format!("{}ch", c.count())) .unwrap_or_else(|| "?ch".into()); let lossless = codec_name.starts_with("pcm") || matches!( codec_name, "flac" | "alac" | "wavpack" | "monkeys-audio" | "tta" | "shorten" ); let kind = if lossless { "LOSSLESS" } else { "lossy" }; crate::app_deprintln!( "[stream] {tag}: codec={codec_name} ({kind}) {bits} {rate} {ch} container={}", container_hint.unwrap_or("?") ); } /// Max retries for IO/packet-read errors (fatal — network drop, truncated file). const DECODE_MAX_RETRIES: usize = 3; /// Max *consecutive* DecodeErrors before giving up on a file. /// Non-fatal errors like "invalid main_data offset" are silently dropped up to /// this limit so a handful of corrupt MP3 frames never aborts an otherwise /// playable track (VLC-style frame dropping). const MAX_CONSECUTIVE_DECODE_ERRORS: usize = 100; struct SizedDecoder { decoder: Box, current_frame_offset: usize, format: Box, total_duration: Option