解析与编码

pub(crate) trait Http1Transaction {
    type Incoming;
    type Outgoing: Default;
    #[cfg(feature = "tracing")]
    const LOG: &'static str;
    fn parse(bytes: &mut BytesMut, ctx: ParseContext<'_>) -> ParseResult<Self::Incoming>;
    fn encode(enc: Encode<'_, Self::Outgoing>, dst: &mut Vec<u8>) -> crate::Result<Encoder>;

    fn on_error(err: &crate::Error) -> Option<MessageHead<Self::Outgoing>>;

    fn is_client() -> bool {
        !Self::is_server()
    }

    fn is_server() -> bool {
        !Self::is_client()
    }

    fn should_error_on_parse_eof() -> bool {
        Self::is_client()
    }

    fn should_read_first() -> bool {
        Self::is_server()
    }

    fn update_date() {}
}

pub(crate) struct ParsedMessage<T> {
    head: MessageHead<T>,
    decode: DecodedLength,
    expect_continue: bool,
    keep_alive: bool,
    wants_upgrade: bool,
}

fn is_complete_fast(bytes: &[u8], prev_len: usize) -> bool {
    let start = prev_len.saturating_sub(3);
    let bytes = &bytes[start..];

    for (i, b) in bytes.iter().copied().enumerate() {
        if b == b'\r' {
            if bytes[i + 1..].chunks(3).next() == Some(&b"\n\r\n"[..]) {
                return true;
            }
        } else if b == b'\n'
            && (bytes.get(i + 1) == Some(&b'\n')
                || bytes[i + 1..].chunks(2).next() == Some(&b"\r\n"[..]))
        {
            return true;
        }
    }

    false
}

        // Both headers_indices and headers are using uninitialized memory,
        // but we *never* read any of it until after httparse has assigned
        // values into it. By not zeroing out the stack memory, this saves
        // a good ~5% on pipeline benchmarks.
        let mut headers_indices: SmallVec<[MaybeUninit<HeaderIndices>; DEFAULT_MAX_HEADERS]> =
            match ctx.h1_max_headers {
                Some(cap) => smallvec![MaybeUninit::uninit(); cap],
                None => smallvec_inline![MaybeUninit::uninit(); DEFAULT_MAX_HEADERS],
            };
        {
            let mut headers: SmallVec<[MaybeUninit<httparse::Header<'_>>; DEFAULT_MAX_HEADERS]> =
                match ctx.h1_max_headers {
                    Some(cap) => smallvec![MaybeUninit::uninit(); cap],
                    None => smallvec_inline![MaybeUninit::uninit(); DEFAULT_MAX_HEADERS],
                };

        let slice = buf.split_to(len).freeze();

        // According to https://tools.ietf.org/html/rfc7230#section-3.3.3
        // 1. (irrelevant to Request)
        // 2. (irrelevant to Request)
        // 3. Transfer-Encoding: chunked has a chunked body.
        // 4. If multiple differing Content-Length headers or invalid, close connection.
        // 5. Content-Length header has a sized body.
        // 6. Length 0.
        // 7. (irrelevant to Request)

                header::TRANSFER_ENCODING => {
                    // ...
                    if !is_http_11 {
                        debug!("HTTP/1.0 cannot have Transfer-Encoding header");
                        return Err(Parse::transfer_encoding_unexpected());
                    }
                    is_te = true;
                    if is_cl && con_len.take().is_some() {
                        headers.remove(header::CONTENT_LENGTH);
                    }
                    if headers::is_chunked_(&value) {
                        is_te_chunked = true;
                        decoder = DecodedLength::CHUNKED;
                    } else {
                        is_te_chunked = false;
                    }
                }
                header::CONTENT_LENGTH => {
                    is_cl = true;
                    if is_te {
                        continue;
                    }
                    let len = headers::content_length_parse(&value)
                        .ok_or_else(Parse::content_length_invalid)?;
                    if let Some(prev) = con_len {
                        if prev != len {
                            // ...
                            return Err(Parse::content_length_invalid());
                        }
                        // we don't need to append this secondary length
                        continue;
                    }
                    decoder = DecodedLength::checked_new(len)?;
                    con_len = Some(len);
                }

        if is_te && !is_te_chunked {
            debug!("request with transfer-encoding header, but not chunked, bad request");
            return Err(Parse::transfer_encoding_invalid());
        }

        if is_te && is_cl {
            keep_alive = false;
        }

    fn on_error(err: &crate::Error) -> Option<MessageHead<Self::Outgoing>> {
        use crate::error::Kind;
        let status = match *err.kind() {
            Kind::Parse(Parse::Method)
            | Kind::Parse(Parse::Header(_))
            | Kind::Parse(Parse::Uri)
            | Kind::Parse(Parse::Version) => StatusCode::BAD_REQUEST,
            Kind::Parse(Parse::TooLarge) => StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE,
            Kind::Parse(Parse::UriTooLong) => StatusCode::URI_TOO_LONG,
            _ => return None,
        };

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

use std::error::Error as _;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use tokio::time::timeout;

pub fn show(bytes: &[u8]) -> String {
  let s = String::from_utf8_lossy(bytes);
  s.replace("\r\n", "\\r\\n\n")
}

pub fn err_chain(e: &hyper::Error) -> String {
  let mut s = e.to_string();
  let mut src = e.source();
  while let Some(inner) = src {
    s.push_str(&format!(": {inner}"));
    src = inner.source();
  }
  s
}

// Send `req`, read until EOF or 300ms of silence, print it.
pub async fn exchange(addr: std::net::SocketAddr, name: &str, req: &[u8], rx: &mut mpsc::UnboundedReceiver<String>) {
  println!("== {name}");
  let mut tcp = TcpStream::connect(addr).await.unwrap();
  if let Err(e) = tcp.write_all(req).await { println!("(client write error: {e})"); }
  let mut got = Vec::new();
  let mut buf = vec![0u8; 64 * 1024];
  let end = loop {
    match timeout(Duration::from_millis(300), tcp.read(&mut buf)).await {
      Ok(Ok(0)) => break "[EOF]",
      Ok(Ok(n)) => got.extend_from_slice(&buf[..n]),
      Ok(Err(_)) => break "[reset]",
      Err(_) => break "[still open]",
    }
  };
  print!("{}", show(&got));
  if !got.ends_with(b"\r\n") && !got.is_empty() {
    println!();
  }
  println!("{end}");
  drop(tcp);
  match timeout(Duration::from_secs(1), rx.recv()).await {
    Ok(Some(s)) => println!("server: {s}"),
    _ => println!("server: (no result)"),
  }
}

#[path = "../common.rs"]
mod common;

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

use bytes::Bytes;
use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full, StreamBody};
use hyper::body::{Body, Frame, Incoming, SizeHint};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;
use tokio::sync::mpsc;

// Claims 5 bytes, delivers 3.
struct Short(bool);

impl Body for Short {
  type Data = Bytes;
  type Error = Infallible;
  fn poll_frame(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Result<Frame<Bytes>, Infallible>>> {
    if self.0 {
      return Poll::Ready(None);
    }
    self.0 = true;
    Poll::Ready(Some(Ok(Frame::data(Bytes::from_static(b"abc")))))
  }
  fn size_hint(&self) -> SizeHint {
    SizeHint::with_exact(5)
  }
}

type B = BoxBody<Bytes, Infallible>;

async fn handle(req: Request<Incoming>) -> Result<Response<B>, Infallible> {
  let path = req.uri().path().to_string();
  let res = match path.as_str() {
    "/full" => Response::new(Full::new(Bytes::from("hello")).boxed()),
    "/empty" => Response::new(Empty::new().boxed()),
    "/stream" => {
      let frames = vec![
        Ok::<_, Infallible>(Frame::data(Bytes::from("abc"))),
        Ok(Frame::data(Bytes::from("defg"))),
      ];
      Response::new(StreamBody::new(futures::stream::iter(frames)).boxed())
    }
    "/nocontent" => Response::builder()
      .status(StatusCode::NO_CONTENT)
      .body(Empty::new().boxed())
      .unwrap(),
    "/short" => Response::new(Short(false).boxed()),
    "/echo" => {
      let cl = req.headers().get("content-length").cloned();
      let te = req.headers().get("transfer-encoding").cloned();
      let body = req.into_body().collect().await.map(|c| c.to_bytes());
      let text = format!("cl={cl:?} te={te:?} body={body:?}");
      Response::new(Full::new(Bytes::from(text)).boxed())
    }
    _ => Response::new(Empty::new().boxed()),
  };
  Ok(res)
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let (tx, mut rx) = mpsc::unbounded_channel();
  let addr = spawn_server(false, tx.clone()).await;
  let title_addr = spawn_server(true, tx).await;

  let long_uri = format!("GET /{} HTTP/1.1\r\n\r\n", "a".repeat(65534));
  let many: String = (0..101).map(|i| format!("x-h{i}: v\r\n")).collect();
  let many_headers = format!("GET / HTTP/1.1\r\n{many}\r\n");
  let huge = format!("GET / HTTP/1.1\r\nx-big: {}\r\n\r\n", "a".repeat(500 * 1024));

  let cases: Vec<(&str, Vec<u8>)> = vec![
    ("full", b"GET /full HTTP/1.1\r\n\r\n".to_vec()),
    ("stream", b"GET /stream HTTP/1.1\r\n\r\n".to_vec()),
    ("empty", b"GET /empty HTTP/1.1\r\n\r\n".to_vec()),
    ("head full", b"HEAD /full HTTP/1.1\r\n\r\n".to_vec()),
    ("204", b"GET /nocontent HTTP/1.1\r\n\r\n".to_vec()),
    ("stream over 1.0", b"GET /stream HTTP/1.0\r\n\r\n".to_vec()),
    ("short body", b"GET /short HTTP/1.1\r\n\r\n".to_vec()),
    (
      "cl then te",
      b"POST /echo HTTP/1.1\r\ncontent-length: 3\r\ntransfer-encoding: chunked\r\n\r\n3\r\nabc\r\n0\r\n\r\n".to_vec(),
    ),
    ("te gzip", b"POST /echo HTTP/1.1\r\ntransfer-encoding: gzip\r\n\r\n".to_vec()),
    ("te on 1.0", b"POST /echo HTTP/1.0\r\ntransfer-encoding: chunked\r\n\r\n".to_vec()),
    ("cl abc", b"POST /echo HTTP/1.1\r\ncontent-length: abc\r\n\r\n".to_vec()),
    ("cl 3 and 4", b"POST /echo HTTP/1.1\r\ncontent-length: 3\r\ncontent-length: 4\r\n\r\n".to_vec()),
    ("cl u64::MAX", b"POST /echo HTTP/1.1\r\ncontent-length: 18446744073709551615\r\n\r\n".to_vec()),
    ("bad method", b"G@T / HTTP/1.1\r\n\r\n".to_vec()),
    ("bad version", b"GET / HTTP/1.2\r\n\r\n".to_vec()),
    ("bad header", b"GET / HTTP/1.1\r\nbad header: x\r\n\r\n".to_vec()),
    ("uri 65535", long_uri.into_bytes()),
    ("101 headers", many_headers.into_bytes()),
    ("500KiB head", huge.into_bytes()),
    ("h2 preface", b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n".to_vec()),
  ];
  for (name, req) in cases {
    common::exchange(addr, name, &req, &mut rx).await;
  }
  common::exchange(title_addr, "full, title_case_headers(true)", b"GET /full HTTP/1.1\r\n\r\n", &mut rx).await;
}

async fn spawn_server(title_case: bool, tx: mpsc::UnboundedSender<String>) -> std::net::SocketAddr {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      let tx = tx.clone();
      tokio::spawn(async move {
        let r = http1::Builder::new()
          .title_case_headers(title_case)
          .serve_connection(TokioIo::new(stream), service_fn(handle))
          .await;
        let _ = tx.send(match r {
          Ok(()) => "Ok".to_string(),
          Err(e) => format!("Err({})", common::err_chain(&e)),
        });
      });
    }
  });
  addr
}

== cl then te
HTTP/1.1 200 OK\r\n
connection: close\r\n
content-length: 42\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
cl=None te=Some("chunked") body=Ok(b"abc")
[EOF]
server: Ok
== te gzip
HTTP/1.1 400 Bad Request\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(invalid transfer-encoding parsed)
== te on 1.0
HTTP/1.1 400 Bad Request\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(unexpected transfer-encoding parsed)
== cl abc
HTTP/1.1 400 Bad Request\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(invalid content-length parsed)
== cl 3 and 4
HTTP/1.1 400 Bad Request\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(invalid content-length parsed)

== bad method
HTTP/1.1 400 Bad Request\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(invalid HTTP method parsed)
== bad version
HTTP/1.1 400 Bad Request\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(invalid HTTP version parsed)
== bad header
HTTP/1.1 400 Bad Request\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(invalid HTTP header parsed)
== uri 65535
HTTP/1.1 414 URI Too Long\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(URI too long)
== 101 headers
HTTP/1.1 431 Request Header Fields Too Large\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(message head is too large)
== 500KiB head
HTTP/1.1 431 Request Header Fields Too Large\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[reset]
server: Err(message head is too large)
== h2 preface
[EOF]
server: Err(invalid HTTP version parsed (found HTTP2 preface))

        let (ret, is_last) = if msg.head.subject == StatusCode::SWITCHING_PROTOCOLS {
            (Ok(()), true)
        } else if msg.req_method == &Some(Method::CONNECT) && msg.head.subject.is_success() {
            // Sending content-length or transfer-encoding header on 2xx response
            // to CONNECT is forbidden in RFC 7231.
            wrote_len = true;
            (Ok(()), true)
        } else if msg.head.subject.is_informational() {
            warn!("response with 1xx status code not supported");
            *msg.head = MessageHead::default();
            msg.head.subject = StatusCode::INTERNAL_SERVER_ERROR;
            msg.body = None;
            (Err(crate::Error::new_user_unsupported_status_code()), true)
        } else {
            (Ok(()), !msg.keep_alive)
        };

async fn handle(_: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  Ok(Response::builder().status(StatusCode::from_u16(103).unwrap()).body(Full::new(Bytes::from("x"))).unwrap())
}

== 103 from service
HTTP/1.1 500 Internal Server Error\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:17:24 GMT\r\n
\r\n
[EOF]
server: Err(response has 1xx status code, not supported by server)

        if !wrote_len {
            encoder = match msg.body {
                Some(BodyLength::Unknown) => {
                    if msg.head.version == Version::HTTP_10
                        || !Server::can_chunked(msg.req_method.as_ref(), msg.head.subject)
                    {
                        Encoder::close_delimited()
                    } else {
                        header_name_writer.write_full_header_line(
                            dst,
                            "transfer-encoding: chunked\r\n",
                            (header::TRANSFER_ENCODING, ": chunked\r\n"),
                        );
                        Encoder::chunked()
                    }
                }
                None | Some(BodyLength::Known(0)) => {
                    if Server::can_have_implicit_zero_content_length(
                        msg.req_method.as_ref(),
                        msg.head.subject,
                    ) {
                        header_name_writer.write_full_header_line(
                            dst,
                            "content-length: 0\r\n",
                            (header::CONTENT_LENGTH, ": 0\r\n"),
                        );
                    }
                    Encoder::length(0)
                }
                Some(BodyLength::Known(len)) => {
                    if !Server::can_have_content_length(msg.req_method.as_ref(), msg.head.subject) {
                        Encoder::length(0)
                    } else {
                        header_name_writer.write_header_name_with_colon(
                            dst,
                            "content-length: ",
                            header::CONTENT_LENGTH,
                        );
                        extend(dst, ::itoa::Buffer::new().format(len).as_bytes());
                        extend(dst, b"\r\n");
                        Encoder::length(len)
                    }
                }
            };
        }

== full
HTTP/1.1 200 OK\r\n
content-length: 5\r\n
date: Fri, 23 Oct 2026 13:13:14 GMT\r\n
\r\n
hello
[still open]
server: Ok
== stream
HTTP/1.1 200 OK\r\n
transfer-encoding: chunked\r\n
date: Fri, 23 Oct 2026 13:13:15 GMT\r\n
\r\n
3\r\n
abc\r\n
4\r\n
defg\r\n
0\r\n
\r\n
[still open]
server: Ok
== empty
HTTP/1.1 200 OK\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:15 GMT\r\n
\r\n
[still open]
server: Ok
== head full
HTTP/1.1 200 OK\r\n
content-length: 5\r\n
date: Fri, 23 Oct 2026 13:13:15 GMT\r\n
\r\n
[still open]
server: Ok
== 204
HTTP/1.1 204 No Content\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[still open]
server: Ok
== stream over 1.0
HTTP/1.0 200 OK\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
abcdefg
[EOF]
server: Ok

        // cached date is much faster than formatting every request
        // don't force the write if disabled
        if !wrote_date && msg.date_header {
            dst.reserve(date::DATE_VALUE_LENGTH + 8);
            header_name_writer.write_header_name_with_colon(dst, "date: ", header::DATE);
            date::extend(dst);
            extend(dst, b"\r\n\r\n");
        } else {
            extend(dst, b"\r\n");
        }

thread_local!(static CACHED: RefCell<CachedDate> = RefCell::new(CachedDate::new()));

    fn update(&mut self, now: SystemTime) {
        let nanos = now
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .subsec_nanos();

        self.render(now);
        self.next_update = now + Duration::new(1, 0) - Duration::from_nanos(u64::from(nanos));
    }

== full, title_case_headers(true)
HTTP/1.1 200 OK\r\n
Content-Length: 5\r\n
Date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
hello
[still open]
server: Ok

== cl u64::MAX
HTTP/1.1 431 Request Header Fields Too Large\r\n
connection: close\r\n
content-length: 0\r\n
date: Fri, 23 Oct 2026 13:13:16 GMT\r\n
\r\n
[EOF]
server: Err(message head is too large)

    pub(crate) fn checked_new(len: u64) -> Result<Self, crate::error::Parse> {
        if len <= MAX_LEN {
            Ok(DecodedLength(len))
        } else {
            warn!("content-length bigger than maximum: {} > {}", len, MAX_LEN);
            Err(crate::error::Parse::TooLarge)
        }
    }