Updated hyper and http in test_common

tkmcmaster · tkmcmaster · commit 6829949149c8 · 2024-07-18T13:48:19.000-06:00
- Major changes. Many of the utility functions have been moved to the new hyper-util - See https://hyper.rs/guides/1/upgrading/
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/lib/test_common/Cargo.toml b/lib/test_common/Cargo.toml
@@ -10,10 +10,13 @@ doctest = false
 path = "test_common.rs"
 
 [dependencies]
+bytes = "1"
 futures = "0.3"
 futures-timer = "3"
-hyper = { version = "0.14", features = ["server"] }
-http = "0.2"
+hyper = { version = "1.4", features = ["http1"] }
+hyper-util = { version = "0.1", features = ["tokio", "server", "http1"] }
+http = "1.1"
+http-body-util = "0.1"
 parking_lot = "0.12"
 tokio = { version = "1", features = ["full"] }
 url = "2"
diff --git a/lib/test_common/test_common.rs b/lib/test_common/test_common.rs
@@ -1,18 +1,21 @@
 use log::{debug, info};
-use std::{future::Future, io, str::FromStr, sync::Arc, time::Duration};
+use std::{future::Future, io, net::SocketAddr, str::FromStr, sync::Arc, time::Duration};
+use tokio::net::TcpListener;
 
+use bytes::Bytes;
 use futures::{channel::oneshot, future::select, FutureExt};
 use futures_timer::Delay;
 use http::{header, StatusCode};
-use hyper::{
-    server::conn::AddrStream,
-    service::{make_service_fn, service_fn},
-    Body, Error, Request, Response, Server,
+use http_body_util::{combinators::BoxBody, BodyExt, Empty};
+use hyper::{body::Incoming as Body, service::service_fn, Error, Request, Response};
+use hyper_util::{
+    rt::{TokioExecutor, TokioIo},
+    server::conn::auto::Builder as HyperBuilder,
 };
 use parking_lot::Mutex;
 use url::Url;
 
-async fn echo_route(req: Request<Body>) -> Response<Body> {
+async fn echo_route(req: Request<Body>) -> Response<BoxBody<Bytes, hyper::Error>> {
     let headers = req.headers();
     let content_type = headers
         .get(header::CONTENT_TYPE)
@@ -36,24 +39,24 @@ async fn echo_route(req: Request<Body>) -> Response<Body> {
     if echo.is_some() {
         debug!("Echo Body = {}", echo.clone().unwrap_or_default());
     }
-    let mut response = match (req.method(), echo) {
+    let mut response: Response<BoxBody<Bytes, Error>> = match (req.method(), echo) {
         (&http::Method::GET, Some(b)) => Response::builder()
             .status(StatusCode::OK)
             .header(header::CONTENT_TYPE, content_type)
-            .body(b.into())
+            .body(b.map_err(|never| match never {}).boxed())
             .unwrap(),
         (&http::Method::POST, _) | (&http::Method::PUT, _) => Response::builder()
             .status(StatusCode::OK)
             .header(header::CONTENT_TYPE, content_type)
-            .body(req.into_body())
+            .body(req.into_body().boxed())
             .unwrap(),
         _ => Response::builder()
             .status(StatusCode::NO_CONTENT)
-            .body(Body::empty())
+            .body(empty())
             .unwrap(),
     };
     let ms = wait.and_then(|c| FromStr::from_str(&c).ok()).unwrap_or(0);
-    let old_body = std::mem::replace(response.body_mut(), Body::empty());
+    let old_body = std::mem::replace(response.body_mut(), empty());
     if ms > 0 {
         debug!("waiting {} ms", ms);
     }
@@ -62,13 +65,22 @@ async fn echo_route(req: Request<Body>) -> Response<Body> {
     response
 }
 
-pub fn start_test_server(
+fn empty() -> BoxBody<Bytes, hyper::Error> {
+    Empty::<Bytes>::new()
+        .map_err(|never| match never {})
+        .boxed()
+}
+
+pub async fn start_test_server(
     port: Option<u16>,
 ) -> (u16, oneshot::Sender<()>, impl Future<Output = ()>) {
     let port = port.unwrap_or(0);
-    let address = ([127, 0, 0, 1], port).into();
+    let address: SocketAddr = ([127, 0, 0, 1], port).into();
+
+    let listener = TcpListener::bind(address).await.unwrap();
+    let local_addr = listener.local_addr().unwrap();
 
-    let make_svc = make_service_fn(|_: &AddrStream| async {
+    let server = tokio::spawn(async move {
         let service = service_fn(|req: Request<Body>| async {
             debug!("{:?}", req);
             let method = req.method().to_string();
@@ -78,7 +90,7 @@ pub fn start_test_server(
                 "/" => echo_route(req).await,
                 _ => Response::builder()
                     .status(StatusCode::NOT_FOUND)
-                    .body(Body::empty())
+                    .body(empty())
                     .unwrap(),
             };
             debug!("{:?}", response);
@@ -92,14 +104,20 @@ pub fn start_test_server(
             );
             Ok::<_, Error>(response)
         });
-        Ok::<_, Error>(service)
+
+        loop {
+            let (stream, _) = listener.accept().await.unwrap();
+            let stream = TokioIo::new(stream);
+            tokio::task::spawn(async move {
+                let builder = HyperBuilder::new(TokioExecutor::new());
+                builder.serve_connection(stream, service).await.unwrap();
+            });
+        }
     });
 
     let (tx, rx) = oneshot::channel();
 
-    let server = Server::bind(&address).serve(make_svc);
-
-    let port = server.local_addr().port();
+    let port = local_addr.port();
 
     let future = select(server, rx);
 
diff --git a/src/bin/test_server.rs b/src/bin/test_server.rs
@@ -8,7 +8,7 @@ fn main() {
     rt.block_on(async {
         let port = std::env::var("PORT").ok().and_then(|s| s.parse().ok());
         debug!("port = {}", port.unwrap_or_default());
-        let (port, rx, handle) = start_test_server(port);
+        let (port, rx, handle) = start_test_server(port).await;
 
         println!("Listening on port {port}");
 
diff --git a/src/request/request_maker.rs b/src/request/request_maker.rs
@@ -401,7 +401,7 @@ mod tests {
     fn sends_request() {
         let rt = Runtime::new().unwrap();
         rt.block_on(async move {
-            let (port, ..) = test_common::start_test_server(None);
+            let (port, ..) = test_common::start_test_server(None).await;
             let url = Template::simple(&format!("https://127.0.0.1:{}", port));
             let method = Method::GET;
             let headers = Vec::new();
diff --git a/tests/integration.rs b/tests/integration.rs
@@ -7,7 +7,7 @@ use tokio::runtime::Runtime;
 fn run_test(path: &str) -> (bool, String, String) {
     let rt = Runtime::new().unwrap();
     rt.block_on(async move {
-        let (port, kill_server, _) = start_test_server(None);
+        let (port, kill_server, _) = start_test_server(None).await;
         env::set_var("PORT", port.to_string());
 
         let (_, ctrlc_channel) = futures::channel::mpsc::unbounded();
