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

oxideav_http/
lib.rs
