feat: allow fetch to be initialized with user-provided client

modules/llrt_http/src/fetch.rs

@@ -1,10 +1,14 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
+use std::convert::Infallible;
 use std::{collections::HashSet, time::Instant};
 
 use bytes::Bytes;
+use http_body_util::combinators::BoxBody;
 use http_body_util::Full;
 use hyper::{header::HeaderName, Method, Request, Uri};
+use hyper_util::client::legacy::connect::Connect;
+use hyper_util::client::legacy::Client;
 use llrt_abort::AbortSignal;
 use llrt_utils::{
     bytes::ObjectBytes, encoding::bytes_from_b64, mc_oneshot, result::ResultExt, VERSION,
@@ -17,19 +21,18 @@ use rquickjs::{
 };
 use tokio::select;
 
-use super::{
-    blob::Blob, headers::Headers, response::Response, security::ensure_url_access, HTTP_CLIENT,
-};
+use super::{blob::Blob, headers::Headers, response::Response, security::ensure_url_access};
 
 const MAX_REDIRECT_COUNT: u32 = 20;
 
-pub(crate) fn init(ctx: &Ctx<'_>, globals: &Object) -> Result<()> {
-    //init eagerly
-    let client = HTTP_CLIENT.as_ref().or_throw(ctx)?;
-
+pub(crate) fn init<C>(client: Client<C, BoxBody<Bytes, Infallible>>, globals: &Object) -> Result<()>
+where
+    C: Clone + Send + Sync + Connect + 'static,
+{
     globals.set(
         "fetch",
         Func::from(Async(move |ctx, resource, args| {
+            let client = client.clone();
             let start = Instant::now();
             let options = get_fetch_options(&ctx, resource, args);
 
@@ -162,7 +165,7 @@ fn build_request(
     body: Option<&BodyBytes>,
     prev_status: &u16,
     initial_uri: &Uri,
-) -> Result<Request<Full<Bytes>>> {
+) -> Result<Request<BoxBody<Bytes, Infallible>>> {
     let same_origin = is_same_origin(uri, initial_uri);
 
     let change_method = should_change_method(*prev_status, method);
@@ -200,8 +203,10 @@ fn build_request(
         req = req.header("accept", "*/*");
     }
 
-    req.body(body.map(|b| b.body.clone()).unwrap_or_default())
-        .or_throw(ctx)
+    req.body(BoxBody::new(
+        body.map(|b| b.body.clone()).unwrap_or_default(),
+    ))
+    .or_throw(ctx)
 }
 
 fn is_same_origin(uri: &Uri, initial_uri: &Uri) -> bool {

modules/llrt_http/src/lib.rs

@@ -1,15 +1,17 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
+use std::convert::Infallible;
 use std::{io, sync::OnceLock, time::Duration};
 
 use bytes::Bytes;
-use http_body_util::Full;
+use http_body_util::combinators::BoxBody;
 use hyper_rustls::HttpsConnector;
 use hyper_util::{
     client::legacy::{connect::HttpConnector, Client},
     rt::{TokioExecutor, TokioTimer},
 };
 use llrt_utils::class::CustomInspectExtension;
+use llrt_utils::result::ResultExt;
 use once_cell::sync::Lazy;
 use rquickjs::{Class, Ctx, Result};
 use rustls::{
@@ -117,36 +119,39 @@ fn get_http_version() -> HttpVersion {
     })
 }
 
-pub static HTTP_CLIENT: Lazy<io::Result<Client<HttpsConnector<HttpConnector>, Full<Bytes>>>> =
-    Lazy::new(|| {
-        let pool_idle_timeout = get_pool_idle_timeout();
+type HttpsClient = Client<HttpsConnector<HttpConnector>, BoxBody<Bytes, Infallible>>;
+pub static HTTP_CLIENT: Lazy<
+    io::Result<HttpsClient>,
+> = Lazy::new(|| {
+    let pool_idle_timeout = get_pool_idle_timeout();
 
-        let maybe_tls_config = match &*TLS_CONFIG {
-            Ok(tls_config) => io::Result::Ok(tls_config.clone()),
-            Err(e) => io::Result::Err(io::Error::new(e.kind(), e.to_string())),
-        };
+    let maybe_tls_config = match &*TLS_CONFIG {
+        Ok(tls_config) => io::Result::Ok(tls_config.clone()),
+        Err(e) => io::Result::Err(io::Error::new(e.kind(), e.to_string())),
+    };
 
-        let builder = hyper_rustls::HttpsConnectorBuilder::new()
-            .with_tls_config(maybe_tls_config?)
-            .https_or_http();
+    let builder = hyper_rustls::HttpsConnectorBuilder::new()
+        .with_tls_config(maybe_tls_config?)
+        .https_or_http();
 
-        let https = match get_http_version() {
-            #[cfg(feature = "http1")]
-            HttpVersion::Http1_1 => builder.enable_http1().build(),
-            #[cfg(feature = "http2")]
-            HttpVersion::Http2 => builder.enable_all_versions().build(),
-        };
+    let https = match get_http_version() {
+        #[cfg(feature = "http1")]
+        HttpVersion::Http1_1 => builder.enable_http1().build(),
+        #[cfg(feature = "http2")]
+        HttpVersion::Http2 => builder.enable_all_versions().build(),
+    };
 
-        Ok(Client::builder(TokioExecutor::new())
-            .pool_idle_timeout(pool_idle_timeout)
-            .pool_timer(TokioTimer::new())
-            .build(https))
-    });
+    Ok(Client::builder(TokioExecutor::new())
+        .pool_idle_timeout(pool_idle_timeout)
+        .pool_timer(TokioTimer::new())
+        .build(https))
+});
 
 pub fn init(ctx: &Ctx) -> Result<()> {
     let globals = ctx.globals();
 
-    fetch::init(ctx, &globals)?;
+    //init eagerly
+    fetch::init(HTTP_CLIENT.as_ref().or_throw(ctx)?.clone(), &globals)?;
 
     Class::<Request>::define(&globals)?;
     Class::<Response>::define(&globals)?;