http-body-util

impl<D> Full<D>
where
    D: Buf,
{
    /// Create a new `Full`.
    pub fn new(data: D) -> Self {
        let data = if data.has_remaining() {
            Some(data)
        } else {
            None
        };
        Full { data }
    }
}

impl<D> Body for Full<D>
where
    D: Buf,
{
    type Data = D;
    type Error = Infallible;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<D>, Self::Error>>> {
        Poll::Ready(self.data.take().map(|d| Ok(Frame::data(d))))
    }

    fn is_end_stream(&self) -> bool {
        self.data.is_none()
    }

    fn size_hint(&self) -> SizeHint {
        self.data
            .as_ref()
            .map(|data| SizeHint::with_exact(u64::try_from(data.remaining()).unwrap()))
            .unwrap_or_else(|| SizeHint::with_exact(0))
    }
}

impl<T: Body + ?Sized> Future for Collect<T> {
    type Output = Result<crate::Collected<T::Data>, T::Error>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> std::task::Poll<Self::Output> {
        let mut me = self.project();

        loop {
            let frame = ready!(me.body.as_mut().poll_frame(cx));

            let frame = if let Some(frame) = frame {
                frame?
            } else {
                return Poll::Ready(Ok(me.collected.take().expect("polled after complete")));
            };

            me.collected.as_mut().unwrap().push_frame(frame);
        }
    }
}

impl<B> Body for Limited<B>
where
    B: Body,
    B::Error: Into<Box<dyn Error + Send + Sync>>,
{
    type Data = B::Data;
    type Error = Box<dyn Error + Send + Sync>;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        let this = self.project();
        let res = match this.inner.poll_frame(cx) {
            Poll::Pending => return Poll::Pending,
            Poll::Ready(None) => None,
            Poll::Ready(Some(Ok(frame))) => {
                if let Some(data) = frame.data_ref() {
                    if data.remaining() > *this.remaining {
                        *this.remaining = 0;
                        Some(Err(LengthLimitError.into()))
                    } else {
                        *this.remaining -= data.remaining();
                        Some(Ok(frame))
                    }
                } else {
                    Some(Ok(frame))
                }
            }
            Poll::Ready(Some(Err(err))) => Some(Err(err.into())),
        };

        Poll::Ready(res)
    }

    fn size_hint(&self) -> SizeHint {
        use std::convert::TryFrom;
        match u64::try_from(self.remaining) {
            Ok(n) => {
                let mut hint = self.inner.size_hint();
                if hint.lower() >= n {
                    hint.set_exact(n)
                } else if let Some(max) = hint.upper() {
                    hint.set_upper(n.min(max))
                } else {
                    hint.set_upper(n)
                }
                hint
            }
            Err(_) => self.inner.size_hint(),
        }
    }

use std::convert::Infallible;
use http_body_util::{Empty, Full};
use hyper::{body::{Body, Bytes, Incoming}, Request, Response};

async fn handler(req: Request<Incoming>) -> Result<Response<impl Body>, Infallible> {
  if req.uri().path() == "/" {
    Ok(Response::new(Full::new(Bytes::from_static(b"hi"))))
  } else {
    Ok(Response::new(Empty::<Bytes>::new()))
  }
}

error[E0308]: mismatched types
   --> src/bin/box_err.rs:9:22
    |
  9 |     Ok(Response::new(Empty::<Bytes>::new()))
    |        ------------- ^^^^^^^^^^^^^^^^^^^^^ expected `Full<Bytes>`, found `Empty<Bytes>`
    |        |
    |        arguments to this function are incorrect
    |
    = note: expected struct `http_body_util::Full<hyper::body::Bytes>`
               found struct `http_body_util::Empty<hyper::body::Bytes>`

impl<D, E> Body for BoxBody<D, E>
where
    D: Buf,
{
    type Data = D;
    type Error = E;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        self.inner.as_mut().poll_frame(cx)
    }

    fn is_end_stream(&self) -> bool {
        self.inner.is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        self.inner.size_hint()
    }
}

hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
http-body-util = "=0.1.5"
tokio = { version = "=1.54.1", features = ["full"] }
futures = "0.3"

use std::{convert::Infallible, time::Duration};
use futures::stream;
use http_body_util::{BodyExt, Empty, Full, StreamBody, combinators::BoxBody};
use hyper::{
  Request, Response,
  body::{Body, Bytes, Frame, Incoming},
  server::conn::http1,
  service::service_fn,
};
use hyper_util::rt::TokioIo;
use tokio::{
  io::{AsyncReadExt, AsyncWriteExt},
  net::{TcpListener, TcpStream},
  time::timeout,
};

async fn handler(req: Request<Incoming>) -> Result<Response<BoxBody<Bytes, Infallible>>, Infallible> {
  let body = match req.uri().path() {
    "/full" => Full::new(Bytes::from_static(b"hello")).boxed(),
    "/empty" => Empty::<Bytes>::new().boxed(),
    _ => {
      let chunks = ["hel", "lo"].map(|s| Ok::<_, Infallible>(Frame::data(Bytes::from_static(s.as_bytes()))));
      StreamBody::new(stream::iter(chunks)).boxed()
    }
  };
  let hint = body.size_hint();
  println!("{:<8} size_hint exact {:?}, is_end_stream {}", req.uri().path(), hint.exact(), body.is_end_stream());
  Ok(Response::new(body))
}

#[tokio::main]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    http1::Builder::new()
      .auto_date_header(false)
      .serve_connection(TokioIo::new(stream), service_fn(handler))
      .await
      .unwrap();
  });
  let mut tcp = TcpStream::connect(addr).await.unwrap();
  for path in ["/full", "/empty", "/stream"] {
    tcp.write_all(format!("GET {path} HTTP/1.1\r\nHost: x\r\n\r\n").as_bytes()).await.unwrap();
    let mut got = Vec::new();
    let mut buf = [0u8; 1024];
    while let Ok(Ok(n)) = timeout(Duration::from_millis(100), tcp.read(&mut buf)).await {
      got.extend_from_slice(&buf[..n]);
    }
    println!("  wire: {:?}", String::from_utf8_lossy(&got));
  }
}

/full    size_hint exact Some(5), is_end_stream false
  wire: "HTTP/1.1 200 OK\r\ncontent-length: 5\r\n\r\nhello"
/empty   size_hint exact Some(0), is_end_stream true
  wire: "HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n"
/stream  size_hint exact None, is_end_stream false
  wire: "HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n3\r\nhel\r\n2\r\nlo\r\n0\r\n\r\n"

== content-length 1000
  server: Limited size_hint lower 1000 upper Some(1000)
  server: frame 1000 bytes, accepted so far 1000
  client: HTTP/1.1 200 OK | body "1000 bytes" | then [still open]
  server: connection ended Ok(())
== content-length 4096
  server: Limited size_hint lower 1024 upper Some(1024)
  server: error Display "length limit exceeded", Debug LengthLimitError, is LengthLimitError: true
  client: HTTP/1.1 413 Payload Too Large | body "length limit exceeded" | then [still open]
  server: connection ended Ok(())
== chunked, 8 chunks of 512
  server: Limited size_hint lower 0 upper Some(1024)
  server: frame 512 bytes, accepted so far 512
  server: frame 512 bytes, accepted so far 1024
  server: error Display "length limit exceeded", Debug LengthLimitError, is LengthLimitError: true
  server: connection ended Ok(())
  client: HTTP/1.1 413 Payload Too Large | body "length limit exceeded" | then [EOF]