body 的解码与编码

enum Kind {
    /// A Reader used when a Content-Length header is passed with a positive integer.
    Length(u64),
    /// A Reader used when Transfer-Encoding is `chunked`.
    Chunked {
        state: ChunkedState,
        chunk_len: u64,
        extensions_cnt: u64,
        trailers_buf: Option<BytesMut>,
        trailers_cnt: usize,
        h1_max_headers: Option<usize>,
        h1_max_header_size: Option<usize>,
    },
    /// A Reader used for responses that don't indicate a length or chunked.
    ///
    /// The bool tracks when EOF is seen on the transport.
    ///
    /// Note: This should only used for `Response`s. It is illegal for a
    /// `Request` to be made with both `Content-Length` and
    /// `Transfer-Encoding: chunked` missing, as explained from the spec:
    // ...
    Eof(bool),
}

    pub(crate) fn decode<R: MemRead>(
        &mut self,
        cx: &mut Context<'_>,
        body: &mut R,
    ) -> Poll<Result<Frame<Bytes>, io::Error>> {

    fn read_mem(&mut self, cx: &mut Context<'_>, len: usize) -> Poll<io::Result<Bytes>> {
        if !self.read_buf.is_empty() {
            let n = std::cmp::min(len, self.read_buf.len());
            Poll::Ready(Ok(self.read_buf.split_to(n).freeze()))
        } else {
            let n = ready!(self.poll_read_from_io(cx))?;
            Poll::Ready(Ok(self.read_buf.split_to(::std::cmp::min(len, n)).freeze()))
        }
    }

            Length(remaining) => {
                if *remaining == 0 {
                    Poll::Ready(Ok(Frame::data(Bytes::new())))
                } else {
                    let to_read = usize::try_from(*remaining).unwrap_or(usize::MAX);
                    let buf = ready!(body.read_mem(cx, to_read))?;
                    let num = buf.as_ref().len() as u64;
                    if num > *remaining {
                        *remaining = 0;
                    } else if num == 0 {
                        return Poll::Ready(Err(io::Error::new(
                            io::ErrorKind::UnexpectedEof,
                            IncompleteBody,
                        )));
                    } else {
                        *remaining -= num;
                    }
                    Poll::Ready(Ok(Frame::data(buf)))
                }
            }

            Eof(is_eof) => {
                if *is_eof {
                    Poll::Ready(Ok(Frame::data(Bytes::new())))
                } else {
                    // 8192 chosen because its about 2 packets, there probably
                    // won't be that much available, so don't have MemReaders
                    // allocate buffers to big
                    body.read_mem(cx, 8192).map_ok(|slice| {
                        *is_eof = slice.is_empty();
                        Frame::data(slice)
                    })
                }
            }

5;name=val\r\n
hello\r\n
0\r\n
x-sum: 42\r\n
\r\n

macro_rules! byte (
    ($rdr:ident, $cx:expr) => ({
        let buf = ready!($rdr.read_mem($cx, 1))?;
        if !buf.is_empty() {
            buf[0]
        } else {
            return Poll::Ready(Err(io::Error::new(io::ErrorKind::UnexpectedEof,
                                      "unexpected EOF during chunk size line")));
        }
    })
);

        // cap remaining bytes at the max capacity of usize
        let to_read = usize::try_from(*rem).unwrap_or(usize::MAX);
        let slice = ready!(rdr.read_mem(cx, to_read))?;
        let count = slice.len();

        if count == 0 {
            *rem = 0;
            return Poll::Ready(Err(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                IncompleteBody,
            )));
        }
        *buf = Some(slice);
        *rem -= count as u64;

        // We don't care about extensions really at all. Just ignore them.
        // They "end" at the next CRLF.
        //
        // However, some implementations may not check for the CR, so to save
        // them from themselves, we reject extensions containing plain LF as
        // well.
        match byte!(rdr, cx) {
            b'\r' => Poll::Ready(Ok(ChunkedState::SizeLf)),
            b'\n' => Poll::Ready(Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "invalid chunk extension contains newline",
            ))),
            _ => {
                *extensions_cnt += 1;
                if *extensions_cnt >= CHUNKED_EXTENSIONS_LIMIT {
                    Poll::Ready(Err(io::Error::new(
                        io::ErrorKind::InvalidData,
                        "chunk extensions over limit",
                    )))
                } else {
                    Poll::Ready(Ok(ChunkedState::Extension))
                }
            } // no supported extensions
        }

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

