refactor: finish test_util server cleanup, simplify dependencies (#21714)

Closes https://github.com/denoland/deno/issues/21578

1 files changed, 146 insertions, 0 deletions

diff --git a/test_util/src/servers/hyper_utils.rs b/test_util/src/servers/hyper_utils.rs
new file mode 100644
index 000000000..b8e9472a4
--- /dev/null
+++ b/test_util/src/servers/hyper_utils.rs
@@ -0,0 +1,146 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+use bytes::Bytes;
+use futures::Future;
+use futures::FutureExt;
+use futures::Stream;
+use futures::StreamExt;
+use http;
+use http::Request;
+use http::Response;
+use http_body_util::combinators::UnsyncBoxBody;
+use hyper1 as hyper;
+use hyper_util::rt::TokioIo;
+use std::convert::Infallible;
+use std::io;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::result::Result;
+use tokio::net::TcpListener;
+
+#[derive(Debug, Clone, Copy)]
+pub enum ServerKind {
+  Auto,
+  OnlyHttp1,
+  OnlyHttp2,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct ServerOptions {
+  pub error_msg: &'static str,
+  pub addr: SocketAddr,
+  pub kind: ServerKind,
+}
+
+type HandlerOutput =
+  Result<Response<UnsyncBoxBody<Bytes, Infallible>>, anyhow::Error>;
+
+pub async fn run_server<F, S>(options: ServerOptions, handler: F)
+where
+  F: Fn(Request<hyper::body::Incoming>) -> S + Copy + 'static,
+  S: Future<Output = HandlerOutput> + 'static,
+{
+  let fut: Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>> =
+    async move {
+      let listener = TcpListener::bind(options.addr).await?;
+      loop {
+        let (stream, _) = listener.accept().await?;
+        let io = TokioIo::new(stream);
+        deno_unsync::spawn(hyper1_serve_connection(
+          io,
+          handler,
+          options.error_msg,
+          options.kind,
+        ));
+      }
+    }
+    .boxed_local();
+
+  if let Err(e) = fut.await {
+    eprintln!("{}: {:?}", options.error_msg, e);
+  }
+}
+
+pub async fn run_server_with_acceptor<'a, A, F, S>(
+  mut acceptor: Pin<Box<A>>,
+  handler: F,
+  error_msg: &'static str,
+  kind: ServerKind,
+) where
+  A: Stream<Item = io::Result<rustls_tokio_stream::TlsStream>> + ?Sized,
+  F: Fn(Request<hyper::body::Incoming>) -> S + Copy + 'static,
+  S: Future<Output = HandlerOutput> + 'static,
+{
+  let fut: Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>> =
+    async move {
+      while let Some(result) = acceptor.next().await {
+        let stream = result?;
+        let io = TokioIo::new(stream);
+        deno_unsync::spawn(hyper1_serve_connection(
+          io, handler, error_msg, kind,
+        ));
+      }
+      Ok(())
+    }
+    .boxed_local();
+
+  if let Err(e) = fut.await {
+    eprintln!("{}: {:?}", error_msg, e);
+  }
+}
+
+async fn hyper1_serve_connection<I, F, S>(
+  io: I,
+  handler: F,
+  error_msg: &'static str,
+  kind: ServerKind,
+) where
+  I: hyper::rt::Read + hyper::rt::Write + Unpin + 'static,
+  F: Fn(Request<hyper::body::Incoming>) -> S + Copy + 'static,
+  S: Future<Output = HandlerOutput> + 'static,
+{
+  let service = hyper1::service::service_fn(handler);
+
+  let result: Result<(), anyhow::Error> = match kind {
+    ServerKind::Auto => {
+      let builder =
+        hyper_util::server::conn::auto::Builder::new(DenoUnsyncExecutor);
+      builder
+        .serve_connection(io, service)
+        .await
+        .map_err(|e| anyhow::anyhow!("{}", e))
+    }
+    ServerKind::OnlyHttp1 => {
+      let builder = hyper1::server::conn::http1::Builder::new();
+      builder
+        .serve_connection(io, service)
+        .await
+        .map_err(|e| e.into())
+    }
+    ServerKind::OnlyHttp2 => {
+      let builder =
+        hyper1::server::conn::http2::Builder::new(DenoUnsyncExecutor);
+      builder
+        .serve_connection(io, service)
+        .await
+        .map_err(|e| e.into())
+    }
+  };
+
+  if let Err(e) = result {
+    eprintln!("{}: {:?}", error_msg, e);
+  }
+}
+
+#[derive(Clone)]
+struct DenoUnsyncExecutor;
+
+impl<Fut> hyper::rt::Executor<Fut> for DenoUnsyncExecutor
+where
+  Fut: Future + 'static,
+  Fut::Output: 'static,
+{
+  fn execute(&self, fut: Fut) {
+    deno_unsync::spawn(fut);
+  }
+}