Rename lib to binstalk (#361)

crates/binstalk/src/helpers/download/async_extracter.rs

@@ -0,0 +1,114 @@
+use std::{
+    fmt::Debug,
+    fs,
+    io::{copy, Read, Seek},
+    path::Path,
+};
+
+use bytes::Bytes;
+use futures_util::stream::Stream;
+use log::debug;
+use scopeguard::{guard, ScopeGuard};
+use tar::Entries;
+use tempfile::tempfile;
+use tokio::task::block_in_place;
+
+use super::{extracter::*, stream_readable::StreamReadable};
+use crate::{errors::BinstallError, manifests::cargo_toml_binstall::TarBasedFmt};
+
+pub async fn extract_bin<S, E>(stream: S, path: &Path) -> Result<(), BinstallError>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
+    BinstallError: From<E>,
+{
+    let mut reader = StreamReadable::new(stream).await;
+    block_in_place(move || {
+        fs::create_dir_all(path.parent().unwrap())?;
+
+        let mut file = fs::File::create(path)?;
+
+        // remove it unless the operation isn't aborted and no write
+        // fails.
+        let remove_guard = guard(&path, |path| {
+            fs::remove_file(path).ok();
+        });
+
+        copy(&mut reader, &mut file)?;
+
+        // Operation isn't aborted and all writes succeed,
+        // disarm the remove_guard.
+        ScopeGuard::into_inner(remove_guard);
+
+        Ok(())
+    })
+}
+
+pub async fn extract_zip<S, E>(stream: S, path: &Path) -> Result<(), BinstallError>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
+    BinstallError: From<E>,
+{
+    let mut reader = StreamReadable::new(stream).await;
+    block_in_place(move || {
+        fs::create_dir_all(path.parent().unwrap())?;
+
+        let mut file = tempfile()?;
+
+        copy(&mut reader, &mut file)?;
+
+        // rewind it so that we can pass it to unzip
+        file.rewind()?;
+
+        unzip(file, path)
+    })
+}
+
+pub async fn extract_tar_based_stream<S, E>(
+    stream: S,
+    path: &Path,
+    fmt: TarBasedFmt,
+) -> Result<(), BinstallError>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
+    BinstallError: From<E>,
+{
+    let reader = StreamReadable::new(stream).await;
+    block_in_place(move || {
+        fs::create_dir_all(path.parent().unwrap())?;
+
+        debug!("Extracting from {fmt} archive to {path:#?}");
+
+        create_tar_decoder(reader, fmt)?.unpack(path)?;
+
+        Ok(())
+    })
+}
+
+/// Visitor must iterate over all entries.
+/// Entires can be in arbitary order.
+pub trait TarEntriesVisitor {
+    type Target;
+
+    fn visit<R: Read>(&mut self, entries: Entries<'_, R>) -> Result<(), BinstallError>;
+    fn finish(self) -> Result<Self::Target, BinstallError>;
+}
+
+pub async fn extract_tar_based_stream_and_visit<S, V, E>(
+    stream: S,
+    fmt: TarBasedFmt,
+    mut visitor: V,
+) -> Result<V::Target, BinstallError>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
+    V: TarEntriesVisitor + Debug + Send + 'static,
+    BinstallError: From<E>,
+{
+    let reader = StreamReadable::new(stream).await;
+    block_in_place(move || {
+        debug!("Extracting from {fmt} archive to process it in memory");
+
+        let mut tar = create_tar_decoder(reader, fmt)?;
+        visitor.visit(tar.entries()?)?;
+        visitor.finish()
+    })
+}

crates/binstalk/src/helpers/download/extracter.rs

@@ -0,0 +1,46 @@
+use std::{
+    fs::File,
+    io::{self, BufRead, Read},
+    path::Path,
+};
+
+use bzip2::bufread::BzDecoder;
+use flate2::bufread::GzDecoder;
+use log::debug;
+use tar::Archive;
+use xz2::bufread::XzDecoder;
+use zip::read::ZipArchive;
+use zstd::stream::Decoder as ZstdDecoder;
+
+use crate::{errors::BinstallError, manifests::cargo_toml_binstall::TarBasedFmt};
+
+pub fn create_tar_decoder(
+    dat: impl BufRead + 'static,
+    fmt: TarBasedFmt,
+) -> io::Result<Archive<Box<dyn Read>>> {
+    use TarBasedFmt::*;
+
+    let r: Box<dyn Read> = match fmt {
+        Tar => Box::new(dat),
+        Tbz2 => Box::new(BzDecoder::new(dat)),
+        Tgz => Box::new(GzDecoder::new(dat)),
+        Txz => Box::new(XzDecoder::new(dat)),
+        Tzstd => {
+            // The error can only come from raw::Decoder::with_dictionary as of zstd 0.10.2 and
+            // 0.11.2, which is specified as `&[]` by `ZstdDecoder::new`, thus `ZstdDecoder::new`
+            // should not return any error.
+            Box::new(ZstdDecoder::with_buffer(dat)?)
+        }
+    };
+
+    Ok(Archive::new(r))
+}
+
+pub fn unzip(dat: File, dst: &Path) -> Result<(), BinstallError> {
+    debug!("Decompressing from zip archive to `{dst:?}`");
+
+    let mut zip = ZipArchive::new(dat)?;
+    zip.extract(dst)?;
+
+    Ok(())
+}

crates/binstalk/src/helpers/download/stream_readable.rs

@@ -0,0 +1,83 @@
+use std::{
+    cmp::min,
+    io::{self, BufRead, Read},
+};
+
+use bytes::{Buf, Bytes};
+use futures_util::stream::{Stream, StreamExt};
+use tokio::runtime::Handle;
+
+use crate::errors::BinstallError;
+
+/// This wraps an AsyncIterator as a `Read`able.
+/// It must be used in non-async context only,
+/// meaning you have to use it with
+/// `tokio::task::{block_in_place, spawn_blocking}` or
+/// `std::thread::spawn`.
+#[derive(Debug)]
+pub struct StreamReadable<S> {
+    stream: S,
+    handle: Handle,
+    bytes: Bytes,
+}
+
+impl<S> StreamReadable<S> {
+    pub(super) async fn new(stream: S) -> Self {
+        Self {
+            stream,
+            handle: Handle::current(),
+            bytes: Bytes::new(),
+        }
+    }
+}
+
+impl<S, E> Read for StreamReadable<S>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin,
+    BinstallError: From<E>,
+{
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        if buf.is_empty() {
+            return Ok(0);
+        }
+
+        if self.fill_buf()?.is_empty() {
+            return Ok(0);
+        }
+
+        let bytes = &mut self.bytes;
+
+        // copy_to_slice requires the bytes to have enough remaining bytes
+        // to fill buf.
+        let n = min(buf.len(), bytes.remaining());
+
+        bytes.copy_to_slice(&mut buf[..n]);
+
+        Ok(n)
+    }
+}
+impl<S, E> BufRead for StreamReadable<S>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin,
+    BinstallError: From<E>,
+{
+    fn fill_buf(&mut self) -> io::Result<&[u8]> {
+        let bytes = &mut self.bytes;
+
+        if !bytes.has_remaining() {
+            match self.handle.block_on(async { self.stream.next().await }) {
+                Some(Ok(new_bytes)) => *bytes = new_bytes,
+                Some(Err(e)) => {
+                    let e: BinstallError = e.into();
+                    return Err(io::Error::new(io::ErrorKind::Other, e));
+                }
+                None => (),
+            }
+        }
+        Ok(&*bytes)
+    }
+
+    fn consume(&mut self, amt: usize) {
+        self.bytes.advance(amt);
+    }
+}