diff --git a/rustfs/src/admin/rpc.rs b/rustfs/src/admin/rpc.rs index d650e5c51..16cd5be3f 100644 --- a/rustfs/src/admin/rpc.rs +++ b/rustfs/src/admin/rpc.rs @@ -6,7 +6,7 @@ use ecstore::disk::DiskAPI; use ecstore::disk::WalkDirOptions; use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; use ecstore::store::find_local_disk; -use futures::TryStreamExt; +use futures::StreamExt; use http::StatusCode; use hyper::Method; use matchit::Params; @@ -17,8 +17,8 @@ use s3s::S3Result; use s3s::dto::StreamingBlob; use s3s::s3_error; use serde_urlencoded::from_bytes; +use tokio::io::AsyncWriteExt; use tokio_util::io::ReaderStream; -use tokio_util::io::StreamReader; use tracing::warn; pub const RPC_PREFIX: &str = "/rustfs/rpc"; @@ -194,11 +194,12 @@ impl Operation for PutFile { .map_err(|e| s3_error!(InternalError, "read file err {}", e))? }; - let mut body = StreamReader::new(req.input.into_stream().map_err(std::io::Error::other)); - - tokio::io::copy(&mut body, &mut file) - .await - .map_err(|e| s3_error!(InternalError, "copy err {}", e))?; + let mut body = req.input; + while let Some(item) = body.next().await { + let bytes = item.map_err(|e| s3_error!(InternalError, "body stream err {}", e))?; + let result = file.write_all(&bytes).await; + result.map_err(|e| s3_error!(InternalError, "write file err {}", e))?; + } Ok(S3Response::new((StatusCode::OK, Body::empty()))) }