utils/
range_source.rs

1//! HTTP Range-backed seekable byte source.
2//!
3//! Used for YouTube Music (and any other URL where the server returns
4//! `Accept-Ranges: bytes`). Unlike [`crate::stream_buffer::StreamBuffer`],
5//! this never downloads the file linearly — every miss in the rolling
6//! window cache becomes a `Range: bytes=N-M` request. Symphonia can seek
7//! freely: to the end (Matroska Cues), to scrub targets, anywhere.
8//!
9//! Architecture:
10//! - One rolling 512 KiB window, anchored at `chunk_start`.
11//! - `seek()` is a constant-time pointer move.
12//! - `read()` fetches a fresh window only when `pos` falls outside the
13//!   currently-cached window. Sequential playback stays inside the window
14//!   90%+ of the time.
15//! - HTTP calls happen via `reqwest::blocking` inside whatever thread is
16//!   calling `Read::read` — callers MUST already be on a blocking-friendly
17//!   thread (`spawn_blocking` or similar).
18//!
19//! `byte_len()` is determined once upfront from `Content-Range` of the
20//! initial probe fetch. If the server doesn't include it, we fall back to
21//! a HEAD request. If both fail, this source can't be constructed.
22
23use std::io::{Error as IoError, ErrorKind, Read, Result as IoResult, Seek, SeekFrom};
24use std::time::Duration;
25
26const CHUNK: usize = 512 * 1024;
27const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
28
29pub struct RangeStreamSource {
30    url: String,
31    client: reqwest::blocking::Client,
32    total_size: u64,
33    pos: u64,
34    chunk: Vec<u8>,
35    chunk_start: u64,
36}
37
38impl RangeStreamSource {
39    /// Probe the URL with a `Range: bytes=0-0` HEAD-equivalent to learn its
40    /// total size and confirm Range support. Returns the source positioned
41    /// at byte 0 with an empty cache.
42    pub fn new(url: String, user_agent: Option<String>) -> IoResult<Self> {
43        let ua =
44            user_agent.unwrap_or_else(|| concat!("Kopuz/", env!("CARGO_PKG_VERSION")).to_string());
45        let client = reqwest::blocking::Client::builder()
46            .tcp_nodelay(true)
47            .user_agent(ua)
48            .timeout(REQUEST_TIMEOUT)
49            .build()
50            .map_err(IoError::other)?;
51
52        // One-byte probe — cheap, and the server returns the full
53        // `Content-Range: bytes 0-0/<TOTAL>` we want.
54        let resp = client
55            .get(&url)
56            .header("Range", "bytes=0-0")
57            .send()
58            .map_err(IoError::other)?;
59        let status = resp.status();
60        if !status.is_success() {
61            return Err(IoError::other(format!("range probe HTTP {status}")));
62        }
63        let total_size = parse_total_size(&resp)
64            .ok_or_else(|| IoError::other("server didn't expose total size on range probe"))?;
65
66        Ok(Self {
67            url,
68            client,
69            total_size,
70            pos: 0,
71            chunk: Vec::with_capacity(CHUNK),
72            chunk_start: 0,
73        })
74    }
75
76    pub fn total_size(&self) -> u64 {
77        self.total_size
78    }
79
80    fn fetch_chunk(&mut self, start: u64) -> IoResult<()> {
81        let end = (start + CHUNK as u64 - 1).min(self.total_size - 1);
82        let resp = self
83            .client
84            .get(&self.url)
85            .header("Range", format!("bytes={start}-{end}"))
86            .send()
87            .map_err(IoError::other)?;
88        if !resp.status().is_success() {
89            return Err(IoError::other(format!(
90                "range fetch {start}-{end} HTTP {}",
91                resp.status()
92            )));
93        }
94        let bytes = resp.bytes().map_err(IoError::other)?;
95        self.chunk.clear();
96        self.chunk.extend_from_slice(&bytes);
97        self.chunk_start = start;
98        Ok(())
99    }
100
101    fn pos_in_cache(&self, pos: u64) -> bool {
102        !self.chunk.is_empty()
103            && pos >= self.chunk_start
104            && pos < self.chunk_start + self.chunk.len() as u64
105    }
106}
107
108impl Read for RangeStreamSource {
109    fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
110        if self.pos >= self.total_size {
111            return Ok(0);
112        }
113        if !self.pos_in_cache(self.pos) {
114            self.fetch_chunk(self.pos)?;
115            if self.chunk.is_empty() {
116                return Ok(0);
117            }
118        }
119        let offset = (self.pos - self.chunk_start) as usize;
120        let available = self.chunk.len() - offset;
121        let to_copy = available.min(buf.len());
122        buf[..to_copy].copy_from_slice(&self.chunk[offset..offset + to_copy]);
123        self.pos += to_copy as u64;
124        Ok(to_copy)
125    }
126}
127
128impl Seek for RangeStreamSource {
129    fn seek(&mut self, p: SeekFrom) -> IoResult<u64> {
130        let new_pos: i64 = match p {
131            SeekFrom::Start(n) => n as i64,
132            SeekFrom::Current(n) => self.pos as i64 + n,
133            SeekFrom::End(n) => self.total_size as i64 + n,
134        };
135        if new_pos < 0 {
136            return Err(IoError::new(
137                ErrorKind::InvalidInput,
138                "seek to negative position",
139            ));
140        }
141        self.pos = new_pos as u64;
142        Ok(self.pos)
143    }
144}
145
146fn parse_total_size(resp: &reqwest::blocking::Response) -> Option<u64> {
147    // Prefer Content-Range: "bytes 0-0/12345" — the part after '/' is the
148    // total. Fall back to Content-Length only if Range wasn't honoured (in
149    // which case the server gave us the whole body, and Content-Length is
150    // the full file size).
151    if let Some(v) = resp.headers().get("content-range")
152        && let Ok(s) = v.to_str()
153        && let Some(slash) = s.rfind('/')
154    {
155        let tail = &s[slash + 1..];
156        if tail != "*"
157            && let Ok(n) = tail.parse()
158        {
159            return Some(n);
160        }
161    }
162    resp.content_length()
163}
utils/range_source.rs

utils/
range_source.rs
