Simplify tftpd-targz.rs example

Author: oblique

Cargo.toml

@@ -31,19 +31,19 @@ futures-util = "0.3.5"
 
 [dev-dependencies]
 anyhow = "1.0.32"
-log = "0.4.11"
 structopt = "0.3.16"
 rand = { version = "0.7.3", features = ["small_rng"] }
 md5 = "0.7.0"
 tempfile = "3.1.0"
 matches = "0.1.8"
 fern = "0.6.0"
-piper = "0.1.3"
-tar = "0.4.29"
-flate2 = "1.0.16"
-threadpool = "1.8.1"
 async-executor = "0.1.2"
 async-channel = "1.4.0"
 
+# deps for tftpd-targz.rs
+async-std = "1.6.3"
+async-tar = "0.3.0"
+async-compression = { version = "0.3.5", features = ["gzip", "futures-io"] }
+
 [features]
 external-client-tests = []

examples/tftpd-targz.rs

@@ -1,28 +1,25 @@
 use anyhow::Result;
 use structopt::StructOpt;
 
+use async_compression::futures::bufread::GzipDecoder;
+use async_std::fs::File;
+use async_std::io::{BufReader, Sink};
+use async_std::path::{Path, PathBuf};
+use async_std::stream::StreamExt;
+use async_std::task::block_on;
+use async_tar::{Archive, Entry};
 use async_tftp::packet;
 use async_tftp::server::{Handler, TftpServerBuilder};
-use flate2::read::GzDecoder;
-use futures_lite::future::block_on;
-use futures_lite::io::{AsyncWrite, AsyncWriteExt, Sink};
-use std::fs::File;
-use std::io::Read;
 use std::net::SocketAddr;
-use std::path::{Path, PathBuf};
-use tar::Archive;
-use threadpool::ThreadPool;
 
 struct TftpdTarGzHandler {
     archive_path: PathBuf,
-    thread_pool: ThreadPool,
 }
 
 impl TftpdTarGzHandler {
     fn new(path: impl AsRef<Path>) -> Self {
         TftpdTarGzHandler {
             archive_path: path.as_ref().to_owned(),
-            thread_pool: ThreadPool::new(5),
         }
     }
 }
@@ -32,108 +29,48 @@ fn strip_path_prefixes(path: &Path) -> &Path {
     path.strip_prefix("/").or_else(|_| path.strip_prefix("./")).unwrap_or(path)
 }
 
-fn io_copy(
-    mut src: impl Read,
-    mut dest: impl AsyncWrite + Unpin,
-) -> std::io::Result<()> {
-    let mut buf = [0u8; 1024];
-
-    loop {
-        match src.read(&mut buf[..]) {
-            Ok(0) => break,
-            Ok(len) => block_on(dest.write_all(&buf[..len]))?,
-            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
-            Err(e) => return Err(e),
-        }
-    }
-
-    Ok(())
-}
-
-// This macro sends `Err` over a channel, it is used within `thread_pool`.
-macro_rules! try_or_send {
-    ($e:expr, $tx:ident) => {{
-        match $e {
-            Ok(x) => x,
-            Err(e) => {
-                let _ = $tx.send(Err(e.into()));
-                return;
-            }
-        }
-    }};
-}
-
 #[async_tftp::async_trait]
 impl Handler for TftpdTarGzHandler {
-    type Reader = piper::Reader;
+    type Reader = Entry<Archive<GzipDecoder<BufReader<File>>>>;
     type Writer = Sink;
 
     async fn read_req_open(
         &mut self,
         _client: &SocketAddr,
-        path: &Path,
+        path: &std::path::Path,
     ) -> Result<(Self::Reader, Option<u64>), packet::Error> {
-        let req_path = strip_path_prefixes(path).to_owned();
-        let archive_path = self.archive_path.clone();
-
-        let (pipe_r, pipe_w) = piper::pipe(65536);
-        let (open_res_tx, open_res_rx) = async_channel::bounded(1);
+        let req_path = strip_path_prefixes(path.into()).to_owned();
 
-        // We need to use our own thread pool to handle blocking IO
-        // of `tar::Entry`.
-        self.thread_pool.execute(move || {
-            let file = try_or_send!(File::open(archive_path), open_res_tx);
+        let file = File::open(self.archive_path.clone()).await?;
+        let archive = Archive::new(GzipDecoder::new(BufReader::new(file)));
 
-            let mut archive = Archive::new(GzDecoder::new(file));
-            let entries = try_or_send!(archive.entries(), open_res_tx);
+        let mut entries = archive.entries()?;
 
-            for entry in entries {
-                let entry = try_or_send!(entry, open_res_tx);
+        while let Some(Ok(entry)) = entries.next().await {
+            if entry
+                .path()
+                .map(|p| strip_path_prefixes(&*p) == req_path)
+                .unwrap_or(false)
+            {
+                // We manage to find the entry.
 
-                // If entry path is the same with requested path.
-                if entry
-                    .path()
-                    .map(|p| strip_path_prefixes(&p) == req_path)
-                    .unwrap_or(false)
+                // Check if it is a regular file.
+                if entry.header().entry_type() != async_tar::EntryType::Regular
                 {
-                    // We manage to find the entry.
-
-                    // Check if it is a regular file.
-                    if entry.header().entry_type() != tar::EntryType::Regular {
-                        break;
-                    }
-
-                    // Inform handler to continue on serving the data.
-                    if open_res_tx.try_send(Ok(())).is_err() {
-                        // Do not transfer data if handler task is canceled.
-                        return;
-                    }
-
-                    // Forward data to handler.
-                    let _ = io_copy(entry, pipe_w);
-
-                    return;
+                    break;
                 }
 
-                // Requested path not found within the archive.
-                let _ = open_res_tx.send(Err(packet::Error::FileNotFound));
+                return Ok((entry, None));
             }
-        });
-
-        // Wait for the above task to find the requested path and
-        // starts transferring data.
-        open_res_rx
-            .recv()
-            .await
-            .unwrap_or_else(|_| Err(packet::Error::FileNotFound))?;
+        }
 
-        Ok((pipe_r, None))
+        Err(packet::Error::FileNotFound)
     }
 
     async fn write_req_open(
         &mut self,
         _client: &SocketAddr,
-        _path: &Path,
+        _path: &std::path::Path,
         _size: Option<u64>,
     ) -> Result<Self::Writer, packet::Error> {
         Err(packet::Error::IllegalOperation)