mirror of
https://github.com/kilyabin/psysonic.git
synced 2026-07-21 22:15:40 +00:00
cucadmuh
5d06738ce1
* fix(audio): honor manual skip when next track is gapless-pre-chained * fix(audio): match gapless chain and preload by stream id, not full URL
2552 lines
99 KiB
Rust
2552 lines
99 KiB
Rust
use std::io::{Cursor, Read, Seek, SeekFrom};
|
||
use std::sync::{Arc, Mutex};
|
||
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
|
||
use std::time::{Duration, Instant};
|
||
|
||
use ringbuf::{HeapConsumer, HeapProducer, HeapRb};
|
||
|
||
use biquad::{Biquad, Coefficients, DirectForm2Transposed, ToHertz, Type as FilterType};
|
||
use rodio::{Sink, Source};
|
||
use rodio::source::UniformSourceIterator;
|
||
use serde::Serialize;
|
||
use symphonia::core::{
|
||
audio::{AudioBufferRef, SampleBuffer, SignalSpec},
|
||
codecs::{DecoderOptions, CODEC_TYPE_NULL},
|
||
formats::{FormatOptions, FormatReader, SeekMode, SeekTo},
|
||
io::{MediaSource, MediaSourceStream},
|
||
meta::MetadataOptions,
|
||
probe::Hint,
|
||
units::{self, Time},
|
||
};
|
||
use futures_util::StreamExt;
|
||
use tauri::{AppHandle, Emitter, State};
|
||
|
||
// ─── 10-Band Graphic Equalizer ────────────────────────────────────────────────
|
||
|
||
const EQ_BANDS_HZ: [f32; 10] = [31.0, 62.0, 125.0, 250.0, 500.0, 1000.0, 2000.0, 4000.0, 8000.0, 16000.0];
|
||
const EQ_Q: f32 = 1.41;
|
||
const EQ_CHECK_INTERVAL: usize = 1024;
|
||
|
||
struct EqSource<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> {
|
||
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>>.
|
||
|
||
struct DynSource {
|
||
inner: Box<dyn Source<Item = f32> + Send>,
|
||
channels: u16,
|
||
sample_rate: u32,
|
||
}
|
||
|
||
impl DynSource {
|
||
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.
|
||
|
||
struct EqualPowerFadeIn<S: Source<Item = f32>> {
|
||
inner: S,
|
||
sample_count: u64,
|
||
fade_samples: u64,
|
||
}
|
||
|
||
impl<S: Source<Item = f32>> EqualPowerFadeIn<S> {
|
||
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.
|
||
|
||
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> {
|
||
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()`.
|
||
|
||
struct NotifyingSource<S: Source<Item = f32>> {
|
||
inner: S,
|
||
done: Arc<AtomicBool>,
|
||
signalled: bool,
|
||
}
|
||
|
||
impl<S: Source<Item = f32>> NotifyingSource<S> {
|
||
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.
|
||
|
||
struct CountingSource<S: Source<Item = f32>> {
|
||
inner: S,
|
||
counter: Arc<AtomicU64>,
|
||
}
|
||
|
||
impl<S: Source<Item = f32>> CountingSource<S> {
|
||
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
|
||
}
|
||
}
|
||
|
||
// ─── Internet Radio v2 — Lock-Free SPSC + ICY Metadata + Hybrid Pause ────────
|
||
//
|
||
// HTTP task (tokio)
|
||
// └─[IcyInterceptor]─► HeapProducer<u8>
|
||
// │ (4 MB HeapRb, lock-free)
|
||
// HeapConsumer<u8>
|
||
// │
|
||
// AudioStreamReader (Read + Seek + MediaSource)
|
||
// │
|
||
// SizedDecoder (symphonia)
|
||
// │
|
||
// rodio Sink
|
||
//
|
||
// Pause modes:
|
||
// Logical pause — sink.pause(); download task keeps filling (time-shift).
|
||
// Hard pause — buffer ≥ RADIO_HARD_PAUSE_THRESH full + paused ≥ 5 s
|
||
// → TCP disconnect, is_hard_paused = true.
|
||
// Resume (warm) — sink.play(); buffer drains seamlessly.
|
||
// Resume (cold) — new HeapRb + new GET; consumer swapped in AudioStreamReader.
|
||
//
|
||
// New Tauri event: "radio:metadata" → String (ICY StreamTitle)
|
||
|
||
/// 256 KB on the heap — ≈16 s at 128 kbps, ≈6 s at 320 kbps.
|
||
/// Small enough that stale audio drains within a few seconds on reconnect;
|
||
/// large enough to absorb brief network hiccups without stuttering.
|
||
const RADIO_BUF_CAPACITY: usize = 256 * 1024;
|
||
/// Seconds at stall threshold while paused before hard-disconnect.
|
||
const RADIO_HARD_PAUSE_SECS: u64 = 5;
|
||
/// AudioStreamReader timeout: if no audio bytes arrive for this long → EOF.
|
||
const RADIO_READ_TIMEOUT_SECS: u64 = 15;
|
||
/// Sleep interval when ring buffer is empty (prevents CPU spin).
|
||
const RADIO_YIELD_MS: u64 = 2;
|
||
|
||
// ── ICY Metadata State Machine ────────────────────────────────────────────────
|
||
//
|
||
// Shoutcast/Icecast embed metadata every `metaint` audio bytes:
|
||
//
|
||
// ┌──────────────────────┬───┬─────────────┐
|
||
// │ audio × metaint │ N │ meta × N×16 │ (repeating)
|
||
// └──────────────────────┴───┴─────────────┘
|
||
//
|
||
// N = 0 → no metadata this block. Metadata bytes are stripped so only
|
||
// pure audio reaches the ring buffer and Symphonia never sees text bytes.
|
||
|
||
enum IcyState {
|
||
/// Forwarding audio bytes; `remaining` counts down to the next boundary.
|
||
ReadingAudio { remaining: usize },
|
||
/// Next byte is the metadata length multiplier N.
|
||
ReadingLengthByte,
|
||
/// Accumulating N×16 metadata bytes.
|
||
ReadingMetadata { remaining: usize, buf: Vec<u8> },
|
||
}
|
||
|
||
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.
|
||
|
||
struct AudioStreamReader {
|
||
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().
|
||
new_cons_rx: Mutex<std::sync::mpsc::Receiver<HeapConsumer<u8>>>,
|
||
deadline: std::time::Instant,
|
||
gen_arc: Arc<AtomicU64>,
|
||
gen: u64,
|
||
/// Monotonic byte offset for SeekFrom::Current(0) "tell" (Symphonia probe).
|
||
pos: u64,
|
||
}
|
||
|
||
impl Read for AudioStreamReader {
|
||
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<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 std::time::Instant::now() >= self.deadline {
|
||
eprintln!(
|
||
"[radio] AudioStreamReader: {}s without data → EOF",
|
||
RADIO_READ_TIMEOUT_SECS
|
||
);
|
||
return Err(std::io::Error::new(
|
||
std::io::ErrorKind::TimedOut,
|
||
"radio: no data received",
|
||
));
|
||
}
|
||
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,
|
||
"radio stream is not seekable",
|
||
)),
|
||
}
|
||
}
|
||
}
|
||
|
||
impl MediaSource for AudioStreamReader {
|
||
fn is_seekable(&self) -> bool { false }
|
||
fn byte_len(&self) -> Option<u64> { None }
|
||
}
|
||
|
||
// ── Pause / Reconnect Coordination ───────────────────────────────────────────
|
||
|
||
pub(crate) struct RadioSharedFlags {
|
||
/// Set by audio_pause; cleared by audio_resume.
|
||
is_paused: AtomicBool,
|
||
/// Set by download task on hard disconnect; cleared on resume-reconnect.
|
||
is_hard_paused: AtomicBool,
|
||
/// Delivers a fresh HeapConsumer<u8> to AudioStreamReader on reconnect.
|
||
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"]
|
||
|
||
fn try_make_radio_decoder(
|
||
params: &symphonia::core::codecs::CodecParameters,
|
||
opts: &DecoderOptions,
|
||
) -> Result<Box<dyn symphonia::core::codecs::Decoder>, symphonia::core::errors::Error> {
|
||
symphonia::default::get_codecs().make(params, opts)
|
||
}
|
||
|
||
// ── Async HTTP Download Task ──────────────────────────────────────────────────
|
||
//
|
||
// Lifecycle:
|
||
// 'outer loop — reconnect on TCP drop (up to MAX_RECONNECTS)
|
||
// 'inner loop — read HTTP chunks → ICY interceptor → push audio to ring buffer
|
||
//
|
||
// Hard-pause detection: if push_slice() returns 0 (buffer full) AND sink is
|
||
// paused AND that condition persists for RADIO_HARD_PAUSE_SECS → disconnect.
|
||
// Sets is_hard_paused = true so audio_resume knows it must reconnect.
|
||
|
||
async fn radio_download_task(
|
||
gen: u64,
|
||
gen_arc: Arc<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 {
|
||
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() => {
|
||
eprintln!("[radio] reconnected ({bytes_total} B so far)");
|
||
r
|
||
}
|
||
Ok(r) => {
|
||
eprintln!("[radio] reconnect: HTTP {} — giving up", r.status());
|
||
break 'outer;
|
||
}
|
||
Err(e) => {
|
||
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;
|
||
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 { "" };
|
||
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;
|
||
eprintln!("[radio] stream error: {e} → reconnecting (consecutive #{reconnect_count})");
|
||
break 'inner;
|
||
}
|
||
None => {
|
||
reconnect_count += 1;
|
||
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
|
||
|
||
eprintln!("[radio] download task done ({bytes_total} B total)");
|
||
}
|
||
|
||
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 { None }
|
||
}
|
||
|
||
// ─── 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.
|
||
|
||
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.
|
||
|
||
const DECODE_MAX_RETRIES: usize = 3;
|
||
|
||
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,
|
||
}
|
||
|
||
impl SizedDecoder {
|
||
fn new(data: Vec<u8>, format_hint: Option<&str>) -> Result<Self, String> {
|
||
let data_len = data.len() as u64;
|
||
let source = SizedCursorSource {
|
||
inner: Cursor::new(data),
|
||
len: data_len,
|
||
};
|
||
let mss = MediaSourceStream::new(
|
||
Box::new(source) as Box<dyn MediaSource>,
|
||
Default::default(),
|
||
);
|
||
|
||
let mut hint = Hint::new();
|
||
if let Some(ext) = format_hint {
|
||
hint.with_extension(ext);
|
||
}
|
||
let format_opts = FormatOptions {
|
||
enable_gapless: true,
|
||
..Default::default()
|
||
};
|
||
|
||
let probed = symphonia::default::get_probe()
|
||
.format(&hint, mss, &format_opts, &MetadataOptions::default())
|
||
.map_err(|e| {
|
||
let hint_str = format_hint.unwrap_or("unknown");
|
||
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()
|
||
.find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
|
||
.ok_or_else(|| "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));
|
||
|
||
let mut decoder = symphonia::default::get_codecs()
|
||
.make(&track.codec_params, &DecoderOptions::default())
|
||
.map_err(|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.
|
||
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) => return Err(format!("could not read audio data: {e}")),
|
||
};
|
||
if packet.track_id() != track_id { continue; }
|
||
match decoder.decode(&packet) {
|
||
Ok(decoded) => break decoded,
|
||
Err(symphonia::core::errors::Error::DecodeError(_)) => {
|
||
decode_errors += 1;
|
||
if decode_errors > DECODE_MAX_RETRIES {
|
||
return Err("too many decode errors — file may be corrupt".into());
|
||
}
|
||
}
|
||
Err(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,
|
||
})
|
||
}
|
||
|
||
/// Build a decoder from any `MediaSource` (e.g. `RadioBuffer`).
|
||
/// Uses `enable_gapless: false` — live streams are not seekable; gapless
|
||
/// trimming requires seeking to read the LAME/iTunSMPB end-padding info.
|
||
fn new_streaming(media: Box<dyn MediaSource>, format_hint: Option<&str>) -> Result<Self, String> {
|
||
let mss = MediaSourceStream::new(media, Default::default());
|
||
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!("radio: format probe failed: {e}"))?;
|
||
|
||
let track = probed.format.tracks().iter()
|
||
.find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
|
||
.ok_or_else(|| "radio: no audio track found".to_string())?;
|
||
let track_id = track.id;
|
||
// 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!("radio: 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(_)) => {
|
||
errors += 1;
|
||
if errors > DECODE_MAX_RETRIES { return Err("radio: too many decode errors".into()); }
|
||
}
|
||
Err(e) => return Err(format!("radio: 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 })
|
||
}
|
||
|
||
#[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() {
|
||
let packet = self.format.next_packet().ok()?;
|
||
let mut decoded = self.decoder.decode(&packet);
|
||
for _ in 0..DECODE_MAX_RETRIES {
|
||
if decoded.is_err() {
|
||
let p = self.format.next_packet().ok()?;
|
||
decoded = self.decoder.decode(&p);
|
||
}
|
||
}
|
||
let decoded = decoded.ok()?;
|
||
decoded.spec().clone_into(&mut self.spec);
|
||
self.buffer = Self::make_buffer(decoded, &self.spec);
|
||
self.current_frame_offset = 0;
|
||
}
|
||
|
||
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 576–2112 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.
|
||
|
||
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 }
|
||
}
|
||
}
|
||
|
||
fn find_subsequence(data: &[u8], needle: &[u8]) -> Option<usize> {
|
||
data.windows(needle.len()).position(|w| w == needle)
|
||
}
|
||
|
||
fn parse_gapless_info(data: &[u8]) -> GaplessInfo {
|
||
let pos = match find_subsequence(data, b"iTunSMPB") {
|
||
Some(p) => p,
|
||
None => return GaplessInfo::default(),
|
||
};
|
||
|
||
// Collect printable ASCII bytes after the tag (skip nulls / control chars)
|
||
let tail = &data[pos + 8..data.len().min(pos + 8 + 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.
|
||
struct BuiltSource {
|
||
source: CountingSource<NotifyingSource<TriggeredFadeOut<EqualPowerFadeIn<EqSource<DynSource>>>>>,
|
||
duration_secs: f64,
|
||
output_rate: u32,
|
||
output_channels: u16,
|
||
/// Trigger for the sample-level crossfade fade-out.
|
||
fadeout_trigger: Arc<AtomicBool>,
|
||
/// Total samples for the fade-out (set before triggering).
|
||
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.
|
||
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>,
|
||
) -> Result<BuiltSource, String> {
|
||
let gapless = parse_gapless_info(&data);
|
||
|
||
let decoder = SizedDecoder::new(data, format_hint)?;
|
||
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,
|
||
})
|
||
}
|
||
|
||
// ─── Engine state ─────────────────────────────────────────────────────────────
|
||
|
||
pub(crate) struct PreloadedTrack {
|
||
url: String,
|
||
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.
|
||
url: String,
|
||
/// Raw file bytes (shared with the chained decoder). Lets manual skip reuse
|
||
/// them instead of re-downloading after dropping the Sink queue.
|
||
raw_bytes: Arc<Vec<u8>>,
|
||
duration_secs: f64,
|
||
replay_gain_linear: f32,
|
||
base_volume: f32,
|
||
/// Set by NotifyingSource when this chained track's source is exhausted.
|
||
source_done: Arc<AtomicBool>,
|
||
/// Atomic sample counter for this chained source (swapped into
|
||
/// samples_played on transition).
|
||
sample_counter: Arc<AtomicU64>,
|
||
}
|
||
|
||
pub struct AudioEngine {
|
||
pub stream_handle: Arc<rodio::OutputStreamHandle>,
|
||
pub current: Arc<Mutex<AudioCurrent>>,
|
||
/// Monotonically incremented on each audio_play (non-chain) / audio_stop call.
|
||
pub generation: Arc<AtomicU64>,
|
||
pub http_client: reqwest::Client,
|
||
pub eq_gains: Arc<[AtomicU32; 10]>,
|
||
pub eq_enabled: Arc<AtomicBool>,
|
||
pub eq_pre_gain: Arc<AtomicU32>,
|
||
pub preloaded: Arc<Mutex<Option<PreloadedTrack>>>,
|
||
pub crossfade_enabled: Arc<AtomicBool>,
|
||
pub crossfade_secs: Arc<AtomicU32>,
|
||
pub fading_out_sink: Arc<Mutex<Option<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>,
|
||
/// Info about the next-up chained track (gapless mode).
|
||
/// The progress task reads this when `current_source_done` fires.
|
||
pub 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 radio_state: Mutex<Option<RadioLiveState>>,
|
||
}
|
||
|
||
pub struct AudioCurrent {
|
||
pub sink: Option<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
|
||
}
|
||
}
|
||
}
|
||
|
||
pub fn create_engine() -> (AudioEngine, std::thread::JoinHandle<()>) {
|
||
let (tx, rx) = std::sync::mpsc::sync_channel::<rodio::OutputStreamHandle>(0);
|
||
|
||
// Request a larger audio buffer from PipeWire/PulseAudio to reduce ALSA underruns.
|
||
// Only set if the user hasn't already configured these themselves.
|
||
// PIPEWIRE_LATENCY: 4096 frames / 48000 Hz ≈ 85 ms — enough to absorb scheduler jitter.
|
||
#[cfg(target_os = "linux")]
|
||
{
|
||
if std::env::var("PIPEWIRE_LATENCY").is_err() {
|
||
std::env::set_var("PIPEWIRE_LATENCY", "4096/48000");
|
||
}
|
||
if std::env::var("PULSE_LATENCY_MSEC").is_err() {
|
||
std::env::set_var("PULSE_LATENCY_MSEC", "85");
|
||
}
|
||
}
|
||
|
||
// macOS: request a smaller CoreAudio buffer to reduce output latency.
|
||
// Smaller buffers = lower latency between decoded samples and DAC output,
|
||
// which tightens the gap between actual audio and UI event delivery.
|
||
#[cfg(target_os = "macos")]
|
||
{
|
||
if std::env::var("COREAUDIO_BUFFER_SIZE").is_err() {
|
||
std::env::set_var("COREAUDIO_BUFFER_SIZE", "512");
|
||
}
|
||
}
|
||
|
||
let thread = std::thread::Builder::new()
|
||
.name("psysonic-audio-stream".into())
|
||
.spawn(move || match rodio::OutputStream::try_default() {
|
||
Ok((_stream, handle)) => {
|
||
tx.send(handle).ok();
|
||
loop { std::thread::park(); }
|
||
}
|
||
Err(e) => { eprintln!("[psysonic] audio output error: {e}"); }
|
||
})
|
||
.expect("spawn audio stream thread");
|
||
|
||
let stream_handle = rx.recv().expect("audio stream handle");
|
||
|
||
let engine = AudioEngine {
|
||
stream_handle: Arc::new(stream_handle),
|
||
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: reqwest::Client::builder()
|
||
.timeout(Duration::from_secs(30))
|
||
.use_rustls_tls()
|
||
.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)),
|
||
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)),
|
||
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),
|
||
};
|
||
|
||
(engine, thread)
|
||
}
|
||
|
||
// ─── 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.
|
||
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
|
||
}
|
||
|
||
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,
|
||
}
|
||
}
|
||
|
||
/// Fetch track bytes from the preload cache or via HTTP.
|
||
async fn fetch_data(
|
||
url: &str,
|
||
state: &AudioEngine,
|
||
gen: u64,
|
||
app: &AppHandle,
|
||
) -> Result<Option<Vec<u8>>, String> {
|
||
// Check preload cache first.
|
||
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 = state.http_client.get(url).send().await.map_err(|e| e.to_string())?;
|
||
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))
|
||
}
|
||
|
||
/// -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.
|
||
const MASTER_HEADROOM: f32 = 0.891_254;
|
||
|
||
fn compute_gain(
|
||
replay_gain_db: Option<f32>,
|
||
replay_gain_peak: Option<f32>,
|
||
volume: f32,
|
||
) -> (f32, f32) {
|
||
let gain_linear = replay_gain_db
|
||
.map(|db| 10f32.powf(db / 20.0))
|
||
.unwrap_or(1.0);
|
||
let peak = replay_gain_peak.unwrap_or(1.0).max(0.001);
|
||
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)
|
||
}
|
||
|
||
// ─── Commands ─────────────────────────────────────────────────────────────────
|
||
|
||
#[tauri::command]
|
||
pub async fn audio_play(
|
||
url: String,
|
||
volume: f32,
|
||
duration_hint: f64,
|
||
replay_gain_db: Option<f32>,
|
||
replay_gain_peak: Option<f32>,
|
||
manual: bool, // true = user-initiated skip → bypass crossfade, start immediately
|
||
app: AppHandle,
|
||
state: State<'_, AudioEngine>,
|
||
) -> Result<(), String> {
|
||
let gapless = state.gapless_enabled.load(Ordering::Relaxed);
|
||
|
||
// ── Ghost-command guard ───────────────────────────────────────────────────
|
||
// After a gapless auto-advance, the frontend may fire a stale playTrack()
|
||
// call via IPC. If we're within 500 ms of the last gapless switch AND the
|
||
// requested URL matches the already-playing chained track, reject it.
|
||
{
|
||
let switch_ms = state.gapless_switch_at.load(Ordering::SeqCst);
|
||
if switch_ms > 0 {
|
||
let now_ms = std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.unwrap_or_default()
|
||
.as_millis() as u64;
|
||
if now_ms.saturating_sub(switch_ms) < 500 {
|
||
// Within the guard window — suppress this ghost command.
|
||
return Ok(());
|
||
}
|
||
}
|
||
}
|
||
|
||
// ── Gapless pre-chain hit ─────────────────────────────────────────────────
|
||
// audio_chain_preload already appended this URL to the Sink 30 s in
|
||
// advance. The source is live in the queue — just return and let the
|
||
// progress task handle the state transition when the previous source ends.
|
||
//
|
||
// Never for manual skips: the UI already jumped to this track in JS, but
|
||
// the current source is still playing until the chain drains. User-initiated
|
||
// play must clear the chain and start this URL immediately (standard path).
|
||
if gapless && !manual {
|
||
let already_chained = state.chained_info.lock().unwrap()
|
||
.as_ref()
|
||
.map(|c| same_playback_target(&c.url, &url))
|
||
.unwrap_or(false);
|
||
if already_chained {
|
||
return Ok(());
|
||
}
|
||
}
|
||
|
||
// ── Standard (new-sink) path ─────────────────────────────────────────────
|
||
// Used for: manual skip, gapless OFF, first play, or gapless when the
|
||
// proactive chain was not set up in time.
|
||
|
||
// Bump generation first so the old progress task stops before we peel
|
||
// chained_info (avoids a race where it sees current_done + empty chain).
|
||
let gen = state.generation.fetch_add(1, Ordering::SeqCst) + 1;
|
||
|
||
// Manual skip onto the gapless-pre-chained track: reuse raw bytes (no HTTP;
|
||
// preload cache was already consumed when the chain was built). Otherwise
|
||
// clear any stale chain metadata.
|
||
let reuse_chained_bytes: Option<Vec<u8>> = if gapless && manual {
|
||
let mut ci = state.chained_info.lock().unwrap();
|
||
if ci.as_ref().is_some_and(|c| same_playback_target(&c.url, &url)) {
|
||
ci.take().map(|info| {
|
||
Arc::try_unwrap(info.raw_bytes).unwrap_or_else(|a| (*a).clone())
|
||
})
|
||
} else {
|
||
*ci = None;
|
||
None
|
||
}
|
||
} else {
|
||
*state.chained_info.lock().unwrap() = None;
|
||
None
|
||
};
|
||
|
||
// Stop fading-out sink from previous crossfade.
|
||
if let Some(old) = state.fading_out_sink.lock().unwrap().take() {
|
||
old.stop();
|
||
}
|
||
|
||
// Fetch bytes (preload cache) unless we reused the chained download above.
|
||
let data = if let Some(d) = reuse_chained_bytes {
|
||
Some(d)
|
||
} else {
|
||
fetch_data(&url, &state, gen, &app).await?
|
||
};
|
||
let data = match data {
|
||
Some(d) => d,
|
||
None => return Ok(()), // superseded while downloading
|
||
};
|
||
|
||
if state.generation.load(Ordering::SeqCst) != gen {
|
||
return Ok(());
|
||
}
|
||
|
||
let (gain_linear, effective_volume) = compute_gain(replay_gain_db, replay_gain_peak, volume);
|
||
|
||
// Manual skips (user-initiated) bypass crossfade — the track should start immediately.
|
||
let crossfade_enabled = state.crossfade_enabled.load(Ordering::Relaxed) && !manual;
|
||
let crossfade_secs_val = f32::from_bits(state.crossfade_secs.load(Ordering::Relaxed)).clamp(0.5, 12.0);
|
||
|
||
// Measure how much audio Track A actually has left right now.
|
||
// By the time audio_play is called, near_end_ticks (2×500ms) + IPC latency
|
||
// have consumed ~500–800ms from Track A's tail — so its true remaining time
|
||
// is always less than crossfade_secs_val. Using the measured remaining time
|
||
// for BOTH fade-out (Track A) and fade-in (Track B) keeps them in sync and
|
||
// guarantees Track A reaches 0 exactly when its source exhausts.
|
||
let actual_fade_secs: f32 = if crossfade_enabled {
|
||
let cur = state.current.lock().unwrap();
|
||
let remaining = (cur.duration_secs - cur.position()) as f32;
|
||
remaining.clamp(0.1, crossfade_secs_val)
|
||
} else {
|
||
0.0
|
||
};
|
||
|
||
// Fade-in duration for Track B:
|
||
// crossfade → equal-power sin(t·π/2) over actual remaining time of Track A
|
||
// hard cut → 5 ms micro-fade to suppress DC-offset click
|
||
let fade_in_dur = if crossfade_enabled {
|
||
Duration::from_secs_f32(actual_fade_secs)
|
||
} else {
|
||
Duration::from_millis(5)
|
||
};
|
||
|
||
// Build source: decode → trim → resample → EQ → fade-in → fade-out → notify → count.
|
||
let done_flag = Arc::new(AtomicBool::new(false));
|
||
// Reset sample counter for the new track.
|
||
state.samples_played.store(0, Ordering::Relaxed);
|
||
// Always 0 — no application-level resampling. Rodio handles conversion to
|
||
// the output device rate internally; we let every track play at its native rate.
|
||
let target_rate: u32 = 0;
|
||
// Extract format hint from URL for better symphonia probing.
|
||
let format_hint = url.rsplit('.').next()
|
||
.and_then(|ext| ext.split('?').next())
|
||
.map(|s| s.to_lowercase());
|
||
let built = build_source(
|
||
data,
|
||
duration_hint,
|
||
state.eq_gains.clone(),
|
||
state.eq_enabled.clone(),
|
||
state.eq_pre_gain.clone(),
|
||
done_flag.clone(),
|
||
fade_in_dur,
|
||
state.samples_played.clone(),
|
||
target_rate,
|
||
format_hint.as_deref(),
|
||
).map_err(|e| { app.emit("audio:error", &e).ok(); e })?;
|
||
let source = built.source;
|
||
let duration_secs = built.duration_secs;
|
||
let output_rate = built.output_rate;
|
||
let output_channels = built.output_channels;
|
||
|
||
// Store the actual output rate/channels for position calculation.
|
||
state.current_sample_rate.store(output_rate, Ordering::Relaxed);
|
||
state.current_channels.store(output_channels as u32, Ordering::Relaxed);
|
||
|
||
if state.generation.load(Ordering::SeqCst) != gen {
|
||
return Ok(());
|
||
}
|
||
|
||
let sink = Sink::try_new(&*state.stream_handle).map_err(|e| e.to_string())?;
|
||
sink.set_volume(effective_volume);
|
||
|
||
// Gapless OFF: prepend a short silence so tracks are clearly separated.
|
||
// Only when this is an auto-advance (near end), not on manual skip.
|
||
if !gapless {
|
||
let cur_pos = {
|
||
let cur = state.current.lock().unwrap();
|
||
cur.position()
|
||
};
|
||
let cur_dur = {
|
||
let cur = state.current.lock().unwrap();
|
||
cur.duration_secs
|
||
};
|
||
let is_auto_advance = cur_dur > 3.0 && cur_pos >= cur_dur - 3.0;
|
||
if is_auto_advance {
|
||
let silence = rodio::source::Zero::<f32>::new(
|
||
source.channels(),
|
||
source.sample_rate(),
|
||
).take_duration(Duration::from_millis(500));
|
||
sink.append(silence);
|
||
}
|
||
}
|
||
|
||
sink.append(source);
|
||
|
||
// Atomically swap sinks — extract old sink + its fade-out trigger.
|
||
let (old_sink, old_fadeout_trigger, old_fadeout_samples) = {
|
||
let mut cur = state.current.lock().unwrap();
|
||
let old = cur.sink.take();
|
||
let old_fo_trigger = cur.fadeout_trigger.take();
|
||
let old_fo_samples = cur.fadeout_samples.take();
|
||
cur.sink = Some(sink);
|
||
cur.duration_secs = duration_secs;
|
||
cur.seek_offset = 0.0;
|
||
cur.play_started = Some(Instant::now());
|
||
cur.paused_at = None;
|
||
cur.replay_gain_linear = gain_linear;
|
||
cur.base_volume = volume.clamp(0.0, 1.0);
|
||
cur.fadeout_trigger = Some(built.fadeout_trigger);
|
||
cur.fadeout_samples = Some(built.fadeout_samples);
|
||
(old, old_fo_trigger, old_fo_samples)
|
||
};
|
||
|
||
// Handle old sink: symmetric crossfade or immediate stop.
|
||
if crossfade_enabled {
|
||
if let Some(old) = old_sink {
|
||
// Trigger sample-level fade-out on Track A via TriggeredFadeOut.
|
||
// Calculate total fade samples from the measured actual_fade_secs.
|
||
let rate = state.current_sample_rate.load(Ordering::Relaxed);
|
||
let ch = state.current_channels.load(Ordering::Relaxed);
|
||
let fade_total = (actual_fade_secs as f64 * rate as f64 * ch as f64) as u64;
|
||
|
||
if let (Some(trigger), Some(samples)) = (old_fadeout_trigger, old_fadeout_samples) {
|
||
samples.store(fade_total.max(1), Ordering::SeqCst);
|
||
trigger.store(true, Ordering::SeqCst);
|
||
}
|
||
|
||
// Keep old sink alive until the fade completes + small margin,
|
||
// then drop it. No volume stepping needed — the fade-out runs
|
||
// at sample level inside the audio thread.
|
||
*state.fading_out_sink.lock().unwrap() = Some(old);
|
||
let fo_arc = state.fading_out_sink.clone();
|
||
let cleanup_dur = Duration::from_secs_f32(actual_fade_secs + 0.5);
|
||
tokio::spawn(async move {
|
||
tokio::time::sleep(cleanup_dur).await;
|
||
if let Some(s) = fo_arc.lock().unwrap().take() {
|
||
s.stop();
|
||
}
|
||
});
|
||
}
|
||
} else if let Some(old) = old_sink {
|
||
old.stop();
|
||
}
|
||
|
||
app.emit("audio:playing", duration_secs).ok();
|
||
|
||
// ── Progress + ended detection ────────────────────────────────────────────
|
||
spawn_progress_task(
|
||
gen,
|
||
state.generation.clone(),
|
||
state.current.clone(),
|
||
state.chained_info.clone(),
|
||
state.crossfade_enabled.clone(),
|
||
state.crossfade_secs.clone(),
|
||
done_flag,
|
||
app,
|
||
state.samples_played.clone(),
|
||
state.current_sample_rate.clone(),
|
||
state.current_channels.clone(),
|
||
state.gapless_switch_at.clone(),
|
||
);
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Proactively appends the next track to the current Sink ~30 s before the
|
||
/// current track ends. Called from JS at the same trigger point as preload.
|
||
///
|
||
/// Because this runs well before the track boundary, the IPC round-trip is
|
||
/// irrelevant — by the time the current track actually ends, the next source
|
||
/// is already live in the Sink queue and rodio transitions at sample accuracy.
|
||
///
|
||
/// audio_play() checks chained_info.url on arrival: if it matches, it returns
|
||
/// immediately without touching the Sink (pure no-op on the audio path).
|
||
#[tauri::command]
|
||
pub async fn audio_chain_preload(
|
||
url: String,
|
||
volume: f32,
|
||
duration_hint: f64,
|
||
replay_gain_db: Option<f32>,
|
||
replay_gain_peak: Option<f32>,
|
||
state: State<'_, AudioEngine>,
|
||
) -> Result<(), String> {
|
||
// Idempotent: already chained this track → nothing to do.
|
||
{
|
||
let chained = state.chained_info.lock().unwrap();
|
||
if chained.as_ref().is_some_and(|c| same_playback_target(&c.url, &url)) {
|
||
return Ok(());
|
||
}
|
||
}
|
||
|
||
// Gapless must be enabled and a sink must exist.
|
||
if !state.gapless_enabled.load(Ordering::Relaxed) {
|
||
return Ok(());
|
||
}
|
||
|
||
let snapshot_gen = state.generation.load(Ordering::SeqCst);
|
||
|
||
// Fetch bytes — use preload cache if available, otherwise HTTP.
|
||
let data: Vec<u8> = {
|
||
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(d) = cached {
|
||
d
|
||
} else {
|
||
if let Some(path) = url.strip_prefix("psysonic-local://") {
|
||
tokio::fs::read(path).await.map_err(|e| e.to_string())?
|
||
} else {
|
||
let resp = state.http_client.get(&url).send().await
|
||
.map_err(|e| e.to_string())?;
|
||
if !resp.status().is_success() {
|
||
return Ok(()); // silently fail — audio_play will retry
|
||
}
|
||
let hint = resp.content_length().unwrap_or(0) as usize;
|
||
let mut stream = resp.bytes_stream();
|
||
let mut buf = Vec::with_capacity(hint);
|
||
while let Some(chunk) = stream.next().await {
|
||
if state.generation.load(Ordering::SeqCst) != snapshot_gen {
|
||
return Ok(()); // superseded by manual skip — abort download
|
||
}
|
||
buf.extend_from_slice(&chunk.map_err(|e| e.to_string())?);
|
||
}
|
||
buf
|
||
}
|
||
}
|
||
};
|
||
|
||
// Bail if the user skipped to a different track while we were downloading.
|
||
if state.generation.load(Ordering::SeqCst) != snapshot_gen {
|
||
return Ok(());
|
||
}
|
||
|
||
let raw_bytes = Arc::new(data);
|
||
|
||
let (gain_linear, effective_volume) = compute_gain(replay_gain_db, replay_gain_peak, volume);
|
||
|
||
let done_next = Arc::new(AtomicBool::new(false));
|
||
// Use a dedicated counter for the chained source — it will be swapped into
|
||
// samples_played when the chained track becomes active.
|
||
let chain_counter = Arc::new(AtomicU64::new(0));
|
||
// Always 0 — no application-level resampling (same as audio_play).
|
||
let target_rate: u32 = 0;
|
||
let format_hint = url.rsplit('.').next()
|
||
.and_then(|ext| ext.split('?').next())
|
||
.map(|s| s.to_lowercase());
|
||
let built = build_source(
|
||
(*raw_bytes).clone(),
|
||
duration_hint,
|
||
state.eq_gains.clone(),
|
||
state.eq_enabled.clone(),
|
||
state.eq_pre_gain.clone(),
|
||
done_next.clone(),
|
||
Duration::ZERO, // gapless: no fade-in — sample-accurate boundary, no click
|
||
chain_counter.clone(),
|
||
target_rate,
|
||
format_hint.as_deref(),
|
||
).map_err(|e| e.to_string())?;
|
||
let source = built.source;
|
||
let duration_secs = built.duration_secs;
|
||
|
||
// Final gen check — reject if a manual skip happened during decode.
|
||
if state.generation.load(Ordering::SeqCst) != snapshot_gen {
|
||
return Ok(());
|
||
}
|
||
|
||
// Append to the existing Sink. The audio hardware stream never stalls.
|
||
{
|
||
let cur = state.current.lock().unwrap();
|
||
match &cur.sink {
|
||
Some(sink) => {
|
||
sink.set_volume(effective_volume);
|
||
sink.append(source);
|
||
}
|
||
None => return Ok(()), // playback stopped — bail
|
||
}
|
||
}
|
||
|
||
*state.chained_info.lock().unwrap() = Some(ChainedInfo {
|
||
url,
|
||
raw_bytes,
|
||
duration_secs,
|
||
replay_gain_linear: gain_linear,
|
||
base_volume: volume.clamp(0.0, 1.0),
|
||
source_done: done_next,
|
||
sample_counter: chain_counter,
|
||
});
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Spawns the per-generation progress + ended-detection task.
|
||
///
|
||
/// The task owns a local `done: Arc<AtomicBool>` reference that starts as
|
||
/// the current track's done flag. When the progress task detects that the
|
||
/// done flag is set AND `chained_info` has data, it swaps `done` to the
|
||
/// chained source's flag and transitions state — all without creating a new
|
||
/// task or changing the generation counter.
|
||
///
|
||
/// Key changes from the previous implementation:
|
||
/// • 100 ms tick (was 500 ms) — halves worst-case event latency
|
||
/// • Position from atomic sample counter (no wall-clock drift)
|
||
/// • Immediate `audio:track_switched` event at decoder boundary
|
||
/// • `audio:ended` only fires when no chained successor exists
|
||
fn spawn_progress_task(
|
||
gen: u64,
|
||
gen_counter: Arc<AtomicU64>,
|
||
current_arc: Arc<Mutex<AudioCurrent>>,
|
||
chained_arc: Arc<Mutex<Option<ChainedInfo>>>,
|
||
crossfade_enabled_arc: Arc<AtomicBool>,
|
||
crossfade_secs_arc: Arc<AtomicU32>,
|
||
initial_done: Arc<AtomicBool>,
|
||
app: AppHandle,
|
||
samples_played: Arc<AtomicU64>,
|
||
sample_rate_arc: Arc<AtomicU32>,
|
||
channels_arc: Arc<AtomicU32>,
|
||
gapless_switch_at: Arc<AtomicU64>,
|
||
) {
|
||
tokio::spawn(async move {
|
||
let mut near_end_ticks: u32 = 0;
|
||
// Local done-flag reference; swapped on gapless transition.
|
||
let mut current_done = initial_done;
|
||
// Local sample counter; swapped to chained source's counter on transition.
|
||
let mut samples_played = samples_played;
|
||
|
||
loop {
|
||
// 100 ms tick — tight enough for responsive UI, low enough CPU cost.
|
||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||
|
||
if gen_counter.load(Ordering::SeqCst) != gen {
|
||
break;
|
||
}
|
||
|
||
// ── Gapless transition detection ─────────────────────────────────
|
||
// If the current source is exhausted AND we have a chained track
|
||
// ready, transition seamlessly: swap tracking state, emit
|
||
// audio:track_switched for the new track, and continue the loop.
|
||
if current_done.load(Ordering::SeqCst) {
|
||
// Radio (dur == 0): stream exhausted / connection dropped → stop.
|
||
let cur_dur = current_arc.lock().unwrap().duration_secs;
|
||
if cur_dur <= 0.0 {
|
||
eprintln!("[radio] current_done fired → emitting audio:ended (dur=0)");
|
||
gen_counter.fetch_add(1, Ordering::SeqCst);
|
||
app.emit("audio:ended", ()).ok();
|
||
break;
|
||
}
|
||
|
||
let chained = chained_arc.lock().unwrap().take();
|
||
if let Some(info) = chained {
|
||
// Swap to the chained source's done flag.
|
||
current_done = info.source_done;
|
||
|
||
// Swap to the chained source's sample counter.
|
||
// The chained CountingSource increments its own Arc,
|
||
// so we must rebind our local reference to it —
|
||
// a one-time value copy would go stale immediately.
|
||
samples_played = info.sample_counter;
|
||
|
||
// Update tracking state.
|
||
{
|
||
let mut cur = current_arc.lock().unwrap();
|
||
cur.replay_gain_linear = info.replay_gain_linear;
|
||
cur.base_volume = info.base_volume;
|
||
cur.duration_secs = info.duration_secs;
|
||
cur.seek_offset = 0.0;
|
||
cur.play_started = Some(Instant::now());
|
||
}
|
||
|
||
// Record the gapless switch timestamp for ghost-command guard.
|
||
let switch_ts = std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.unwrap_or_default()
|
||
.as_millis() as u64;
|
||
gapless_switch_at.store(switch_ts, Ordering::SeqCst);
|
||
|
||
// Emit the new track_switched event — this is immediate,
|
||
// not delayed by 500 ms like the old audio:playing was.
|
||
app.emit("audio:track_switched", info.duration_secs).ok();
|
||
near_end_ticks = 0;
|
||
continue;
|
||
}
|
||
// Current source exhausted but no chain queued — the Sink is
|
||
// likely draining; audio:ended will fire on the next tick via
|
||
// the near-end logic below.
|
||
}
|
||
|
||
// ── Position from atomic sample counter ──────────────────────────
|
||
let rate = sample_rate_arc.load(Ordering::Relaxed) as f64;
|
||
let ch = channels_arc.load(Ordering::Relaxed) as f64;
|
||
let samples = samples_played.load(Ordering::Relaxed) as f64;
|
||
let divisor = (rate * ch).max(1.0);
|
||
|
||
let dur = {
|
||
let cur = current_arc.lock().unwrap();
|
||
cur.duration_secs
|
||
};
|
||
let is_paused = {
|
||
let cur = current_arc.lock().unwrap();
|
||
cur.paused_at.is_some()
|
||
};
|
||
|
||
let pos = if is_paused {
|
||
let cur = current_arc.lock().unwrap();
|
||
cur.paused_at.unwrap_or(0.0)
|
||
} else {
|
||
(samples / divisor).min(dur.max(0.001))
|
||
};
|
||
|
||
app.emit("audio:progress", ProgressPayload { current_time: pos, duration: dur }).ok();
|
||
|
||
if is_paused {
|
||
continue;
|
||
}
|
||
|
||
let cf_enabled = crossfade_enabled_arc.load(Ordering::Relaxed);
|
||
let cf_secs = f32::from_bits(crossfade_secs_arc.load(Ordering::Relaxed)).clamp(0.5, 12.0) as f64;
|
||
let end_threshold = if cf_enabled { cf_secs.max(1.0) } else { 1.0 };
|
||
|
||
if dur > end_threshold && pos >= dur - end_threshold {
|
||
near_end_ticks += 1;
|
||
// At 100 ms ticks, 10 ticks ≈ 1 s — equivalent to the old 2×500ms.
|
||
if near_end_ticks >= 10 {
|
||
// If a gapless chain is pending, the source hasn't
|
||
// exhausted yet — duration_hint (integer seconds from
|
||
// Subsonic) is shorter than the actual audio content.
|
||
// Don't emit audio:ended; let the gapless transition
|
||
// handle it when current_done fires.
|
||
let has_chain = chained_arc.lock().unwrap().is_some();
|
||
if has_chain {
|
||
continue;
|
||
}
|
||
gen_counter.fetch_add(1, Ordering::SeqCst);
|
||
app.emit("audio:ended", ()).ok();
|
||
break;
|
||
}
|
||
} else {
|
||
near_end_ticks = 0;
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_pause(state: State<'_, AudioEngine>) {
|
||
let mut cur = state.current.lock().unwrap();
|
||
if let Some(sink) = &cur.sink {
|
||
if !sink.is_paused() {
|
||
let pos = cur.position();
|
||
sink.pause();
|
||
cur.paused_at = Some(pos);
|
||
cur.play_started = None;
|
||
}
|
||
}
|
||
// Notify the download task so it can start measuring the hard-pause stall timer.
|
||
if let Some(rs) = state.radio_state.lock().unwrap().as_ref() {
|
||
rs.flags.is_paused.store(true, Ordering::Release);
|
||
}
|
||
}
|
||
|
||
/// Resume playback.
|
||
///
|
||
/// **Warm resume** (`is_hard_paused = false`): download task is still running,
|
||
/// buffer has buffered audio. `sink.play()` suffices.
|
||
///
|
||
/// **Cold resume** (`is_hard_paused = true`): TCP was dropped. A fresh 4 MB
|
||
/// ring buffer is created, its consumer is sent to `AudioStreamReader` (which
|
||
/// swaps it in on the next `read()`), and a new download task is spawned.
|
||
#[tauri::command]
|
||
pub async fn audio_resume(state: State<'_, AudioEngine>, app: AppHandle) -> Result<(), String> {
|
||
// Detect radio hard-disconnect.
|
||
let reconnect_info = {
|
||
let guard = state.radio_state.lock().unwrap();
|
||
guard
|
||
.as_ref()
|
||
.filter(|rs| rs.flags.is_hard_paused.load(Ordering::Acquire))
|
||
.map(|rs| (rs.url.clone(), rs.gen, rs.flags.clone()))
|
||
};
|
||
|
||
if let Some((url, gen, flags)) = reconnect_info {
|
||
let rb = HeapRb::<u8>::new(RADIO_BUF_CAPACITY);
|
||
let (new_prod, new_cons) = rb.split();
|
||
|
||
// Send new consumer to AudioStreamReader (non-blocking; unbounded channel).
|
||
let ok = flags.new_cons_tx.lock().unwrap().send(new_cons).is_ok();
|
||
|
||
if ok {
|
||
let new_task = tokio::spawn(radio_download_task(
|
||
gen,
|
||
state.generation.clone(),
|
||
None, // task performs its own fresh GET
|
||
state.http_client.clone(),
|
||
url,
|
||
new_prod,
|
||
flags.clone(),
|
||
app,
|
||
));
|
||
if let Some(rs) = state.radio_state.lock().unwrap().as_mut() {
|
||
let old = std::mem::replace(&mut rs.task, new_task);
|
||
old.abort(); // ensure any lingering old task is gone
|
||
rs.flags.is_hard_paused.store(false, Ordering::Release);
|
||
rs.flags.is_paused.store(false, Ordering::Release);
|
||
}
|
||
} else {
|
||
eprintln!("[radio] resume: AudioStreamReader gone — skipping reconnect");
|
||
}
|
||
}
|
||
|
||
// Resume the rodio Sink (works for both warm and cold resume).
|
||
{
|
||
let mut cur = state.current.lock().unwrap();
|
||
if let Some(sink) = &cur.sink {
|
||
if sink.is_paused() {
|
||
let pos = cur.paused_at.unwrap_or(cur.seek_offset);
|
||
sink.play();
|
||
cur.seek_offset = pos;
|
||
cur.play_started = Some(Instant::now());
|
||
cur.paused_at = None;
|
||
}
|
||
}
|
||
}
|
||
if let Some(rs) = state.radio_state.lock().unwrap().as_ref() {
|
||
rs.flags.is_paused.store(false, Ordering::Release);
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_stop(state: State<'_, AudioEngine>) {
|
||
state.generation.fetch_add(1, Ordering::SeqCst);
|
||
*state.chained_info.lock().unwrap() = None;
|
||
// Drop RadioLiveState → triggers Drop → task.abort() → TCP released.
|
||
drop(state.radio_state.lock().unwrap().take());
|
||
let mut cur = state.current.lock().unwrap();
|
||
if let Some(sink) = cur.sink.take() { sink.stop(); }
|
||
cur.duration_secs = 0.0;
|
||
cur.seek_offset = 0.0;
|
||
cur.play_started = None;
|
||
cur.paused_at = None;
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_seek(seconds: f64, state: State<'_, AudioEngine>) -> Result<(), String> {
|
||
// Ghost-command guard: reject seeks within 500 ms of a gapless auto-advance.
|
||
{
|
||
let switch_ms = state.gapless_switch_at.load(Ordering::SeqCst);
|
||
if switch_ms > 0 {
|
||
let now_ms = std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.unwrap_or_default()
|
||
.as_millis() as u64;
|
||
if now_ms.saturating_sub(switch_ms) < 500 {
|
||
return Ok(());
|
||
}
|
||
}
|
||
}
|
||
|
||
// Seeking back invalidates any pending gapless chain.
|
||
let cur_pos = {
|
||
let cur = state.current.lock().unwrap();
|
||
cur.position()
|
||
};
|
||
if seconds < cur_pos - 1.0 {
|
||
*state.chained_info.lock().unwrap() = None;
|
||
}
|
||
|
||
let mut cur = state.current.lock().unwrap();
|
||
if cur.sink.is_none() { return Ok(()); }
|
||
|
||
cur.sink.as_ref().unwrap()
|
||
.try_seek(Duration::from_secs_f64(seconds.max(0.0)))
|
||
.map_err(|e| e.to_string())?;
|
||
|
||
if cur.paused_at.is_some() {
|
||
cur.paused_at = Some(seconds);
|
||
} else {
|
||
cur.seek_offset = seconds;
|
||
cur.play_started = Some(Instant::now());
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_set_volume(volume: f32, state: State<'_, AudioEngine>) {
|
||
let mut cur = state.current.lock().unwrap();
|
||
cur.base_volume = volume.clamp(0.0, 1.0);
|
||
if let Some(sink) = &cur.sink {
|
||
sink.set_volume((cur.base_volume * cur.replay_gain_linear * MASTER_HEADROOM).clamp(0.0, 1.0));
|
||
}
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_update_replay_gain(
|
||
volume: f32,
|
||
replay_gain_db: Option<f32>,
|
||
replay_gain_peak: Option<f32>,
|
||
state: State<'_, AudioEngine>,
|
||
) {
|
||
let (gain_linear, effective) = compute_gain(replay_gain_db, replay_gain_peak, volume);
|
||
let mut cur = state.current.lock().unwrap();
|
||
cur.replay_gain_linear = gain_linear;
|
||
cur.base_volume = volume.clamp(0.0, 1.0);
|
||
if let Some(sink) = &cur.sink {
|
||
sink.set_volume(effective);
|
||
}
|
||
}
|
||
|
||
/// Proxy: fetches https://autoeq.app/entries via Rust to bypass WebView CORS restrictions.
|
||
#[tauri::command]
|
||
pub async fn autoeq_entries(state: State<'_, AudioEngine>) -> Result<String, String> {
|
||
state.http_client
|
||
.get("https://autoeq.app/entries")
|
||
.send().await.map_err(|e| e.to_string())?
|
||
.text().await.map_err(|e| e.to_string())
|
||
}
|
||
|
||
/// Fetches the AutoEQ FixedBandEQ profile for a specific headphone from GitHub raw content.
|
||
///
|
||
/// Directory layout in the AutoEQ repo:
|
||
/// results/{source}/{form}/{name}/{name} FixedBandEQ.txt (most sources)
|
||
/// results/{source}/{rig} {form}/{name}/{name} FixedBandEQ.txt (crinacle — rig-prefixed dir)
|
||
///
|
||
/// We try the rig-prefixed path first (when rig is present), then fall back to form-only.
|
||
#[tauri::command]
|
||
pub async fn autoeq_fetch_profile(
|
||
name: String,
|
||
source: String,
|
||
rig: Option<String>,
|
||
form: String,
|
||
state: State<'_, AudioEngine>,
|
||
) -> Result<String, String> {
|
||
let base = "https://raw.githubusercontent.com/jaakkopasanen/AutoEq/master/results";
|
||
let filename = format!("{} FixedBandEQ.txt", name);
|
||
|
||
let candidates: Vec<String> = if let Some(ref r) = rig {
|
||
vec![
|
||
format!("{}/{}/{} {}/{}/{}", base, source, r, form, name, filename),
|
||
format!("{}/{}/{}/{}/{}", base, source, form, name, filename),
|
||
]
|
||
} else {
|
||
vec![format!("{}/{}/{}/{}/{}", base, source, form, name, filename)]
|
||
};
|
||
|
||
for url in &candidates {
|
||
let resp = state.http_client.get(url).send().await.map_err(|e| e.to_string())?;
|
||
if resp.status().is_success() {
|
||
return resp.text().await.map_err(|e| e.to_string());
|
||
}
|
||
}
|
||
|
||
Err(format!("FixedBandEQ profile not found for '{}'", name))
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_set_eq(gains: [f32; 10], enabled: bool, pre_gain: f32, state: State<'_, AudioEngine>) {
|
||
state.eq_enabled.store(enabled, Ordering::Relaxed);
|
||
state.eq_pre_gain.store(pre_gain.clamp(-30.0, 6.0).to_bits(), Ordering::Relaxed);
|
||
for (i, &gain) in gains.iter().enumerate() {
|
||
state.eq_gains[i].store(gain.clamp(-12.0, 12.0).to_bits(), Ordering::Relaxed);
|
||
}
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub async fn audio_preload(
|
||
url: String,
|
||
duration_hint: f64,
|
||
state: State<'_, AudioEngine>,
|
||
) -> Result<(), String> {
|
||
{
|
||
let preloaded = state.preloaded.lock().unwrap();
|
||
if preloaded.as_ref().is_some_and(|p| same_playback_target(&p.url, &url)) {
|
||
return Ok(());
|
||
}
|
||
}
|
||
let data: Vec<u8> = if let Some(path) = url.strip_prefix("psysonic-local://") {
|
||
tokio::fs::read(path).await.map_err(|e| e.to_string())?
|
||
} else {
|
||
let response = state.http_client.get(&url).send().await.map_err(|e| e.to_string())?;
|
||
if !response.status().is_success() {
|
||
return Ok(());
|
||
}
|
||
response.bytes().await.map_err(|e| e.to_string())?.into()
|
||
};
|
||
let _ = duration_hint; // kept in API for compatibility
|
||
*state.preloaded.lock().unwrap() = Some(PreloadedTrack { url, data });
|
||
Ok(())
|
||
}
|
||
|
||
/// Play a live internet radio stream.
|
||
///
|
||
/// Sends `Icy-MetaData: 1` to request inline ICY metadata.
|
||
/// Emits `audio:playing` with `duration = 0.0` (sentinel for live stream)
|
||
/// and `radio:metadata` whenever the StreamTitle changes.
|
||
#[tauri::command]
|
||
pub async fn audio_play_radio(
|
||
url: String,
|
||
volume: f32,
|
||
app: AppHandle,
|
||
state: State<'_, AudioEngine>,
|
||
) -> Result<(), String> {
|
||
let gen = state.generation.fetch_add(1, Ordering::SeqCst) + 1;
|
||
|
||
// Abort any previous radio task before stopping the sink.
|
||
drop(state.radio_state.lock().unwrap().take());
|
||
|
||
*state.chained_info.lock().unwrap() = None;
|
||
{
|
||
let mut cur = state.current.lock().unwrap();
|
||
if let Some(old) = cur.sink.take() { old.stop(); }
|
||
}
|
||
if let Some(old) = state.fading_out_sink.lock().unwrap().take() { old.stop(); }
|
||
|
||
// ── Open initial HTTP connection ──────────────────────────────────────────
|
||
let response = state.http_client
|
||
.get(&url)
|
||
.header("Icy-MetaData", "1")
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
let m = format!("radio: connection failed: {e}");
|
||
app.emit("audio:error", &m).ok();
|
||
m
|
||
})?;
|
||
|
||
if !response.status().is_success() {
|
||
let m = format!("radio: HTTP {}", response.status());
|
||
app.emit("audio:error", &m).ok();
|
||
return Err(m);
|
||
}
|
||
|
||
let fmt_hint = content_type_to_hint(
|
||
response.headers()
|
||
.get("content-type")
|
||
.and_then(|v| v.to_str().ok())
|
||
.unwrap_or(""),
|
||
);
|
||
|
||
// ── Build 4 MB lock-free SPSC ring buffer ─────────────────────────────────
|
||
let rb = HeapRb::<u8>::new(RADIO_BUF_CAPACITY);
|
||
let (prod, cons) = rb.split();
|
||
|
||
let (new_cons_tx, new_cons_rx) = std::sync::mpsc::channel::<HeapConsumer<u8>>();
|
||
let flags = Arc::new(RadioSharedFlags {
|
||
is_paused: AtomicBool::new(false),
|
||
is_hard_paused: AtomicBool::new(false),
|
||
new_cons_tx: Mutex::new(new_cons_tx),
|
||
});
|
||
|
||
// ── Spawn download task ───────────────────────────────────────────────────
|
||
let task = tokio::spawn(radio_download_task(
|
||
gen,
|
||
state.generation.clone(),
|
||
Some(response),
|
||
state.http_client.clone(),
|
||
url.clone(),
|
||
prod,
|
||
flags.clone(),
|
||
app.clone(),
|
||
));
|
||
|
||
*state.radio_state.lock().unwrap() = Some(RadioLiveState {
|
||
url: url.clone(),
|
||
gen,
|
||
task,
|
||
flags: flags.clone(),
|
||
});
|
||
|
||
// ── Build Symphonia decoder in a blocking thread ──────────────────────────
|
||
let reader = AudioStreamReader {
|
||
cons,
|
||
new_cons_rx: Mutex::new(new_cons_rx),
|
||
deadline: std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS),
|
||
gen_arc: state.generation.clone(),
|
||
gen,
|
||
pos: 0,
|
||
};
|
||
|
||
if state.generation.load(Ordering::SeqCst) != gen { return Ok(()); }
|
||
|
||
let hint_clone = fmt_hint.clone();
|
||
let decoder = tokio::task::spawn_blocking(move || {
|
||
SizedDecoder::new_streaming(Box::new(reader), hint_clone.as_deref())
|
||
})
|
||
.await
|
||
.map_err(|e| e.to_string())??;
|
||
|
||
if state.generation.load(Ordering::SeqCst) != gen { return Ok(()); }
|
||
|
||
let sample_rate = decoder.sample_rate();
|
||
let channels = decoder.channels();
|
||
let done_flag = Arc::new(AtomicBool::new(false));
|
||
let fadeout_trigger = Arc::new(AtomicBool::new(false));
|
||
let fadeout_samples = Arc::new(AtomicU64::new(0));
|
||
state.samples_played.store(0, Ordering::Relaxed);
|
||
|
||
// Radio: no gapless trim, no ReplayGain, 5 ms fade-in to suppress click.
|
||
let dyn_src = DynSource::new(decoder.convert_samples::<f32>());
|
||
let eq_src = EqSource::new(dyn_src, state.eq_gains.clone(),
|
||
state.eq_enabled.clone(), state.eq_pre_gain.clone());
|
||
let fade_in = EqualPowerFadeIn::new(eq_src, Duration::from_millis(5));
|
||
let fade_out = TriggeredFadeOut::new(fade_in, fadeout_trigger.clone(), fadeout_samples.clone());
|
||
let notifying = NotifyingSource::new(fade_out, done_flag.clone());
|
||
let counting = CountingSource::new(notifying, state.samples_played.clone());
|
||
|
||
if state.generation.load(Ordering::SeqCst) != gen { return Ok(()); }
|
||
|
||
let sink = Sink::try_new(&*state.stream_handle).map_err(|e| e.to_string())?;
|
||
sink.set_volume((volume.clamp(0.0, 1.0) * MASTER_HEADROOM).clamp(0.0, 1.0));
|
||
sink.append(counting);
|
||
|
||
{
|
||
let mut cur = state.current.lock().unwrap();
|
||
if let Some(old) = cur.sink.take() { old.stop(); }
|
||
cur.sink = Some(sink);
|
||
cur.duration_secs = 0.0; // sentinel: live stream
|
||
cur.seek_offset = 0.0;
|
||
cur.play_started = Some(Instant::now());
|
||
cur.paused_at = None;
|
||
cur.replay_gain_linear = 1.0;
|
||
cur.base_volume = volume.clamp(0.0, 1.0);
|
||
cur.fadeout_trigger = Some(fadeout_trigger);
|
||
cur.fadeout_samples = Some(fadeout_samples);
|
||
}
|
||
|
||
state.current_sample_rate.store(sample_rate, Ordering::Relaxed);
|
||
state.current_channels.store(channels as u32, Ordering::Relaxed);
|
||
|
||
app.emit("audio:playing", 0.0f64).ok();
|
||
|
||
spawn_progress_task(
|
||
gen,
|
||
state.generation.clone(),
|
||
state.current.clone(),
|
||
state.chained_info.clone(),
|
||
state.crossfade_enabled.clone(),
|
||
state.crossfade_secs.clone(),
|
||
done_flag,
|
||
app,
|
||
state.samples_played.clone(),
|
||
state.current_sample_rate.clone(),
|
||
state.current_channels.clone(),
|
||
state.gapless_switch_at.clone(),
|
||
);
|
||
|
||
Ok(())
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_set_crossfade(enabled: bool, secs: f32, state: State<'_, AudioEngine>) {
|
||
state.crossfade_enabled.store(enabled, Ordering::Relaxed);
|
||
state.crossfade_secs.store(secs.clamp(0.1, 12.0).to_bits(), Ordering::Relaxed);
|
||
}
|
||
|
||
#[tauri::command]
|
||
pub fn audio_set_gapless(enabled: bool, state: State<'_, AudioEngine>) {
|
||
state.gapless_enabled.store(enabled, Ordering::Relaxed);
|
||
}
|