gwseq_io/source/

mod.rs

//! Byte sources and sinks.
//!
//! [`ByteSource`] reads are *positioned* and take `&self`, so one handle
//! serves every worker thread. That removes the need for a pool of handles —
//! a checkout semaphore, worker threads, a promise per buffer — and leaves the
//! part that earns its keep, the aligned LRU block cache, as the
//! [`CachedSource`] decorator.
//!
//! Sources compose by wrapping:
//!
//! ```no_run
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! use gwseq_io::source::{CachedSource, UrlSource};
//!
//! let url = "https://example.org/track.bigwig";
//! let src = CachedSource::new(UrlSource::open(url)?, 1 << 20, 128);
//! # let _ = src;
//! # Ok(())
//! # }
//! ```

use std::ops::Range;

use bytes::Bytes;

use crate::error::{Error, Result};

mod cache;
mod gzip;
mod local;
mod sink;
#[cfg(feature = "url")]
pub(crate) mod url;

pub use cache::CachedSource;
pub use gzip::{is_gzipped, open_text, GzipReader, TextInput};
pub use local::LocalSource;
pub use sink::{ByteSink, LocalSink, SinkSource};
#[cfg(feature = "url")]
pub use url::{http_get_text, RetryPolicy, UrlSource};

/// Recommended block size for the cache in front of a local file.
pub const LOCAL_BLOCK_SIZE: u64 = 32 * 1024;
/// Recommended block size for the cache in front of a URL: a range request has
/// a fixed cost worth amortising over more bytes.
pub const URL_BLOCK_SIZE: u64 = 1024 * 1024;
/// Recommended cache capacity, in blocks.
pub const MAX_BLOCKS: usize = 128;

/// A random-access source of bytes.
///
/// `Send + Sync` and read by offset, so the readers hand `&dyn ByteSource` to
/// every worker rather than checking a handle out of a pool.
///
/// `Debug` is a supertrait so that the readers, which hold an
/// `Arc<dyn ByteSource>`, can derive it — and so a source turns up legibly in a
/// failed assertion rather than as an opaque pointer.
pub trait ByteSource: Send + Sync + std::fmt::Debug {
    /// Path or URL this reads, for error messages.
    fn path(&self) -> &str;

    /// Total length in bytes.
    fn len(&self) -> Result<u64>;

    fn is_empty(&self) -> Result<bool> {
        Ok(self.len()? == 0)
    }

    /// Read up to `len` bytes at `offset`, returning fewer at end of file.
    fn read_at(&self, offset: u64, len: usize) -> Result<Bytes>;

    /// Read exactly `len` bytes at `offset`; a short read is [`Error::Corrupt`].
    fn read_exact_at(&self, offset: u64, len: usize) -> Result<Bytes> {
        let data = self.read_at(offset, len)?;
        if data.len() != len {
            return Err(Error::corrupt(
                self.path(),
                offset,
                format!("wanted {len} bytes, got {}", data.len()),
            ));
        }
        Ok(data)
    }

    /// Read from `offset` to end of file.
    fn read_to_end(&self, offset: u64) -> Result<Bytes> {
        let len = self.len()?.saturating_sub(offset);
        self.read_at(offset, len as usize)
    }

    /// Announce ranges that are about to be read, so a source that can fetch
    /// them together does. A no-op unless overridden.
    ///
    /// This is where read-ahead lives: the R-tree walk knows every leaf it is
    /// about to visit before it visits any of them, and over HTTP that
    /// knowledge is worth a multi-range request.
    fn prefetch(&self, _ranges: &[Range<u64>]) {}

    /// Release the underlying handle. Idempotent; reads after it fail with
    /// [`Error::Closed`].
    fn close(&self) {}
}

impl<T: ByteSource + ?Sized> ByteSource for std::sync::Arc<T> {
    fn path(&self) -> &str {
        (**self).path()
    }
    fn len(&self) -> Result<u64> {
        (**self).len()
    }
    fn read_at(&self, offset: u64, len: usize) -> Result<Bytes> {
        (**self).read_at(offset, len)
    }
    fn prefetch(&self, ranges: &[Range<u64>]) {
        (**self).prefetch(ranges)
    }
    fn close(&self) {
        (**self).close()
    }
}

