oxideav_http/
lib.rs

1//! HTTP/HTTPS source driver for oxideav.
2//!
3//! Implements `Read + Seek` over HTTP `Range` requests so any container
4//! demuxer that takes a `Box<dyn ReadSeek>` can read directly from a URL.
5//! Servers must support `Range: bytes=…` (most static-file hosts do; we
6//! verify with a `HEAD` at construction).
7//!
8//! Wire it into a [`oxideav_source::SourceRegistry`] with [`register`]:
9//!
10//! ```no_run
11//! let mut reg = oxideav_source::SourceRegistry::with_defaults();
12//! oxideav_http::register(&mut reg);
13//! let _r = reg.open("https://example.com/clip.mp4").unwrap();
14//! ```
15
16use std::io::{self, Read, Seek, SeekFrom};
17use std::sync::OnceLock;
18
19use oxideav_container::ReadSeek;
20use oxideav_core::{Error, Result};
21use oxideav_source::SourceRegistry;
22use ureq::Agent;
23
24/// Register the `http` and `https` schemes on the registry.
25pub fn register(registry: &mut SourceRegistry) {
26    registry.register("http", open_http);
27    registry.register("https", open_http);
28}
29
30/// Open a URL as a seekable source.
31pub fn open_http(uri: &str) -> Result<Box<dyn ReadSeek>> {
32    let src = HttpSource::open(uri)?;
33    Ok(Box::new(src))
34}
35
36fn agent() -> &'static Agent {
37    static A: OnceLock<Agent> = OnceLock::new();
38    A.get_or_init(|| Agent::config_builder().build().new_agent())
39}
40
41/// `ReadSeek` over an HTTP/HTTPS resource, using `Range` requests.
42pub struct HttpSource {
43    uri: String,
44    total_len: u64,
45    pos: u64,
46    /// Active response body for the current contiguous read run, if any.
47    body: Option<Box<dyn Read + Send>>,
48}
49
50impl HttpSource {
51    pub fn open(uri: &str) -> Result<Self> {
52        let head = agent()
53            .head(uri)
54            .call()
55            .map_err(|e| Error::other(format!("HTTP HEAD {uri}: {e}")))?;
56
57        let status = head.status();
58        if !status.is_success() {
59            return Err(Error::other(format!("HTTP HEAD {uri}: status {status}")));
60        }
61        let headers = head.headers();
62        let total_len = headers
63            .get("content-length")
64            .and_then(|v| v.to_str().ok())
65            .and_then(|s| s.parse::<u64>().ok())
66            .ok_or_else(|| {
67                Error::Unsupported(format!("HTTP HEAD {uri}: missing Content-Length"))
68            })?;
69        let accept_ranges = headers
70            .get("accept-ranges")
71            .and_then(|v| v.to_str().ok())
72            .unwrap_or("");
73        if !accept_ranges.eq_ignore_ascii_case("bytes") {
74            return Err(Error::Unsupported(format!(
75                "HTTP HEAD {uri}: server does not advertise byte ranges (Accept-Ranges: '{accept_ranges}')"
76            )));
77        }
78        Ok(Self {
79            uri: uri.to_owned(),
80            total_len,
81            pos: 0,
82            body: None,
83        })
84    }
85
86    pub fn len(&self) -> u64 {
87        self.total_len
88    }
89
90    pub fn is_empty(&self) -> bool {
91        self.total_len == 0
92    }
93
94    fn issue_range(&mut self) -> io::Result<()> {
95        if self.pos >= self.total_len {
96            self.body = None;
97            return Ok(());
98        }
99        let range = format!("bytes={}-", self.pos);
100        let resp = agent()
101            .get(&self.uri)
102            .header("Range", &range)
103            .call()
104            .map_err(|e| io::Error::other(format!("HTTP GET {} {}: {e}", self.uri, range)))?;
105        let status = resp.status();
106        if !(status == 206 || status == 200) {
107            return Err(io::Error::other(format!(
108                "HTTP GET {} {}: status {status}",
109                self.uri, range
110            )));
111        }
112        // ureq 3: Body owns the stream; into_body().into_reader() yields
113        // a `Read` that pulls from the wire as bytes are requested.
114        let reader = resp.into_body().into_reader();
115        self.body = Some(Box::new(reader));
116        Ok(())
117    }
118}
119
120impl Read for HttpSource {
121    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
122        if out.is_empty() {
123            return Ok(0);
124        }
125        if self.pos >= self.total_len {
126            return Ok(0);
127        }
128        loop {
129            if self.body.is_none() {
130                self.issue_range()?;
131            }
132            let body = self.body.as_mut().expect("body just issued");
133            let n = body.read(out)?;
134            if n > 0 {
135                self.pos += n as u64;
136                return Ok(n);
137            }
138            self.body = None;
139            if self.pos >= self.total_len {
140                return Ok(0);
141            }
142        }
143    }
144}
145
146impl Seek for HttpSource {
147    fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
148        let new_pos: u64 = match from {
149            SeekFrom::Start(n) => n,
150            SeekFrom::Current(d) => add_signed(self.pos, d)?,
151            SeekFrom::End(d) => add_signed(self.total_len, d)?,
152        };
153        if new_pos > self.total_len {
154            return Err(io::Error::new(io::ErrorKind::InvalidInput, "seek past end"));
155        }
156        if new_pos != self.pos {
157            self.body = None;
158            self.pos = new_pos;
159        }
160        Ok(new_pos)
161    }
162}
163
164fn add_signed(base: u64, delta: i64) -> io::Result<u64> {
165    if delta >= 0 {
166        base.checked_add(delta as u64)
167            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "seek overflow"))
168    } else {
169        base.checked_sub(delta.unsigned_abs())
170            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "seek before start"))
171    }
172}
oxideav_http/lib.rs

oxideav_http/
lib.rs