// Like `exchange`, but writes `parts` 50ms apart, optionally half-closing after.
pub async fn exchange_parts(
  addr: std::net::SocketAddr,
  name: &str,
  parts: &[&[u8]],
  shutdown: bool,
  rx: &mut mpsc::UnboundedReceiver<String>,
) {
  println!("== {name}");
  let mut tcp = TcpStream::connect(addr).await.unwrap();
  for (i, p) in parts.iter().enumerate() {
    if i > 0 {
      tokio::time::sleep(Duration::from_millis(50)).await;
    }
    tcp.write_all(p).await.unwrap();
  }
  if shutdown {
    tcp.shutdown().await.unwrap();
  }
  // ... the rest is the same as `exchange`
}

// Claims 3 bytes, delivers 6.
struct Long(bool);

impl Body for Long {
  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"abcdef")))))
  }
  fn size_hint(&self) -> SizeHint {
    SizeHint::with_exact(3)
  }
}

type B = BoxBody<Bytes, Infallible>;

async fn handle(req: Request<Incoming>) -> Result<Response<B>, Infallible> {
  match req.uri().path() {
    "/frames" => {
      // Report every frame the decoder hands us.
      let mut body = req.into_body();
      let mut seen = Vec::new();
      while let Some(f) = body.frame().await {
        match f {
          Ok(f) if f.is_data() => seen.push(format!("data({})", f.into_data().unwrap().len())),
          Ok(f) => seen.push(format!("trailers{:?}", f.into_trailers().unwrap())),
          Err(e) => {
            seen.push(format!("err({})", common::err_chain(&e)));
            break;
          }
        }
      }
      Ok(Response::new(Full::new(Bytes::from(seen.join(" "))).boxed()))
    }
    "/trailers" => {
      let mut t = HeaderMap::new();
      t.insert("x-checksum", "42".parse().unwrap());
      t.insert("set-cookie", "a=b".parse().unwrap());
      let frames = vec![Ok::<_, Infallible>(Frame::data(Bytes::from("abc"))), Ok(Frame::trailers(t))];
      Ok(
        Response::builder()
          .header("trailer", "x-checksum, set-cookie")
          .body(StreamBody::new(futures::stream::iter(frames)).boxed())
          .unwrap(),
      )
    }
    "/long" => Ok(Response::new(Long(false).boxed())),
    _ => Ok(Response::new(Full::new(Bytes::new()).boxed())),
  }
}

  const CHUNKED: &[u8] = b"POST /frames HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n";
  let ext = format!("1;{}\r\n", "e".repeat(16 * 1024));
  let big_trailer = format!("1\r\na\r\n0\r\nx-t: {}\r\n\r\n", "t".repeat(16 * 1024));
  let r = &mut rx;

  common::exchange_parts(
    addr,
    "chunked: ext + 2 chunks + trailer, one write",
    &[&[CHUNKED, b"5;name=val\r\nhello\r\n6\r\n world\r\n0\r\nx-sum: 42\r\n\r\n"].concat()],
    false,
    r,
  )
  .await;
  common::exchange_parts(addr, "chunked: one chunk split across writes", &[CHUNKED, b"a\r\n0123", b"456789\r\n0\r\n\r\n"], false, r).await;
  common::exchange_parts(
    addr,
    "length 7 split across writes",
    &[b"POST /frames HTTP/1.1\r\ncontent-length: 7\r\n\r\nabc", b"defg"],
    false,
    r,
  )
  .await;
  common::exchange_parts(addr, "length 10, peer half-closes after 3", &[b"POST /frames HTTP/1.1\r\ncontent-length: 10\r\n\r\nabc"], true, r).await;
  common::exchange_parts(addr, "chunk size overflow", &[CHUNKED, b"10000000000000000\r\n"], false, r).await;
  common::exchange_parts(addr, "chunk data without CRLF", &[CHUNKED, b"3\r\nabcX"], false, r).await;
  common::exchange_parts(addr, "16KiB chunk extension", &[CHUNKED, ext.as_bytes()], false, r).await;
  common::exchange_parts(addr, "16KiB trailer", &[CHUNKED, big_trailer.as_bytes()], false, r).await;
  common::exchange_parts(addr, "response trailers, no te", &[b"GET /trailers HTTP/1.1\r\n\r\n"], false, r).await;
  common::exchange_parts(addr, "response trailers, te: trailers", &[b"GET /trailers HTTP/1.1\r\nte: trailers\r\n\r\n"], false, r).await;
  common::exchange_parts(addr, "body longer than size_hint", &[b"GET /long HTTP/1.1\r\n\r\n"], false, r).await;

