spatialrust_io/copc/
http.rs

1//! HTTP range-request byte source for remote COPC files.
2
3use std::io::Read;
4use std::sync::Arc;
5
6use copc_streaming::ByteSource;
7
8use spatialrust_core::PointCloud;
9
10use crate::copc::query::{CopcFileInfo, CopcQuery};
11use crate::copc::reader::{read_copc_from_byte_source, read_header_info};
12use crate::error::{copc_format, IoError};
13
14const DEFAULT_MAX_PARALLEL_RANGES: usize = 8;
15
16/// Random-access COPC byte source backed by HTTP range requests.
17///
18/// Clones share one HTTP connection pool so repeated hierarchy and point-chunk
19/// ranges can reuse established TCP/TLS connections across request batches.
20#[derive(Clone, Debug)]
21pub struct HttpByteSource {
22    url: String,
23    max_parallel_ranges: usize,
24    agent: ureq::Agent,
25}
26
27impl PartialEq for HttpByteSource {
28    fn eq(&self, other: &Self) -> bool {
29        self.url == other.url && self.max_parallel_ranges == other.max_parallel_ranges
30    }
31}
32
33impl Eq for HttpByteSource {}
34
35impl HttpByteSource {
36    /// Creates an HTTP byte source for a remote COPC URL.
37    pub fn new(url: impl Into<String>) -> Result<Self, IoError> {
38        let url = url.into();
39        validate_http_url(&url)?;
40        Ok(Self {
41            url,
42            max_parallel_ranges: DEFAULT_MAX_PARALLEL_RANGES,
43            agent: ureq::Agent::new(),
44        })
45    }
46
47    /// Limits how many HTTP range requests are in flight at once.
48    #[must_use]
49    pub fn with_max_parallel_ranges(mut self, max_parallel_ranges: usize) -> Self {
50        self.max_parallel_ranges = max_parallel_ranges.max(1);
51        self
52    }
53
54    /// Returns the source URL.
55    #[must_use]
56    pub fn url(&self) -> &str {
57        &self.url
58    }
59
60    /// Returns the configured maximum number of parallel range requests.
61    #[must_use]
62    pub fn max_parallel_ranges(&self) -> usize {
63        self.max_parallel_ranges
64    }
65}
66
67impl ByteSource for HttpByteSource {
68    async fn read_range(
69        &self,
70        offset: u64,
71        length: u64,
72    ) -> Result<Vec<u8>, copc_streaming::CopcError> {
73        fetch_http_range(&self.agent, &self.url, offset, length)
74    }
75
76    async fn read_ranges(
77        &self,
78        ranges: &[(u64, u64)],
79    ) -> Result<Vec<Vec<u8>>, copc_streaming::CopcError> {
80        fetch_http_ranges_parallel(&self.agent, &self.url, ranges, self.max_parallel_ranges)
81    }
82
83    async fn size(&self) -> Result<Option<u64>, copc_streaming::CopcError> {
84        fetch_http_size(&self.agent, &self.url)
85    }
86}
87
88/// Reads all points from a remote COPC URL.
89pub fn read_copc_url(url: &str) -> Result<PointCloud, IoError> {
90    read_copc_url_with_query(url, None)
91}
92
93/// Reads points from a remote COPC URL using an optional spatial query.
94pub fn read_copc_url_with_query(
95    url: &str,
96    query: Option<&CopcQuery>,
97) -> Result<PointCloud, IoError> {
98    if let Some(query) = query {
99        query.validate()?;
100    }
101    validate_http_url(url)?;
102    let source = HttpByteSource::new(url)?;
103    pollster::block_on(read_copc_from_byte_source(source, query))
104}
105
106/// Reads COPC header metadata from a remote URL without loading points.
107pub fn read_copc_url_info(url: &str) -> Result<CopcFileInfo, IoError> {
108    validate_http_url(url)?;
109    let source = HttpByteSource::new(url)?;
110    pollster::block_on(async { read_header_info(source).await.map(|(_, info)| info) })
111}
112
113fn fetch_http_range(
114    agent: &ureq::Agent,
115    url: &str,
116    offset: u64,
117    length: u64,
118) -> Result<Vec<u8>, copc_streaming::CopcError> {
119    if length == 0 {
120        return Ok(Vec::new());
121    }
122
123    let end = offset.saturating_add(length.saturating_sub(1));
124    let response = agent
125        .get(url)
126        .set("Range", &format!("bytes={offset}-{end}"))
127        .call()
128        .map_err(|error| copc_streaming::CopcError::ByteSource(Box::new(error)))?;
129
130    let status = response.status();
131    if status != 200 && status != 206 {
132        return Err(copc_streaming::CopcError::ByteSource(Box::new(std::io::Error::new(
133            std::io::ErrorKind::InvalidData,
134            format!("unexpected HTTP status {status} for range request"),
135        ))));
136    }
137
138    let mut bytes = Vec::with_capacity(length as usize);
139    response
140        .into_reader()
141        .take(length)
142        .read_to_end(&mut bytes)
143        .map_err(copc_streaming::CopcError::Io)?;
144    Ok(bytes)
145}
146
147fn fetch_http_ranges_parallel(
148    agent: &ureq::Agent,
149    url: &str,
150    ranges: &[(u64, u64)],
151    max_parallel_ranges: usize,
152) -> Result<Vec<Vec<u8>>, copc_streaming::CopcError> {
153    if ranges.is_empty() {
154        return Ok(Vec::new());
155    }
156
157    let mut results = Vec::with_capacity(ranges.len());
158    let url = Arc::new(url.to_owned());
159
160    for batch in ranges.chunks(max_parallel_ranges.max(1)) {
161        let batch_results = read_range_batch(agent.clone(), Arc::clone(&url), batch)?;
162        results.extend(batch_results);
163    }
164
165    Ok(results)
166}
167
168fn read_range_batch(
169    agent: ureq::Agent,
170    url: Arc<String>,
171    ranges: &[(u64, u64)],
172) -> Result<Vec<Vec<u8>>, copc_streaming::CopcError> {
173    if ranges.len() == 1 {
174        let (offset, length) = ranges[0];
175        return Ok(vec![fetch_http_range(&agent, url.as_str(), offset, length)?]);
176    }
177
178    std::thread::scope(|scope| {
179        let mut handles = Vec::with_capacity(ranges.len());
180        for (index, &(offset, length)) in ranges.iter().enumerate() {
181            let agent = agent.clone();
182            let url = Arc::clone(&url);
183            handles.push(scope.spawn(move || {
184                let bytes = fetch_http_range(&agent, url.as_str(), offset, length)?;
185                Ok::<_, copc_streaming::CopcError>((index, bytes))
186            }));
187        }
188
189        let mut batch = vec![Vec::new(); ranges.len()];
190        for handle in handles {
191            let (index, bytes) = handle.join().map_err(|_| {
192                copc_streaming::CopcError::ByteSource(Box::new(std::io::Error::other(
193                    "parallel HTTP range worker panicked",
194                )))
195            })??;
196            batch[index] = bytes;
197        }
198        Ok(batch)
199    })
200}
201
202fn fetch_http_size(
203    agent: &ureq::Agent,
204    url: &str,
205) -> Result<Option<u64>, copc_streaming::CopcError> {
206    if let Ok(response) = agent.head(url).call() {
207        if let Some(total) = response.header("Content-Length").and_then(parse_u64_header) {
208            return Ok(Some(total));
209        }
210    }
211
212    let response = agent
213        .get(url)
214        .set("Range", "bytes=0-0")
215        .call()
216        .map_err(|error| copc_streaming::CopcError::ByteSource(Box::new(error)))?;
217
218    if let Some(total) = response.header("Content-Range").and_then(parse_content_range_total) {
219        return Ok(Some(total));
220    }
221
222    if let Some(total) = response.header("Content-Length").and_then(parse_u64_header) {
223        return Ok(Some(total));
224    }
225
226    Ok(None)
227}
228
229fn validate_http_url(url: &str) -> Result<(), IoError> {
230    if url.starts_with("http://") || url.starts_with("https://") {
231        Ok(())
232    } else {
233        Err(copc_format(format!(
234            "COPC HTTP sources require an http:// or https:// URL, got `{url}`"
235        )))
236    }
237}
238
239fn parse_u64_header(value: &str) -> Option<u64> {
240    value.trim().parse().ok()
241}
242
243fn parse_content_range_total(value: &str) -> Option<u64> {
244    value.split('/').nth(1)?.trim().parse().ok()
245}
246
247#[cfg(test)]
248mod tests {
249    use super::{
250        fetch_http_ranges_parallel, parse_content_range_total, read_range_batch, validate_http_url,
251        HttpByteSource,
252    };
253    use copc_streaming::ByteSource;
254    use std::io::{Read, Write};
255    use std::net::{TcpListener, TcpStream};
256    use std::sync::atomic::{AtomicUsize, Ordering};
257    use std::sync::Arc;
258    use std::thread;
259    use std::time::Duration;
260
261    #[test]
262    fn validates_http_urls() {
263        assert!(validate_http_url("https://example.com/cloud.copc.laz").is_ok());
264        assert!(validate_http_url("/tmp/local.copc.laz").is_err());
265    }
266
267    #[test]
268    fn parses_content_range_total() {
269        assert_eq!(parse_content_range_total("bytes 0-0/12345"), Some(12345));
270    }
271
272    #[test]
273    fn constructs_http_source() {
274        let source = HttpByteSource::new("https://example.com/cloud.copc.laz").unwrap();
275        assert_eq!(source.url(), "https://example.com/cloud.copc.laz");
276        assert_eq!(source.max_parallel_ranges(), 8);
277        assert_eq!(source.clone(), source);
278        assert_ne!(source.clone().with_max_parallel_ranges(4), source);
279    }
280
281    #[test]
282    fn read_ranges_fetches_multiple_byte_ranges() {
283        let payload = b"0123456789ABCDEF";
284        let requests = Arc::new(AtomicUsize::new(0));
285        let requests_server = Arc::clone(&requests);
286
287        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
288        listener.set_nonblocking(true).unwrap();
289        let addr = listener.local_addr().unwrap();
290
291        let server = thread::spawn(move || {
292            let deadline = std::time::Instant::now() + Duration::from_secs(5);
293            while requests_server.load(Ordering::SeqCst) < 3 {
294                if std::time::Instant::now() > deadline {
295                    panic!("timed out waiting for HTTP range requests");
296                }
297                let Ok((mut stream, _)) = listener.accept() else {
298                    thread::sleep(Duration::from_millis(10));
299                    continue;
300                };
301                serve_test_range(&mut stream, payload, &requests_server);
302            }
303        });
304
305        let url = format!("http://{addr}/cloud.copc.laz");
306        let source = HttpByteSource::new(&url).unwrap().with_max_parallel_ranges(3);
307        let ranges = vec![(0, 4), (4, 4), (8, 4)];
308        let results = pollster::block_on(source.read_ranges(&ranges)).unwrap();
309
310        assert_eq!(results.len(), 3);
311        assert_eq!(results[0], b"0123".to_vec());
312        assert_eq!(results[1], b"4567".to_vec());
313        assert_eq!(results[2], b"89AB".to_vec());
314        assert_eq!(requests.load(Ordering::SeqCst), 3);
315        server.join().unwrap();
316    }
317
318    #[test]
319    fn fetch_ranges_batches_by_parallelism_limit() {
320        let url = "https://example.com/cloud.copc.laz";
321        let ranges = vec![(0, 1); 5];
322        let err = fetch_http_ranges_parallel(&ureq::Agent::new(), url, &ranges, 2).unwrap_err();
323        assert!(matches!(
324            err,
325            copc_streaming::CopcError::ByteSource(_) | copc_streaming::CopcError::Io(_)
326        ));
327    }
328
329    #[test]
330    fn single_range_batch_delegates_to_fetch() {
331        let result = read_range_batch(
332            ureq::Agent::new(),
333            Arc::new("https://invalid.test/not-found.copc.laz".to_owned()),
334            &[(0, 1)],
335        );
336        assert!(result.is_err());
337    }
338
339    #[cfg(feature = "streaming")]
340    #[test]
341    fn bounded_http_copc_source_reads_range_served_file() {
342        use std::sync::atomic::AtomicBool;
343        use std::time::Duration;
344
345        use spatialrust_core::PointCloudBuilder;
346        use spatialrust_records::{
347            BoundedSpatialRecordSource, CancellationToken, MemoryBudget, StreamOptions,
348        };
349
350        let mut builder = PointCloudBuilder::xyz();
351        for index in 0..100 {
352            builder.push_point([index as f32, 0.0, 0.0]).unwrap();
353        }
354        let cloud = builder.build().unwrap();
355        let path = std::env::temp_dir()
356            .join(format!("spatialrust_http_stream_{}.copc.laz", std::process::id()));
357        crate::copc::write_copc_file(&path, &cloud).unwrap();
358        let payload = Arc::new(std::fs::read(&path).unwrap());
359        std::fs::remove_file(path).unwrap();
360
361        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
362        listener.set_nonblocking(true).unwrap();
363        let addr = listener.local_addr().unwrap();
364        let stopped = Arc::new(AtomicBool::new(false));
365        let stopped_server = Arc::clone(&stopped);
366        let payload_server = Arc::clone(&payload);
367        let server = std::thread::spawn(move || {
368            let requests = AtomicUsize::new(0);
369            while !stopped_server.load(Ordering::Acquire) {
370                match listener.accept() {
371                    Ok((mut stream, _)) => {
372                        serve_test_range(&mut stream, &payload_server, &requests);
373                    }
374                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
375                        std::thread::sleep(Duration::from_millis(1));
376                    }
377                    Err(error) => panic!("HTTP test server failed: {error}"),
378                }
379            }
380            requests.load(Ordering::SeqCst)
381        });
382
383        let url = format!("http://{addr}/cloud.copc.laz");
384        let options = StreamOptions::new(17, MemoryBudget::new(4 * 1024 * 1024).unwrap()).unwrap();
385        let mut source =
386            crate::CopcChunkSource::open_url(&url, None, options, CancellationToken::default())
387                .unwrap();
388        let mut point_count = 0;
389        while let Some(chunk) = source.next_chunk() {
390            point_count += chunk.unwrap().record().cloud().len();
391        }
392        stopped.store(true, Ordering::Release);
393        assert_eq!(point_count, 100);
394        assert!(server.join().unwrap() >= 3);
395    }
396
397    fn serve_test_range(stream: &mut TcpStream, payload: &[u8], requests: &AtomicUsize) {
398        // On macOS an accepted socket can inherit `O_NONBLOCK` from the
399        // listener. The fixture serves synchronously, so make that contract
400        // explicit before reading the request.
401        stream.set_nonblocking(false).unwrap();
402        let mut buffer = [0_u8; 512];
403        let read = stream.read(&mut buffer).unwrap();
404        let request = std::str::from_utf8(&buffer[..read]).unwrap();
405        if request.starts_with("HEAD ") {
406            requests.fetch_add(1, Ordering::SeqCst);
407            let response = format!(
408                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nAccept-Ranges: bytes\r\n\r\n",
409                payload.len()
410            );
411            stream.write_all(response.as_bytes()).unwrap();
412            return;
413        }
414        let range = request
415            .lines()
416            .find_map(|line| line.strip_prefix("Range: bytes="))
417            .expect("missing Range header");
418        let (start, end) = range
419            .split_once('-')
420            .and_then(|(start, end)| Some((start.parse::<u64>().ok()?, end.parse::<u64>().ok()?)))
421            .expect("invalid Range header");
422        let start = start as usize;
423        let end = end as usize;
424        let body = payload[start..=end].to_vec();
425
426        requests.fetch_add(1, Ordering::SeqCst);
427        let response = format!(
428            "HTTP/1.1 206 Partial Content\r\nContent-Length: {}\r\nContent-Range: bytes {start}-{end}/{}\r\n\r\n",
429            body.len(),
430            payload.len()
431        );
432        stream.write_all(response.as_bytes()).unwrap();
433        stream.write_all(&body).unwrap();
434    }
435}
spatialrust_io/copc/http.rs

spatialrust_io/copc/
http.rs
