use anyhow::Result; use std::io::{self, Read, Seek, SeekFrom}; use std::sync::{ atomic::{AtomicBool, Ordering}, Arc, }; use symphonia::core::{ codecs::DecoderOptions, errors::Error as SymphoniaError, formats::FormatOptions, io::{MediaSource, MediaSourceStream}, meta::MetadataOptions, probe::Hint, }; use crate::player::{output::AudioOutput, PlayerStatus}; /// First 512 KiB of stream kept in memory to support backward seeks during probing. const HEAD_SIZE: usize = 512 * 1024; struct HttpStreamSource { reader: reqwest::blocking::Response, head: Vec, reader_pos: u64, pos: u64, content_length: Option, } impl HttpStreamSource { fn new(response: reqwest::blocking::Response, content_length: Option) -> Self { Self { reader: response, head: Vec::new(), reader_pos: 0, pos: 0, content_length, } } } impl Read for HttpStreamSource { fn read(&mut self, buf: &mut [u8]) -> io::Result { let pos = self.pos as usize; if pos < self.head.len() { let avail = self.head.len() - pos; let n = buf.len().min(avail); buf[..n].copy_from_slice(&self.head[pos..pos + n]); self.pos += n as u64; return Ok(n); } let n = self.reader.read(buf)?; if n > 0 { if self.reader_pos < HEAD_SIZE as u64 { let capacity = HEAD_SIZE.saturating_sub(self.head.len()); let to_buf = n.min(capacity); if to_buf > 0 { self.head.extend_from_slice(&buf[..to_buf]); } } self.reader_pos += n as u64; self.pos += n as u64; } Ok(n) } } impl Seek for HttpStreamSource { fn seek(&mut self, from: SeekFrom) -> io::Result { let cl = self.content_length.unwrap_or(u64::MAX); let target: u64 = match from { SeekFrom::Start(n) => n, SeekFrom::End(n) if n < 0 => cl.saturating_sub((-n) as u64), SeekFrom::End(_) => cl, SeekFrom::Current(n) if n >= 0 => self.pos.saturating_add(n as u64), SeekFrom::Current(n) => self.pos.saturating_sub((-n) as u64), }; if target == self.pos { return Ok(self.pos); } if target < self.reader_pos { if target < self.head.len() as u64 { self.pos = target; return Ok(self.pos); } return Err(io::Error::new( io::ErrorKind::InvalidInput, "backward seek past head buffer", )); } let mut remaining = target - self.reader_pos; while remaining > 0 { let mut discard = [0u8; 8192]; let want = (remaining as usize).min(discard.len()); match self.reader.read(&mut discard[..want]) { Ok(0) => break, Ok(n) => { if self.reader_pos < HEAD_SIZE as u64 { let capacity = HEAD_SIZE.saturating_sub(self.head.len()); let to_buf = n.min(capacity); if to_buf > 0 { self.head.extend_from_slice(&discard[..to_buf]); } } self.reader_pos += n as u64; remaining -= n as u64; } Err(e) => return Err(e), } } self.pos = self.reader_pos; Ok(self.pos) } } impl MediaSource for HttpStreamSource { fn is_seekable(&self) -> bool { true } fn byte_len(&self) -> Option { self.content_length } } /// Stream and decode audio from `url`. Runs on a dedicated OS thread. pub fn play_track( url: &str, status: &PlayerStatus, stop: &Arc, paused: &Arc, ) -> Result<()> { let response = reqwest::blocking::get(url)?; let content_length = response.content_length(); let source = HttpStreamSource::new(response, content_length); let mss = MediaSourceStream::new(Box::new(source), Default::default()); let hint = Hint::new(); let probed = symphonia::default::get_probe() .format(&hint, mss, &FormatOptions::default(), &MetadataOptions::default()) .map_err(|e| anyhow::anyhow!("probe failed: {e}"))?; let mut format = probed.format; let track = format .tracks() .iter() .find(|t| t.codec_params.codec != symphonia::core::codecs::CODEC_TYPE_NULL) .ok_or_else(|| anyhow::anyhow!("no audio tracks"))? .clone(); let track_id = track.id; let sample_rate = track.codec_params.sample_rate.unwrap_or(44100); let channels = track.codec_params.channels.map(|c| c.count()).unwrap_or(2); let mut decoder = symphonia::default::get_codecs() .make(&track.codec_params, &DecoderOptions::default()) .map_err(|e| anyhow::anyhow!("decoder init failed: {e}"))?; let mut audio_output = AudioOutput::try_open(sample_rate, channels)?; loop { if stop.load(Ordering::SeqCst) { break; } while paused.load(Ordering::SeqCst) { std::thread::sleep(std::time::Duration::from_millis(50)); if stop.load(Ordering::SeqCst) { return Ok(()); } } let packet = match format.next_packet() { Ok(p) => p, Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { break; } Err(SymphoniaError::ResetRequired) => { decoder.reset(); continue; } Err(e) => return Err(anyhow::anyhow!("format error: {e}")), }; if packet.track_id() != track_id { continue; } if let Some(ts) = packet.ts().checked_div(sample_rate as u64) { status.position_secs.store(ts, std::sync::atomic::Ordering::Relaxed); } match decoder.decode(&packet) { Ok(decoded) => { let volume = status.volume.load(Ordering::Relaxed) as f32 / 100.0; audio_output.write(decoded, volume, stop)?; } Err(SymphoniaError::IoError(_)) => break, Err(SymphoniaError::DecodeError(e)) => eprintln!("decode error: {e}"), Err(e) => return Err(anyhow::anyhow!("decode error: {e}")), } } Ok(()) }