== chunked: one chunk split across writes
HTTP/1.1 200 OK\r\n
content-length: 15\r\n
date: Fri, 23 Oct 2026 13:11:28 GMT\r\n
\r\n
data(4) data(6)
[still open]
server: Ok
== length 7 split across writes
HTTP/1.1 200 OK\r\n
content-length: 15\r\n
date: Fri, 23 Oct 2026 13:11:29 GMT\r\n
\r\n
data(3) data(4)
[still open]
server: Ok
== length 10, peer half-closes after 3
HTTP/1.1 200 OK\r\n
content-length: 92\r\n
date: Fri, 23 Oct 2026 13:11:29 GMT\r\n
\r\n
data(3) err(error reading a body from connection: end of file before message length reached)
[EOF]
server: Ok
== chunk size overflow
HTTP/1.1 200 OK\r\n
content-length: 71\r\n
date: Fri, 23 Oct 2026 13:11:29 GMT\r\n
\r\n
err(error reading a body from connection: invalid chunk size: overflow)
[EOF]
server: Ok
== chunk data without CRLF
HTTP/1.1 200 OK\r\n
content-length: 72\r\n
date: Fri, 23 Oct 2026 13:11:29 GMT\r\n
\r\n
data(3) err(error reading a body from connection: Invalid chunk body CR)
[EOF]
server: Ok
== 16KiB chunk extension
HTTP/1.1 200 OK\r\n
content-length: 70\r\n
date: Fri, 23 Oct 2026 13:11:29 GMT\r\n
\r\n
err(error reading a body from connection: chunk extensions over limit)
[EOF]
server: Ok
== 16KiB trailer
HTTP/1.1 200 OK\r\n
content-length: 82\r\n
date: Fri, 23 Oct 2026 13:11:29 GMT\r\n
\r\n
data(1) err(error reading a body from connection: chunk trailers bytes over limit)
[EOF]
server: Ok

pub(crate) struct Encoder {
    kind: Kind,
    is_last: bool,
}

