(opendal) 01/01: feat(core): Use Buffer as http_body::Body directly

[xu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xuanwo pushed a commit to branch optimize-http
in repository https://gitbox.apache.org/repos/asf/opendal.git

commit 47990b55c8ce883a779f5cf9d2546ffa9c58d08e
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Apr 15 18:40:14 2025 +0800

    feat(core): Use Buffer as http_body::Body directly
    
    Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
 core/Cargo.lock                  |  1 +
 core/Cargo.toml                  |  5 ++---
 core/src/raw/http_util/client.rs | 34 +++++++++++++++++++++++++++++++++-
 3 files changed, 36 insertions(+), 4 deletions(-)

diff --git a/core/Cargo.lock b/core/Cargo.lock
index 6623bcd50..7a90449cf 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -5441,6 +5441,7 @@ dependencies = [
  "hdrs",
  "hmac",
  "http 1.3.1",
+ "http-body 1.0.1",
  "js-sys",
  "libtest-mimic",
  "log",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index ce7d322e6..45141d418 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -252,13 +252,12 @@ futures = { version = "0.3", default-features = false, features = [
 ] }
 ghac = { version = "0.2.0", optional = true }
 http = "1.1"
+http-body = "1"
 log = "0.4"
 md-5 = "0.10"
 percent-encoding = "2"
 quick-xml = { version = "0.37", features = ["serialize", "overlapped-lists"] }
-reqwest = { version = "0.12.2", features = [
-  "stream",
-], default-features = false }
+reqwest = { version = "0.12.8", default-features = false }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1.27", features = ["sync", "io-util"] }
diff --git a/core/src/raw/http_util/client.rs b/core/src/raw/http_util/client.rs
index 60038b8f2..ada330ca9 100644
--- a/core/src/raw/http_util/client.rs
+++ b/core/src/raw/http_util/client.rs
@@ -15,18 +15,25 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::convert::Infallible;
 use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::future;
 use std::mem;
 use std::ops::Deref;
+use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::Arc;
+use std::task::Context;
+use std::task::Poll;
 
+use bytes::Bytes;
 use futures::Future;
 use futures::TryStreamExt;
 use http::Request;
 use http::Response;
+use http_body::Frame;
+use http_body::SizeHint;
 use raw::oio::Read;
 use std::sync::LazyLock;
 
@@ -171,7 +178,7 @@ impl HttpFetch for reqwest::Client {
         if !body.is_empty() {
             #[cfg(not(target_arch = "wasm32"))]
             {
-                req_builder = req_builder.body(reqwest::Body::wrap_stream(body))
+                req_builder = req_builder.body(reqwest::Body::wrap(HttpBufferBody(body)))
             }
             #[cfg(target_arch = "wasm32")]
             {
@@ -240,3 +247,28 @@ fn is_temporary_error(err: &reqwest::Error) -> bool {
     // error decoding response body, for example, connection reset.
     err.is_decode()
 }
+
+struct HttpBufferBody(Buffer);
+
+impl http_body::Body for HttpBufferBody {
+    type Data = Bytes;
+    type Error = Infallible;
+
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        match self.0.next() {
+            Some(bs) => Poll::Ready(Some(Ok(Frame::data(bs)))),
+            None => Poll::Ready(None),
+        }
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.0.is_empty()
+    }
+
+    fn size_hint(&self) -> SizeHint {
+        SizeHint::with_exact(self.0.len() as u64)
+    }
+}