support graceful shutdown for http listener

NOTE that for auto http server this is not yet actually
shutting down the connection gracefully

due to
lack of support with hyperium/hyper-util#66

examples/tokio_tcp_http_hello.rs

@@ -44,8 +44,8 @@ async fn main() {
 
         let mut http_server = http::HttpServer::auto();
 
-        http_server.http1().preserve_header_case(true);
-        http_server.h2().adaptive_window(true);
+        http_server.http1_config().preserve_header_case(true);
+        http_server.h2_config().adaptive_window(true);
 
         let web_server = http_server
             .compression()

examples/tokio_tls_https_hello.rs

@@ -56,7 +56,7 @@ async fn main() {
 
     // create tls proxy
     shutdown.spawn_task_fn(|guard| async move {
-        let web_server = http::HttpServer::auto()
+        let web_server = http::HttpServer::http1()
             .compression()
             .trace()
             .timeout(Duration::from_secs(10))

src/http/server/hyper_conn.rs

@@ -4,7 +4,8 @@ use hyper::server::conn::http1::Builder as Http1Builder;
 use hyper::server::conn::http2::Builder as Http2Builder;
 use hyper_util::server::conn::auto::Builder as AutoBuilder;
 
-use crate::tcp::TcpStream;
+use crate::rt::{graceful::ShutdownGuard, pin, select};
+use crate::{state::Extendable, tcp::TcpStream};
 
 use super::{GlobalExecutor, HyperIo, Response, ServeResult};
 
@@ -53,11 +54,32 @@ impl HyperConnServer for Http1Builder {
         Body::Error: Into<Box<dyn StdError + Send + Sync>>,
     {
         let io = Box::pin(io);
+        let guard = io.extensions().get::<ShutdownGuard>().cloned();
+
         let stream = HyperIo::new(io);
-        self.serve_connection(stream, service)
-            .with_upgrades()
-            .await?;
-        Ok(())
+
+        let conn = self.serve_connection(stream, service).with_upgrades();
+
+        if let Some(guard) = guard {
+            pin!(conn);
+
+            loop {
+                select! {
+                    _ = guard.cancelled() => {
+                        tracing::trace!("signal received: initiate graceful shutdown");
+                        conn.as_mut().graceful_shutdown();
+                    }
+                    result = conn.as_mut() => {
+                        tracing::trace!("connection finished");
+                        result?;
+                        return Ok(());
+                    }
+                }
+            }
+        } else {
+            conn.await?;
+            Ok(())
+        }
     }
 }
 
@@ -83,9 +105,32 @@ impl HyperConnServer for Http2Builder<GlobalExecutor> {
         Body::Error: Into<Box<dyn StdError + Send + Sync>>,
     {
         let io = Box::pin(io);
+        let guard = io.extensions().get::<ShutdownGuard>().cloned();
+
         let stream = HyperIo::new(io);
-        self.serve_connection(stream, service).await?;
-        Ok(())
+
+        let conn = self.serve_connection(stream, service);
+
+        if let Some(guard) = guard {
+            pin!(conn);
+
+            loop {
+                select! {
+                    _ = guard.cancelled() => {
+                        tracing::trace!("signal received: initiate graceful shutdown");
+                        conn.as_mut().graceful_shutdown();
+                    }
+                    result = conn.as_mut() => {
+                        tracing::trace!("connection finished");
+                        result?;
+                        return Ok(());
+                    }
+                }
+            }
+        } else {
+            conn.await?;
+            Ok(())
+        }
     }
 }
 
@@ -111,8 +156,33 @@ impl HyperConnServer for AutoBuilder<GlobalExecutor> {
         Body::Error: Into<Box<dyn StdError + Send + Sync>>,
     {
         let io = Box::pin(io);
+        let guard = io.extensions().get::<ShutdownGuard>().cloned();
+
         let stream = HyperIo::new(io);
-        self.serve_connection_with_upgrades(stream, service).await?;
-        Ok(())
+
+        let conn = self.serve_connection_with_upgrades(stream, service);
+
+        if let Some(guard) = guard {
+            pin!(conn);
+
+            loop {
+                select! {
+                    _ = guard.cancelled() => {
+                        tracing::trace!("signal received: nop: graceful shutdown not supported for auto builder");
+                        // TODO: support once it is implemented:
+                        // https://github.com/hyperium/hyper-util/pull/66
+                        // conn.as_mut().graceful_shutdown();
+                    }
+                    result = conn.as_mut() => {
+                        tracing::trace!("connection finished");
+                        result?;
+                        return Ok(());
+                    }
+                }
+            }
+        } else {
+            conn.await?;
+            Ok(())
+        }
     }
 }

src/http/server/service.rs

@@ -286,14 +286,14 @@ impl HttpServer<AutoBuilder<GlobalExecutor>, Identity> {
 
 impl<E, L> HttpServer<AutoBuilder<E>, L> {
     /// Http1 configuration.
-    pub fn http1(&mut self) -> AutoHttp1Builder<'_, E> {
+    pub fn http1_config(&mut self) -> AutoHttp1Builder<'_, E> {
         AutoHttp1Builder {
             inner: self.builder.http1(),
         }
     }
 
     /// H2 configuration.
-    pub fn h2(&mut self) -> AutoH2Builder<'_, E> {
+    pub fn h2_config(&mut self) -> AutoH2Builder<'_, E> {
         AutoH2Builder {
             inner: self.builder.http2(),
         }
@@ -1241,8 +1241,6 @@ impl<B, S> Clone for HttpService<B, S> {
     }
 }
 
-// TODO: support graceful service...
-
 impl<B, T, S, Body> Service<TcpStream<T>> for HttpService<B, S>
 where
     B: HyperConnServer,