From e520299c4bcb0f3c814c2e2a30dc84db8c132ce9 Mon Sep 17 00:00:00 2001 From: Nugine Date: Tue, 17 Jun 2025 16:22:55 +0800 Subject: [PATCH 1/6] refactor(ecstore): `DiskAPI::write_all` use `Bytes` --- ecstore/src/disk/local.rs | 30 +++++++++++++++--------------- ecstore/src/disk/mod.rs | 4 ++-- ecstore/src/disk/remote.rs | 4 ++-- ecstore/src/heal/heal_commands.rs | 2 +- ecstore/src/store_init.rs | 2 +- rustfs/src/grpc.rs | 2 +- 6 files changed, 22 insertions(+), 22 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index a66d64078..004aaa61d 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -68,7 +68,7 @@ use uuid::Uuid; #[derive(Debug)] pub struct FormatInfo { pub id: Option, - pub data: Vec, + pub data: Bytes, pub file_info: Option, pub last_check: Option, } @@ -153,7 +153,7 @@ impl LocalDisk { let format_info = FormatInfo { id, - data: format_data, + data: format_data.into(), file_info: format_meta, last_check: format_last_check, }; @@ -629,7 +629,7 @@ impl LocalDisk { } // write_all_public for trail - async fn write_all_public(&self, volume: &str, path: &str, data: Vec) -> Result<()> { + async fn write_all_public(&self, volume: &str, path: &str, data: Bytes) -> Result<()> { if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { let mut format_info = self.format_info.write().await; format_info.data.clone_from(&data); @@ -637,7 +637,7 @@ impl LocalDisk { let volume_dir = self.get_bucket_path(volume)?; - self.write_all_private(volume, path, data.into(), true, &volume_dir).await?; + self.write_all_private(volume, path, data, true, &volume_dir).await?; Ok(()) } @@ -1131,7 +1131,7 @@ impl DiskAPI for LocalDisk { format_info.id = Some(disk_id); format_info.file_info = Some(file_meta); - format_info.data = b; + format_info.data = b.into(); format_info.last_check = Some(OffsetDateTime::now_utc()); Ok(Some(disk_id)) @@ -1151,7 +1151,7 @@ impl DiskAPI for LocalDisk { if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { let format_info = self.format_info.read().await; if !format_info.data.is_empty() { - return Ok(format_info.data.clone()); + return Ok(format_info.data.to_vec()); } } // TOFIX: @@ -1162,7 +1162,7 @@ impl DiskAPI for LocalDisk { } #[tracing::instrument(level = "debug", skip_all)] - async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { + async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> { self.write_all_public(volume, path, data).await } @@ -1331,7 +1331,7 @@ impl DiskAPI for LocalDisk { rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?; - self.write_all(dst_volume, format!("{}.meta", dst_path).as_str(), meta.to_vec()) + self.write_all(dst_volume, format!("{}.meta", dst_path).as_str(), meta) .await?; if let Some(parent) = src_file_path.parent() { @@ -1700,7 +1700,7 @@ impl DiskAPI for LocalDisk { let new_dst_buf = xlmeta.marshal_msg()?; - self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf) + self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf.into()) .await?; if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() { let no_inline = fi.data.is_none() && fi.size > 0; @@ -1902,7 +1902,7 @@ impl DiskAPI for LocalDisk { let fm_data = meta.marshal_msg()?; - self.write_all(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), fm_data) + self.write_all(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), fm_data.into()) .await?; Ok(()) @@ -2461,8 +2461,8 @@ mod test { disk.make_volume("test-volume").await.unwrap(); // Test write and read operations - let test_data = vec![1, 2, 3, 4, 5]; - disk.write_all("test-volume", "test-file.txt", test_data.clone()) + let test_data: Vec = vec![1, 2, 3, 4, 5]; + disk.write_all("test-volume", "test-file.txt", test_data.clone().into()) .await .unwrap(); @@ -2587,7 +2587,7 @@ mod test { // Valid format info let valid_format_info = FormatInfo { id: Some(Uuid::new_v4()), - data: vec![1, 2, 3], + data: vec![1, 2, 3].into(), file_info: Some(fs::metadata(".").await.unwrap()), last_check: Some(now), }; @@ -2596,7 +2596,7 @@ mod test { // Invalid format info (missing id) let invalid_format_info = FormatInfo { id: None, - data: vec![1, 2, 3], + data: vec![1, 2, 3].into(), file_info: Some(fs::metadata(".").await.unwrap()), last_check: Some(now), }; @@ -2606,7 +2606,7 @@ mod test { let old_time = OffsetDateTime::now_utc() - time::Duration::seconds(10); let old_format_info = FormatInfo { id: Some(Uuid::new_v4()), - data: vec![1, 2, 3], + data: vec![1, 2, 3].into(), file_info: Some(fs::metadata(".").await.unwrap()), last_check: Some(old_time), }; diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 8fc016017..821e362f3 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -364,7 +364,7 @@ impl DiskAPI for Disk { } #[tracing::instrument(skip(self))] - async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { + async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await, Disk::Remote(remote_disk) => remote_disk.write_all(volume, path, data).await, @@ -504,7 +504,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { // ReadParts async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; // CleanAbandonedData - async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; + async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>; async fn read_all(&self, volume: &str, path: &str) -> Result>; async fn disk_info(&self, opts: &DiskInfoOptions) -> Result; async fn ns_scanner( diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 25fd11ebe..eb3794a47 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -804,7 +804,7 @@ impl DiskAPI for RemoteDisk { } #[tracing::instrument(skip(self))] - async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { + async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> { info!("write_all"); let mut client = node_service_time_out_client(&self.addr) .await @@ -813,7 +813,7 @@ impl DiskAPI for RemoteDisk { disk: self.endpoint.to_string(), volume: volume.to_string(), path: path.to_string(), - data: data.into(), + data, }); let response = client.write_all(request).await?.into_inner(); diff --git a/ecstore/src/heal/heal_commands.rs b/ecstore/src/heal/heal_commands.rs index daa434a3b..7dd798a9a 100644 --- a/ecstore/src/heal/heal_commands.rs +++ b/ecstore/src/heal/heal_commands.rs @@ -232,7 +232,7 @@ impl HealingTracker { if let Some(disk) = &self.disk { let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME); - disk.write_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), htracker_bytes) + disk.write_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), htracker_bytes.into()) .await?; } Ok(()) diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs index 68a6b72ba..9cb781b1e 100644 --- a/ecstore/src/store_init.rs +++ b/ecstore/src/store_init.rs @@ -311,7 +311,7 @@ pub async fn save_format_file(disk: &Option, format: &Option) -> Result, Status> { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { - match disk.write_all(&request.volume, &request.path, request.data.into()).await { + match disk.write_all(&request.volume, &request.path, request.data).await { Ok(_) => Ok(tonic::Response::new(WriteAllResponse { success: true, error: None, From 39e988537c138c447197be6667d48efbb7940f96 Mon Sep 17 00:00:00 2001 From: Nugine Date: Tue, 17 Jun 2025 16:22:55 +0800 Subject: [PATCH 2/6] refactor(ecstore): `DiskAPI::read_all` use `Bytes` --- ecstore/src/disk/local.rs | 26 +++++++++++++------------- ecstore/src/disk/mod.rs | 4 ++-- ecstore/src/disk/remote.rs | 4 ++-- ecstore/src/store_init.rs | 2 +- rustfs/src/grpc.rs | 2 +- 5 files changed, 19 insertions(+), 19 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 004aaa61d..1cfd28e2a 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -138,7 +138,7 @@ impl LocalDisk { let mut format_last_check = None; if !format_data.is_empty() { - let s = format_data.as_slice(); + let s = format_data.as_ref(); let fm = FormatV3::try_from(s).map_err(Error::other)?; let (set_idx, disk_idx) = fm.find_disk_index_by_disk_id(fm.erasure.this)?; @@ -153,7 +153,7 @@ impl LocalDisk { let format_info = FormatInfo { id, - data: format_data.into(), + data: format_data, file_info: format_meta, last_check: format_last_check, }; @@ -980,13 +980,13 @@ fn is_root_path(path: impl AsRef) -> bool { } // 过滤 std::io::ErrorKind::NotFound -pub async fn read_file_exists(path: impl AsRef) -> Result<(Vec, Option)> { +pub async fn read_file_exists(path: impl AsRef) -> Result<(Bytes, Option)> { let p = path.as_ref(); let (data, meta) = match read_file_all(&p).await { Ok((data, meta)) => (data, Some(meta)), Err(e) => { if e == Error::FileNotFound { - (Vec::new(), None) + (Bytes::new(), None) } else { return Err(e); } @@ -1001,13 +1001,13 @@ pub async fn read_file_exists(path: impl AsRef) -> Result<(Vec, Option Ok((data, meta)) } -pub async fn read_file_all(path: impl AsRef) -> Result<(Vec, Metadata)> { +pub async fn read_file_all(path: impl AsRef) -> Result<(Bytes, Metadata)> { let p = path.as_ref(); let meta = read_file_metadata(&path).await?; let data = fs::read(&p).await.map_err(to_file_error)?; - Ok((data, meta)) + Ok((data.into(), meta)) } pub async fn read_file_metadata(p: impl AsRef) -> Result { @@ -1147,11 +1147,11 @@ impl DiskAPI for LocalDisk { } #[tracing::instrument(skip(self))] - async fn read_all(&self, volume: &str, path: &str) -> Result> { + async fn read_all(&self, volume: &str, path: &str) -> Result { if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { let format_info = self.format_info.read().await; if !format_info.data.is_empty() { - return Ok(format_info.data.to_vec()); + return Ok(format_info.data.clone()); } } // TOFIX: @@ -1866,11 +1866,11 @@ impl DiskAPI for LocalDisk { } })?; - if !FileMeta::is_xl2_v1_format(buf.as_slice()) { + if !FileMeta::is_xl2_v1_format(buf.as_ref()) { return Err(DiskError::FileVersionNotFound); } - let mut xl_meta = FileMeta::load(buf.as_slice())?; + let mut xl_meta = FileMeta::load(buf.as_ref())?; xl_meta.update_object_version(fi)?; @@ -2076,7 +2076,7 @@ impl DiskAPI for LocalDisk { } res.exists = true; - res.data = data; + res.data = data.into(); res.mod_time = match meta.modified() { Ok(md) => Some(OffsetDateTime::from(md)), Err(_) => { @@ -2627,7 +2627,7 @@ mod test { // Test existing file let (data, metadata) = read_file_exists(test_file).await.unwrap(); - assert_eq!(data, b"test content"); + assert_eq!(data.as_ref(), b"test content"); assert!(metadata.is_some()); // Clean up @@ -2644,7 +2644,7 @@ mod test { // Test reading file let (data, metadata) = read_file_all(test_file).await.unwrap(); - assert_eq!(data, test_content); + assert_eq!(data.as_ref(), test_content); assert!(metadata.is_file()); assert_eq!(metadata.len(), test_content.len() as u64); diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 821e362f3..a345732f9 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -372,7 +372,7 @@ impl DiskAPI for Disk { } #[tracing::instrument(skip(self))] - async fn read_all(&self, volume: &str, path: &str) -> Result> { + async fn read_all(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.read_all(volume, path).await, Disk::Remote(remote_disk) => remote_disk.read_all(volume, path).await, @@ -505,7 +505,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; // CleanAbandonedData async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>; - async fn read_all(&self, volume: &str, path: &str) -> Result>; + async fn read_all(&self, volume: &str, path: &str) -> Result; async fn disk_info(&self, opts: &DiskInfoOptions) -> Result; async fn ns_scanner( &self, diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index eb3794a47..9495a14ea 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -826,7 +826,7 @@ impl DiskAPI for RemoteDisk { } #[tracing::instrument(skip(self))] - async fn read_all(&self, volume: &str, path: &str) -> Result> { + async fn read_all(&self, volume: &str, path: &str) -> Result { info!("read_all {}/{}", volume, path); let mut client = node_service_time_out_client(&self.addr) .await @@ -843,7 +843,7 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } - Ok(response.data.into()) + Ok(response.data) } #[tracing::instrument(skip(self))] diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs index 9cb781b1e..97c23cb02 100644 --- a/ecstore/src/store_init.rs +++ b/ecstore/src/store_init.rs @@ -256,7 +256,7 @@ pub async fn load_format_erasure(disk: &DiskStore, heal: bool) -> disk::error::R _ => e, })?; - let mut fm = FormatV3::try_from(data.as_slice())?; + let mut fm = FormatV3::try_from(data.as_ref())?; if heal { let info = disk diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index ee7bd1366..95ca12c2e 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -277,7 +277,7 @@ impl Node for NodeService { match disk.read_all(&request.volume, &request.path).await { Ok(data) => Ok(tonic::Response::new(ReadAllResponse { success: true, - data: data.into(), + data, error: None, })), Err(err) => Ok(tonic::Response::new(ReadAllResponse { From da4a4e7cbe79d98782e8344b565016f9a91fefa5 Mon Sep 17 00:00:00 2001 From: Nugine Date: Tue, 17 Jun 2025 16:22:55 +0800 Subject: [PATCH 3/6] feat(ecstore): erasure encode reuse buf --- ecstore/src/erasure_coding/encode.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/ecstore/src/erasure_coding/encode.rs b/ecstore/src/erasure_coding/encode.rs index c9dcac1be..6517c26e0 100644 --- a/ecstore/src/erasure_coding/encode.rs +++ b/ecstore/src/erasure_coding/encode.rs @@ -104,8 +104,8 @@ impl Erasure { let task = tokio::spawn(async move { let block_size = self.block_size; let mut total = 0; + let mut buf = vec![0u8; block_size]; loop { - let mut buf = vec![0u8; block_size]; match rustfs_utils::read_full(&mut reader, &mut buf).await { Ok(n) if n > 0 => { total += n; @@ -122,7 +122,6 @@ impl Erasure { return Err(e); } } - buf.clear(); } Ok((reader, total)) From 086eab8c7030cf32d116c7b3e8d15f0341ef5b1b Mon Sep 17 00:00:00 2001 From: Nugine Date: Tue, 17 Jun 2025 16:22:55 +0800 Subject: [PATCH 4/6] feat(admin): PutFile stream write file --- rustfs/src/admin/rpc.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) 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()))) } From 4cadc4c12d69d7026b910fe999c6cf05ed8ba17c Mon Sep 17 00:00:00 2001 From: Nugine Date: Tue, 17 Jun 2025 16:22:55 +0800 Subject: [PATCH 5/6] feat(ecstore): MultiWriter concurrent write --- ecstore/src/erasure_coding/encode.rs | 44 +++++++++++++++++----------- 1 file changed, 27 insertions(+), 17 deletions(-) diff --git a/ecstore/src/erasure_coding/encode.rs b/ecstore/src/erasure_coding/encode.rs index 6517c26e0..899b8f570 100644 --- a/ecstore/src/erasure_coding/encode.rs +++ b/ecstore/src/erasure_coding/encode.rs @@ -4,6 +4,8 @@ use crate::disk::error::Error; use crate::disk::error_reduce::count_errs; use crate::disk::error_reduce::{OBJECT_OP_IGNORED_ERRS, reduce_write_quorum_errs}; use bytes::Bytes; +use futures::StreamExt; +use futures::stream::FuturesUnordered; use std::sync::Arc; use std::vec; use tokio::io::AsyncRead; @@ -26,33 +28,41 @@ impl<'a> MultiWriter<'a> { } } - #[allow(clippy::needless_range_loop)] - pub async fn write(&mut self, data: Vec) -> std::io::Result<()> { - for i in 0..self.writers.len() { - if self.errs[i].is_some() { - continue; // Skip if we already have an error for this writer - } - - let writer_opt = &mut self.writers[i]; - let shard = &data[i]; - - if let Some(writer) = writer_opt { + async fn write_shard(writer_opt: &mut Option, err: &mut Option, shard: &Bytes) { + match writer_opt { + Some(writer) => { match writer.write(shard).await { Ok(n) => { if n < shard.len() { - self.errs[i] = Some(Error::ShortWrite); - self.writers[i] = None; // Mark as failed + *err = Some(Error::ShortWrite); + *writer_opt = None; // Mark as failed } else { - self.errs[i] = None; + *err = None; } } Err(e) => { - self.errs[i] = Some(Error::from(e)); + *err = Some(Error::from(e)); } } - } else { - self.errs[i] = Some(Error::DiskNotFound); } + None => { + *err = Some(Error::DiskNotFound); + } + } + } + + pub async fn write(&mut self, data: Vec) -> std::io::Result<()> { + assert_eq!(data.len(), self.writers.len()); + + { + let mut futures = FuturesUnordered::new(); + for ((writer_opt, err), shard) in self.writers.iter_mut().zip(self.errs.iter_mut()).zip(data.iter()) { + if err.is_some() { + continue; // Skip if we already have an error for this writer + } + futures.push(Self::write_shard(writer_opt, err, shard)); + } + while let Some(()) = futures.next().await {} } let nil_count = self.errs.iter().filter(|&e| e.is_none()).count(); From 4a786618d47c0321c4c044f34eb1f65ec5256999 Mon Sep 17 00:00:00 2001 From: Nugine Date: Tue, 17 Jun 2025 16:22:55 +0800 Subject: [PATCH 6/6] refactor(rio): HttpReader use StreamReader --- crates/rio/src/http_reader.rs | 92 +++++++++++------------------------ 1 file changed, 28 insertions(+), 64 deletions(-) diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index e0cfc89c6..80801d05d 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -1,15 +1,17 @@ use bytes::Bytes; -use futures::{Stream, StreamExt}; +use futures::{Stream, TryStreamExt as _}; use http::HeaderMap; use pin_project_lite::pin_project; use reqwest::{Client, Method, RequestBuilder}; use std::error::Error as _; use std::io::{self, Error}; +use std::ops::Not as _; use std::pin::Pin; use std::sync::LazyLock; use std::task::{Context, Poll}; -use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, DuplexStream, ReadBuf}; -use tokio::sync::{mpsc, oneshot}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio::sync::mpsc; +use tokio_util::io::StreamReader; use crate::{EtagResolvable, HashReaderDetector, HashReaderMut}; @@ -38,8 +40,7 @@ pin_project! { url:String, method: Method, headers: HeaderMap, - inner: DuplexStream, - err_rx: oneshot::Receiver, + inner: StreamReader>+Send+Sync>>, Bytes>, } } @@ -54,11 +55,11 @@ impl HttpReader { method: Method, headers: HeaderMap, body: Option>, - mut read_buf_size: usize, + _read_buf_size: usize, ) -> io::Result { http_log!( "[HttpReader::with_capacity] url: {url}, method: {method:?}, headers: {headers:?}, buf_size: {}", - read_buf_size + _read_buf_size ); // First, check if the connection is available (HEAD) let client = get_http_client(); @@ -76,59 +77,30 @@ impl HttpReader { } } - let url_clone = url.clone(); - let method_clone = method.clone(); - let headers_clone = headers.clone(); - - if read_buf_size == 0 { - read_buf_size = 8192; // Default buffer size + let client = get_http_client(); + let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone()); + if let Some(body) = body { + request = request.body(body); } - let (rd, mut wd) = tokio::io::duplex(read_buf_size); - let (err_tx, err_rx) = oneshot::channel::(); - tokio::spawn(async move { - let client = get_http_client(); - let mut request: RequestBuilder = client.request(method_clone, url_clone).headers(headers_clone); - if let Some(body) = body { - request = request.body(body); - } - let response = request.send().await; - match response { - Ok(resp) => { - if resp.status().is_success() { - let mut stream = resp.bytes_stream(); - while let Some(chunk) = stream.next().await { - match chunk { - Ok(data) => { - if let Err(e) = wd.write_all(&data).await { - let _ = err_tx.send(Error::other(format!("HttpReader write error: {}", e))); - break; - } - } - Err(e) => { - let _ = err_tx.send(Error::other(format!("HttpReader stream error: {}", e))); - break; - } - } - } - } else { - http_log!("[HttpReader::spawn] HTTP request failed with status: {}", resp.status()); - let _ = err_tx.send(Error::other(format!( - "HttpReader HTTP request failed with non-200 status {}", - resp.status() - ))); - } - } - Err(e) => { - let _ = err_tx.send(Error::other(format!("HttpReader HTTP request error: {}", e))); - } - } + let resp = request + .send() + .await + .map_err(|e| Error::other(format!("HttpReader HTTP request error: {}", e)))?; + + if resp.status().is_success().not() { + return Err(Error::other(format!( + "HttpReader HTTP request failed with non-200 status {}", + resp.status() + ))); + } + + let stream = resp + .bytes_stream() + .map_err(|e| Error::other(format!("HttpReader stream error: {}", e))); - http_log!("[HttpReader::spawn] HTTP request completed, exiting"); - }); Ok(Self { - inner: rd, - err_rx, + inner: StreamReader::new(Box::pin(stream)), url, method, headers, @@ -153,14 +125,6 @@ impl AsyncRead for HttpReader { self.method, buf.remaining() ); - // Check for errors from the request - match Pin::new(&mut self.err_rx).try_recv() { - Ok(e) => return Poll::Ready(Err(e)), - Err(oneshot::error::TryRecvError::Empty) => {} - Err(oneshot::error::TryRecvError::Closed) => { - // return Poll::Ready(Err(Error::new(ErrorKind::Other, "HTTP request closed"))); - } - } // Read from the inner stream Pin::new(&mut self.inner).poll_read(cx, buf) }