Enable streaming tarball download

Also add a non-compressed tar option

4 files changed, 151 insertions, 104 deletions

diff --git a/src/archive.rs b/src/archive.rs
index b62c3bd..ca22d28 100644
--- a/src/archive.rs
+++ b/src/archive.rs
@@ -1,8 +1,6 @@
 use actix_web::http::ContentEncoding;
-use bytes::Bytes;
 use libflate::gzip::Encoder;
 use serde::Deserialize;
-use std::io;
 use std::path::Path;
 use strum_macros::{Display, EnumIter, EnumString};
 use tar::Builder;
@@ -10,70 +8,83 @@ use tar::Builder;
 use crate::errors::ContextualError;
 
 /// Available compression methods
-#[derive(Deserialize, Clone, EnumIter, EnumString, Display)]
+#[derive(Deserialize, Clone, Copy, EnumIter, EnumString, Display)]
 #[serde(rename_all = "snake_case")]
 #[strum(serialize_all = "snake_case")]
 pub enum CompressionMethod {
-    /// TAR GZ
+    /// Gzipped tarball
     TarGz,
+
+    /// Regular tarball
+    Tar,
 }
 
 impl CompressionMethod {
-    pub fn extension(&self) -> String {
-        match &self {
+    pub fn extension(self) -> String {
+        match self {
             CompressionMethod::TarGz => "tar.gz",
+            CompressionMethod::Tar => "tar",
         }
         .to_string()
     }
 
-    pub fn content_type(&self) -> String {
-        match &self {
+    pub fn content_type(self) -> String {
+        match self {
             CompressionMethod::TarGz => "application/gzip",
+            CompressionMethod::Tar => "application/tar",
         }
         .to_string()
     }
 
-    pub fn content_encoding(&self) -> ContentEncoding {
-        match &self {
+    pub fn content_encoding(self) -> ContentEncoding {
+        match self {
             CompressionMethod::TarGz => ContentEncoding::Gzip,
+            CompressionMethod::Tar => ContentEncoding::Identity,
         }
     }
-    /// Creates an archive of a folder, using the algorithm the user chose from the web interface
-    /// This method returns the archive as a stream of bytes
-    pub fn create_archive<T: AsRef<Path>>(
-        &self,
+
+    pub fn create_archive<T, W>(
+        self,
         dir: T,
         skip_symlinks: bool,
-    ) -> Result<(String, Bytes), ContextualError> {
+        out: W,
+    ) -> Result<(), ContextualError>
+    where
+        T: AsRef<Path>,
+        W: std::io::Write,
+    {
+        let dir = dir.as_ref();
         match self {
-            CompressionMethod::TarGz => tgz_compress(dir, skip_symlinks),
+            CompressionMethod::TarGz => tar_gz(dir, skip_symlinks, out),
+            CompressionMethod::Tar => tar_dir(dir, skip_symlinks, out),
         }
     }
 }
 
-/// Compresses a given folder in .tar.gz format, and returns the result as a stream of bytes
-fn tgz_compress<T: AsRef<Path>>(
-    dir: T,
-    skip_symlinks: bool,
-) -> Result<(String, Bytes), ContextualError> {
-    if let Some(inner_folder) = dir.as_ref().file_name() {
-        if let Some(directory) = inner_folder.to_str() {
-            let dst_filename = format!("{}.tar", directory);
-            let dst_tgz_filename = format!("{}.gz", dst_filename);
-            let mut tgz_data = Bytes::new();
+fn tar_gz<W>(dir: &Path, skip_symlinks: bool, out: W) -> Result<(), ContextualError>
+where
+    W: std::io::Write,
+{
+    let mut out = Encoder::new(out).map_err(|e| ContextualError::IOError("GZIP".to_string(), e))?;
 
-            let tar_data =
-                tar(dir.as_ref(), directory.to_string(), skip_symlinks).map_err(|e| {
-                    ContextualError::ArchiveCreationError("tarball".to_string(), Box::new(e))
-                })?;
+    tar_dir(dir, skip_symlinks, &mut out)?;
 
-            let gz_data = gzip(&tar_data).map_err(|e| {
-                ContextualError::ArchiveCreationError("GZIP archive".to_string(), Box::new(e))
-            })?;
+    out.finish()
+        .into_result()
+        .map_err(|e| ContextualError::IOError("GZIP finish".to_string(), e))?;
 
-            tgz_data.extend_from_slice(&gz_data);
+    Ok(())
+}
 
-            Ok((dst_tgz_filename, tgz_data))
+fn tar_dir<W>(dir: &Path, skip_symlinks: bool, out: W) -> Result<(), ContextualError>
+where
+    W: std::io::Write,
+{
+    if let Some(inner_folder) = dir.file_name() {
+        if let Some(directory) = inner_folder.to_str() {
+            tar(dir, directory.to_string(), skip_symlinks, out).map_err(|e| {
+                ContextualError::ArchiveCreationError("tarball".to_string(), Box::new(e))
+            })
         } else {
             // https://doc.rust-lang.org/std/ffi/struct.OsStr.html#method.to_str
             Err(ContextualError::InvalidPathError(
@@ -88,45 +99,36 @@ fn tgz_compress<T: AsRef<Path>>(
     }
 }
 
-/// Creates a TAR archive of a folder, and returns it as a stream of bytes
-fn tar<T: AsRef<Path>>(
-    src_dir: T,
+fn tar<W>(
+    src_dir: &Path,
     inner_folder: String,
     skip_symlinks: bool,
-) -> Result<Vec<u8>, ContextualError> {
-    let mut tar_builder = Builder::new(Vec::new());
+    out: W,
+) -> Result<(), ContextualError>
+where
+    W: std::io::Write,
+{
+    let mut tar_builder = Builder::new(out);
 
     tar_builder.follow_symlinks(!skip_symlinks);
+
     // Recursively adds the content of src_dir into the archive stream
     tar_builder
-        .append_dir_all(inner_folder, src_dir.as_ref())
+        .append_dir_all(inner_folder, src_dir)
         .map_err(|e| {
             ContextualError::IOError(
                 format!(
                     "Failed to append the content of {} to the TAR archive",
-                    src_dir.as_ref().to_str().unwrap_or("file")
+                    src_dir.to_str().unwrap_or("file")
                 ),
                 e,
             )
         })?;
 
-    let tar_content = tar_builder.into_inner().map_err(|e| {
+    // Finish the archive
+    tar_builder.into_inner().map_err(|e| {
         ContextualError::IOError("Failed to finish writing the TAR archive".to_string(), e)
     })?;
 
-    Ok(tar_content)
-}
-
-/// Compresses a stream of bytes using the GZIP algorithm, and returns the resulting stream
-fn gzip(mut data: &[u8]) -> Result<Vec<u8>, ContextualError> {
-    let mut encoder = Encoder::new(Vec::new())
-        .map_err(|e| ContextualError::IOError("Failed to create GZIP encoder".to_string(), e))?;
-    io::copy(&mut data, &mut encoder)
-        .map_err(|e| ContextualError::IOError("Failed to write GZIP data".to_string(), e))?;
-    let data = encoder
-        .finish()
-        .into_result()
-        .map_err(|e| ContextualError::IOError("Failed to write GZIP trailer".to_string(), e))?;
-
-    Ok(data)
+    Ok(())
 }
diff --git a/src/listing.rs b/src/listing.rs
index 2ffcf2f..ba2e58e 100644
--- a/src/listing.rs
+++ b/src/listing.rs
@@ -1,7 +1,7 @@
-use actix_web::http::StatusCode;
-use actix_web::{fs, http, Body, FromRequest, HttpRequest, HttpResponse, Query, Result};
+use actix_web::{fs, Body, FromRequest, HttpRequest, HttpResponse, Query, Result};
 use bytesize::ByteSize;
-use futures::stream::once;
+use failure::Fail;
+use futures::Stream;
 use htmlescape::encode_minimal as escape_html_entity;
 use percent_encoding::{utf8_percent_encode, DEFAULT_ENCODE_SET};
 use serde::Deserialize;
@@ -235,46 +235,46 @@ pub fn directory_listing<S>(
 
     let color_scheme = query_params.theme.unwrap_or(default_color_scheme);
 
-    if let Some(compression_method) = &query_params.download {
+    if let Some(compression_method) = query_params.download {
         log::info!(
             "Creating an archive ({extension}) of {path}...",
             extension = compression_method.extension(),
             path = &dir.path.display().to_string()
         );
-        match compression_method.create_archive(&dir.path, skip_symlinks) {
-            Ok((filename, content)) => {
-                log::info!("{file} successfully created !", file = &filename);
-                Ok(HttpResponse::Ok()
-                    .content_type(compression_method.content_type())
-                    .content_encoding(compression_method.content_encoding())
-                    .header("Content-Transfer-Encoding", "binary")
-                    .header(
-                        "Content-Disposition",
-                        format!("attachment; filename={:?}", filename),
-                    )
-                    .chunked()
-                    .body(Body::Streaming(Box::new(once(Ok(content))))))
-            }
-            Err(err) => {
-                errors::log_error_chain(err.to_string());
-                Ok(HttpResponse::Ok()
-                    .status(http::StatusCode::INTERNAL_SERVER_ERROR)
-                    .body(
-                        renderer::render_error(
-                            &err.to_string(),
-                            StatusCode::INTERNAL_SERVER_ERROR,
-                            serve_path,
-                            query_params.sort,
-                            query_params.order,
-                            color_scheme,
-                            default_color_scheme,
-                            false,
-                            true,
-                        )
-                        .into_string(),
-                    ))
+
+        let filename = format!(
+            "{}.{}",
+            dir.path.file_name().unwrap().to_str().unwrap(),
+            compression_method.extension()
+        );
+
+        // Create a pipe to connect the archive creation thread and the response.
+        // Include 10 messages of buffer for erratic connection speeds.
+        let (tx, rx) = futures::sync::mpsc::channel(10);
+        let pipe = crate::pipe::Pipe::new(tx);
+
+        // Start the actual archive creation in a separate thread.
+        let dir = dir.path.to_path_buf();
+        std::thread::spawn(move || {
+            if let Err(err) = compression_method.create_archive(dir, skip_symlinks, pipe) {
+                log::error!("Error during archive creation: {:?}", err);
             }
-        }
+        });
+
+        // `<rx as Stream>::Error == ()` but we want `actix_web::error::Error`
+        // It can't happen, so let's just please the type checker.
+        let rx = rx.map_err(|_| unreachable!("pipes never fail"));
+
+        Ok(HttpResponse::Ok()
+            .content_type(compression_method.content_type())
+            .content_encoding(compression_method.content_encoding())
+            .header("Content-Transfer-Encoding", "binary")
+            .header(
+                "Content-Disposition",
+                format!("attachment; filename={:?}", filename),
+            )
+            .chunked()
+            .body(Body::Streaming(Box::new(rx))))
     } else {
         Ok(HttpResponse::Ok()
             .content_type("text/html; charset=utf-8")
@@ -302,7 +302,7 @@ pub fn extract_query_parameters<S>(req: &HttpRequest<S>) -> QueryParameters {
         Ok(query) => QueryParameters {
             sort: query.sort,
             order: query.order,
-            download: query.download.clone(),
+            download: query.download,
             theme: query.theme,
             path: query.path.clone(),
         },
diff --git a/src/main.rs b/src/main.rs
index f26369a..ddf25e1 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -16,6 +16,7 @@ mod auth;
 mod errors;
 mod file_upload;
 mod listing;
+mod pipe;
 mod renderer;
 mod themes;
 
@@ -79,17 +80,21 @@ fn run() -> Result<(), ContextualError> {
         TermLogger::init(LevelFilter::Error, Config::default())
     };
 
-    if miniserve_config.no_symlinks
-        && miniserve_config
+    if miniserve_config.no_symlinks {
+        let is_symlink = miniserve_config
             .path
             .symlink_metadata()
-            .map_err(|e| ContextualError::IOError("Failed to retrieve symlink's metadata".to_string(), e))?
+            .map_err(|e| {
+                ContextualError::IOError("Failed to retrieve symlink's metadata".to_string(), e)
+            })?
             .file_type()
-            .is_symlink()
-    {
-        return Err(ContextualError::from(
-            "The no-symlinks option cannot be used with a symlink path".to_string(),
-        ));
+            .is_symlink();
+
+        if is_symlink {
+            return Err(ContextualError::from(
+                "The no-symlinks option cannot be used with a symlink path".to_string(),
+            ));
+        }
     }
 
     let inside_config = miniserve_config.clone();
diff --git a/src/pipe.rs b/src/pipe.rs
new file mode 100644
index 0000000..710be1f
--- /dev/null
+++ b/src/pipe.rs
@@ -0,0 +1,40 @@
+use bytes::{Bytes, BytesMut};
+use futures::sink::{Sink, Wait};
+use futures::sync::mpsc::Sender;
+use std::io::{Error, ErrorKind, Result, Write};
+
+pub struct Pipe {
+    dest: Wait<Sender<Bytes>>,
+    bytes: BytesMut,
+}
+
+impl Pipe {
+    pub fn new(destination: Sender<Bytes>) -> Self {
+        Pipe {
+            dest: destination.wait(),
+            bytes: BytesMut::new(),
+        }
+    }
+}
+
+impl Drop for Pipe {
+    fn drop(&mut self) {
+        let _ = self.dest.close();
+    }
+}
+
+impl Write for Pipe {
+    fn write(&mut self, buf: &[u8]) -> Result<usize> {
+        self.bytes.extend_from_slice(buf);
+        match self.dest.send(self.bytes.take().into()) {
+            Ok(_) => Ok(buf.len()),
+            Err(e) => Err(Error::new(ErrorKind::UnexpectedEof, e)),
+        }
+    }
+
+    fn flush(&mut self) -> Result<()> {
+        self.dest
+            .flush()
+            .map_err(|e| Error::new(ErrorKind::UnexpectedEof, e))
+    }
+}