2 files changed, 88 insertions, 20 deletions

diff --git a/ext/http/http_next.rs b/ext/http/http_next.rs
index 1251f00cc..7dbac6021 100644
--- a/ext/http/http_next.rs
+++ b/ext/http/http_next.rs
@@ -18,6 +18,7 @@ use crate::service::HttpServerState;
 use crate::service::SignallingRc;
 use crate::websocket_upgrade::WebSocketUpgrade;
 use crate::LocalExecutor;
+use crate::Options;
 use cache_control::CacheControl;
 use deno_core::external;
 use deno_core::futures::future::poll_fn;
@@ -821,10 +822,16 @@ fn serve_http11_unconditional(
   io: impl HttpServeStream,
   svc: impl HttpService<Incoming, ResBody = HttpRecordResponse> + 'static,
   cancel: Rc<CancelHandle>,
+  http1_builder_hook: Option<fn(http1::Builder) -> http1::Builder>,
 ) -> impl Future<Output = Result<(), hyper::Error>> + 'static {
-  let conn = http1::Builder::new()
-    .keep_alive(true)
-    .writev(*USE_WRITEV)
+  let mut builder = http1::Builder::new();
+  builder.keep_alive(true).writev(*USE_WRITEV);
+
+  if let Some(http1_builder_hook) = http1_builder_hook {
+    builder = http1_builder_hook(builder);
+  }
+
+  let conn = builder
     .serve_connection(TokioIo::new(io), svc)
     .with_upgrades();
 
@@ -843,9 +850,17 @@ fn serve_http2_unconditional(
   io: impl HttpServeStream,
   svc: impl HttpService<Incoming, ResBody = HttpRecordResponse> + 'static,
   cancel: Rc<CancelHandle>,
+  http2_builder_hook: Option<
+    fn(http2::Builder<LocalExecutor>) -> http2::Builder<LocalExecutor>,
+  >,
 ) -> impl Future<Output = Result<(), hyper::Error>> + 'static {
