SoulStream/crates/player/src/source.rs

//! Growing-file media source: reads a file that is still being written,
//! blocking (instead of returning EOF) when playback catches up to the
//! download's write head.

use std::io::{Read, Seek, SeekFrom};
use std::path::Path;
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;

/// Everything rodio's Decoder needs from a source.
pub trait MediaRead: Read + Seek + Send + Sync {}
impl<T: Read + Seek + Send + Sync> MediaRead for T {}

#[derive(Debug, Default)]
struct ProgressState {
    received: u64,
    done: bool,
    failed: Option<String>,
    aborted: bool,
}

/// How a transfer ended. A reader cannot tell these apart on its own —
/// [`GrowingFile::read`] reports an abort as a graceful end-of-file so
/// playback stops cleanly — so anything that cares whether it saw the WHOLE
/// file (loudness measurement) has to ask the transfer itself.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Terminal {
    /// Still arriving.
    Running,
    /// Every byte arrived. An end-of-file after this one is the real end.
    Done,
    /// Stopped on purpose — the track was switched or playback stopped.
    Aborted,
    /// The transfer died before it finished.
    Failed,
}

/// Shared write-progress of an in-flight download. The downloader calls
/// [`update`](Self::update); the audio thread's reads wait on it.
#[derive(Debug, Default)]
pub struct StreamProgress {
    state: Mutex<ProgressState>,
    cv: Condvar,
}

impl StreamProgress {
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    pub fn update(&self, received: u64, done: bool, failed: Option<String>) {
        let mut st = self.state.lock().unwrap();
        st.received = st.received.max(received);
        st.done |= done;
        if st.failed.is_none() {
            st.failed = failed;
        }
        self.cv.notify_all();
    }

    /// Current (received bytes, done) — diagnostics/observability readout.
    pub fn received_done(&self) -> (u64, bool) {
        let st = self.state.lock().unwrap();
        (st.received, st.done)
    }

    /// How this transfer ended, if it has. `Done` wins over `Aborted`: the
    /// player aborts the stream when a track ends normally, and by then
    /// every byte has already arrived — a reader that hit EOF read the
    /// whole file, so the abort says nothing about coverage.
    pub fn terminal(&self) -> Terminal {
        let st = self.state.lock().unwrap();
        if st.done {
            Terminal::Done
        } else if st.aborted {
            Terminal::Aborted
        } else if st.failed.is_some() {
            Terminal::Failed
        } else {
            Terminal::Running
        }
    }

    /// Unblock any waiting reads permanently (track switched / stopped).
    pub fn abort(&self) {
        let mut st = self.state.lock().unwrap();
        st.aborted = true;
        self.cv.notify_all();
    }

    /// Wait until at least `wanted` bytes exist (or terminal state).
    /// Returns Ok(available bytes) or Err(reason) on failure/abort.
    fn wait_for(&self, wanted: u64) -> Result<u64, String> {
        let mut st = self.state.lock().unwrap();
        loop {
            if st.aborted {
                return Err("stream aborted".into());
            }
            if let Some(reason) = &st.failed {
                return Err(reason.clone());
            }
            if st.done || st.received >= wanted {
                return Ok(st.received);
            }
            let (guard, _) = self
                .cv
                .wait_timeout(st, Duration::from_millis(250))
                .map_err(|_| "progress lock poisoned".to_string())?;
            st = guard;
        }
    }

    fn snapshot(&self) -> (u64, bool) {
        let st = self.state.lock().unwrap();
        (st.received, st.done)
    }
}

/// A file being written by a download, exposed as a blocking Read + Seek.
pub struct GrowingFile {
    file: std::fs::File,
    pos: u64,
    total: u64,
    progress: Arc<StreamProgress>,
}

impl GrowingFile {
    pub fn open(
        path: &Path,
        total: u64,
        progress: Arc<StreamProgress>,
    ) -> std::io::Result<GrowingFile> {
        // The file may not exist until the remote queue clears; wait for
        // the first bytes before opening.
        progress.wait_for(1).map_err(std::io::Error::other)?;
        let file = std::fs::File::open(path)?;
        Ok(GrowingFile {
            file,
            pos: 0,
            total,
            progress,
        })
    }
}

impl Read for GrowingFile {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let (received, done) = self.progress.snapshot();

        // At the write head and still downloading: wait for more.
        if self.pos >= received && !done {
            match self.progress.wait_for(self.pos + 1) {
                Ok(_) => {}
                Err(e) if e == "stream aborted" => return Ok(0), // graceful EOF
                Err(e) => return Err(std::io::Error::other(e)),
            }
        }

        let n = self.file.read(buf)?;
        if n == 0 {
            // Raced the writer: data was promised but the handle hasn't
            // caught up; only report EOF once the download is done.
            let (_, done) = self.progress.snapshot();
            if !done {
                std::thread::sleep(Duration::from_millis(50));
                let n2 = self.file.read(buf)?;
                self.pos += n2 as u64;
                return Ok(n2);
            }
        }
        self.pos += n as u64;
        Ok(n)
    }
}

impl Seek for GrowingFile {
    fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
        // Normalize End against the *final* size so seek math is stable
        // even while the file is still growing.
        let target = match pos {
            SeekFrom::Start(o) => o,
            SeekFrom::Current(d) => self.pos.saturating_add_signed(d),
            SeekFrom::End(d) => self.total.saturating_add_signed(d),
        };
        // Block until the target region exists, so decoder probes into
        // not-yet-downloaded space stall instead of erroring.
        if target > 0 {
            self.progress
                .wait_for(target.min(self.total))
                .map_err(std::io::Error::other)?;
        }
        self.pos = self.file.seek(SeekFrom::Start(target))?;
        Ok(self.pos)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;

    #[test]
    fn growing_read_blocks_until_bytes_arrive() {
        let dir = std::env::temp_dir().join("ssplus-player-test");
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("grow.bin");
        std::fs::write(&path, b"").unwrap();

        let progress = StreamProgress::new();
        let writer_progress = progress.clone();
        let writer_path = path.clone();

        let writer = std::thread::spawn(move || {
            let mut f = std::fs::OpenOptions::new()
                .write(true)
                .open(&writer_path)
                .unwrap();
            for chunk in 0..4u8 {
                std::thread::sleep(Duration::from_millis(60));
                f.write_all(&[chunk; 100]).unwrap();
                f.flush().unwrap();
                writer_progress.update((chunk as u64 + 1) * 100, chunk == 3, None);
            }
        });

        let mut g = GrowingFile::open(&path, 400, progress).unwrap();
        let mut out = Vec::new();
        g.read_to_end(&mut out).unwrap();
        writer.join().unwrap();

        assert_eq!(out.len(), 400);
        assert_eq!(out[350], 3);
    }

    #[test]
    fn abort_unblocks_reader() {
        let dir = std::env::temp_dir().join("ssplus-player-test");
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("abort.bin");
        std::fs::write(&path, b"xx").unwrap();

        let progress = StreamProgress::new();
        progress.update(2, false, None);
        let mut g = GrowingFile::open(&path, 1000, progress.clone()).unwrap();

        let killer = std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(100));
            progress.abort();
        });

        let mut out = Vec::new();
        // Reads the 2 bytes, then blocks at the write head until abort.
        g.read_to_end(&mut out).unwrap();
        killer.join().unwrap();
        assert_eq!(out.len(), 2);
    }
}