// ...
enum Kind {
    /// An Encoder for when Transfer-Encoding includes `chunked`.
    Chunked(Option<Vec<HeaderName>>),
    /// An Encoder for when Content-Length is set.
    ///
    /// Enforces that the body is not longer than the Content-Length header.
    Length(u64),
    /// An Encoder for when neither Content-Length nor Chunked encoding is set.
    ///
    /// This is mostly only used with HTTP/1.0 with a length. This kind requires
    /// the connection to be closed when the body is finished.
    #[cfg(feature = "server")]
    CloseDelimited,
}

        let kind = match &mut self.kind {
            Kind::Chunked(_) => {
                trace!("encoding chunked {}B", len);
                let buf = ChunkSize::new(len)
                    .chain(msg)
                    .chain(b"\r\n" as &'static [u8]);
                BufKind::Chunked(buf)
            }
            Kind::Length(remaining) => {
                trace!("sized write, len = {}", len);
                match usize::try_from(*remaining) {
                    // Holding more than is owed, so write only what is left.
                    Ok(limit) if limit < len => {
                        *remaining = 0;
                        BufKind::Limited(msg.take(limit))
                    }
                    // Ok(_) => Owed at least what we hold, write all of it.
                    // Err(_) => Owed more than `usize` can represent, write all of it.
                    Ok(_) | Err(_) => {
                        *remaining -= len as u64;
                        BufKind::Exact(msg)
                    }
                }
            }
            #[cfg(feature = "server")]
            Kind::CloseDelimited => {
                trace!("close delimited write {}B", len);
                BufKind::Exact(msg)
            }
        };

const USIZE_BYTES: usize = std::mem::size_of::<usize>();

// each byte will become 2 hex
const CHUNK_SIZE_MAX_BYTES: usize = USIZE_BYTES * 2;

#[derive(Clone, Copy)]
struct ChunkSize {
    bytes: [u8; CHUNK_SIZE_MAX_BYTES + 2],
    pos: u8,
    len: u8,
}

    pub(crate) fn end<B>(&self) -> Result<Option<EncodedBuf<B>>, NotEof> {
        match self.kind {
            Kind::Length(0) => Ok(None),
            Kind::Chunked(_) => Ok(Some(EncodedBuf {
                kind: BufKind::ChunkedEnd(b"0\r\n\r\n"),
            })),
            #[cfg(feature = "server")]
            Kind::CloseDelimited => Ok(None),
            Kind::Length(n) => Err(NotEof(n)),
        }
    }

== body longer than size_hint
HTTP/1.1 200 OK\r\n
content-length: 3\r\n
date: Fri, 23 Oct 2026 13:11:30 GMT\r\n
\r\n
abc
[still open]
server: Ok

== short body
[EOF]
server: Err(user body write aborted: early end, expected 2 more bytes)

// Claims 5 bytes, delivers 3, and is Pending once before ending.
struct ShortPending(u8);

impl Body for ShortPending {
  type Data = Bytes;
  type Error = Infallible;
  fn poll_frame(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<Frame<Bytes>, Infallible>>> {
    self.0 += 1;
    match self.0 {
      1 => Poll::Ready(Some(Ok(Frame::data(Bytes::from_static(b"abc"))))),
      2 => {
        cx.waker().wake_by_ref();
        Poll::Pending
      }
      _ => Poll::Ready(None),
    }
  }
  fn size_hint(&self) -> SizeHint {
    SizeHint::with_exact(5)
  }
}

== short body, pending once
HTTP/1.1 200 OK\r\n
content-length: 5\r\n
date: Fri, 23 Oct 2026 13:20:53 GMT\r\n
\r\n
abc
[EOF]
server: Err(user body write aborted: early end, expected 2 more bytes)

    pub(crate) fn write_trailers(&mut self, trailers: HeaderMap) {
        if T::is_server() && !self.state.allow_trailer_fields {
            debug!("trailers not allowed to be sent");
            return;
        }

== response trailers, no te
HTTP/1.1 200 OK\r\n
trailer: x-checksum, set-cookie\r\n
transfer-encoding: chunked\r\n
date: Fri, 23 Oct 2026 13:11:29 GMT\r\n
\r\n
3\r\n
abc\r\n
0\r\n
\r\n
[still open]
server: Ok
== response trailers, te: trailers
HTTP/1.1 200 OK\r\n
trailer: x-checksum, set-cookie\r\n
transfer-encoding: chunked\r\n
date: Fri, 23 Oct 2026 13:11:30 GMT\r\n
\r\n
3\r\n
abc\r\n
0\r\n
x-checksum: 42\r\n
\r\n
[still open]
server: Ok

== chunked: ext + 2 chunks + trailer, one write
HTTP/1.1 200 OK\r\n
content-length: 39\r\n
date: Fri, 23 Oct 2026 13:11:28 GMT\r\n
\r\n
data(5) data(6) trailers{"x-sum": "42"}
[still open]
server: Ok