服务端 Builder 与 Connection

loop {
  let (stream, _) = listener.accept().await?;
  tokio::spawn(async move {
    let io = TokioIo::new(stream);
    if let Err(e) = http1::Builder::new().serve_connection(io, service_fn(handler)).await {
      eprintln!("connection error: {e}");
    }
  });
}

    pub fn serve_connection<I, S>(&self, io: I, service: S) -> Connection<I, S>
    where
        S: HttpService<IncomingBody>,
        S::Error: Into<Box<dyn StdError + Send + Sync>>,
        S::ResBody: 'static,
        <S::ResBody as Body>::Error: Into<Box<dyn StdError + Send + Sync>>,
        I: Read + Write + Unpin,
    {
        let mut conn = proto::Conn::new(io);
        conn.set_h1_parser_config(self.h1_parser_config.clone());
        conn.set_timer(self.timer.clone());
        if !self.h1_keep_alive {
            conn.disable_keep_alive();
        }
        // ...
        if let Some(dur) = self
            .timer
            .check(self.h1_header_read_timeout, "header_read_timeout")
        {
            conn.set_http1_header_read_timeout(dur);
        }
        // ...
        if !self.date_header {
            conn.disable_date_header();
        }
        let sd = proto::h1::dispatch::Server::new(service);
        let proto = proto::h1::Dispatcher::new(sd, conn);
        Connection { conn: proto }
    }

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match ready!(Pin::new(&mut self.conn).poll(cx)) {
            Ok(done) => {
                match done {
                    proto::Dispatched::Shutdown => {}
                    proto::Dispatched::Upgrade(pending) => {
                        // With no `Send` bound on `I`, we can't try to do
                        // upgrades here. In case a user was trying to use
                        // `Body::on_upgrade` with this API, send a special
                        // error letting them know about that.
                        pending.manual();
                    }
                }
                Poll::Ready(Ok(()))
            }
            Err(e) => Poll::Ready(Err(e)),
        }
    }

hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
http-body-util = "0.1"
bytes = "1"
tokio = { version = "=1.54.1", features = ["full"] }

use std::convert::Infallible;

use bytes::Bytes;
use http_body_util::Full;
use hyper::{body::Incoming, server::conn::http1, service::service_fn, Request, Response};
use hyper_util::rt::TokioIo;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

async fn handler(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  println!("  {} handled in task {}", req.uri().path(), tokio::task::id());
  Ok(Response::new(Full::new(Bytes::from_static(b"ok"))))
}

async fn pair() -> (TcpStream, TcpStream) {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let client = TcpStream::connect(listener.local_addr().unwrap()).await.unwrap();
  let (server, _) = listener.accept().await.unwrap();
  (server, client)
}

#[tokio::main]
async fn main() {
  // 1. every request on one connection runs inside the task that polls Connection
  println!("1. three pipelined requests");
  let (srv, mut cli) = pair().await;
  let task = tokio::spawn(async move {
    println!("  connection task {}", tokio::task::id());
    http1::Builder::new().serve_connection(TokioIo::new(srv), service_fn(handler)).await
  });
  cli
    .write_all(
      b"GET /a HTTP/1.1\r\nHost: x\r\n\r\n\
        GET /b HTTP/1.1\r\nHost: x\r\n\r\n\
        GET /c HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",
    )
    .await
    .unwrap();
  let mut out = String::new();
  cli.read_to_string(&mut out).await.unwrap();
  println!("  result {:?}", task.await.unwrap().map_err(|e| e.to_string()));
  let first = out.split("\r\n\r\n").next().unwrap();
  println!("  first response head: {first:?}");

  // 2. without_shutdown hands back the socket and unread bytes
  println!("2. keep_alive(false) + without_shutdown");
  let (srv, mut cli) = pair().await;
  let task = tokio::spawn(async move {
    let conn = http1::Builder::new()
      .keep_alive(false)
      .serve_connection(TokioIo::new(srv), service_fn(handler));
    let parts = conn.without_shutdown().await.unwrap();
    println!("  read_buf = {:?}", parts.read_buf);
    let mut io = parts.io.into_inner();
    io.write_all(b"raw bytes after HTTP\n").await.unwrap();
  });
  cli
    .write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\n")
    .await
    .unwrap();
  task.await.unwrap();
  let mut out = String::new();
  cli.read_to_string(&mut out).await.unwrap();
  let lines: Vec<&str> = out
    .split("\r\n")
    .filter(|l| l.starts_with("HTTP/") || l.starts_with("connection") || l.contains("raw"))
    .collect();
  println!("  client got {lines:?}");
}

1. three pipelined requests
  connection task 13
  /a handled in task 13
  /b handled in task 13
  /c handled in task 13
  result Ok(())
  first response head: "HTTP/1.1 200 OK\r\ncontent-length: 2\r\ndate: Fri, 23 Oct 2026 13:22:01 GMT"
2. keep_alive(false) + without_shutdown
  /a handled in task 14
  read_buf = b"GET /b HTTP/1.1\r\nHost: x\r\n\r\n"
  client got ["HTTP/1.1 200 OK", "connection: close", "okraw bytes after HTTP\n"]

    pub fn graceful_shutdown(mut self: Pin<&mut Self>) {
        self.conn.disable_keep_alive();
    }

    pub(crate) fn disable_keep_alive(&mut self) {
        self.conn.disable_keep_alive();

        // If keep alive has been disabled and no read or write has been seen on
        // the connection yet, we must be in a state where the server is being asked to
        // shut down before any data has been seen on the connection
        if self.conn.is_write_closed() || self.conn.has_initial_read_write_state() {
            self.close();
        }
    }

    pub fn into_parts(self) -> Parts<I, S> {
        let (io, read_buf, dispatch) = self.conn.into_inner();
        Parts {
            io,
            read_buf,
            service: dispatch.into_service(),
        }
    }

                Ok(proto::Dispatched::Upgrade(pending)) => {
                    let (io, buf, _) = self
                        .inner
                        .take()
                        .expect("upgradeable server connection missing after upgrade")
                        .conn
                        .into_inner();
                    pending.fulfill(Upgraded::new(io, buf));
                    Poll::Ready(Ok(()))
                }

idle:
  after graceful_shutdown: Ok(()) in 0 ms
  client got []
in flight:
  /slow handled
  after graceful_shutdown: Ok(()) in 151 ms
  client got ["HTTP/1.1 200 OK", "connection: close"]