从零写一个 HTTP/1.1 服务端

httparse = "=1.10.1"
bytes = "1"
tokio = { version = "=1.54.1", features = ["net", "io-util", "rt-multi-thread", "macros", "time"] }

hyper = { version = "=1.12.0", features = ["client", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
http-body-util = "0.1"
futures = "0.3"

const MAX_HEAD: usize = 8 * 1024;
const MAX_BODY: usize = 1024 * 1024;
const MAX_HEADERS: usize = 32;

struct Head {
  method: String,
  path: String,
  minor: u8,
  content_length: usize,
  keep_alive: bool,
  chunked: bool,
}

/// Ok(None)：头还没收全。Err(status)：用这个状态码拒绝并关连接。
fn parse_head(buf: &[u8]) -> Result<Option<(Head, usize)>, u16> {
  let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
  let mut req = httparse::Request::new(&mut headers);
  let head_len = match req.parse(buf) {
    Ok(httparse::Status::Complete(n)) => n,
    Ok(httparse::Status::Partial) => return Ok(None),
    Err(httparse::Error::TooManyHeaders) => return Err(431),
    Err(_) => return Err(400),
  };
  // Complete 保证 method、path、version 都是 Some
  let minor = req.version.unwrap_or(1);
  let mut head = Head {
    method: req.method.unwrap_or_default().to_owned(),
    path: req.path.unwrap_or_default().to_owned(),
    minor,
    content_length: 0,
    keep_alive: minor == 1, // 1.1 默认 keep-alive，1.0 默认关闭
    chunked: false,
  };
  let mut seen_length = false;
  let (mut saw_close, mut saw_keep_alive) = (false, false);
  for h in req.headers.iter() {
    if h.name.eq_ignore_ascii_case("content-length") {
      let n = std::str::from_utf8(h.value)
        .ok()
        .and_then(|s| s.trim().parse::<usize>().ok())
        .ok_or(400u16)?;
      // 两个不同的长度：请求走私的入口
      if seen_length && n != head.content_length {
        return Err(400);
      }
      seen_length = true;
      head.content_length = n;
    } else if h.name.eq_ignore_ascii_case("transfer-encoding") {
      head.chunked = true;
    } else if h.name.eq_ignore_ascii_case("connection") {
      for token in h.value.split(|&b| b == b',') {
        let token = token.trim_ascii();
        saw_close |= token.eq_ignore_ascii_case(b"close");
        saw_keep_alive |= token.eq_ignore_ascii_case(b"keep-alive");
      }
    }
  }
  // 所有 Connection 行都看完再定：任何一处 close 都算数
  if saw_close {
    head.keep_alive = false;
  } else if saw_keep_alive {
    head.keep_alive = true;
  }
  Ok(Some((head, head_len)))
}

                header::CONNECTION => {
                    // A message may carry more than one `Connection` line. Once
                    // any of them asks to close, a later `keep-alive` must not
                    // undo it.
                    if saw_connection_close || headers::connection_close(&value) {
                        saw_connection_close = true;
                        keep_alive = false;
                    } else if !keep_alive {
                        // HTTP/1.0, which defaults to close
                        keep_alive = headers::connection_keep_alive(&value);
                    }
                }

async fn serve(mut stream: TcpStream, id: usize) -> io::Result<()> {
  let mut buf = BytesMut::with_capacity(4096);
  loop {
    // 1. 读，直到缓冲区里有一个完整的头
    let (head, head_len) = loop {
      match parse_head(&buf) {
        Ok(Some(parsed)) => break parsed,
        Ok(None) if buf.len() >= MAX_HEAD => return reject(&mut stream, 431, id).await,
        Ok(None) => {}
        Err(status) => return reject(&mut stream, status, id).await,
      }
      if read_more(&mut stream, &mut buf, id).await? == 0 {
        // 两个请求之间读到 EOF 是正常关闭；头读到一半就不是
        println!("[srv c{id}] eof, {} bytes of partial head", buf.len());
        return Ok(());
      }
    };
    if head.chunked {
      // 打出客户端发来的字节，然后放弃：这里没有 chunked 解码器
      println!("[srv c{id}] buffer: {}", escape(&buf));
      return reject(&mut stream, 501, id).await;
    }
    buf.advance(head_len);
    if head.content_length > MAX_BODY {
      return reject(&mut stream, 413, id).await;
    }

    // 2. body：正好 Content-Length 个字节，后面的是下一个请求
    while buf.len() < head.content_length {
      if read_more(&mut stream, &mut buf, id).await? == 0 {
        return Err(io::ErrorKind::UnexpectedEof.into());
      }
    }
    let body = buf.split_to(head.content_length);

    // 3. 调「服务」，拼好整个响应一次写出
    let resp = handle(&head, &body);
    let mut out = format!("HTTP/1.1 200 OK\r\ncontent-length: {}\r\n", resp.len());
    if !head.keep_alive {
      out.push_str("connection: close\r\n");
    } else if head.minor == 0 {
      out.push_str("connection: keep-alive\r\n");
    }
    out.push_str("\r\n");
    out.push_str(&resp);
    stream.write_all(out.as_bytes()).await?;
    // ...（打印一行日志）

    // 4. keep-alive：再转一圈，缓冲区里可能已经躺着下一个请求
    if !head.keep_alive {
      return Ok(());
    }
  }
}

type Body = BoxBody<Bytes, Infallible>;

async fn hyper_client(addr: SocketAddr, reqs: Vec<Request<Body>>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
  let stream = TcpStream::connect(addr).await?;
  let (mut sender, conn) = hyper::client::conn::http1::handshake::<_, Body>(TokioIo::new(stream)).await?;
  let conn_task = tokio::spawn(conn);
  for req in reqs {
    let line = format!("{} {}", req.method(), req.uri());
    if let Err(e) = sender.ready().await {
      println!("[cli] {line}: not sent, {e}");
      break;
    }
    match sender.send_request(req).await {
      // Ok：打印状态码、响应头和收齐的 body；Err：打印错误
      // ...
    }
  }
  drop(sender);
  let res = conn_task.await?;
  tokio::time::sleep(Duration::from_millis(20)).await; // 等服务端那一行日志先打出来
  println!("[cli] connection task: {res:?}");
  Ok(())
}

// main 里：
let reqs = vec![
  Request::get("/hello").header("host", "mini").body(Full::new(Bytes::new()).boxed())?,
  Request::post("/echo").header("host", "mini").body(Full::new(Bytes::from("ping")).boxed())?,
];
hyper_client(addr, reqs).await?;

== hyper client (client::conn::http1), two requests on one connection
[srv c1] read 35 bytes
[srv c1] GET /hello -> 200, keep-alive, 0 bytes left in buffer
[cli] GET /hello -> 200 OK [content-length: 36] b"GET /hello HTTP/1.1, body 0 bytes: \n"
[srv c1] read 58 bytes
[srv c1] POST /echo -> 200, keep-alive, 0 bytes left in buffer
[cli] POST /echo -> 200 OK [content-length: 40] b"POST /echo HTTP/1.1, body 4 bytes: ping\n"
[srv c1] read 0 bytes
[srv c1] eof, 0 bytes of partial head
[cli] connection task: Ok(())

async fn raw(addr: SocketAddr, label: &str, writes: &[&[u8]]) -> io::Result<()> {
  println!("== raw: {label}");
  let mut stream = TcpStream::connect(addr).await?;
  for (i, chunk) in writes.iter().enumerate() {
    if i > 0 {
      tokio::time::sleep(Duration::from_millis(50)).await;
    }
    stream.write_all(chunk).await?;
  }
  let mut out = Vec::new();
  stream.read_to_end(&mut out).await?;
  // ...（逐行打印）
  Ok(())
}

raw(
  addr,
  "three pipelined requests in one write",
  &[b"GET /a HTTP/1.1\r\nHost: x\r\n\r\nPOST /b HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n\r\nhelloGET /c HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"],
)
.await?;

== raw: three pipelined requests in one write
[srv c2] read 128 bytes
[srv c2] GET /a -> 200, keep-alive, 100 bytes left in buffer
[srv c2] POST /b -> 200, keep-alive, 47 bytes left in buffer
[srv c2] GET /c -> 200, close, 0 bytes left in buffer
[raw] HTTP/1.1 200 OK\r\n
[raw] content-length: 32\r\n
[raw] \r\n
[raw] GET /a HTTP/1.1, body 0 bytes: \n
[raw] HTTP/1.1 200 OK\r\n
[raw] content-length: 38\r\n
[raw] \r\n
[raw] POST /b HTTP/1.1, body 5 bytes: hello\n
[raw] HTTP/1.1 200 OK\r\n
[raw] content-length: 32\r\n
[raw] connection: close\r\n
[raw] \r\n
[raw] GET /c HTTP/1.1, body 0 bytes: \n
[raw] eof after 238 bytes

raw(
  addr,
  "head split across two writes, body across the next",
  &[b"POST /split HTTP/1.1\r\nHo", b"st: x\r\nContent-Length: 10\r\nConnection: close\r\n\r\n01234", b"56789"],
)
.await?;

== raw: head split across two writes, body across the next
[srv c3] read 24 bytes
[srv c3] read 53 bytes
[srv c3] read 5 bytes
[srv c3] POST /split -> 200, close, 0 bytes left in buffer
[raw] HTTP/1.1 200 OK\r\n
[raw] content-length: 48\r\n
[raw] connection: close\r\n
[raw] \r\n
[raw] POST /split HTTP/1.1, body 10 bytes: 0123456789\n
[raw] eof after 106 bytes

raw(addr, "HTTP/1.0, no keep-alive", &[b"GET /old HTTP/1.0\r\n\r\n"]).await?;
raw(
  addr,
  "Connection: close, then Connection: keep-alive on a second line",
  &[b"GET /two-lines HTTP/1.1\r\nHost: x\r\nConnection: close\r\nConnection: keep-alive\r\n\r\n"],
)
.await?;
raw(addr, "malformed header", &[b"GET / HTTP/1.1\r\nHost x\r\n\r\n"]).await?;

== raw: HTTP/1.0, no keep-alive
[srv c4] read 21 bytes
[srv c4] GET /old -> 200, close, 0 bytes left in buffer
[raw] HTTP/1.1 200 OK\r\n
[raw] content-length: 34\r\n
[raw] connection: close\r\n
[raw] \r\n
[raw] GET /old HTTP/1.0, body 0 bytes: \n
[raw] eof after 92 bytes
== raw: Connection: close, then Connection: keep-alive on a second line
[srv c5] read 79 bytes
[srv c5] GET /two-lines -> 200, close, 0 bytes left in buffer
[raw] HTTP/1.1 200 OK\r\n
[raw] content-length: 40\r\n
[raw] connection: close\r\n
[raw] \r\n
[raw] GET /two-lines HTTP/1.1, body 0 bytes: \n
[raw] eof after 98 bytes
== raw: malformed header
[srv c6] read 26 bytes
[srv c6] reject 400, close
[raw] HTTP/1.1 400 Bad Request\r\n
[raw] content-length: 0\r\n
[raw] connection: close\r\n
[raw] \r\n
[raw] eof after 66 bytes

== hyper client, body without a known length
[srv c7] read 83 bytes
[srv c7] buffer: POST /stream HTTP/1.1\r\nhost: mini\r\ntransfer-encoding: chunked\r\n\r\n8\r\nstreamed\r\n0\r\n\r\n
[srv c7] reject 501, close
[cli] POST /stream -> 501 Not Implemented [content-length: 0, connection: close] b""
[cli] connection task: Ok(())