use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;
use symphonia::core::audio::SampleBuffer;
use symphonia::core::codecs::{
CODEC_TYPE_AAC, CODEC_TYPE_ALAC, CODEC_TYPE_FLAC, CODEC_TYPE_MP3, CODEC_TYPE_OPUS,
CODEC_TYPE_PCM_F32LE, CODEC_TYPE_PCM_S16LE, CODEC_TYPE_PCM_S24LE, CODEC_TYPE_PCM_S32LE,
CODEC_TYPE_VORBIS, CODEC_TYPE_WAVPACK, CodecType, DecoderOptions,
};
use symphonia::core::formats::{FormatOptions, SeekMode, SeekTo};
use symphonia::core::io::MediaSourceStream;
use symphonia::core::meta::MetadataOptions;
use symphonia::core::probe::Hint;
use symphonia::core::units::Time;
use thiserror::Error;
use crate::audio::viz::VizBuffer;
use crate::config::ReplayGainMode;
use crate::player::state::QueueItemId;
#[derive(Debug, Error)]
pub enum DecodeError {
#[error("failed to open file: {0}")]
Io(#[from] std::io::Error),
#[error("no supported audio track found")]
NoTrack,
#[error("unsupported codec")]
UnsupportedCodec,
#[error("decode error: {0}")]
Decode(String),
}
#[derive(Debug, Clone)]
pub struct StreamInfo {
pub codec: String,
pub sample_rate: u32,
pub channels: u16,
pub bit_depth: u16,
pub duration_ms: u64,
}
pub struct DecodeHandle {
stop: Arc<AtomicBool>,
thread: Option<thread::JoinHandle<()>>,
}
impl DecodeHandle {
pub fn signal_stop(&self) {
self.stop.store(true, Ordering::Relaxed);
}
pub fn stop(&mut self) {
self.signal_stop();
if let Some(handle) = self.thread.take()
&& let Err(payload) = handle.join()
{
let msg = payload
.downcast_ref::<String>()
.map(|s| s.as_str())
.or_else(|| payload.downcast_ref::<&str>().copied())
.unwrap_or("unknown");
log::error!("decode thread panicked: {}", msg);
}
}
}
impl Drop for DecodeHandle {
fn drop(&mut self) {
self.stop();
}
}
#[derive(Debug, Clone)]
pub struct TrackBoundary {
pub id: QueueItemId,
pub path: PathBuf,
pub info: StreamInfo,
pub sample_offset: u64,
pub samples_written: u64,
pub seek_samples: u64,
}
pub struct PlaybackTimeline {
boundaries: parking_lot::RwLock<Vec<TrackBoundary>>,
samples_written: AtomicU64,
pub samples_played: Arc<AtomicU64>,
}
impl PlaybackTimeline {
pub fn new() -> Arc<Self> {
Arc::new(Self {
boundaries: parking_lot::RwLock::new(Vec::new()),
samples_written: AtomicU64::new(0),
samples_played: Arc::new(AtomicU64::new(0)),
})
}
fn push_boundary(&self, boundary: TrackBoundary) {
self.boundaries.write().push(boundary);
}
fn add_written(&self, count: u64) {
self.samples_written.fetch_add(count, Ordering::Relaxed);
let mut bounds = self.boundaries.write();
if let Some(last) = bounds.last_mut() {
last.samples_written += count;
}
}
pub fn reset(&self) {
self.boundaries.write().clear();
self.samples_written.store(0, Ordering::Relaxed);
self.samples_played.store(0, Ordering::Relaxed);
}
pub fn samples_played_counter(&self) -> Arc<AtomicU64> {
self.samples_played.clone()
}
pub fn current_playback(&self) -> Option<(QueueItemId, PathBuf, StreamInfo, u64)> {
let bounds = self.boundaries.read();
if bounds.is_empty() {
return None;
}
let played = self.samples_played.load(Ordering::Acquire);
let idx = bounds.partition_point(|b| b.sample_offset <= played);
let current = if idx > 0 {
&bounds[idx - 1]
} else {
return None;
};
let ch = current.info.channels as u64;
let rate = current.info.sample_rate as u64;
if ch == 0 || rate == 0 {
return None;
}
let track_samples = played.saturating_sub(current.sample_offset);
let position_ms =
(track_samples / ch) * 1000 / rate + (current.seek_samples / ch) * 1000 / rate;
Some((
current.id,
current.path.clone(),
current.info.clone(),
position_ms,
))
}
}
pub struct SourceEntry {
pub id: QueueItemId,
pub path: PathBuf,
pub hint: Hint,
pub make_mss: Box<dyn FnOnce() -> std::io::Result<MediaSourceStream> + Send>,
}
impl SourceEntry {
pub fn from_file(id: QueueItemId, path: PathBuf) -> Self {
let ext = path
.extension()
.and_then(|e| e.to_str())
.unwrap_or("")
.to_string();
let path_clone = path.clone();
let mut hint = Hint::new();
if !ext.is_empty() {
hint.with_extension(&ext);
}
Self {
id,
path,
hint,
make_mss: Box::new(move || {
let file = File::open(&path_clone)?;
Ok(MediaSourceStream::new(Box::new(file), Default::default()))
}),
}
}
}
pub fn probe_source(mss: MediaSourceStream, hint: &Hint) -> Result<StreamInfo, DecodeError> {
probe_mss(mss, hint)
}
pub fn probe_file(path: &Path) -> Result<StreamInfo, DecodeError> {
let file = File::open(path)?;
let mss = MediaSourceStream::new(Box::new(file), Default::default());
let mut hint = Hint::new();
if let Some(ext) = path.extension().and_then(|e| e.to_str()) {
hint.with_extension(ext);
}
probe_mss(mss, &hint)
}
fn probe_mss(mss: MediaSourceStream, hint: &Hint) -> Result<StreamInfo, DecodeError> {
let probed = symphonia::default::get_probe()
.format(
hint,
mss,
&FormatOptions::default(),
&MetadataOptions::default(),
)
.map_err(|e| DecodeError::Decode(e.to_string()))?;
let reader = probed.format;
let track = reader.default_track().ok_or(DecodeError::NoTrack)?;
let codec_params = &track.codec_params;
let sample_rate = codec_params.sample_rate.unwrap_or(44100);
let channels = codec_params.channels.map(|c| c.count() as u16).unwrap_or(2);
let bit_depth = codec_params.bits_per_sample.unwrap_or(16) as u16;
let duration_ms = track
.codec_params
.n_frames
.map(|frames| frames * 1000 / sample_rate as u64)
.unwrap_or(0);
let codec = codec_name(codec_params.codec);
Ok(StreamInfo {
codec,
sample_rate,
channels,
bit_depth,
duration_ms,
})
}
#[allow(clippy::too_many_arguments)]
pub fn start_decode<N>(
first: SourceEntry,
producer: rtrb::Producer<f32>,
seek_ms: u64,
next_track: N,
timeline: Arc<PlaybackTimeline>,
viz_buffer: Option<Arc<VizBuffer>>,
rg_mode: ReplayGainMode,
pre_amp_db: f64,
) -> Result<(StreamInfo, DecodeHandle), DecodeError>
where
N: Fn() -> Option<SourceEntry> + Send + 'static,
{
let stop = Arc::new(AtomicBool::new(false));
let stop_clone = stop.clone();
let thread = thread::Builder::new()
.name("koan-decode".into())
.spawn(move || {
decode_queue_loop(
first,
producer,
&stop_clone,
seek_ms,
&next_track,
&timeline,
viz_buffer.as_deref(),
rg_mode,
pre_amp_db,
);
})
.map_err(DecodeError::Io)?;
let placeholder = StreamInfo {
codec: String::from("?"),
sample_rate: 44100,
channels: 2,
bit_depth: 16,
duration_ms: 0,
};
Ok((
placeholder,
DecodeHandle {
stop,
thread: Some(thread),
},
))
}
#[allow(clippy::too_many_arguments)]
pub fn start_decode_file<N>(
initial_id: QueueItemId,
path: &Path,
producer: rtrb::Producer<f32>,
seek_ms: u64,
next_track: N,
timeline: Arc<PlaybackTimeline>,
viz_buffer: Option<Arc<VizBuffer>>,
rg_mode: ReplayGainMode,
pre_amp_db: f64,
) -> Result<(StreamInfo, DecodeHandle), DecodeError>
where
N: Fn() -> Option<(QueueItemId, PathBuf)> + Send + 'static,
{
let info = probe_file(path)?;
let first = SourceEntry::from_file(initial_id, path.to_path_buf());
let (_, handle) = start_decode(
first,
producer,
seek_ms,
move || {
let (id, p) = next_track()?;
Some(SourceEntry::from_file(id, p))
},
timeline,
viz_buffer,
rg_mode,
pre_amp_db,
)?;
Ok((info, handle))
}
#[allow(clippy::too_many_arguments)]
fn decode_queue_loop<N>(
first: SourceEntry,
mut producer: rtrb::Producer<f32>,
stop: &AtomicBool,
initial_seek_ms: u64,
next_track: &N,
timeline: &PlaybackTimeline,
viz_buffer: Option<&VizBuffer>,
rg_mode: ReplayGainMode,
pre_amp_db: f64,
) where
N: Fn() -> Option<SourceEntry>,
{
let path = first.path.clone();
let hint = first.hint.clone();
let mss = match (first.make_mss)() {
Ok(mss) => mss,
Err(e) => {
if !stop.load(Ordering::Relaxed) {
log::error!("failed to open {}: {}", path.display(), e);
}
return;
}
};
if let Err(e) = decode_single(
first.id,
&path,
&hint,
mss,
&mut producer,
stop,
initial_seek_ms,
timeline,
viz_buffer,
rg_mode,
pre_amp_db,
) {
if !stop.load(Ordering::Relaxed) {
log::error!("decode error on {}: {}", path.display(), e);
}
return;
}
while !stop.load(Ordering::Relaxed) {
let Some(entry) = (next_track)() else {
log::info!("playlist exhausted, decode thread finishing");
break;
};
log::info!("gapless transition → {}", entry.path.display());
let next_path = entry.path.clone();
let next_hint = entry.hint.clone();
let next_mss = match (entry.make_mss)() {
Ok(mss) => mss,
Err(e) => {
if !stop.load(Ordering::Relaxed) {
log::error!("failed to open {}: {}", next_path.display(), e);
}
break;
}
};
if let Err(e) = decode_single(
entry.id,
&next_path,
&next_hint,
next_mss,
&mut producer,
stop,
0,
timeline,
viz_buffer,
rg_mode,
pre_amp_db,
) {
if !stop.load(Ordering::Relaxed) {
log::error!("decode error on {}: {}", next_path.display(), e);
}
break;
}
}
}
#[allow(clippy::too_many_arguments)]
fn decode_single(
queue_item_id: QueueItemId,
path: &Path,
hint: &Hint,
mss: MediaSourceStream,
producer: &mut rtrb::Producer<f32>,
stop: &AtomicBool,
seek_ms: u64,
timeline: &PlaybackTimeline,
viz_buffer: Option<&VizBuffer>,
rg_mode: ReplayGainMode,
pre_amp_db: f64,
) -> Result<(), DecodeError> {
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| DecodeError::Decode(e.to_string()))?;
let mut reader = probed.format;
let track = reader.default_track().ok_or(DecodeError::NoTrack)?;
let track_id = track.id;
let codec_params = &track.codec_params;
let sample_rate = codec_params.sample_rate.unwrap_or(44100);
let channels = codec_params.channels.map(|c| c.count() as u16).unwrap_or(2);
let info = StreamInfo {
codec: codec_name(codec_params.codec),
sample_rate,
channels,
bit_depth: codec_params.bits_per_sample.unwrap_or(16) as u16,
duration_ms: codec_params
.n_frames
.map(|f| f * 1000 / sample_rate as u64)
.unwrap_or(0),
};
let seek_samples = seek_ms * sample_rate as u64 * channels as u64 / 1000;
let write_offset = timeline.samples_written.load(Ordering::Relaxed);
timeline.push_boundary(TrackBoundary {
id: queue_item_id,
path: path.to_path_buf(),
info,
sample_offset: write_offset,
samples_written: 0,
seek_samples,
});
let mut decoder = symphonia::default::get_codecs()
.make(&track.codec_params, &DecoderOptions::default())
.map_err(|_| DecodeError::UnsupportedCodec)?;
if seek_ms > 0 {
let secs = seek_ms / 1000;
let frac = (seek_ms % 1000) as f64 / 1000.0;
reader
.seek(
SeekMode::Coarse,
SeekTo::Time {
time: Time::new(secs, frac),
track_id: Some(track_id),
},
)
.map_err(|e| DecodeError::Decode(format!("seek failed: {}", e)))?;
decoder.reset();
}
let rg_gain = if rg_mode != ReplayGainMode::Off {
match crate::audio::replaygain::read_tags(path) {
Ok(rg_info) => {
let selected = crate::audio::replaygain::select_gain(&rg_info, rg_mode);
if let Some((gain_db, _)) = selected {
log::info!(
"replaygain: applying {:.2} dB ({:?}) to {}",
gain_db,
rg_mode,
path.display()
);
}
selected
}
Err(e) => {
log::debug!("replaygain: no tags for {}: {}", path.display(), e);
None
}
}
} else {
None
};
let mut rg_scratch: Vec<f32> = Vec::new();
let mut sample_buf: Option<SampleBuffer<f32>> = None;
loop {
if stop.load(Ordering::Relaxed) {
return Ok(());
}
let packet = match reader.next_packet() {
Ok(p) => p,
Err(symphonia::core::errors::Error::IoError(ref e))
if e.kind() == std::io::ErrorKind::UnexpectedEof =>
{
return Ok(());
}
Err(e) => return Err(DecodeError::Decode(e.to_string())),
};
if packet.track_id() != track_id {
continue;
}
let decoded = match decoder.decode(&packet) {
Ok(d) => d,
Err(symphonia::core::errors::Error::DecodeError(e)) => {
log::warn!("decode error (skipping packet): {}", e);
continue;
}
Err(e) => return Err(DecodeError::Decode(e.to_string())),
};
let spec = *decoded.spec();
let duration = decoded.capacity();
let sbuf = sample_buf.get_or_insert_with(|| SampleBuffer::new(duration as u64, spec));
sbuf.copy_interleaved_ref(decoded);
let samples = if let Some((gain_db, peak)) = rg_gain {
rg_scratch.clear();
rg_scratch.extend_from_slice(sbuf.samples());
crate::audio::replaygain::apply_gain(&mut rg_scratch, gain_db, peak, pre_amp_db);
&rg_scratch[..]
} else {
sbuf.samples()
};
if let Some(viz) = viz_buffer {
viz.push_samples(samples, channels, sample_rate);
}
let mut offset = 0;
while offset < samples.len() {
if stop.load(Ordering::Relaxed) {
return Ok(());
}
let slots = producer.slots();
if slots == 0 {
thread::sleep(std::time::Duration::from_micros(500));
continue;
}
let chunk_size = slots.min(samples.len() - offset);
if let Ok(mut chunk) = producer.write_chunk_uninit(chunk_size) {
let to_write = &samples[offset..offset + chunk_size];
let (first, second) = chunk.as_mut_slices();
let first_len = first.len().min(to_write.len());
for (slot, &val) in first.iter_mut().zip(&to_write[..first_len]) {
slot.write(val);
}
if first_len < to_write.len() {
for (slot, &val) in second.iter_mut().zip(&to_write[first_len..]) {
slot.write(val);
}
}
unsafe { chunk.commit_all() };
offset += chunk_size;
}
}
timeline.add_written(samples.len() as u64);
}
}
pub fn codec_name(codec: CodecType) -> String {
match codec {
CODEC_TYPE_FLAC => "FLAC",
CODEC_TYPE_MP3 => "MP3",
CODEC_TYPE_AAC => "AAC",
CODEC_TYPE_VORBIS => "Vorbis",
CODEC_TYPE_OPUS => "Opus",
CODEC_TYPE_ALAC => "ALAC",
CODEC_TYPE_WAVPACK => "WavPack",
CODEC_TYPE_PCM_S16LE => "PCM/16",
CODEC_TYPE_PCM_S24LE => "PCM/24",
CODEC_TYPE_PCM_S32LE => "PCM/32",
CODEC_TYPE_PCM_F32LE => "PCM/f32",
other => return format!("Unknown({:?})", other),
}
.to_string()
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use std::sync::atomic::Ordering;
use super::*;
use crate::player::state::QueueItemId;
fn make_info(sample_rate: u32, channels: u16) -> StreamInfo {
StreamInfo {
codec: "FLAC".to_string(),
sample_rate,
channels,
bit_depth: 16,
duration_ms: 10_000,
}
}
fn make_boundary(
id: QueueItemId,
sample_offset: u64,
seek_samples: u64,
channels: u16,
sample_rate: u32,
) -> TrackBoundary {
TrackBoundary {
id,
path: PathBuf::from("/music/track.flac"),
info: make_info(sample_rate, channels),
sample_offset,
samples_written: 0,
seek_samples,
}
}
#[test]
fn test_timeline_single_track() {
let timeline = PlaybackTimeline::new();
let id = QueueItemId::new();
timeline.push_boundary(make_boundary(id, 0, 0, 2, 44100));
timeline.add_written(88200);
timeline.samples_played.store(88200, Ordering::Relaxed);
let result = timeline.current_playback();
assert!(
result.is_some(),
"expected Some for single track with samples played"
);
let (result_id, _path, _info, position_ms) = result.unwrap();
assert_eq!(result_id, id);
assert_eq!(
position_ms, 1000,
"1 second of 44100 Hz stereo should be 1000 ms"
);
}
#[test]
fn test_timeline_gapless_transition() {
let timeline = PlaybackTimeline::new();
let id1 = QueueItemId::new();
let id2 = QueueItemId::new();
timeline.push_boundary(make_boundary(id1, 0, 0, 2, 44100));
timeline.add_written(88200);
timeline.push_boundary(make_boundary(id2, 88200, 0, 2, 44100));
timeline.add_written(44100);
timeline.samples_played.store(90000, Ordering::Relaxed);
let result = timeline.current_playback();
assert!(result.is_some());
let (result_id, _path, _info, position_ms) = result.unwrap();
assert_eq!(
result_id, id2,
"playback head past boundary should report second track"
);
assert_eq!(position_ms, 20, "position within track 2 should be ~20 ms");
}
#[test]
fn test_timeline_zero_samples() {
let timeline = PlaybackTimeline::new();
let id = QueueItemId::new();
timeline.push_boundary(make_boundary(id, 0, 0, 2, 44100));
timeline.add_written(1000);
timeline.samples_played.store(0, Ordering::Relaxed);
let result = timeline.current_playback();
assert!(
result.is_some(),
"expected Some at 0 samples played with a boundary at offset 0"
);
let (result_id, _path, _info, position_ms) = result.unwrap();
assert_eq!(result_id, id);
assert_eq!(position_ms, 0);
}
#[test]
fn test_timeline_past_all_boundaries() {
let timeline = PlaybackTimeline::new();
let id1 = QueueItemId::new();
let id2 = QueueItemId::new();
timeline.push_boundary(make_boundary(id1, 0, 0, 2, 44100));
timeline.add_written(88200);
timeline.push_boundary(make_boundary(id2, 88200, 0, 2, 44100));
timeline.add_written(88200);
timeline
.samples_played
.store(999_999_999, Ordering::Relaxed);
let result = timeline.current_playback();
assert!(result.is_some());
let (result_id, _path, _info, _position_ms) = result.unwrap();
assert_eq!(
result_id, id2,
"samples past all boundaries should report the last track"
);
}
#[test]
fn test_timeline_seek_offset() {
let timeline = PlaybackTimeline::new();
let id = QueueItemId::new();
let seek_samples = 88200u64; timeline.push_boundary(make_boundary(id, 0, seek_samples, 2, 44100));
timeline.add_written(44100); timeline.samples_played.store(0, Ordering::Relaxed);
let result = timeline.current_playback();
assert!(result.is_some());
let (_result_id, _path, _info, position_ms) = result.unwrap();
assert_eq!(
position_ms, 1000,
"position should include seek offset of 1000 ms"
);
}
#[test]
fn test_timeline_reset() {
let timeline = PlaybackTimeline::new();
let id = QueueItemId::new();
timeline.push_boundary(make_boundary(id, 0, 0, 2, 44100));
timeline.add_written(88200);
timeline.samples_played.store(44100, Ordering::Relaxed);
assert!(timeline.current_playback().is_some());
timeline.reset();
assert!(
timeline.current_playback().is_none(),
"after reset, current_playback should return None"
);
assert_eq!(
timeline.samples_played.load(Ordering::Relaxed),
0,
"samples_played should be 0 after reset"
);
assert_eq!(
timeline.samples_written.load(Ordering::Relaxed),
0,
"samples_written should be 0 after reset"
);
}
}