-  let conn =
-    http2::Builder::new(LocalExecutor).serve_connection(TokioIo::new(io), svc);
+  let mut builder = http2::Builder::new(LocalExecutor);
+
+  if let Some(http2_builder_hook) = http2_builder_hook {
+    builder = http2_builder_hook(builder);
+  }
+
+  let conn = builder.serve_connection(TokioIo::new(io), svc);
   async {
     match conn.or_abort(cancel).await {
       Err(mut conn) => {
@@ -861,15 +876,16 @@ async fn serve_http2_autodetect(
   io: impl HttpServeStream,
   svc: impl HttpService<Incoming, ResBody = HttpRecordResponse> + 'static,
   cancel: Rc<CancelHandle>,
+  options: Options,
 ) -> Result<(), HttpNextError> {
   let prefix = NetworkStreamPrefixCheck::new(io, HTTP2_PREFIX);
   let (matches, io) = prefix.match_prefix().await?;
   if matches {
-    serve_http2_unconditional(io, svc, cancel)
+    serve_http2_unconditional(io, svc, cancel, options.http2_builder_hook)
       .await
       .map_err(HttpNextError::Hyper)
   } else {
-    serve_http11_unconditional(io, svc, cancel)
+    serve_http11_unconditional(io, svc, cancel, options.http1_builder_hook)
       .await
       .map_err(HttpNextError::Hyper)
   }
@@ -880,6 +896,7 @@ fn serve_https(
   request_info: HttpConnectionProperties,
   lifetime: HttpLifetime,
   tx: tokio::sync::mpsc::Sender<Rc<HttpRecord>>,
+  options: Options,
 ) -> JoinHandle<Result<(), HttpNextError>> {
   let HttpLifetime {
     server_state,
@@ -891,21 +908,31 @@ fn serve_https(
     handle_request(req, request_info.clone(), server_state.clone(), tx.clone())
   });
   spawn(
-    async {
+    async move {
       let handshake = io.handshake().await?;
       // If the client specifically negotiates a protocol, we will use it. If not, we'll auto-detect
       // based on the prefix bytes
       let handshake = handshake.alpn;
       if Some(TLS_ALPN_HTTP_2) == handshake.as_deref() {
-        serve_http2_unconditional(io, svc, listen_cancel_handle)
-          .await
-          .map_err(HttpNextError::Hyper)
+        serve_http2_unconditional(
+          io,
+          svc,
+          listen_cancel_handle,
+          options.http2_builder_hook,
+        )
+        .await
+        .map_err(HttpNextError::Hyper)
       } else if Some(TLS_ALPN_HTTP_11) == handshake.as_deref() {
-        serve_http11_unconditional(io, svc, listen_cancel_handle)
-          .await
-          .map_err(HttpNextError::Hyper)
+        serve_http11_unconditional(
+          io,
+          svc,
+          listen_cancel_handle,
+          options.http1_builder_hook,
+        )
+        .await
+        .map_err(HttpNextError::Hyper)
       } else {
-        serve_http2_autodetect(io, svc, listen_cancel_handle).await
+        serve_http2_autodetect(io, svc, listen_cancel_handle, options).await
       }
     }
     .try_or_cancel(connection_cancel_handle),
@@ -917,6 +944,7 @@ fn serve_http(
   request_info: HttpConnectionProperties,
   lifetime: HttpLifetime,
   tx: tokio::sync::mpsc::Sender<Rc<HttpRecord>>,
+  options: Options,
 ) -> JoinHandle<Result<(), HttpNextError>> {
   let HttpLifetime {
     server_state,
@@ -928,7 +956,7 @@ fn serve_http(
     handle_request(req, request_info.clone(), server_state.clone(), tx.clone())
   });
   spawn(
-    serve_http2_autodetect(io, svc, listen_cancel_handle)
+    serve_http2_autodetect(io, svc, listen_cancel_handle, options)
       .try_or_cancel(connection_cancel_handle),
   )
 }
@@ -938,6 +966,7 @@ fn serve_http_on<HTTP>(
   listen_properties: &HttpListenProperties,
   lifetime: HttpLifetime,
   tx: tokio::sync::mpsc::Sender<Rc<HttpRecord>>,
+  options: Options,
 ) -> JoinHandle<Result<(), HttpNextError>>
 where
   HTTP: HttpPropertyExtractor,
@@ -949,14 +978,14 @@ where
 
   match network_stream {
     NetworkStream::Tcp(conn) => {
-      serve_http(conn, connection_properties, lifetime, tx)
+      serve_http(conn, connection_properties, lifetime, tx, options)
     }
     NetworkStream::Tls(conn) => {
-      serve_https(conn, connection_properties, lifetime, tx)
+      serve_https(conn, connection_properties, lifetime, tx, options)
     }
     #[cfg(unix)]
     NetworkStream::Unix(conn) => {
-      serve_http(conn, connection_properties, lifetime, tx)
+      serve_http(conn, connection_properties, lifetime, tx, options)
     }
   }
 }
@@ -1045,6 +1074,11 @@ where
 
   let lifetime = resource.lifetime();
 
+  let options = {
+    let state = state.borrow();
+    *state.borrow::<Options>()
+  };
+
   let listen_properties_clone: HttpListenProperties = listen_properties.clone();
   let handle = spawn(async move {
     loop {
@@ -1057,6 +1091,7 @@ where
         &listen_properties_clone,
         lifetime.clone(),
         tx.clone(),
+        options,
       );
     }
     #[allow(unreachable_code)]
@@ -1093,11 +1128,17 @@ where
   let (tx, rx) = tokio::sync::mpsc::channel(10);
   let resource: Rc<HttpJoinHandle> = Rc::new(HttpJoinHandle::new(rx));
 
+  let options = {
+    let state = state.borrow();
+    *state.borrow::<Options>()
+  };
+
   let handle = serve_http_on::<HTTP>(
     connection,
     &listen_properties,
     resource.lifetime(),
     tx,
+    options,
   );
 
   // Set the handle after we start the future
diff --git a/ext/http/lib.rs b/ext/http/lib.rs
index 49893b1b9..39b0bbc2a 100644
--- a/ext/http/lib.rs
+++ b/ext/http/lib.rs
@@ -39,6 +39,8 @@ use deno_net::raw::NetworkStream;
 use deno_websocket::ws_create_server_stream;
 use flate2::write::GzEncoder;
 use flate2::Compression;
+use hyper::server::conn::http1;
+use hyper::server::conn::http2;
 use hyper_util::rt::TokioIo;
 use hyper_v014::body::Bytes;
 use hyper_v014::body::HttpBody;
@@ -96,6 +98,25 @@ pub use request_properties::HttpRequestProperties;
 pub use service::UpgradeUnavailableError;
 pub use websocket_upgrade::WebSocketUpgradeError;
 
+#[derive(Debug, Default, Clone, Copy)]
+pub struct Options {
+  /// By passing a hook function, the caller can customize various configuration
+  /// options for the HTTP/2 server.
+  /// See [`http2::Builder`] for what parameters can be customized.
+  ///
+  /// If `None`, the default configuration provided by hyper will be used. Note
+  /// that the default configuration is subject to change in future versions.
+  pub http2_builder_hook:
+    Option<fn(http2::Builder<LocalExecutor>) -> http2::Builder<LocalExecutor>>,
+  /// By passing a hook function, the caller can customize various configuration
+  /// options for the HTTP/1 server.
+  /// See [`http1::Builder`] for what parameters can be customized.
+  ///
+  /// If `None`, the default configuration provided by hyper will be used. Note
+  /// that the default configuration is subject to change in future versions.
+  pub http1_builder_hook: Option<fn(http1::Builder) -> http1::Builder>,
+}
+
 deno_core::extension!(
   deno_http,
   deps = [deno_web, deno_net, deno_fetch, deno_websocket],
@@ -135,6 +156,12 @@ deno_core::extension!(
     http_next::op_http_cancel,
   ],
   esm = ["00_serve.ts", "01_http.js", "02_websocket.ts"],
+  options = {
+    options: Options,
+  },
+  state = |state, options| {
+    state.put::<Options>(options.options);
+  }
 );
 
 #[derive(Debug, thiserror::Error)]
@@ -1117,7 +1144,7 @@ async fn op_http_upgrade_websocket(
 
 // Needed so hyper can use non Send futures
 #[derive(Clone)]
-struct LocalExecutor;
+pub struct LocalExecutor;
 
 impl<Fut> hyper_v014::rt::Executor<Fut> for LocalExecutor
 where