From 8900dde7e70fcfd237d73660697585dce246a931 Mon Sep 17 00:00:00 2001 From: sigoden Date: Sun, 29 May 2022 06:57:16 +0800 Subject: [PATCH] refactor: handler zip --- src/server.rs | 23 +++++------------------ 1 file changed, 5 insertions(+), 18 deletions(-) diff --git a/src/server.rs b/src/server.rs index af04c40..ff03047 100644 --- a/src/server.rs +++ b/src/server.rs @@ -5,7 +5,6 @@ use async_zip::write::{EntryOptions, ZipFileWriter}; use async_zip::Compression; use futures::stream::StreamExt; use futures::TryStreamExt; -use hyper::body::Bytes; use hyper::header::HeaderValue; use hyper::service::{make_service_fn, service_fn}; use hyper::{Body, Method, StatusCode}; @@ -16,10 +15,10 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::SystemTime; use tokio::fs::File; -use tokio::io::{AsyncReadExt, AsyncWrite}; +use tokio::io::AsyncWrite; use tokio::{fs, io}; use tokio_util::codec::{BytesCodec, FramedRead}; -use tokio_util::io::StreamReader; +use tokio_util::io::{ReaderStream, StreamReader}; type Request = hyper::Request; type Response = hyper::Response; @@ -188,27 +187,15 @@ impl InnerService { } async fn handle_send_dir_zip(&self, path: &Path) -> BoxResult { - let (mut tx, body) = Body::channel(); - let (mut writer, mut reader) = tokio::io::duplex(BUF_SIZE); + let (mut writer, reader) = tokio::io::duplex(BUF_SIZE); let path = path.to_owned(); tokio::spawn(async move { if let Err(e) = dir_zip(&mut writer, &path).await { error!("Fail to zip {}, {}", path.display(), e.to_string()); } }); - tokio::spawn(async move { - // Reuse this buffer - let mut buf = [0_u8; BUF_SIZE]; - loop { - let n = reader.read(&mut buf).await.unwrap(); - if n == 0 { - break; - } - if (tx.send_data(Bytes::from(buf[..n].to_vec())).await).is_err() { - break; - } - } - }); + let stream = ReaderStream::new(reader); + let body = Body::wrap_stream(stream); Ok(Response::new(body)) }