一个请求的一生

hyper = { version = "=1.12.0", features = ["client", "server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
tokio = { version = "=1.54.1", features = ["macros", "rt", "net"] }

use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};

use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::net::{TcpListener, TcpStream};

tokio::task_local! {
  // 每个任务自己的名字，日志里用它标出「这一行是哪个任务执行的」
  static TASK: &'static str;
}

fn log(msg: impl AsRef<str>) {
  let task = TASK.try_with(|t| *t).unwrap_or("?");
  println!("[{task:<13}] {}", msg.as_ref());
}

// 只取第一行，Date 头每次都不同，不打印它
fn first_line(bytes: &[u8]) -> String {
  let text = String::from_utf8_lossy(bytes);
  text.lines().next().unwrap_or("").to_string()
}

// 包住 TcpStream，记录每一次真正碰到 socket 的读写
struct LogIo {
  side: &'static str,
  inner: TcpStream,
}

impl AsyncRead for LogIo {
  fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
    let before = buf.filled().len();
    let side = self.side;
    let res = Pin::new(&mut self.inner).poll_read(cx, buf);
    match &res {
      Poll::Ready(Ok(())) => {
        let got = &buf.filled()[before..];
        log(format!("{side} socket read  {:>3} bytes  {:?}", got.len(), first_line(got)));
      }
      Poll::Pending => log(format!("{side} socket read  Pending")),
      Poll::Ready(Err(e)) => log(format!("{side} socket read  error {e}")),
    }
    res
  }
}

impl AsyncWrite for LogIo {
  fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
    let side = self.side;
    let res = Pin::new(&mut self.inner).poll_write(cx, buf);
    if let Poll::Ready(Ok(n)) = &res {
      log(format!("{side} socket write {n:>3} bytes  {:?}", first_line(&buf[..*n])));
    }
    res
  }

  fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
    Pin::new(&mut self.inner).poll_flush(cx)
  }

  fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
    log(format!("{} socket shutdown", self.side));
    Pin::new(&mut self.inner).poll_shutdown(cx)
  }
}

async fn handle(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, hyper::Error> {
  log(format!("service future: got {} {}", req.method(), req.uri()));
  let body = req.into_body().collect().await?.to_bytes();
  log(format!("service future: request body {:?}", body));
  Ok(Response::new(Full::new(Bytes::from(format!("echo: {}", String::from_utf8_lossy(&body))))))
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
  let listener = TcpListener::bind("127.0.0.1:0").await?;
  let addr = listener.local_addr()?;

  // 服务端：accept 一条连接，为它 spawn 一个任务去跑 Connection future
  let server = tokio::spawn(TASK.scope("server-accept", async move {
    let (stream, _) = listener.accept().await.unwrap();
    log("accepted, spawn server-conn");
    let io = TokioIo::new(LogIo { side: "srv", inner: stream });
    let svc = hyper::service::service_fn(|req| {
      log("service.call(req)");
      handle(req)
    });
    let conn = hyper::server::conn::http1::Builder::new().serve_connection(io, svc);
    tokio::spawn(TASK.scope("server-conn", async move {
      let res = conn.await;
      log(format!("Connection future done: {res:?}"));
    }))
  }));

  TASK.scope("main", async move {
    // 客户端：连上之后握手，得到 SendRequest 和 Connection；Connection 交给另一个任务
    let stream = TcpStream::connect(addr).await?;
    let io = TokioIo::new(LogIo { side: "cli", inner: stream });
    let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await?;
    log("handshake done, spawn client-conn");
    let client_conn = tokio::spawn(TASK.scope("client-conn", async move {
      let res = conn.await;
      log(format!("Connection future done: {res:?}"));
    }));

    let req = Request::post("/echo")
      .header("host", addr.to_string())
      .body(Full::new(Bytes::from_static(b"hello")))?;
    log("send_request(req)");
    let resp = sender.send_request(req).await?;
    log(format!("got response head: {}", resp.status()));
    let body = resp.into_body().collect().await?.to_bytes();
    log(format!("response body {:?}", body));

    drop(sender);
    log("dropped SendRequest");
    client_conn.await?;
    server.await?.await?;
    Ok::<_, Box<dyn std::error::Error>>(())
  })
  .await
}

[main         ] handshake done, spawn client-conn
[main         ] send_request(req)
[client-conn  ] cli socket read  Pending
[client-conn  ] cli socket write  70 bytes  "POST /echo HTTP/1.1"
[server-accept] accepted, spawn server-conn
[server-conn  ] srv socket read   70 bytes  "POST /echo HTTP/1.1"
[server-conn  ] service.call(req)
[server-conn  ] srv socket read  Pending
[server-conn  ] service future: got POST /echo
[server-conn  ] service future: request body b"hello"
[server-conn  ] srv socket write  87 bytes  "HTTP/1.1 200 OK"
[server-conn  ] srv socket read  Pending
[client-conn  ] cli socket read   87 bytes  "HTTP/1.1 200 OK"
[client-conn  ] cli socket read  Pending
[client-conn  ] cli socket read  Pending
[client-conn  ] cli socket read  Pending
[main         ] got response head: 200 OK
[main         ] response body b"echo: hello"
[main         ] dropped SendRequest
[client-conn  ] cli socket read  Pending
[client-conn  ] cli socket shutdown
[client-conn  ] Connection future done: Ok(())
[server-conn  ] srv socket read    0 bytes  ""
[server-conn  ] srv socket shutdown
[server-conn  ] Connection future done: Ok(())

        async move {
            trace!("client handshake HTTP/1");

            let (tx, rx) = dispatch::channel();
            let mut conn = proto::Conn::new(io);
            conn.set_h1_parser_config(opts.h1_parser_config);
            // ... 按配置设置 conn 的各个选项 ...
            let cd = proto::h1::dispatch::Client::new(rx);
            let proto = proto::h1::Dispatcher::new(cd, conn);

            Ok((SendRequest { dispatch: tx }, Connection { inner: proto }))
        }

        let sent = self.dispatch.send(req);

        async move {
            match sent {
                Ok(rx) => match rx.await {
                    Ok(Ok(resp)) => Ok(resp),
                    Ok(Err(err)) => Err(err),
                    // this is definite bug if it happens, but it shouldn't happen!
                    Err(_canceled) => unreachable!("dispatch dropped without returning error"),
                },
                Err(_req) => {
                    debug!("connection was not ready");
                    Err(crate::Error::new_canceled().with("connection was not ready"))
                }
            }
        }

        fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, IncomingBody)>) -> crate::Result<()> {
            let (msg, body) = msg?;
            let mut req = Request::new(body);
            *req.method_mut() = msg.subject.0;
            *req.uri_mut() = msg.subject.1;
            *req.headers_mut() = msg.headers;
            *req.version_mut() = msg.version;
            *req.extensions_mut() = msg.extensions;
            let fut = self.service.call(req);
            self.in_flight.set(Some(fut));
            Ok(())
        }

without conn: timed out after 1s

conn dropped first: Ok(Err(hyper::Error(Canceled, "connection was not ready")))