refactor(audio): split monolithic audio module into submodules

Move audio engine functionality from a single large file into focused Rust modules while preserving behavior and command surface. This keeps boundaries clearer for future extraction and the upcoming lib.rs split.
This commit is contained in:
Maxim Isaev
2026-05-02 04:07:46 +03:00
committed by cucadmuh
parent 274ac5b3b4
commit c917ee1071
15 changed files with 5550 additions and 5469 deletions
File diff suppressed because it is too large Load Diff
+21
View File
@@ -0,0 +1,21 @@
//! Symphonia codec registry (incl. Opus) and radio decoder factory.
use std::sync::OnceLock;
use symphonia::core::codecs::{CodecRegistry, DecoderOptions};
pub(crate) fn psysonic_codec_registry() -> &'static CodecRegistry {
static REGISTRY: OnceLock<CodecRegistry> = OnceLock::new();
REGISTRY.get_or_init(|| {
let mut registry = CodecRegistry::new();
symphonia::default::register_enabled_codecs(&mut registry);
registry.register_all::<symphonia_adapter_libopus::OpusDecoder>();
registry
})
}
pub(crate) fn try_make_radio_decoder(
params: &symphonia::core::codecs::CodecParameters,
opts: &DecoderOptions,
) -> Result<Box<dyn symphonia::core::codecs::Decoder>, symphonia::core::errors::Error> {
psysonic_codec_registry().make(params, opts)
}
File diff suppressed because it is too large Load Diff
+682
View File
@@ -0,0 +1,682 @@
//! Symphonia `SizedDecoder`, gapless trim, and `build_source` / `build_streaming_source`.
use std::io::{Cursor, Read, Seek};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64};
use std::sync::Arc;
use std::time::Duration;
use rodio::source::UniformSourceIterator;
use rodio::Source;
use symphonia::core::{
audio::{AudioBufferRef, SampleBuffer, SignalSpec},
codecs::{DecoderOptions, CODEC_TYPE_NULL},
formats::{FormatOptions, FormatReader, SeekMode, SeekTo},
io::{MediaSource, MediaSourceStream, MediaSourceStreamOptions},
meta::MetadataOptions,
probe::Hint,
units::{self, Time},
};
use super::codec::{psysonic_codec_registry, try_make_radio_decoder};
use super::sources::*;
// ─── SizedCursorSource — correct byte_len for seekable in-memory sources ──────
//
// rodio's internal ReadSeekSource wraps Cursor<Vec<u8>> 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.
pub(crate) struct SizedCursorSource {
inner: Cursor<Vec<u8>>,
len: u64,
}
impl Read for SizedCursorSource {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.inner.read(buf)
}
}
impl Seek for SizedCursorSource {
fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
self.inner.seek(pos)
}
}
impl MediaSource for SizedCursorSource {
fn is_seekable(&self) -> bool { true }
fn byte_len(&self) -> Option<u64> { 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<Item = i16> + 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.
pub(crate) 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;
pub(crate) struct SizedDecoder {
decoder: Box<dyn symphonia::core::codecs::Decoder>,
current_frame_offset: usize,
format: Box<dyn FormatReader>,
total_duration: Option<Time>,
buffer: SampleBuffer<i16>,
spec: SignalSpec,
/// Counts consecutive DecodeErrors in the hot-path. Reset to 0 on every
/// successfully decoded frame. Used to detect fully undecodable streams.
consecutive_decode_errors: usize,
}
impl SizedDecoder {
pub(crate) fn new(data: Vec<u8>, format_hint: Option<&str>, hi_res: bool) -> Result<Self, String> {
let data_len = data.len() as u64;
let source = SizedCursorSource {
inner: Cursor::new(data),
len: data_len,
};
// Hi-Res: 4 MB read-ahead so Symphonia demuxes fewer Read calls for
// high-bitrate files (88.2 kHz/24-bit FLAC ≈ 1800 kbps).
// Standard: 512 KB is plenty for MP3/AAC — larger buffers waste allocation
// and compete with the playback thread at track start.
let buf_len = if hi_res { 4 * 1024 * 1024 } else { 512 * 1024 };
let mss = MediaSourceStream::new(
Box::new(source) as Box<dyn MediaSource>,
MediaSourceStreamOptions { buffer_len: buf_len },
);
let mut hint = Hint::new();
if let Some(ext) = format_hint {
hint.with_extension(ext);
}
let format_opts = FormatOptions {
// Disable gapless parsing — Symphonia 0.5.5 crashes on `edts` atoms
// present in older iTunes-purchased M4A files.
enable_gapless: false,
..Default::default()
};
let meta_opts = symphonia::core::meta::MetadataOptions {
// Cap embedded cover art at 8 MiB so oversized MJPEG images in
// iTunes M4A files don't choke the parser.
limit_visual_bytes: symphonia::core::meta::Limit::Maximum(8 * 1024 * 1024),
..Default::default()
};
let probed = symphonia::default::get_probe()
.format(&hint, mss, &format_opts, &meta_opts)
.map_err(|e| {
let hint_str = format_hint.unwrap_or("unknown");
// Always print the raw Symphonia error to the terminal for diagnosis.
crate::app_eprintln!("[psysonic] probe failed (hint={hint_str}): {e}");
if e.to_string().to_lowercase().contains("unsupported") {
format!("unsupported format: .{hint_str} files cannot be played (no demuxer)")
} else {
format!("could not open audio stream (.{hint_str}): {e}")
}
})?;
let track = probed.format
.tracks()
.iter()
// Explicitly select only audio tracks: must have a valid codec and a
// sample_rate. This skips MJPEG cover-art streams that iTunes M4A
// files embed as a secondary video track.
.find(|t| {
t.codec_params.codec != CODEC_TYPE_NULL
&& t.codec_params.sample_rate.is_some()
})
.ok_or_else(|| {
crate::app_eprintln!("[psysonic] no audio track found among {} tracks", probed.format.tracks().len());
"no playable audio track found in file".to_string()
})?;
let track_id = track.id;
let total_duration = track.codec_params.time_base
.zip(track.codec_params.n_frames)
.map(|(base, frames)| base.calc_time(frames));
log_codec_resolution("bytes", &track.codec_params, format_hint);
let mut decoder = psysonic_codec_registry()
.make(&track.codec_params, &DecoderOptions::default())
.map_err(|e| {
crate::app_eprintln!("[psysonic] codec init failed: {e}");
if e.to_string().to_lowercase().contains("unsupported") {
"unsupported codec: no decoder available for this audio format".to_string()
} else {
format!("failed to initialise audio decoder: {e}")
}
})?;
let mut format = probed.format;
// Decode the first packet to initialise spec + buffer.
// DecodeErrors (e.g. "invalid main_data offset") are non-fatal: drop the
// frame and try the next packet up to MAX_CONSECUTIVE_DECODE_ERRORS times.
let mut decode_errors: usize = 0;
let decoded = loop {
let packet = match format.next_packet() {
Ok(p) => p,
Err(symphonia::core::errors::Error::IoError(_)) => {
break decoder.last_decoded();
}
Err(e) => {
crate::app_eprintln!("[psysonic] next_packet error: {e}");
return Err(format!("could not read audio data: {e}"));
}
};
if packet.track_id() != track_id {
crate::app_eprintln!("[psysonic] skipping packet for track {} (want {})", packet.track_id(), track_id);
continue;
}
match decoder.decode(&packet) {
Ok(decoded) => break decoded,
Err(symphonia::core::errors::Error::DecodeError(ref msg)) => {
decode_errors += 1;
crate::app_eprintln!("[psysonic] init: dropped corrupt frame #{decode_errors}: {msg}");
if decode_errors >= MAX_CONSECUTIVE_DECODE_ERRORS {
return Err("too many consecutive decode errors during init — file may be corrupt".into());
}
}
Err(e) => {
crate::app_eprintln!("[psysonic] fatal decode error: {e}");
return Err(format!("audio decode error: {e}"));
}
}
};
let spec = decoded.spec().to_owned();
let buffer = Self::make_buffer(decoded, &spec);
Ok(SizedDecoder {
decoder,
current_frame_offset: 0,
format,
total_duration,
buffer,
spec,
consecutive_decode_errors: 0,
})
}
/// Build a decoder from any `MediaSource` (e.g. track-stream or radio).
/// Uses `enable_gapless: false` — live streams are not seekable; gapless
/// trimming requires seeking to read the LAME/iTunSMPB end-padding info.
pub(crate) fn new_streaming(
media: Box<dyn MediaSource>,
format_hint: Option<&str>,
source_tag: &str,
) -> Result<Self, String> {
// Larger read-ahead buffer for the live streaming SPSC consumer — reduces
// read() call frequency into the ring buffer, easing I/O spikes.
let mss = MediaSourceStream::new(media, MediaSourceStreamOptions { buffer_len: 512 * 1024 });
let mut hint = Hint::new();
if let Some(ext) = format_hint { hint.with_extension(ext); }
let format_opts = FormatOptions { enable_gapless: false, ..Default::default() };
let probed = symphonia::default::get_probe()
.format(&hint, mss, &format_opts, &MetadataOptions::default())
.map_err(|e| format!("{source_tag}: format probe failed: {e}"))?;
let track = probed.format.tracks().iter()
.find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
.ok_or_else(|| format!("{source_tag}: no audio track found"))?;
let track_id = track.id;
log_codec_resolution(source_tag, &track.codec_params, format_hint);
// Live streams have no known total frame count → total_duration = None.
let total_duration = None;
let mut decoder = try_make_radio_decoder(&track.codec_params, &DecoderOptions::default())
.map_err(|e| format!("{source_tag}: codec init failed: {e}"))?;
let mut format = probed.format;
let mut errors = 0usize;
let decoded = loop {
let packet = match format.next_packet() {
Ok(p) => p,
Err(_) => break decoder.last_decoded(),
};
if packet.track_id() != track_id { continue; }
match decoder.decode(&packet) {
Ok(d) => break d,
Err(symphonia::core::errors::Error::DecodeError(ref msg)) => {
errors += 1;
crate::app_eprintln!("[psysonic] {source_tag} init: dropped corrupt frame #{errors}: {msg}");
if errors >= MAX_CONSECUTIVE_DECODE_ERRORS {
return Err(format!("{source_tag}: too many consecutive decode errors"));
}
}
Err(e) => return Err(format!("{source_tag}: decode error: {e}")),
}
};
let spec = decoded.spec().to_owned();
let buffer = Self::make_buffer(decoded, &spec);
Ok(SizedDecoder { decoder, current_frame_offset: 0, format, total_duration, buffer, spec, consecutive_decode_errors: 0 })
}
#[inline]
fn make_buffer(decoded: AudioBufferRef, spec: &SignalSpec) -> SampleBuffer<i16> {
let duration = units::Duration::from(decoded.capacity() as u64);
let mut buffer = SampleBuffer::<i16>::new(duration, *spec);
buffer.copy_interleaved_ref(decoded);
buffer
}
/// Refine position after a coarse seek — decode packets until we reach the
/// exact requested timestamp.
fn refine_position(
&mut self,
seek_res: symphonia::core::formats::SeekedTo,
) -> Result<(), String> {
let mut samples_to_pass = seek_res.required_ts - seek_res.actual_ts;
let packet = loop {
let candidate = self.format.next_packet()
.map_err(|e| format!("refine seek: {e}"))?;
if candidate.dur() > samples_to_pass {
break candidate;
}
samples_to_pass -= candidate.dur();
};
let mut decoded = self.decoder.decode(&packet);
for _ in 0..DECODE_MAX_RETRIES {
if decoded.is_err() {
let p = self.format.next_packet()
.map_err(|e| format!("refine retry: {e}"))?;
decoded = self.decoder.decode(&p);
}
}
let decoded = decoded.map_err(|e| format!("refine decode: {e}"))?;
decoded.spec().clone_into(&mut self.spec);
self.buffer = Self::make_buffer(decoded, &self.spec);
self.current_frame_offset = samples_to_pass as usize * self.spec.channels.count();
Ok(())
}
}
impl Iterator for SizedDecoder {
type Item = i16;
#[inline]
fn next(&mut self) -> Option<i16> {
if self.current_frame_offset >= self.buffer.len() {
// Loop until a decodable packet is found or the stream ends.
// DecodeErrors (e.g. MP3 "invalid main_data offset") are non-fatal:
// drop the frame and advance to the next packet. IO errors and a
// clean end-of-stream both terminate the iterator normally.
loop {
let packet = self.format.next_packet().ok()?;
match self.decoder.decode(&packet) {
Ok(decoded) => {
self.consecutive_decode_errors = 0;
decoded.spec().clone_into(&mut self.spec);
self.buffer = Self::make_buffer(decoded, &self.spec);
self.current_frame_offset = 0;
break;
}
Err(symphonia::core::errors::Error::DecodeError(ref msg)) => {
#[cfg(not(debug_assertions))]
let _ = msg;
self.consecutive_decode_errors += 1;
// Log sparingly: first drop, then every 10th to avoid spam.
if self.consecutive_decode_errors == 1
|| self.consecutive_decode_errors % 10 == 0
{
crate::app_deprintln!(
"[psysonic] dropped corrupt frame #{}: {msg}",
self.consecutive_decode_errors
);
}
if self.consecutive_decode_errors >= MAX_CONSECUTIVE_DECODE_ERRORS {
crate::app_deprintln!(
"[psysonic] {MAX_CONSECUTIVE_DECODE_ERRORS} consecutive decode \
failures — stream appears unrecoverable, stopping"
);
return None;
}
// continue → fetch next packet
}
Err(_) => return None, // IO error or fatal codec error → end of stream
}
}
}
let sample = *self.buffer.samples().get(self.current_frame_offset)?;
self.current_frame_offset += 1;
Some(sample)
}
}
impl Source for SizedDecoder {
#[inline]
fn current_frame_len(&self) -> Option<usize> {
Some(self.buffer.samples().len())
}
#[inline]
fn channels(&self) -> u16 {
self.spec.channels.count() as u16
}
#[inline]
fn sample_rate(&self) -> u32 {
self.spec.rate
}
#[inline]
fn total_duration(&self) -> Option<Duration> {
self.total_duration.map(|Time { seconds, frac }| {
Duration::new(seconds, (frac * 1_000_000_000.0) as u32)
})
}
fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
let seek_beyond_end = self
.total_duration()
.is_some_and(|dur| dur.saturating_sub(pos).as_millis() < 1);
let time: Time = if seek_beyond_end {
let t = self.total_duration.unwrap_or(pos.as_secs_f64().into());
// Step back a tiny bit — some demuxers can't seek to the exact end.
let mut secs = t.seconds;
let mut frac = t.frac - 0.0001;
if frac < 0.0 {
secs = secs.saturating_sub(1);
frac = 1.0 - frac;
}
Time { seconds: secs, frac }
} else {
pos.as_secs_f64().into()
};
let to_skip = self.current_frame_offset % self.channels() as usize;
let seek_res = self
.format
.seek(SeekMode::Accurate, SeekTo::Time { time, track_id: None })
.map_err(|e| rodio::source::SeekError::Other(
Box::new(std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))
))?;
self.refine_position(seek_res)
.map_err(|e| rodio::source::SeekError::Other(
Box::new(std::io::Error::new(std::io::ErrorKind::Other, e))
))?;
self.current_frame_offset += to_skip;
Ok(())
}
}
// ─── Encoder-gap trimming (iTunSMPB) ─────────────────────────────────────────
//
// MP3/AAC encoders prepend an "encoder delay" (typically 5762112 silent
// samples for LAME) and append end-padding to fill the final frame.
// iTunes embeds the exact counts in an ID3v2 COMM frame with description
// "iTunSMPB". Format: " 00000000 DELAY PADDING TOTAL ..." (space-separated hex)
//
// Parsing strategy: scan raw bytes for the ASCII marker, then extract the
// first whitespace-separated hex tokens after it.
pub(crate) struct GaplessInfo {
delay_samples: u64,
total_valid_samples: Option<u64>,
}
impl Default for GaplessInfo {
fn default() -> Self {
Self { delay_samples: 0, total_valid_samples: None }
}
}
pub(crate) fn find_subsequence(data: &[u8], needle: &[u8]) -> Option<usize> {
data.windows(needle.len()).position(|w| w == needle)
}
pub(crate) fn parse_gapless_info(data: &[u8]) -> GaplessInfo {
let pos = match find_subsequence(data, b"iTunSMPB") {
Some(p) => p,
None => return GaplessInfo::default(),
};
// In M4A/iTunes files the key is followed by a binary 'data' atom header
// (16 bytes: size[4] + "data"[4] + type_flags[4] + locale[4]) before the
// actual value string. Search for the " 00000000 " sentinel that every
// iTunSMPB value starts with to locate the true start of the text.
let search_end = data.len().min(pos + 8 + 128);
let search_window = &data[pos + 8..search_end];
let value_start = find_subsequence(search_window, b" 00000000 ")
.map(|off| pos + 8 + off)
.unwrap_or(pos + 8);
let tail = &data[value_start..data.len().min(value_start + 256)];
let text: String = tail.iter()
.map(|&b| b as char)
.filter(|c| c.is_ascii_hexdigit() || *c == ' ')
.collect();
let parts: Vec<&str> = text.split_whitespace().collect();
// parts[0] = "00000000", parts[1] = delay, parts[2] = padding, parts[3] = total
if parts.len() < 3 {
return GaplessInfo::default();
}
let delay = u64::from_str_radix(parts.get(1).unwrap_or(&"0"), 16).unwrap_or(0);
let padding = u64::from_str_radix(parts.get(2).unwrap_or(&"0"), 16).unwrap_or(0);
let total_raw = parts.get(3).and_then(|s| u64::from_str_radix(s, 16).ok());
let total_valid = total_raw.map(|t| t).filter(|&t| t > 0).or_else(|| {
// Derive from delay + padding if total not available:
// Not possible without knowing total encoded samples, so just use None.
let _ = padding;
None
});
GaplessInfo { delay_samples: delay, total_valid_samples: total_valid }
}
/// Result of build_source: the fully-wrapped source plus metadata and control Arcs.
pub(crate) struct BuiltSource {
pub(crate) source: CountingSource<NotifyingSource<TriggeredFadeOut<EqualPowerFadeIn<EqSource<DynSource>>>>>,
pub(crate) duration_secs: f64,
pub(crate) output_rate: u32,
pub(crate) output_channels: u16,
/// Trigger for the sample-level crossfade fade-out.
pub(crate) fadeout_trigger: Arc<AtomicBool>,
/// Total samples for the fade-out (set before triggering).
pub(crate) fadeout_samples: Arc<AtomicU64>,
}
/// Build a fully-prepared playback source:
/// decode → trim → resample → EQ → fade-in → triggered-fade-out → notify → count
///
/// `fade_in_dur`:
/// • `Duration::ZERO` — unity gain; used for gapless chain (no click)
/// • `Duration::from_millis(5)` — micro-fade; used for hard cuts (anti-click)
/// • `Duration::from_secs_f32(cf)` — full equal-power fade-in for crossfade
///
/// `sample_counter`: atomic counter incremented per sample for drift-free position.
/// `target_rate`: canonical output sample rate for resampling (0 = no resampling).
/// `format_hint`: optional file extension (e.g. "flac", "mp3") to help symphonia probe.
pub(crate) fn build_source(
data: Vec<u8>,
duration_hint: f64,
eq_gains: Arc<[AtomicU32; 10]>,
eq_enabled: Arc<AtomicBool>,
eq_pre_gain: Arc<AtomicU32>,
done_flag: Arc<AtomicBool>,
fade_in_dur: Duration,
sample_counter: Arc<AtomicU64>,
target_rate: u32,
format_hint: Option<&str>,
hi_res: bool,
) -> Result<BuiltSource, String> {
let gapless = parse_gapless_info(&data);
let decoder = SizedDecoder::new(data, format_hint, hi_res)?;
let sample_rate = decoder.sample_rate();
let channels = decoder.channels();
// Determine effective duration.
// Prefer hint from Subsonic API (reliable) over decoder (unreliable for VBR MP3).
let effective_dur = if duration_hint > 1.0 {
duration_hint
} else {
decoder.total_duration()
.map(|d| d.as_secs_f64())
.unwrap_or(duration_hint)
};
// Apply encoder-delay trim and optional end-padding trim,
// then resample to the canonical target rate if needed.
let dyn_src: DynSource = if gapless.delay_samples > 0 || gapless.total_valid_samples.is_some() {
let delay_dur = Duration::from_secs_f64(
gapless.delay_samples as f64 / sample_rate as f64
);
let base = decoder.convert_samples::<f32>().skip_duration(delay_dur);
if let Some(total) = gapless.total_valid_samples {
let valid_dur = Duration::from_secs_f64(total as f64 / sample_rate as f64);
let trimmed = base.take_duration(valid_dur);
if target_rate > 0 && sample_rate != target_rate {
DynSource::new(UniformSourceIterator::new(trimmed, channels, target_rate))
} else {
DynSource::new(trimmed)
}
} else {
if target_rate > 0 && sample_rate != target_rate {
DynSource::new(UniformSourceIterator::new(base, channels, target_rate))
} else {
DynSource::new(base)
}
}
} else {
let converted = decoder.convert_samples::<f32>();
if target_rate > 0 && sample_rate != target_rate {
DynSource::new(UniformSourceIterator::new(converted, channels, target_rate))
} else {
DynSource::new(converted)
}
};
let output_rate = if target_rate > 0 && sample_rate != target_rate { target_rate } else { sample_rate };
let fadeout_trigger = Arc::new(AtomicBool::new(false));
let fadeout_samples = Arc::new(AtomicU64::new(0));
let eq_src = EqSource::new(dyn_src, eq_gains, eq_enabled, eq_pre_gain);
let fade_in = EqualPowerFadeIn::new(eq_src, fade_in_dur);
let fade_out = TriggeredFadeOut::new(fade_in, fadeout_trigger.clone(), fadeout_samples.clone());
let notifying = NotifyingSource::new(fade_out, done_flag);
let counting = CountingSource::new(notifying, sample_counter);
Ok(BuiltSource {
source: counting,
duration_secs: effective_dur,
output_rate,
output_channels: channels,
fadeout_trigger,
fadeout_samples,
})
}
/// Streaming variant of `build_source`: uses a live `SizedDecoder` source
/// (non-seekable) and skips iTunSMPB parsing, but preserves the same EQ/fade/
/// counting wrappers and output metadata.
pub(crate) fn build_streaming_source(
decoder: SizedDecoder,
duration_hint: f64,
eq_gains: Arc<[AtomicU32; 10]>,
eq_enabled: Arc<AtomicBool>,
eq_pre_gain: Arc<AtomicU32>,
done_flag: Arc<AtomicBool>,
fade_in_dur: Duration,
sample_counter: Arc<AtomicU64>,
target_rate: u32,
) -> Result<BuiltSource, String> {
let sample_rate = decoder.sample_rate();
let channels = decoder.channels();
// For streaming starts prefer server-provided duration when available.
let effective_dur = if duration_hint > 1.0 {
duration_hint
} else {
decoder
.total_duration()
.map(|d| d.as_secs_f64())
.unwrap_or(duration_hint)
};
let converted = decoder.convert_samples::<f32>();
let dyn_src: DynSource = if target_rate > 0 && sample_rate != target_rate {
DynSource::new(UniformSourceIterator::new(converted, channels, target_rate))
} else {
DynSource::new(converted)
};
let output_rate = if target_rate > 0 && sample_rate != target_rate {
target_rate
} else {
sample_rate
};
let fadeout_trigger = Arc::new(AtomicBool::new(false));
let fadeout_samples = Arc::new(AtomicU64::new(0));
let eq_src = EqSource::new(dyn_src, eq_gains, eq_enabled, eq_pre_gain);
let fade_in = EqualPowerFadeIn::new(eq_src, fade_in_dur);
let fade_out = TriggeredFadeOut::new(fade_in, fadeout_trigger.clone(), fadeout_samples.clone());
let notifying = NotifyingSource::new(fade_out, done_flag);
let counting = CountingSource::new(notifying, sample_counter);
Ok(BuiltSource {
source: counting,
duration_secs: effective_dur,
output_rate,
output_channels: channels,
fadeout_trigger,
fadeout_samples,
})
}
+88
View File
@@ -0,0 +1,88 @@
//! Output device enumeration with suppressed ALSA stderr noise.
#[cfg(unix)]
use libc;
// `rodio::cpal` is referenced from the included body.
/// ALSA probes noisy plugins during device queries — suppress stderr on Unix.
#[cfg(unix)]
pub(crate) fn with_suppressed_alsa_stderr<R>(f: impl FnOnce() -> R) -> R {
struct StderrGuard(i32);
impl Drop for StderrGuard {
fn drop(&mut self) {
unsafe { libc::dup2(self.0, 2); libc::close(self.0); }
}
}
let _guard = unsafe {
let saved = libc::dup(2);
let devnull = libc::open(b"/dev/null\0".as_ptr() as *const libc::c_char, libc::O_WRONLY);
libc::dup2(devnull, 2);
libc::close(devnull);
StderrGuard(saved)
};
f()
}
#[cfg(not(unix))]
#[inline]
pub(crate) fn with_suppressed_alsa_stderr<R>(f: impl FnOnce() -> R) -> R {
f()
}
pub(crate) fn enumerate_output_device_names() -> Vec<String> {
use rodio::cpal::traits::{DeviceTrait, HostTrait};
with_suppressed_alsa_stderr(|| {
let host = rodio::cpal::default_host();
host.output_devices()
.map(|iter| iter.filter_map(|d| d.name().ok()).collect())
.unwrap_or_default()
})
}
/// Linux ALSA-style cpal names: same physical sink can appear with different suffixes;
/// busy devices are sometimes omitted from `output_devices()` while playback works.
#[cfg(target_os = "linux")]
pub(crate) fn linux_alsa_sink_fingerprint(name: &str) -> Option<(String, String, u32)> {
const IFACES: &[&str] = &[
"hdmi", "hw", "plughw", "sysdefault", "iec958", "front", "dmix", "surround40",
"surround51", "surround71",
];
let colon = name.find(':')?;
let iface = name[..colon].to_ascii_lowercase();
if !IFACES.iter().any(|&i| i == iface.as_str()) {
return None;
}
let card = name.split("CARD=").nth(1)?.split(',').next()?.to_string();
let dev = name
.split("DEV=")
.nth(1)
.and_then(|s| s.split(',').next())
.and_then(|s| s.parse().ok())
.unwrap_or(0);
Some((iface, card, dev))
}
#[cfg(not(target_os = "linux"))]
#[inline]
pub(crate) fn linux_alsa_sink_fingerprint(_name: &str) -> Option<(String, String, u32)> {
None
}
pub(crate) fn output_devices_logically_same(a: &str, b: &str) -> bool {
if a == b {
return true;
}
match (
linux_alsa_sink_fingerprint(a),
linux_alsa_sink_fingerprint(b),
) {
(Some(fa), Some(fb)) => fa == fb,
_ => false,
}
}
/// True if `pinned` is the same sink as some entry (exact or Linux ALSA logical match).
pub(crate) fn output_enumeration_includes_pinned(available: &[String], pinned: &str) -> bool {
available
.iter()
.any(|d| output_devices_logically_same(d, pinned))
}
+150
View File
@@ -0,0 +1,150 @@
//! Poll default output device and pinned-device presence; reopen stream when needed.
use std::sync::atomic::Ordering;
use std::time::Duration;
use tauri::Emitter;
use super::engine::AudioEngine;
pub fn start_device_watcher(engine: &AudioEngine, app: tauri::AppHandle) {
let reopen_tx = engine.stream_reopen_tx.clone();
let stream_handle = engine.stream_handle.clone();
let stream_rate = engine.stream_sample_rate.clone();
let current = engine.current.clone();
let fading_out = engine.fading_out_sink.clone();
let selected_device = engine.selected_device.clone();
tauri::async_runtime::spawn(async move {
let mut last_default: Option<String> = tauri::async_runtime::spawn_blocking(|| {
use rodio::cpal::traits::{DeviceTrait, HostTrait};
rodio::cpal::default_host()
.default_output_device()
.and_then(|d| d.name().ok())
}).await.unwrap_or(None);
// macOS/Windows: consecutive polls where a pinned device is absent from cpal's list.
#[cfg(not(target_os = "linux"))]
let mut pinned_miss_count: u32 = 0;
loop {
tokio::time::sleep(Duration::from_secs(3)).await;
// Enumerate all available output devices and the current default.
// Suppress stderr on Unix to avoid ALSA probing noise (JACK, OSS, dmix).
let (current_default, available) = tauri::async_runtime::spawn_blocking(|| {
use rodio::cpal::traits::{DeviceTrait, HostTrait};
#[cfg(unix)]
let _guard = unsafe {
struct StderrGuard(i32);
impl Drop for StderrGuard {
fn drop(&mut self) { unsafe { libc::dup2(self.0, 2); libc::close(self.0); } }
}
let saved = libc::dup(2);
let devnull = libc::open(b"/dev/null\0".as_ptr() as *const libc::c_char, libc::O_WRONLY);
libc::dup2(devnull, 2);
libc::close(devnull);
StderrGuard(saved)
};
let host = rodio::cpal::default_host();
let default = host.default_output_device().and_then(|d| d.name().ok());
let available: Vec<String> = host
.output_devices()
.map(|iter| iter.filter_map(|d| d.name().ok()).collect())
.unwrap_or_default();
(default, available)
}).await.unwrap_or((None, vec![]));
// Empty list almost always means a transient enumeration failure, not
// that every output device vanished. Treating it as "pinned missing"
// caused false audio:device-reset (UI jumped back to system default)
// when switching to external USB / class-compliant interfaces.
if available.is_empty() {
continue;
}
let pinned = selected_device.lock().unwrap().clone();
#[cfg(target_os = "linux")]
if pinned.is_some() {
// Do not infer "unplugged" from `output_devices()` when a device is pinned.
// ALSA/cpal often omit the active HDMI/USB sink from enumeration for the
// whole session — any miss counter eventually tripped audio:device-reset.
// Clearing the pin is left to the user (Settings → System Default) or
// to a future explicit error signal from the output stream.
continue;
}
// ── Case 2 (non-Linux): pinned device disappeared from enumeration ─
#[cfg(not(target_os = "linux"))]
if let Some(ref dev_name) = pinned {
if !output_enumeration_includes_pinned(&available, dev_name) {
pinned_miss_count += 1;
if pinned_miss_count < 3 {
continue;
}
crate::app_eprintln!("[psysonic] device-watcher: pinned device '{dev_name}' disconnected, falling back to system default");
pinned_miss_count = 0;
*selected_device.lock().unwrap() = None;
tokio::time::sleep(Duration::from_millis(500)).await;
let rate = stream_rate.load(Ordering::Relaxed);
let reopen_tx2 = reopen_tx.clone();
let new_handle = tauri::async_runtime::spawn_blocking(move || {
let (reply_tx, reply_rx) =
std::sync::mpsc::sync_channel::<rodio::OutputStreamHandle>(0);
if reopen_tx2.send((rate, false, None, reply_tx)).is_err() {
return None;
}
reply_rx.recv_timeout(Duration::from_secs(5)).ok()
}).await.unwrap_or(None);
if let Some(handle) = new_handle {
*stream_handle.lock().unwrap() = handle;
if let Some(s) = current.lock().unwrap().sink.take() { s.stop(); }
if let Some(s) = fading_out.lock().unwrap().take() { s.stop(); }
app.emit("audio:device-reset", ()).ok();
}
last_default = current_default;
} else {
pinned_miss_count = 0;
}
continue;
}
// ── Case 1: no pinned device, system default changed ──────────────
if current_default == last_default {
continue;
}
last_default = current_default.clone();
let Some(_new_name) = current_default else { continue };
// Debounce: give the OS time to finish configuring the new device.
tokio::time::sleep(Duration::from_millis(500)).await;
let rate = stream_rate.load(Ordering::Relaxed);
let reopen_tx2 = reopen_tx.clone();
let new_handle = tauri::async_runtime::spawn_blocking(move || {
let (reply_tx, reply_rx) =
std::sync::mpsc::sync_channel::<rodio::OutputStreamHandle>(0);
if reopen_tx2.send((rate, false, None, reply_tx)).is_err() {
return None;
}
reply_rx.recv_timeout(Duration::from_secs(5)).ok()
}).await.unwrap_or(None);
let Some(handle) = new_handle else {
crate::app_eprintln!("[psysonic] device-watcher: stream reopen timed out");
continue;
};
*stream_handle.lock().unwrap() = handle;
if let Some(s) = current.lock().unwrap().sink.take() { s.stop(); }
if let Some(s) = fading_out.lock().unwrap().take() { s.stop(); }
app.emit("audio:device-changed", ()).ok();
}
});
}
+429
View File
@@ -0,0 +1,429 @@
//! `AudioEngine` / `AudioCurrent`, stream thread, and HTTP client refresh.
#[cfg(unix)]
use libc;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant};
use rodio::Sink;
use tauri::{AppHandle, Manager};
use super::state::{ChainedInfo, PreloadedTrack};
pub struct AudioEngine {
pub stream_handle: Arc<std::sync::Mutex<rodio::OutputStreamHandle>>,
/// Sample rate the output stream was last opened at (updated on every re-open).
pub stream_sample_rate: Arc<AtomicU32>,
/// The rate the device was opened at on cold start — used to restore the
/// stream when Hi-Res is toggled off while a hi-res rate is active.
pub device_default_rate: u32,
/// Sends `(desired_rate, is_hi_res, device_name, reply_tx)` to the audio-stream
/// thread to re-open the output device. `device_name = None` → system default.
pub stream_reopen_tx: std::sync::mpsc::SyncSender<(u32, bool, Option<String>, std::sync::mpsc::SyncSender<rodio::OutputStreamHandle>)>,
/// User-selected output device name (None = follow system default).
pub selected_device: Arc<Mutex<Option<String>>>,
pub current: Arc<Mutex<AudioCurrent>>,
/// Monotonically incremented on each audio_play (non-chain) / audio_stop call.
pub generation: Arc<AtomicU64>,
pub http_client: Arc<RwLock<reqwest::Client>>,
pub eq_gains: Arc<[AtomicU32; 10]>,
pub eq_enabled: Arc<AtomicBool>,
pub eq_pre_gain: Arc<AtomicU32>,
pub(crate) preloaded: Arc<Mutex<Option<PreloadedTrack>>>,
/// Last fully downloaded manual-stream track bytes (same playback identity),
/// used to recover seek/replay without waiting for network again.
pub(crate) stream_completed_cache: Arc<Mutex<Option<PreloadedTrack>>>,
/// True when the currently playing source supports seeking (in-memory bytes
/// or `RangedHttpSource`); false for the legacy non-seekable streaming
/// fallback (`AudioStreamReader`). `audio_seek` rejects with a "not
/// seekable" error when false so the frontend restart-fallback can engage.
pub(crate) current_is_seekable: Arc<AtomicBool>,
pub crossfade_enabled: Arc<AtomicBool>,
pub crossfade_secs: Arc<AtomicU32>,
pub fading_out_sink: Arc<Mutex<Option<Arc<Sink>>>>,
/// When true, audio_play chains sources to the existing Sink instead of
/// creating a new one, achieving sample-accurate gapless transitions.
pub gapless_enabled: Arc<AtomicBool>,
/// 0=off, 1=replaygain, 2=loudness (future runtime loudness engine).
pub normalization_engine: Arc<AtomicU32>,
/// Target loudness in LUFS for loudness engine (future use).
pub normalization_target_lufs: Arc<AtomicU32>,
/// Extra attenuation (dB) when no loudness DB row exists at decode bind; also seeds streaming heuristics (Settings).
pub loudness_pre_analysis_attenuation_db: Arc<AtomicU32>,
/// Info about the next-up chained track (gapless mode).
/// The progress task reads this when `current_source_done` fires.
pub(crate) chained_info: Arc<Mutex<Option<ChainedInfo>>>,
/// Atomic sample counter — incremented by CountingSource in the audio thread.
/// Progress task reads this for drift-free position tracking.
pub samples_played: Arc<AtomicU64>,
/// Sample rate of the currently playing source (for samples → seconds).
pub current_sample_rate: Arc<AtomicU32>,
/// Channel count of the currently playing source.
pub current_channels: Arc<AtomicU32>,
/// Instant (as nanos since UNIX epoch via Instant hack) of the last gapless
/// auto-advance. Commands arriving within 500 ms are rejected as ghost commands.
pub gapless_switch_at: Arc<AtomicU64>,
/// Active radio session state. None for regular (non-radio) tracks.
/// Dropping the value aborts the HTTP download task via RadioLiveState::Drop.
pub(crate) radio_state: Mutex<Option<crate::audio::stream::RadioLiveState>>,
/// URL last committed to `AudioCurrent` — used so `audio_update_replay_gain` can
/// resolve LUFS / startup trim when the frontend passes `loudnessGainDb: null`
/// (otherwise `compute_gain` would treat that as unity gain and playback "jumps").
pub(crate) current_playback_url: Arc<Mutex<Option<String>>>,
/// Subsonic song id last passed from JS with `audio_play` (trimmed). Used
/// for loudness/waveform cache when the URL is `psysonic-local://…`.
pub(crate) current_analysis_track_id: Arc<Mutex<Option<String>>>,
/// While a `RangedHttpSource` download task is filling the buffer for this
/// `(track_id, play_generation)`, skip `analysis_enqueue_seed_from_url` for the
/// same id — otherwise a parallel full GET + Symphonia competes with playback
/// decode (ALSA underruns). The ranged task clears this on exit; `gen` avoids a
/// late drop clearing a newer play of the same track.
pub(crate) ranged_loudness_seed_hold: Arc<Mutex<Option<(String, u64)>>>,
/// Secondary sink dedicated to track previews. Runs on the same `OutputStream`
/// as the main sink (rodio mixes both internally) so we don't open a second
/// device handle — important on ALSA-exclusive hardware.
pub(crate) preview_sink: Arc<Mutex<Option<Arc<Sink>>>>,
/// Cancel token for the active preview. Bumped on every `audio_preview_play`
/// and `audio_preview_stop` so that orphan timer/progress tasks bail out.
pub(crate) preview_gen: Arc<AtomicU64>,
/// True when `audio_preview_play` paused the main sink and should resume it
/// on preview end. False if the main sink was already paused (or empty).
pub(crate) preview_main_resume: Arc<AtomicBool>,
/// Subsonic song id of the currently playing preview. Echoed back in
/// `audio:preview-end` so the frontend can clear UI state for that row.
pub(crate) preview_song_id: Arc<Mutex<Option<String>>>,
}
pub struct AudioCurrent {
pub sink: Option<Arc<Sink>>,
pub duration_secs: f64,
pub seek_offset: f64,
pub play_started: Option<Instant>,
pub paused_at: Option<f64>,
pub replay_gain_linear: f32,
pub base_volume: f32,
/// Crossfade: trigger for sample-level fade-out of the current source.
pub fadeout_trigger: Option<Arc<AtomicBool>>,
/// Crossfade: total fade samples (set before triggering).
pub fadeout_samples: Option<Arc<AtomicU64>>,
}
impl AudioCurrent {
pub fn position(&self) -> f64 {
if let Some(p) = self.paused_at {
return p;
}
if let Some(t) = self.play_started {
let elapsed = t.elapsed().as_secs_f64();
(self.seek_offset + elapsed).min(self.duration_secs.max(0.001))
} else {
self.seek_offset
}
}
}
/// Open an output device at `desired_rate` Hz (0 = device default).
///
/// `device_name`: exact name from `audio_list_devices`. `None` → system default.
/// Falls back to the system default if the named device is not found.
///
/// Resolution order:
/// 1. Exact rate match in the device's supported config ranges.
/// 2. Highest available rate (for hardware that doesn't support the source rate).
/// 3. Device default.
/// 4. System default (last resort).
///
/// Returns `(OutputStream, OutputStreamHandle, actual_sample_rate)`.
fn open_stream_for_device_and_rate(device_name: Option<&str>, desired_rate: u32) -> (rodio::OutputStream, rodio::OutputStreamHandle, u32) {
use rodio::cpal::traits::{DeviceTrait, HostTrait};
// Suppress ALSA stderr noise while enumerating devices on Unix.
#[cfg(unix)]
let _guard = unsafe {
struct StderrGuard(i32);
impl Drop for StderrGuard {
fn drop(&mut self) { unsafe { libc::dup2(self.0, 2); libc::close(self.0); } }
}
let saved = libc::dup(2);
let devnull = libc::open(b"/dev/null\0".as_ptr() as *const libc::c_char, libc::O_WRONLY);
libc::dup2(devnull, 2);
libc::close(devnull);
StderrGuard(saved)
};
let host = rodio::cpal::default_host();
// Resolve the target device: explicit name first, then (on Linux) prefer
// a "pipewire" or "pulse" ALSA alias before falling back to cpal's system
// default. On PipeWire-based distros the raw ALSA `default` alias can
// route to a null sink at app-start (issue #234 on Debian 13): the stream
// opens cleanly, progress ticks run, no audio reaches the user. The
// named-alias path goes through pipewire-alsa's real sink and just works.
// On systems where neither alias exists (pure ALSA, macOS, Windows),
// `find_by_name` returns None and we drop through to `default_output_device`
// unchanged — no regression.
let find_by_name = |name: &str| -> Option<_> {
host.output_devices().ok()?.find(|d| d.name().ok().as_deref() == Some(name))
};
let device = device_name
.and_then(find_by_name)
.or_else(|| {
#[cfg(target_os = "linux")]
{ find_by_name("pipewire").or_else(|| find_by_name("pulse")) }
#[cfg(not(target_os = "linux"))]
{ None }
})
.or_else(|| host.default_output_device());
if let Some(device) = device {
if desired_rate > 0 {
if let Ok(supported) = device.supported_output_configs() {
let configs: Vec<_> = supported.collect();
// 1. Exact rate match — prefer more channels (stereo > mono).
let exact = configs.iter()
.filter(|c| {
c.min_sample_rate().0 <= desired_rate
&& desired_rate <= c.max_sample_rate().0
})
.max_by_key(|c| c.channels());
if let Some(cfg) = exact {
let config = cfg.clone()
.with_sample_rate(rodio::cpal::SampleRate(desired_rate));
if let Ok((stream, handle)) =
rodio::OutputStream::try_from_device_config(&device, config)
{
crate::app_eprintln!("[psysonic] audio stream opened at {} Hz (exact)", desired_rate);
return (stream, handle, desired_rate);
}
}
// 2. No exact match — use the highest supported rate.
let best = configs.iter()
.max_by_key(|c| c.max_sample_rate().0);
if let Some(cfg) = best {
let rate = cfg.max_sample_rate().0;
let config = cfg.clone()
.with_sample_rate(rodio::cpal::SampleRate(rate));
if let Ok((stream, handle)) =
rodio::OutputStream::try_from_device_config(&device, config)
{
crate::app_eprintln!(
"[psysonic] audio stream opened at {} Hz (highest, wanted {})",
rate, desired_rate
);
return (stream, handle, rate);
}
}
}
}
// 3. Device default.
if let Ok((stream, handle)) = rodio::OutputStream::try_from_device(&device) {
let rate = device
.default_output_config()
.map(|c| c.sample_rate().0)
.unwrap_or(44100);
crate::app_eprintln!("[psysonic] audio stream opened at {} Hz (device default)", rate);
return (stream, handle, rate);
}
}
// 4. Last resort: system default.
crate::app_eprintln!("[psysonic] audio stream falling back to system default");
let (stream, handle) = rodio::OutputStream::try_default()
.expect("cannot open any audio output device");
let rate = rodio::cpal::default_host()
.default_output_device()
.and_then(|d| d.default_output_config().ok())
.map(|c| c.sample_rate().0)
.unwrap_or(44100);
(stream, handle, rate)
}
pub fn create_engine() -> (AudioEngine, std::thread::JoinHandle<()>) {
// macOS: request a smaller CoreAudio buffer to reduce output latency.
#[cfg(target_os = "macos")]
{
if std::env::var("COREAUDIO_BUFFER_SIZE").is_err() {
std::env::set_var("COREAUDIO_BUFFER_SIZE", "512");
}
}
// Channels: main thread ←→ audio-stream thread.
// init_tx/rx : (OutputStreamHandle, actual_rate) sent once at startup.
// reopen_tx/rx: (desired_rate, reply_tx) — triggers a stream re-open.
let (init_tx, init_rx) =
std::sync::mpsc::sync_channel::<(rodio::OutputStreamHandle, u32)>(0);
let (reopen_tx, reopen_rx) =
std::sync::mpsc::sync_channel::<(u32, bool, Option<String>, std::sync::mpsc::SyncSender<rodio::OutputStreamHandle>)>(4);
let thread = std::thread::Builder::new()
.name("psysonic-audio-stream".into())
.spawn(move || {
// Set PipeWire / PulseAudio latency hints before the first open.
#[cfg(target_os = "linux")]
{
// Match cpal ALSA ~200 ms headroom: larger quantum reduces underruns when
// the decoder thread catches up after seek or competes with other work.
if std::env::var("PIPEWIRE_LATENCY").is_err() {
std::env::set_var("PIPEWIRE_LATENCY", "8192/48000");
}
if std::env::var("PULSE_LATENCY_MSEC").is_err() {
std::env::set_var("PULSE_LATENCY_MSEC", "170");
}
}
// Thread priority is kept at default during standard-mode playback.
// It is escalated to Max only when a Hi-Res stream reopen is requested,
// to prevent PipeWire underruns at high quantum sizes (8192 frames).
let (mut _stream, handle, rate) = open_stream_for_device_and_rate(None, 0);
init_tx.send((handle, rate)).ok();
// Keep the stream alive and handle sample-rate / device-switch requests.
while let Ok((desired_rate, is_hi_res, device_name, reply_tx)) = reopen_rx.recv() {
// Escalate to Max for Hi-Res reopens (large PipeWire quanta need
// real-time scheduling to avoid underruns). No escalation for
// standard mode — the thread blocks on recv() between reopens so
// elevated priority would only waste scheduler budget.
if is_hi_res {
thread_priority::set_current_thread_priority(
thread_priority::ThreadPriority::Max
).ok();
}
drop(_stream); // close old stream before opening new one
// Scale the PipeWire quantum with the sample rate so wall-clock
// latency stays roughly constant (≈93 ms) at all rates.
// 8192 frames at 88200 Hz ≈ 92.9 ms (same as 4096 at 48000 Hz).
#[cfg(target_os = "linux")]
{
let frames: u32 = if desired_rate > 48_000 { 8192 } else { 4096 };
std::env::set_var("PIPEWIRE_LATENCY", format!("{frames}/{desired_rate}"));
// Keep PULSE_LATENCY_MSEC in sync so PulseAudio-based setups
// get the same wall-clock quantum as PipeWire.
let latency_ms = (frames as f64 / desired_rate as f64 * 1000.0).round() as u64;
std::env::set_var("PULSE_LATENCY_MSEC", latency_ms.to_string());
}
let (new_stream, new_handle, _actual) = open_stream_for_device_and_rate(device_name.as_deref(), desired_rate);
_stream = new_stream;
reply_tx.send(new_handle).ok();
}
})
.expect("spawn audio stream thread");
let (initial_handle, initial_rate) = init_rx.recv().expect("audio stream handle");
let engine = AudioEngine {
stream_handle: Arc::new(std::sync::Mutex::new(initial_handle)),
stream_sample_rate: Arc::new(AtomicU32::new(initial_rate)),
device_default_rate: initial_rate,
stream_reopen_tx: reopen_tx,
selected_device: Arc::new(Mutex::new(None)),
current: Arc::new(Mutex::new(AudioCurrent {
sink: None,
duration_secs: 0.0,
seek_offset: 0.0,
play_started: None,
paused_at: None,
replay_gain_linear: 1.0,
base_volume: 0.8,
fadeout_trigger: None,
fadeout_samples: None,
})),
generation: Arc::new(AtomicU64::new(0)),
http_client: Arc::new(RwLock::new(
reqwest::Client::builder()
.timeout(Duration::from_secs(30))
.use_rustls_tls()
.user_agent(crate::subsonic_wire_user_agent())
.build()
.unwrap_or_default(),
)),
eq_gains: Arc::new(std::array::from_fn(|_| AtomicU32::new(0f32.to_bits()))),
eq_enabled: Arc::new(AtomicBool::new(false)),
eq_pre_gain: Arc::new(AtomicU32::new(0f32.to_bits())),
preloaded: Arc::new(Mutex::new(None)),
stream_completed_cache: Arc::new(Mutex::new(None)),
current_is_seekable: Arc::new(AtomicBool::new(true)),
crossfade_enabled: Arc::new(AtomicBool::new(false)),
crossfade_secs: Arc::new(AtomicU32::new(3.0f32.to_bits())),
fading_out_sink: Arc::new(Mutex::new(None)),
gapless_enabled: Arc::new(AtomicBool::new(false)),
normalization_engine: Arc::new(AtomicU32::new(0)),
normalization_target_lufs: Arc::new(AtomicU32::new((-16.0f32).to_bits())),
loudness_pre_analysis_attenuation_db: Arc::new(AtomicU32::new((-4.5f32).to_bits())),
chained_info: Arc::new(Mutex::new(None)),
samples_played: Arc::new(AtomicU64::new(0)),
current_sample_rate: Arc::new(AtomicU32::new(0)),
current_channels: Arc::new(AtomicU32::new(2)),
gapless_switch_at: Arc::new(AtomicU64::new(0)),
radio_state: Mutex::new(None),
current_playback_url: Arc::new(Mutex::new(None)),
current_analysis_track_id: Arc::new(Mutex::new(None)),
ranged_loudness_seed_hold: Arc::new(Mutex::new(None)),
preview_sink: Arc::new(Mutex::new(None)),
preview_gen: Arc::new(AtomicU64::new(0)),
preview_main_resume: Arc::new(AtomicBool::new(false)),
preview_song_id: Arc::new(Mutex::new(None)),
};
(engine, thread)
}
/// `analysis_enqueue_seed_from_url` should bail while this track's ranged HTTP buffer
/// is still filling — playback will seed on completion with the same bytes.
pub(crate) fn ranged_loudness_backfill_should_defer(engine: &AudioEngine, track_id: &str) -> bool {
let tid = track_id.trim();
if tid.is_empty() {
return false;
}
let Ok(g) = engine.ranged_loudness_seed_hold.lock() else {
return false;
};
matches!(&*g, Some((t, _)) if t.as_str() == tid)
}
/// Subsonic id pinned for the playing source (`audio_play`). Used to prioritize
/// HTTP loudness backfill for the track the user is listening to.
pub(crate) fn analysis_track_id_is_current_playback(engine: &AudioEngine, track_id: &str) -> bool {
let needle = track_id.trim();
if needle.is_empty() {
return false;
}
let Ok(guard) = engine.current_analysis_track_id.lock() else {
return false;
};
let Some(cur) = guard.as_deref().map(str::trim).filter(|s| !s.is_empty()) else {
return false;
};
cur == needle
}
pub(crate) fn audio_http_client(state: &AudioEngine) -> reqwest::Client {
state
.http_client
.read()
.map(|c| c.clone())
.unwrap_or_default()
}
pub fn refresh_http_user_agent(state: &AudioEngine, ua: &str) {
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(30))
.use_rustls_tls()
.user_agent(ua)
.build()
.unwrap_or_default();
if let Ok(mut slot) = state.http_client.write() {
*slot = client;
}
}
pub(crate) fn analysis_seed_high_priority_for_track(app: &AppHandle, track_id: &str) -> bool {
app.try_state::<AudioEngine>()
.is_some_and(|e| analysis_track_id_is_current_playback(&e, track_id))
}
+563
View File
@@ -0,0 +1,563 @@
//! URL identity, loudness cache resolution, fetch, gain math, and stream analysis helpers.
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use futures_util::StreamExt;
use rodio::Sink;
use serde::Serialize;
use tauri::{AppHandle, Emitter, Manager};
use crate::audio::engine::AudioEngine;
use crate::audio::ipc::{
partial_loudness_should_emit, PartialLoudnessPayload, PARTIAL_LOUDNESS_DELTA_THRESHOLD_DB,
};
pub(crate) 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);
let track_key = playback_identity(url).unwrap_or_else(|| url.to_string());
if !partial_loudness_should_emit(&track_key, gain_db as f32) {
crate::app_deprintln!(
"[normalization] partial-loudness skip reason=delta-below-threshold gain_db={:.2} threshold_db={:.2} track_id={:?}",
gain_db,
PARTIAL_LOUDNESS_DELTA_THRESHOLD_DB,
playback_identity(url)
);
return;
}
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,
},
);
}
pub(crate) fn provisional_loudness_gain_from_progress(
downloaded: usize,
total_size: usize,
target_lufs: f32,
start_db_in: f32,
) -> Option<f32> {
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)
}
pub(crate) fn content_type_to_hint(ct: &str) -> Option<String> {
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 }
}
// ─── Event payloads ───────────────────────────────────────────────────────────
#[derive(Clone, Serialize)]
pub struct ProgressPayload {
pub current_time: f64,
pub duration: f64,
}
// ─── Helpers ──────────────────────────────────────────────────────────────────
/// Subsonic `buildStreamUrl()` uses a fresh random salt on every call, so two
/// URLs for the same track differ in `t`/`s` query params. Compare a stable key.
pub(crate) fn playback_identity(url: &str) -> Option<String> {
if let Some(path) = url.strip_prefix("psysonic-local://") {
return Some(format!("local:{path}"));
}
if !url.contains("stream.view") {
return None;
}
let q = url.split('?').nth(1)?;
for pair in q.split('&') {
if let Some(v) = pair.strip_prefix("id=") {
let v = v.split('&').next().unwrap_or(v);
return Some(format!("stream:{v}"));
}
}
None
}
/// Stable id for analysis cache rows and `analysis:waveform-updated`.
/// Prefer the Subsonic track id from the frontend: `psysonic-local://` URLs
/// only map to `local:path` in `playback_identity`, which does not match
/// `analysis_get_waveform_for_track(trackId)` or the UI's `currentTrack.id`.
pub(crate) fn analysis_cache_track_id(logical_track_id: Option<&str>, url: &str) -> Option<String> {
let logical = logical_track_id
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|s| s.to_string());
logical.or_else(|| playback_identity(url))
}
pub(crate) fn same_playback_target(a_url: &str, b_url: &str) -> bool {
match (playback_identity(a_url), playback_identity(b_url)) {
(Some(a), Some(b)) => a == b,
_ => a_url == b_url,
}
}
#[derive(Clone, Copy)]
pub(crate) struct ResolveLoudnessCacheOpts {
/// When false, skip `get_latest_waveform_for_track` — `audio_update_replay_gain` runs
/// on every partial-LUFS tick; loudness gain does not depend on waveform, and the extra
/// SQLite read was pure overhead on the IPC path.
pub(crate) touch_waveform: bool,
/// When false, omit `cache-miss` / `cache-invalid` debug lines (still log hits and errors).
pub(crate) log_soft_misses: bool,
}
impl Default for ResolveLoudnessCacheOpts {
fn default() -> Self {
Self {
touch_waveform: true,
log_soft_misses: true,
}
}
}
pub(crate) fn resolve_loudness_gain_from_cache(
app: &AppHandle,
url: &str,
target_lufs: f32,
logical_track_id: Option<&str>,
) -> Option<f32> {
resolve_loudness_gain_from_cache_impl(
app,
url,
target_lufs,
logical_track_id,
ResolveLoudnessCacheOpts::default(),
)
}
pub(crate) fn resolve_loudness_gain_from_cache_impl(
app: &AppHandle,
url: &str,
target_lufs: f32,
logical_track_id: Option<&str>,
opts: ResolveLoudnessCacheOpts,
) -> Option<f32> {
// Only a SQLite loudness row counts here. Ephemeral JS hints (`analysis:loudness-partial`)
// are applied in `audio_update_replay_gain` via `loudness_gain_db_or_startup(..., true, _)`.
let Some(track_id) = analysis_cache_track_id(logical_track_id, url) else {
if opts.log_soft_misses {
crate::app_deprintln!(
"[normalization] resolve_loudness_gain source=no-identity url_len={}",
url.len()
);
}
return None;
};
let Some(cache) = app.try_state::<crate::analysis_cache::AnalysisCache>() else {
if opts.log_soft_misses {
crate::app_deprintln!(
"[normalization] resolve_loudness_gain source=no-analysis-cache track_id={}",
track_id
);
}
return None;
};
if opts.touch_waveform {
// Bind / preload: verify waveform context exists alongside loudness lookup.
let _ = cache.get_latest_waveform_for_track(&track_id);
}
match cache.get_latest_loudness_for_track(&track_id) {
Ok(Some(row)) if row.integrated_lufs.is_finite() => {
let recommended = crate::analysis_cache::recommended_gain_for_target(
row.integrated_lufs,
row.true_peak,
target_lufs as f64,
) as f32;
crate::app_deprintln!(
"[normalization] resolve_loudness_gain source=cache track_id={} gain_db={:.2} target_lufs={:.2} integrated_lufs={:.2} updated_at={}",
track_id,
recommended,
target_lufs,
row.integrated_lufs,
row.updated_at
);
Some(recommended)
}
Ok(Some(row)) => {
if opts.log_soft_misses {
crate::app_deprintln!(
"[normalization] resolve_loudness_gain source=cache-invalid track_id={} integrated_lufs={}",
track_id,
row.integrated_lufs
);
}
None
}
Ok(None) => {
if opts.log_soft_misses {
crate::app_deprintln!(
"[normalization] resolve_loudness_gain source=cache-miss track_id={}",
track_id
);
}
None
}
Err(e) => {
crate::app_deprintln!(
"[normalization] resolve_loudness_gain source=cache-error track_id={} err={}",
track_id,
e
);
None
}
}
}
/// Typical integrated LUFS (streaming pivot) when SQLite has no row yet — so target changes
/// still move gain before real analysis completes.
const LOUDNESS_PLACEHOLDER_INTEGRATED_LUFS: f64 = -14.0;
#[inline]
pub(crate) fn loudness_gain_placeholder_until_cache(target_lufs: f32, pre_analysis_attenuation_db: f32) -> f32 {
let pre = pre_analysis_attenuation_db.clamp(-24.0, 0.0).min(0.0);
// `true_peak = 0.0` skips the headroom cap until integrated measurement exists.
let pivot = crate::analysis_cache::recommended_gain_for_target(
LOUDNESS_PLACEHOLDER_INTEGRATED_LUFS,
0.0,
f64::from(target_lufs),
) as f32;
(pivot + pre).clamp(-24.0, 24.0)
}
/// LUFS gain after a single `resolve_loudness_gain_from_cache` result (`None` = miss).
/// Keeps `audio_update_replay_gain` / `audio_play` from resolving twice on the same URL.
/// Until a cache row exists, follow current target (see [`loudness_gain_placeholder_until_cache`]).
pub(crate) fn loudness_gain_db_after_resolve(
resolved_from_cache: Option<f32>,
target_lufs: f32,
pre_analysis_attenuation_db: f32,
allow_js_when_uncached: bool,
js_gain_db: Option<f32>,
) -> Option<f32> {
let uncached = loudness_gain_placeholder_until_cache(target_lufs, pre_analysis_attenuation_db);
match resolved_from_cache {
Some(g) => Some(g),
None => {
if allow_js_when_uncached {
match js_gain_db {
Some(r) if r.is_finite() => Some(r),
_ => Some(uncached),
}
} else {
Some(uncached)
}
}
}
}
/// LUFS: DB-backed integrated LUFS only at bind time (`allow_js_when_uncached = false`);
/// after `analysis:loudness-partial`, `audio_update_replay_gain` passes `true` so finite
/// JS gain applies until SQLite catches up. Must never return `None` or `compute_gain` uses unity.
pub(crate) fn loudness_gain_db_or_startup(
app: &AppHandle,
url: &str,
target_lufs: f32,
logical_track_id: Option<&str>,
pre_analysis_attenuation_db: f32,
allow_js_when_uncached: bool,
js_gain_db: Option<f32>,
) -> Option<f32> {
let resolved = resolve_loudness_gain_from_cache_impl(
app,
url,
target_lufs,
logical_track_id,
ResolveLoudnessCacheOpts::default(),
);
loudness_gain_db_after_resolve(
resolved,
target_lufs,
pre_analysis_attenuation_db,
allow_js_when_uncached,
js_gain_db,
)
}
#[inline]
pub(crate) fn loudness_pre_analysis_db_for_engine(state: &AudioEngine) -> f32 {
f32::from_bits(
state
.loudness_pre_analysis_attenuation_db
.load(Ordering::Relaxed),
)
.clamp(-24.0, 0.0)
.min(0.0)
}
/// Take (consume) completed manual-stream bytes if they correspond to `url`.
pub fn take_stream_completed_for_url(state: &AudioEngine, url: &str) -> Option<Vec<u8>> {
let mut guard = state.stream_completed_cache.lock().unwrap();
if guard
.as_ref()
.is_some_and(|p| same_playback_target(&p.url, url))
{
return guard.take().map(|p| p.data);
}
None
}
/// Fetch track bytes from the preload cache or via HTTP.
pub(crate) async fn fetch_data(
url: &str,
state: &AudioEngine,
gen: u64,
app: &AppHandle,
) -> Result<Option<Vec<u8>>, String> {
// Check completed streamed-track cache first (manual streaming fallback cache).
let streamed_cached = {
let mut streamed = state.stream_completed_cache.lock().unwrap();
if streamed.as_ref().is_some_and(|p| same_playback_target(&p.url, url)) {
streamed.take().map(|p| p.data)
} else {
None
}
};
if let Some(data) = streamed_cached {
return Ok(Some(data));
}
// Check preload cache next.
let cached = {
let mut preloaded = state.preloaded.lock().unwrap();
if preloaded.as_ref().is_some_and(|p| same_playback_target(&p.url, url)) {
preloaded.take().map(|p| p.data)
} else {
None
}
};
if let Some(data) = cached {
return Ok(Some(data));
}
// Offline cache — local file written by download_track_offline.
if let Some(path) = url.strip_prefix("psysonic-local://") {
let data = tokio::fs::read(path).await.map_err(|e| e.to_string())?;
return Ok(Some(data));
}
let response = crate::audio::engine::audio_http_client(&state).get(url).send().await.map_err(|e| e.to_string())?;
let status = response.status();
let ct = response.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.unwrap_or("-");
let server_hdr = response.headers()
.get("server")
.and_then(|v| v.to_str().ok())
.unwrap_or("-");
// Strip auth params from URL before logging.
let safe_url = url.split('?').next().unwrap_or(url);
crate::app_deprintln!(
"[audio] fetch {} → {} | content-type: {} | server: {}",
safe_url, status, ct, server_hdr
);
if !response.status().is_success() {
if state.generation.load(Ordering::SeqCst) != gen {
return Ok(None); // superseded
}
let status = response.status().as_u16();
let msg = format!("HTTP {status}");
app.emit("audio:error", &msg).ok();
return Err(msg);
}
// Stream the body, checking gen between chunks so a rapid manual skip can
// abort a superseded download mid-flight and free bandwidth for the new one.
let hint = response.content_length().unwrap_or(0) as usize;
let mut stream = response.bytes_stream();
let mut data = Vec::with_capacity(hint);
while let Some(chunk) = stream.next().await {
if state.generation.load(Ordering::SeqCst) != gen {
return Ok(None); // superseded — abort
}
data.extend_from_slice(&chunk.map_err(|e| e.to_string())?);
}
Ok(Some(data))
}
/// When playback uses full track bytes already in RAM (gapless `reuse_chained_bytes`,
/// `preloaded`, or `stream_completed_cache` via `fetch_data`), the `psysonic-local`
/// disk-read seed path never runs. Submit the same full-buffer analysis via the cpu-seed queue so waveform /
/// loudness SQLite can fill **offline** without `analysis_enqueue_seed_from_url` HTTP.
pub(crate) fn spawn_analysis_seed_from_in_memory_bytes(
app: &AppHandle,
cache_track_id: Option<&str>,
gen: u64,
gen_arc: &Arc<AtomicU64>,
bytes: &[u8],
) {
let Some(track_id) = cache_track_id.map(str::trim).filter(|s| !s.is_empty()) else {
return;
};
if bytes.is_empty() || bytes.len() > crate::audio::stream::TRACK_STREAM_PROMOTE_MAX_BYTES {
return;
}
let track_id = track_id.to_string();
let bytes = bytes.to_vec();
let app = app.clone();
let gen_arc = gen_arc.clone();
crate::app_deprintln!(
"[stream] in-memory play path: scheduling full-track analysis track_id={} size_mib={:.2}",
track_id,
bytes.len() as f64 / (1024.0 * 1024.0)
);
let high = crate::audio::engine::analysis_seed_high_priority_for_track(&app, &track_id);
tokio::spawn(async move {
if gen_arc.load(Ordering::SeqCst) != gen {
return;
}
if let Err(e) = crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), bytes, high).await {
crate::app_eprintln!(
"[analysis] in-memory play path seed failed for {}: {}",
track_id,
e
);
}
});
}
/// -1 dB headroom applied at full scale to prevent inter-sample clipping.
/// Modern masters are often at 0 dBFS; the EQ biquad chain and resampler
/// can produce inter-sample peaks slightly above ±1.0 → audible distortion.
/// 10^(-1/20) ≈ 0.891 — inaudible volume difference, eliminates clipping.
pub(crate) const MASTER_HEADROOM: f32 = 0.891_254;
pub(crate) const PARTIAL_LOUDNESS_MIN_BYTES: usize = 256 * 1024;
pub(crate) const PARTIAL_LOUDNESS_EMIT_INTERVAL_MS: u64 = 900;
pub(crate) fn compute_gain(
normalization_engine: u32,
replay_gain_db: Option<f32>,
replay_gain_peak: Option<f32>,
loudness_gain_db: Option<f32>,
pre_gain_db: f32,
fallback_db: f32,
volume: f32,
) -> (f32, f32) {
let gain_linear = match normalization_engine {
2 => loudness_gain_db
.map(|db| 10f32.powf(db / 20.0))
.unwrap_or(1.0),
1 => replay_gain_db
.map(|db| 10f32.powf((db + pre_gain_db) / 20.0))
.unwrap_or_else(|| 10f32.powf(fallback_db / 20.0)),
_ => 1.0,
};
let peak = if normalization_engine == 1 {
replay_gain_peak.unwrap_or(1.0).max(0.001)
} else {
1.0
};
let gain_linear = gain_linear.min(1.0 / peak);
let effective = (volume.clamp(0.0, 1.0) * gain_linear * MASTER_HEADROOM).clamp(0.0, 1.0);
(gain_linear, effective)
}
pub(crate) fn normalization_engine_name(mode: u32) -> &'static str {
match mode {
1 => "replaygain",
2 => "loudness",
_ => "off",
}
}
pub(crate) fn gain_linear_to_db(gain_linear: f32) -> Option<f32> {
if gain_linear.is_finite() && gain_linear > 0.0 {
Some(20.0 * gain_linear.log10())
} else {
None
}
}
/// `audio:normalization-state` “Now dB” for the UI: effective applied gain, including
/// loudness pre-analysis trim from settings when no cache row exists yet (matches audible level).
pub(crate) fn loudness_ui_current_gain_db(gain_linear: f32) -> Option<f32> {
gain_linear_to_db(gain_linear)
}
pub(crate) fn ramp_sink_volume(sink: Arc<Sink>, from: f32, to: f32) {
let from = from.clamp(0.0, 1.0);
let to = to.clamp(0.0, 1.0);
if (to - from).abs() < 0.002 {
sink.set_volume(to);
return;
}
static RAMP_GEN: AtomicU64 = AtomicU64::new(0);
let my_gen = RAMP_GEN.fetch_add(1, Ordering::SeqCst) + 1;
std::thread::spawn(move || {
let delta = (to - from).abs();
// Stretch large corrections to avoid audible "step down" moments.
let (steps, step_ms): (usize, u64) = if delta > 0.30 {
(24, 35)
} else if delta > 0.18 {
(18, 30)
} else if delta > 0.10 {
(14, 24)
} else {
(8, 16)
};
for i in 1..=steps {
if RAMP_GEN.load(Ordering::SeqCst) != my_gen {
return;
}
let t = i as f32 / steps as f32;
let v = from + (to - from) * t;
sink.set_volume(v.clamp(0.0, 1.0));
std::thread::sleep(Duration::from_millis(step_ms));
}
});
}
+76
View File
@@ -0,0 +1,76 @@
//! Deduped emits for normalization UI and partial loudness analysis.
use serde::Serialize;
use std::sync::{Mutex, OnceLock};
use tauri::{AppHandle, Emitter};
#[derive(Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PartialLoudnessPayload {
pub(crate) track_id: Option<String>,
pub(crate) gain_db: f32,
pub(crate) target_lufs: f32,
pub(crate) is_partial: bool,
}
#[derive(Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct NormalizationStatePayload {
pub(crate) engine: String,
pub(crate) current_gain_db: Option<f32>,
pub(crate) target_lufs: f32,
}
/// Last `audio:normalization-state` emit, kept so we can suppress duplicate
/// payloads. The frontend already debounces this event, but on Windows
/// (WebView2) the IPC pipe is the bottleneck — every echo we skip here is
/// renderer-thread time we don't pay.
pub(crate) static LAST_NORM_STATE_EMIT: OnceLock<Mutex<Option<NormalizationStatePayload>>> = OnceLock::new();
pub(crate) fn norm_state_lock() -> &'static Mutex<Option<NormalizationStatePayload>> {
LAST_NORM_STATE_EMIT.get_or_init(|| Mutex::new(None))
}
pub(crate) fn norm_state_changed(prev: &NormalizationStatePayload, next: &NormalizationStatePayload) -> bool {
if prev.engine != next.engine { return true; }
if (prev.target_lufs - next.target_lufs).abs() >= 0.02 { return true; }
match (prev.current_gain_db, next.current_gain_db) {
(None, None) => false,
(Some(a), Some(b)) => (a - b).abs() >= 0.05,
_ => true, // None ↔ Some transition is significant
}
}
pub(crate) fn maybe_emit_normalization_state(app: &AppHandle, payload: NormalizationStatePayload) {
let mut guard = norm_state_lock().lock().unwrap();
let should_emit = match guard.as_ref() {
Some(prev) => norm_state_changed(prev, &payload),
None => true,
};
if !should_emit { return; }
*guard = Some(payload.clone());
drop(guard);
let _ = app.emit("audio:normalization-state", payload);
}
/// Last `analysis:loudness-partial` gain emitted per track-identity, used to
/// suppress emits whose gain hasn't moved meaningfully (≥ 0.1 dB). The partial
/// heuristic in `emit_partial_loudness_from_bytes` and the ranged-progress curve
/// both produce values that drift by hundredths of a dB even on identical input,
/// so the time-based throttle alone is not enough to keep the loop quiet.
pub(crate) static LAST_PARTIAL_LOUDNESS_EMIT: OnceLock<Mutex<std::collections::HashMap<String, f32>>> = OnceLock::new();
pub(crate) const PARTIAL_LOUDNESS_DELTA_THRESHOLD_DB: f32 = 0.1;
pub(crate) fn partial_loudness_should_emit(track_key: &str, gain_db: f32) -> bool {
let mut guard = LAST_PARTIAL_LOUDNESS_EMIT
.get_or_init(|| Mutex::new(std::collections::HashMap::new()))
.lock()
.unwrap();
let prev = guard.get(track_key).copied();
if let Some(p) = prev {
if (p - gain_db).abs() < PARTIAL_LOUDNESS_DELTA_THRESHOLD_DB {
return false;
}
}
guard.insert(track_key.to_string(), gain_db);
true
}
+24
View File
@@ -0,0 +1,24 @@
//! Audio playback: Symphonia decode, rodio output, HTTP radio/streaming, gapless, previews.
//!
//! Implementation is split into submodules (`sources`, `decode`, `stream`, `commands`, …)
//! for navigation; behavior matches the historical single `audio.rs` file.
mod codec;
pub mod commands;
mod decode;
mod dev_io;
mod device_watcher;
mod engine;
mod helpers;
mod ipc;
pub mod preview;
mod sources;
mod state;
mod stream;
pub use commands::{audio_default_output_device_name, audio_list_devices_for_engine};
pub use device_watcher::start_device_watcher;
pub use engine::{create_engine, refresh_http_user_agent, AudioEngine};
pub use helpers::take_stream_completed_for_url;
pub(crate) use engine::{analysis_track_id_is_current_playback, ranged_loudness_backfill_should_defer};
+301
View File
@@ -0,0 +1,301 @@
//! Short preview playback on a secondary sink (same output stream).
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::{Duration, Instant};
use rodio::{Sink, Source};
use tauri::{AppHandle, Emitter, State};
use super::decode::SizedDecoder;
use super::engine::{audio_http_client, AudioEngine};
use super::helpers::MASTER_HEADROOM;
// ────────────────────────────────────────────────────────────────────────────
// Preview engine — secondary Sink on the same OutputStream, fed by Symphonia.
// ────────────────────────────────────────────────────────────────────────────
#[derive(Clone, serde::Serialize)]
struct PreviewProgressPayload {
id: String,
elapsed: f64,
duration: f64,
}
#[derive(Clone, serde::Serialize)]
struct PreviewEndPayload {
id: String,
/// "natural" = 30 s timer / source ended; "user" = explicit stop;
/// "interrupted" = a new preview superseded this one.
reason: &'static str,
}
/// Pause main sink and remember whether to resume it after preview ends.
/// Mirrors `audio_pause` semantics so progress timestamps stay consistent.
pub(crate) fn preview_pause_main(state: &AudioEngine) {
let mut cur = state.current.lock().unwrap();
if let Some(sink) = &cur.sink {
if !sink.is_paused() && !sink.empty() {
let pos = cur.position();
sink.pause();
cur.paused_at = Some(pos);
cur.play_started = None;
state.preview_main_resume.store(true, Ordering::Release);
} else {
state.preview_main_resume.store(false, Ordering::Release);
}
} else {
state.preview_main_resume.store(false, Ordering::Release);
}
}
/// Cancel any active preview and clear the resume marker. Called from every
/// command that brings the main sink back to life under its own steam
/// (`audio_play`, `audio_play_radio`, `audio_resume`) — without this the
/// preview would keep playing in parallel and the watchdog would later try
/// to resume a main sink that's already running, double-mixing the audio.
pub(crate) fn preview_clear_for_new_main_playback(state: &AudioEngine, app: &AppHandle) {
// Order matters: clear the resume marker BEFORE bumping the generation
// so the watchdog — if it wakes between our writes — sees no work to do
// and bails without resuming main behind our back.
state.preview_main_resume.store(false, Ordering::Release);
state.preview_gen.fetch_add(1, Ordering::SeqCst);
let sink = state.preview_sink.lock().unwrap().take();
let id = state.preview_song_id.lock().unwrap().take();
if let Some(s) = sink { s.stop(); }
if let Some(id) = id {
app.emit("audio:preview-end", PreviewEndPayload {
id,
reason: "interrupted",
}).ok();
}
}
/// Resume main sink iff `preview_pause_main` paused it. No-op if main was
/// already paused/empty before preview started.
pub(crate) fn preview_resume_main(state: &AudioEngine) {
if !state.preview_main_resume.swap(false, Ordering::AcqRel) {
return;
}
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;
}
}
}
/// Format hint inferred from a Subsonic stream URL. The frontend always passes
/// a `format=flac` query param for `.opus` files (server transcodes); for
/// everything else we guess from the URL's `format=` value or fall back to None.
pub(crate) fn preview_format_hint_from_url(url: &str) -> Option<String> {
url.split('?')
.nth(1)?
.split('&')
.find_map(|kv| {
let (k, v) = kv.split_once('=')?;
if k.eq_ignore_ascii_case("format") { Some(v.to_string()) } else { None }
})
}
#[tauri::command]
pub async fn audio_preview_play(
id: String,
url: String,
start_sec: f64,
duration_sec: f64,
volume: f32,
app: AppHandle,
state: State<'_, AudioEngine>,
) -> Result<(), String> {
let gen = state.preview_gen.fetch_add(1, Ordering::SeqCst) + 1;
// Tear down any existing preview before pausing main (so a rapid preview
// swap doesn't double-pause and double-resume the main sink).
let prev_sink = state.preview_sink.lock().unwrap().take();
let prev_id = state.preview_song_id.lock().unwrap().take();
if let Some(s) = prev_sink { s.stop(); }
if let Some(prev) = prev_id {
app.emit("audio:preview-end", PreviewEndPayload {
id: prev,
reason: "interrupted",
}).ok();
}
// Pause main if and only if we don't already hold a "main was playing"
// marker from a superseded preview. swap_or-style: only pause if the flag
// is currently false.
if !state.preview_main_resume.load(Ordering::Acquire) {
preview_pause_main(&state);
}
// ── Download ─────────────────────────────────────────────────────────────
let bytes = audio_http_client(&state)
.get(&url)
.send()
.await
.map_err(|e| format!("preview: connection failed: {e}"))?
.error_for_status()
.map_err(|e| format!("preview: HTTP {e}"))?
.bytes()
.await
.map_err(|e| format!("preview: read body: {e}"))?
.to_vec();
if state.preview_gen.load(Ordering::SeqCst) != gen {
// A newer preview started while we were downloading — bail.
return Ok(());
}
// ── Decode ───────────────────────────────────────────────────────────────
let hint = preview_format_hint_from_url(&url);
let bytes_for_blocking = bytes;
let hint_for_blocking = hint.clone();
let decoder = tokio::task::spawn_blocking(move || {
SizedDecoder::new(bytes_for_blocking, hint_for_blocking.as_deref(), false)
})
.await
.map_err(|e| format!("preview: decoder thread: {e}"))??;
if state.preview_gen.load(Ordering::SeqCst) != gen { return Ok(()); }
// ── Build source pipeline ────────────────────────────────────────────────
// Minimal pipeline: f32 conversion + take_duration cap. No EQ, no crossfade,
// no ReplayGain — preview is intentionally simple. Volume is set on the Sink.
let dur = Duration::from_secs_f64(duration_sec.max(1.0).min(120.0));
let source = decoder
.convert_samples::<f32>()
.take_duration(dur);
// ── Build secondary sink on the existing OutputStream ────────────────────
let sink = Arc::new(
Sink::try_new(&*state.stream_handle.lock().unwrap())
.map_err(|e| format!("preview: sink new: {e}"))?,
);
sink.set_volume((volume.clamp(0.0, 1.0) * MASTER_HEADROOM).clamp(0.0, 1.0));
sink.append(source);
// ── Best-effort seek to mid-track start position ─────────────────────────
// Symphonia FLAC without SEEKTABLE may fail here; preview then plays from 0
// which is acceptable. Hard timeout 800 ms via a worker thread.
if start_sec > 0.5 {
let start_dur = Duration::from_secs_f64(start_sec);
let seek_sink = sink.clone();
let (tx, rx) = std::sync::mpsc::channel::<()>();
std::thread::spawn(move || {
seek_sink.try_seek(start_dur).ok();
let _ = tx.send(());
});
let _ = rx.recv_timeout(Duration::from_millis(800));
}
*state.preview_sink.lock().unwrap() = Some(sink.clone());
*state.preview_song_id.lock().unwrap() = Some(id.clone());
app.emit("audio:preview-start", id.clone()).ok();
// ── Spawn watchdog: progress emits + auto-end ────────────────────────────
let preview_gen_arc = state.preview_gen.clone();
let preview_sink_arc = state.preview_sink.clone();
let preview_song_arc = state.preview_song_id.clone();
let preview_resume_arc = state.preview_main_resume.clone();
let main_current = state.current.clone();
let app_for_task = app.clone();
let id_for_task = id.clone();
tokio::spawn(async move {
let started = Instant::now();
let mut last_emit = Instant::now() - Duration::from_millis(300);
loop {
tokio::time::sleep(Duration::from_millis(100)).await;
// Cancel: another preview started or audio_preview_stop bumped the gen.
if preview_gen_arc.load(Ordering::SeqCst) != gen { return; }
let elapsed = started.elapsed().as_secs_f64();
let dur_secs = dur.as_secs_f64();
if last_emit.elapsed() >= Duration::from_millis(250) {
last_emit = Instant::now();
app_for_task.emit("audio:preview-progress", PreviewProgressPayload {
id: id_for_task.clone(),
elapsed: elapsed.min(dur_secs),
duration: dur_secs,
}).ok();
}
// Natural end: timer expired OR sink drained early (decode error,
// short track, etc.).
let drained = match preview_sink_arc.lock().unwrap().as_ref() {
Some(s) => s.empty(),
None => true,
};
if elapsed >= dur_secs || drained {
// Re-check generation under the cleanup lock to avoid racing
// a fresh preview that bumped the counter.
if preview_gen_arc.load(Ordering::SeqCst) != gen { return; }
if let Some(s) = preview_sink_arc.lock().unwrap().take() { s.stop(); }
let cleared_id = preview_song_arc.lock().unwrap().take()
.unwrap_or_else(|| id_for_task.clone());
// Resume main if we paused it.
if preview_resume_arc.swap(false, Ordering::AcqRel) {
let mut cur = main_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;
}
}
}
app_for_task.emit("audio:preview-end", PreviewEndPayload {
id: cleared_id,
reason: "natural",
}).ok();
return;
}
}
});
Ok(())
}
#[tauri::command]
pub fn audio_preview_stop(app: AppHandle, state: State<'_, AudioEngine>) {
preview_stop_inner(&app, &state, true);
}
/// Like `audio_preview_stop` but leaves the main sink paused even if it had
/// been paused by `preview_pause_main`. Used by the player-bar Stop button so
/// "stop everything" actually goes silent — without this the engine would
/// auto-resume main playback the moment the preview ends and the user perceives
/// the click as having no effect.
#[tauri::command]
pub fn audio_preview_stop_silent(app: AppHandle, state: State<'_, AudioEngine>) {
preview_stop_inner(&app, &state, false);
}
pub(crate) fn preview_stop_inner(app: &AppHandle, state: &AudioEngine, resume_main: bool) {
state.preview_gen.fetch_add(1, Ordering::SeqCst);
let sink = state.preview_sink.lock().unwrap().take();
let id = state.preview_song_id.lock().unwrap().take();
if let Some(s) = sink { s.stop(); }
if resume_main {
preview_resume_main(state);
} else {
state.preview_main_resume.store(false, Ordering::Release);
}
if let Some(id) = id {
app.emit("audio:preview-end", PreviewEndPayload {
id,
reason: "user",
}).ok();
}
}
+401
View File
@@ -0,0 +1,401 @@
//! Rodio `Source` wrappers: EQ, type erasure, fades, end-of-source notify, sample counter.
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use biquad::{Biquad, Coefficients, DirectForm2Transposed, ToHertz, Type as FilterType};
use rodio::Source;
// ─── 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;
pub(crate) struct EqSource<S: Source<Item = f32>> {
inner: S,
sample_rate: u32,
channels: u16,
gains: Arc<[AtomicU32; 10]>,
enabled: Arc<AtomicBool>,
pre_gain: Arc<AtomicU32>,
filters: [[DirectForm2Transposed<f32>; 2]; 10],
current_gains: [f32; 10],
sample_counter: usize,
channel_idx: usize,
}
impl<S: Source<Item = f32>> EqSource<S> {
pub(crate) fn new(inner: S, gains: Arc<[AtomicU32; 10]>, enabled: Arc<AtomicBool>, pre_gain: Arc<AtomicU32>) -> 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::<f32>::from_params(
FilterType::PeakingEQ(0.0),
(sample_rate as f32).hz(),
freq.hz(),
EQ_Q,
).unwrap_or_else(|_| Coefficients::<f32>::from_params(
FilterType::PeakingEQ(0.0),
(sample_rate as f32).hz(),
1000.0f32.hz(),
EQ_Q,
).unwrap());
DirectForm2Transposed::<f32>::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::<f32>::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<S: Source<Item = f32>> Iterator for EqSource<S> {
type Item = f32;
fn next(&mut self) -> Option<f32> {
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<S: Source<Item = f32>> Source for EqSource<S> {
fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
fn channels(&self) -> u16 { self.channels }
fn sample_rate(&self) -> u32 { self.sample_rate }
fn total_duration(&self) -> Option<Duration> { 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::<f32>::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::<f32>::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<S: Source<Item=f32>>.
pub(crate) struct DynSource {
inner: Box<dyn Source<Item = f32> + Send>,
channels: u16,
sample_rate: u32,
}
impl DynSource {
pub(crate) fn new(src: impl Source<Item = f32> + 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<f32> { self.inner.next() }
}
impl Source for DynSource {
fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
fn channels(&self) -> u16 { self.channels }
fn sample_rate(&self) -> u32 { self.sample_rate }
fn total_duration(&self) -> Option<Duration> { 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.
pub(crate) struct EqualPowerFadeIn<S: Source<Item = f32>> {
inner: S,
sample_count: u64,
fade_samples: u64,
}
impl<S: Source<Item = f32>> EqualPowerFadeIn<S> {
pub(crate) 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<S: Source<Item = f32>> Iterator for EqualPowerFadeIn<S> {
type Item = f32;
fn next(&mut self) -> Option<f32> {
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<S: Source<Item = f32>> Source for EqualPowerFadeIn<S> {
fn current_frame_len(&self) -> Option<usize> { 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<Duration> { 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.
pub(crate) struct TriggeredFadeOut<S: Source<Item = f32>> {
inner: S,
trigger: Arc<AtomicBool>,
fade_total_samples: Arc<AtomicU64>,
fade_progress: u64,
fading: bool,
cached_total: u64,
}
impl<S: Source<Item = f32>> TriggeredFadeOut<S> {
pub(crate) fn new(inner: S, trigger: Arc<AtomicBool>, fade_total_samples: Arc<AtomicU64>) -> Self {
Self {
inner,
trigger,
fade_total_samples,
fade_progress: 0,
fading: false,
cached_total: 0,
}
}
}
impl<S: Source<Item = f32>> Iterator for TriggeredFadeOut<S> {
type Item = f32;
fn next(&mut self) -> Option<f32> {
// 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<S: Source<Item = f32>> Source for TriggeredFadeOut<S> {
fn current_frame_len(&self) -> Option<usize> { 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<Duration> { 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()`.
pub(crate) struct NotifyingSource<S: Source<Item = f32>> {
inner: S,
done: Arc<AtomicBool>,
signalled: bool,
}
impl<S: Source<Item = f32>> NotifyingSource<S> {
pub(crate) fn new(inner: S, done: Arc<AtomicBool>) -> Self {
Self { inner, done, signalled: false }
}
}
impl<S: Source<Item = f32>> Iterator for NotifyingSource<S> {
type Item = f32;
fn next(&mut self) -> Option<f32> {
let sample = self.inner.next();
if sample.is_none() && !self.signalled {
self.signalled = true;
self.done.store(true, Ordering::SeqCst);
}
sample
}
}
impl<S: Source<Item = f32>> Source for NotifyingSource<S> {
fn current_frame_len(&self) -> Option<usize> { 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<Duration> { 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.
pub(crate) struct CountingSource<S: Source<Item = f32>> {
inner: S,
counter: Arc<AtomicU64>,
}
impl<S: Source<Item = f32>> CountingSource<S> {
pub(crate) fn new(inner: S, counter: Arc<AtomicU64>) -> Self {
Self { inner, counter }
}
}
impl<S: Source<Item = f32>> Iterator for CountingSource<S> {
type Item = f32;
fn next(&mut self) -> Option<f32> {
let sample = self.inner.next();
if sample.is_some() {
self.counter.fetch_add(1, Ordering::Relaxed);
}
sample
}
}
impl<S: Source<Item = f32>> Source for CountingSource<S> {
fn current_frame_len(&self) -> Option<usize> { 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<Duration> { 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
}
}
+26
View File
@@ -0,0 +1,26 @@
//! Small shared structs for preload / gapless chain metadata.
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::sync::Arc;
pub(crate) struct PreloadedTrack {
pub(crate) url: String,
pub(crate) data: Vec<u8>,
}
/// Info about the track that has been appended (chained) to the current Sink
/// but whose source has not yet started playing (gapless mode only).
pub(crate) struct ChainedInfo {
/// The URL that was chained — used by audio_play to detect a pre-chain hit.
pub(crate) url: String,
/// Raw file bytes (shared with the chained decoder). Lets manual skip reuse
/// them instead of re-downloading after dropping the Sink queue.
pub(crate) raw_bytes: Arc<Vec<u8>>,
pub(crate) duration_secs: f64,
pub(crate) replay_gain_linear: f32,
pub(crate) base_volume: f32,
/// Set by NotifyingSource when this chained track's source is exhausted.
pub(crate) source_done: Arc<AtomicBool>,
/// Atomic sample counter for this chained source (swapped into
/// samples_played on transition).
pub(crate) sample_counter: Arc<AtomicU64>,
}
+946
View File
@@ -0,0 +1,946 @@
//! Internet radio, ranged/track HTTP readers, ring-buffer producers, and ICY parsing.
use std::io::{Read, Seek, SeekFrom};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use futures_util::StreamExt;
use ringbuf::{HeapConsumer, HeapProducer};
use symphonia::core::io::MediaSource;
use tauri::{AppHandle, Emitter};
use super::state::PreloadedTrack;
/// Clears [`AudioEngine::ranged_loudness_seed_hold`] only if it still matches this play.
struct RangedLoudnessSeedHoldClear {
slot: Arc<Mutex<Option<(String, u64)>>>,
tid: String,
gen: u64,
}
impl Drop for RangedLoudnessSeedHoldClear {
fn drop(&mut self) {
if let Ok(mut g) = self.slot.lock() {
if matches!(&*g, Some((t, gen)) if t == &self.tid && *gen == self.gen) {
*g = None;
}
}
}
}
/// 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.
pub(crate) const RADIO_BUF_CAPACITY: usize = 256 * 1024;
/// Minimum ring buffer for on-demand track streaming starts.
pub(crate) const TRACK_STREAM_MIN_BUF_CAPACITY: usize = 1024 * 1024;
/// Cap ring buffer growth when content-length is known.
pub(crate) 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.
pub(crate) const TRACK_STREAM_PROMOTE_MAX_BYTES: usize = 64 * 1024 * 1024;
/// Hot/offline `psysonic-local://` files are read from disk for waveform/LUFS seeding — not the
/// same heap pressure as retaining a full HTTP capture. FLAC/DSD tracks often exceed 64MiB;
/// using the stream-promote cap here skipped analysis entirely (empty seekbar).
pub(crate) const LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES: usize = 512 * 1024 * 1024;
/// Consecutive body-stream failures tolerated for track streaming before abort.
pub(crate) const TRACK_STREAM_MAX_RECONNECTS: u32 = 3;
/// Seconds at stall threshold while paused before hard-disconnect.
pub(crate) const RADIO_HARD_PAUSE_SECS: u64 = 5;
/// AudioStreamReader timeout: if no audio bytes arrive for this long → EOF.
pub(crate) const RADIO_READ_TIMEOUT_SECS: u64 = 15;
/// Sleep interval when ring buffer is empty (prevents CPU spin).
pub(crate) 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.
pub(crate) 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<u8> },
}
pub(crate) 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<u8>) -> Option<IcyMeta> {
let mut extracted: Option<IcyMeta> = 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<IcyMeta> {
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<u8> (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.
pub(crate) struct AudioStreamReader {
pub(crate) cons: HeapConsumer<u8>,
/// 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().
pub(crate) new_cons_rx: Mutex<std::sync::mpsc::Receiver<HeapConsumer<u8>>>,
pub(crate) deadline: std::time::Instant,
pub(crate) gen_arc: Arc<AtomicU64>,
pub(crate) gen: u64,
/// Diagnostic tag for logs ("radio" or "track-stream").
pub(crate) source_tag: &'static str,
/// Optional completion marker: when true and the ring buffer is empty,
/// return EOF immediately (used by one-shot track streaming).
pub(crate) eof_when_empty: Option<Arc<AtomicBool>>,
/// Monotonic byte offset for SeekFrom::Current(0) "tell" (Symphonia probe).
pub(crate) pos: u64,
}
impl Read for AudioStreamReader {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
// 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<HeapConsumer<u8>> = 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<u64> {
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<u64> { None }
}
// ── RangedHttpSource — seekable HTTP-backed MediaSource ──────────────────────
//
// Pre-allocates a Vec<u8> 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.
pub(crate) struct RangedHttpSource {
/// Pre-allocated buffer of total size. Filled linearly from offset 0.
pub(crate) buf: Arc<Mutex<Vec<u8>>>,
/// Bytes contiguously downloaded from offset 0.
pub(crate) downloaded_to: Arc<AtomicUsize>,
pub(crate) total_size: u64,
pub(crate) pos: u64,
/// Set when the download task terminates (success or hard error).
pub(crate) done: Arc<AtomicBool>,
pub(crate) gen_arc: Arc<AtomicU64>,
pub(crate) gen: u64,
}
impl Read for RangedHttpSource {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
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<u64> {
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<u64> { 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.
pub(crate) struct LocalFileSource {
pub(crate) file: std::fs::File,
pub(crate) len: u64,
}
impl Read for LocalFileSource {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.file.read(buf)
}
}
impl Seek for LocalFileSource {
fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
self.file.seek(pos)
}
}
impl MediaSource for LocalFileSource {
fn is_seekable(&self) -> bool { true }
fn byte_len(&self) -> Option<u64> { Some(self.len) }
}
// ── Pause / Reconnect Coordination ───────────────────────────────────────────
pub(crate) struct RadioSharedFlags {
/// Set by audio_pause; cleared by audio_resume.
pub(crate) is_paused: AtomicBool,
/// Set by download task on hard disconnect; cleared on resume-reconnect.
pub(crate) is_hard_paused: AtomicBool,
/// Delivers a fresh HeapConsumer<u8> to AudioStreamReader on reconnect.
pub(crate) new_cons_tx: Mutex<std::sync::mpsc::Sender<HeapConsumer<u8>>>,
}
/// 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<RadioSharedFlags>,
}
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"]
// ── 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.
pub(crate) async fn radio_download_task(
gen: u64,
gen_arc: Arc<AtomicU64>,
mut initial_response: Option<reqwest::Response>,
http_client: reqwest::Client,
url: String,
mut prod: HeapProducer<u8>,
flags: Arc<RadioSharedFlags>,
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<u8> = 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<usize> = 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<std::time::Instant> = 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.
pub(crate) async fn track_download_task(
gen: u64,
gen_arc: Arc<AtomicU64>,
http_client: reqwest::Client,
app: AppHandle,
url: String,
initial_response: reqwest::Response,
mut prod: HeapProducer<u8>,
done: Arc<AtomicBool>,
promote_cache_slot: Arc<Mutex<Option<PreloadedTrack>>>,
normalization_engine: Arc<AtomicU32>,
normalization_target_lufs: Arc<AtomicU32>,
loudness_pre_analysis_attenuation_db: Arc<AtomicU32>,
cache_track_id: Option<String>,
) {
let mut downloaded: u64 = 0;
let mut reconnects: u32 = 0;
let mut next_response: Option<reqwest::Response> = Some(initial_response);
let mut capture: Vec<u8> = 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(crate::audio::helpers::PARTIAL_LOUDNESS_EMIT_INTERVAL_MS)
{
last_partial_loudness_emit = Instant::now();
if normalization_engine.load(Ordering::Relaxed) == 2 {
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);
super::helpers::emit_partial_loudness_from_bytes(&app, &url, &capture, target_lufs, pre_db);
}
}
offset += pushed;
downloaded += pushed as u64;
}
}
}
if !capture_over_limit && !capture.is_empty() {
if let Some(track_id) = cache_track_id {
crate::app_deprintln!(
"[stream] legacy stream: capture complete track_id={} capture_mib={:.2} — full-track analysis (cpu-seed queue)",
track_id,
capture.len() as f64 / (1024.0 * 1024.0)
);
let high = crate::audio::engine::analysis_seed_high_priority_for_track(&app, &track_id);
if let Err(e) =
crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), capture.clone(), high).await
{
crate::app_eprintln!("[analysis] track seed failed for {}: {}", track_id, e);
}
}
*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.
pub(crate) async fn ranged_download_task(
gen: u64,
gen_arc: Arc<AtomicU64>,
http_client: reqwest::Client,
app: AppHandle,
_duration_hint: f64,
url: String,
initial_response: reqwest::Response,
buf: Arc<Mutex<Vec<u8>>>,
downloaded_to: Arc<AtomicUsize>,
done: Arc<AtomicBool>,
promote_cache_slot: Arc<Mutex<Option<PreloadedTrack>>>,
normalization_engine: Arc<AtomicU32>,
normalization_target_lufs: Arc<AtomicU32>,
loudness_pre_analysis_attenuation_db: Arc<AtomicU32>,
cache_track_id: Option<String>,
// When `Some`, ranged playback seeds on completion — defer HTTP backfill for that
// track; `None` for large files where ranged skips seed (needs backfill).
loudness_seed_hold: Option<Arc<Mutex<Option<(String, u64)>>>>,
) {
let _ranged_loudness_hold_clear = match (loudness_seed_hold.as_ref(), cache_track_id.as_ref()) {
(Some(slot), Some(tid)) => {
let t = tid.clone();
{
let mut g = slot.lock().unwrap();
*g = Some((t.clone(), gen));
}
Some(RangedLoudnessSeedHoldClear {
slot: Arc::clone(slot),
tid: t,
gen,
})
}
_ => None,
};
let total_size = buf.lock().unwrap().len();
let mut downloaded: usize = 0;
let mut reconnects: u32 = 0;
let mut next_response: Option<reqwest::Response> = 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 >= crate::audio::helpers::PARTIAL_LOUDNESS_MIN_BYTES
&& total_size > 0
&& last_partial_loudness_emit.elapsed() >= Duration::from_millis(crate::audio::helpers::PARTIAL_LOUDNESS_EMIT_INTERVAL_MS)
{
last_partial_loudness_emit = Instant::now();
if normalization_engine.load(Ordering::Relaxed) == 2 {
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) =
super::helpers::provisional_loudness_gain_from_progress(downloaded, total_size, target_lufs, start_db)
{
let track_key = crate::audio::helpers::playback_identity(&url).unwrap_or_else(|| url.clone());
if crate::audio::ipc::partial_loudness_should_emit(&track_key, provisional_db) {
let _ = app.emit(
"analysis:loudness-partial",
crate::audio::ipc::PartialLoudnessPayload {
track_id: crate::audio::helpers::playback_identity(&url),
gain_db: provisional_db,
target_lufs,
is_partial: true,
},
);
}
}
}
}
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 {
if let Some(ref tid) = cache_track_id {
crate::app_deprintln!(
"[stream] ranged: HTTP buffer full track_id={} size_mib={:.2} — cloning {} bytes then full-track analysis (cpu-seed queue; this task awaits completion)",
tid,
total_size as f64 / (1024.0 * 1024.0),
total_size
);
}
let t_clone = Instant::now();
let data = buf.lock().unwrap().clone();
if total_size > 32 * 1024 * 1024 {
crate::app_deprintln!(
"[stream] ranged: buffer cloned in_ms={}",
t_clone.elapsed().as_millis()
);
}
if let Some(track_id) = cache_track_id {
let high = crate::audio::engine::analysis_seed_high_priority_for_track(&app, &track_id);
if let Err(e) = crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), data.clone(), high).await {
crate::app_eprintln!("[analysis] ranged seed failed for {}: {}", track_id, e);
}
}
*promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url, data });
crate::app_deprintln!("[stream] promoted to stream_completed_cache for replay");
}
}
+23 -23
View File
@@ -4844,29 +4844,29 @@ pub fn run() {
unregister_global_shortcut,
mpris_set_metadata,
mpris_set_playback,
audio::audio_play,
audio::audio_pause,
audio::audio_resume,
audio::audio_stop,
audio::audio_seek,
audio::audio_set_volume,
audio::audio_update_replay_gain,
audio::audio_set_eq,
audio::autoeq_entries,
audio::autoeq_fetch_profile,
audio::audio_preload,
audio::audio_play_radio,
audio::audio_preview_play,
audio::audio_preview_stop,
audio::audio_preview_stop_silent,
audio::audio_set_crossfade,
audio::audio_set_gapless,
audio::audio_set_normalization,
audio::audio_list_devices,
audio::audio_canonicalize_selected_device,
audio::audio_default_output_device_name,
audio::audio_set_device,
audio::audio_chain_preload,
audio::commands::audio_play,
audio::commands::audio_pause,
audio::commands::audio_resume,
audio::commands::audio_stop,
audio::commands::audio_seek,
audio::commands::audio_set_volume,
audio::commands::audio_update_replay_gain,
audio::commands::audio_set_eq,
audio::commands::autoeq_entries,
audio::commands::autoeq_fetch_profile,
audio::commands::audio_preload,
audio::commands::audio_play_radio,
audio::preview::audio_preview_play,
audio::preview::audio_preview_stop,
audio::preview::audio_preview_stop_silent,
audio::commands::audio_set_crossfade,
audio::commands::audio_set_gapless,
audio::commands::audio_set_normalization,
audio::commands::audio_list_devices,
audio::commands::audio_canonicalize_selected_device,
audio::commands::audio_default_output_device_name,
audio::commands::audio_set_device,
audio::commands::audio_chain_preload,
discord::discord_update_presence,
discord::discord_clear_presence,
lastfm_request,