/// True for a path a [`UrlSource`] handles.
///
/// HTTP and HTTPS, and nothing else. `ftp://` used to be routed here as well,
/// which meant an FTP path was accepted, carried through the retry policy and
/// then refused by `ureq` — an HTTP client — as an unsupported scheme. A path
/// this library cannot read should say so at once, and "no such file" is the
/// clearer thing to say about a scheme it does not speak.
pub fn is_url(path: &str) -> bool {
    let lower = path.to_ascii_lowercase();
    lower.starts_with("http://") || lower.starts_with("https://")
}

/// Open `path` as a local file or a URL, wrapped in a block cache.
///
/// `block_size` and `max_blocks` of `None` take the recommended values, which
/// differ by source kind — this is where the `file_buffer_size = -1` and
/// `max_file_buffer_count = -1` of the Python API are resolved.
pub fn open(
    path: &str,
    block_size: Option<u64>,
    max_blocks: Option<usize>,
) -> Result<std::sync::Arc<dyn ByteSource>> {
    let max_blocks = max_blocks.unwrap_or(MAX_BLOCKS).max(1);
    if is_url(path) {
        #[cfg(feature = "url")]
        {
            let block_size = block_size.unwrap_or(URL_BLOCK_SIZE).max(1);
            let source = UrlSource::open(path)?;
            return Ok(std::sync::Arc::new(CachedSource::new(
                source, block_size, max_blocks,
            )));
        }
        #[cfg(not(feature = "url"))]
        return Err(Error::Unsupported(format!(
            "{path} is a URL and the `url` feature is off"
        )));
    }
    let block_size = block_size.unwrap_or(LOCAL_BLOCK_SIZE).max(1);
    let source = LocalSource::open(path)?;
    Ok(std::sync::Arc::new(CachedSource::new(
        source, block_size, max_blocks,
    )))
}

/// An in-memory [`ByteSource`], for tests that want to hand a parser exact
/// bytes rather than build a file.
///
/// Behind `cfg(test)` for this crate's own tests and behind the `fuzzing`
/// feature for `fuzz/`, which is a separate crate and cannot see the first.
/// Not part of the stable surface either way: a caller with bytes in hand
/// wants `Cursor`, not this.
#[cfg(any(test, feature = "fuzzing"))]
pub mod testing {
    use super::*;

    #[derive(Debug)]
    pub struct MemorySource {
        data: Bytes,
        path: String,
    }

    impl MemorySource {
        pub fn new(data: impl Into<Bytes>) -> Self {
            Self {
                data: data.into(),
                path: "memory".to_string(),
            }
        }

        pub fn named(path: &str, data: impl Into<Bytes>) -> Self {
            Self {
                data: data.into(),
                path: path.to_string(),
            }
        }
    }

    impl ByteSource for MemorySource {
        fn path(&self) -> &str {
            &self.path
        }
        fn len(&self) -> Result<u64> {
            Ok(self.data.len() as u64)
        }
        fn read_at(&self, offset: u64, len: usize) -> Result<Bytes> {
            let start = (offset as usize).min(self.data.len());
            let end = start.saturating_add(len).min(self.data.len());
            Ok(self.data.slice(start..end))
        }
    }
}

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

    #[test]
    fn urls_are_recognised_by_scheme_and_case() {
        for path in [
            "http://example.org/a.bigwig",
            "HTTPS://example.org/a.bigwig",
        ] {
            assert!(is_url(path), "{path}");
        }
        for path in [
            "/data/a.bigwig",
            "a.bigwig",
            "C:/data/a.bigwig",
            "file:///data/a.bigwig",
            "s3://bucket/a.bigwig",
            // Routed here once, and refused by an HTTP client for its scheme
            // after the retry policy had had its say. A scheme this library
            // does not speak is not a URL as far as this is concerned.
            "ftp://example.org/a.bigwig",
        ] {
            assert!(!is_url(path), "{path}");
        }
    }
}