refactor(fetch): reimplement fetch with hyper instead of reqwest (#24237)

This commit re-implements `ext/fetch` and all dependent crates
using `hyper` and `hyper-util`, instead of `reqwest`.

The reasoning is that we want to have greater control and access
to low level `hyper` APIs when implementing `fetch` API as well
as `node:http` module.

---------

Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

1 files changed, 23 insertions, 16 deletions

diff --git a/ext/kv/remote.rs b/ext/kv/remote.rs
index 7541b5a06..922853588 100644
--- a/ext/kv/remote.rs
+++ b/ext/kv/remote.rs
@@ -12,10 +12,8 @@ use bytes::Bytes;
 use deno_core::error::type_error;
 use deno_core::error::AnyError;
 use deno_core::futures::Stream;
-use deno_core::futures::TryStreamExt as _;
 use deno_core::OpState;
 use deno_fetch::create_http_client;
-use deno_fetch::reqwest;
 use deno_fetch::CreateHttpClientOptions;
 use deno_tls::rustls::RootCertStore;
 use deno_tls::Proxy;
@@ -25,6 +23,7 @@ use denokv_remote::MetadataEndpoint;
 use denokv_remote::Remote;
 use denokv_remote::RemoteResponse;
 use denokv_remote::RemoteTransport;
+use http_body_util::BodyExt;
 use url::Url;
 
 #[derive(Clone)]
@@ -109,35 +108,43 @@ impl<P: RemoteDbHandlerPermissions + 'static> denokv_remote::RemotePermissions
 }
 
 #[derive(Clone)]
-pub struct ReqwestClient(reqwest::Client);
-pub struct ReqwestResponse(reqwest::Response);
+pub struct FetchClient(deno_fetch::Client);
+pub struct FetchResponse(http::Response<deno_fetch::ResBody>);
 
-impl RemoteTransport for ReqwestClient {
-  type Response = ReqwestResponse;
+impl RemoteTransport for FetchClient {
+  type Response = FetchResponse;
   async fn post(
     &self,
     url: Url,
     headers: http::HeaderMap,
     body: Bytes,
   ) -> Result<(Url, http::StatusCode, Self::Response), anyhow::Error> {
-    let res = self.0.post(url).headers(headers).body(body).send().await?;
-    let url = res.url().clone();
+    let body = http_body_util::Full::new(body)
+      .map_err(|never| match never {})
+      .boxed();
+    let mut req = http::Request::new(body);
+    *req.method_mut() = http::Method::POST;
+    *req.uri_mut() = url.as_str().parse()?;
+    *req.headers_mut() = headers;
+
+    let res = self.0.clone().send(req).await?;
     let status = res.status();
-    Ok((url, status, ReqwestResponse(res)))
+    Ok((url, status, FetchResponse(res)))
   }
 }
 
-impl RemoteResponse for ReqwestResponse {
+impl RemoteResponse for FetchResponse {
   async fn bytes(self) -> Result<Bytes, anyhow::Error> {
-    Ok(self.0.bytes().await?)
+    Ok(self.0.collect().await?.to_bytes())
   }
   fn stream(
     self,
   ) -> impl Stream<Item = Result<Bytes, anyhow::Error>> + Send + Sync {
-    self.0.bytes_stream().map_err(|e| e.into())
+    self.0.into_body().into_data_stream()
   }
   async fn text(self) -> Result<String, anyhow::Error> {
-    Ok(self.0.text().await?)
+    let bytes = self.bytes().await?;
+    Ok(std::str::from_utf8(&bytes)?.into())
   }
 }
 
@@ -145,7 +152,7 @@ impl RemoteResponse for ReqwestResponse {
 impl<P: RemoteDbHandlerPermissions + 'static> DatabaseHandler
   for RemoteDbHandler<P>
 {
-  type DB = Remote<PermissionChecker<P>, ReqwestClient>;
+  type DB = Remote<PermissionChecker<P>, FetchClient>;
 
   async fn open(
     &self,
@@ -201,14 +208,14 @@ impl<P: RemoteDbHandlerPermissions + 'static> DatabaseHandler
         http2: true,
       },
     )?;
-    let reqwest_client = ReqwestClient(client);
+    let fetch_client = FetchClient(client);
 
     let permissions = PermissionChecker {
       state: state.clone(),
       _permissions: PhantomData,
     };
 
-    let remote = Remote::new(reqwest_client, permissions, metadata_endpoint);
+    let remote = Remote::new(fetch_client, permissions, metadata_endpoint);
 
     Ok(remote)
   }