超时一族

        // Start the `Sleep` if not active.
        let sleep_pinned = if let Some(some) = this.sleep.as_mut().as_pin_mut() {
            some
        } else {
            this.sleep.set(Some(sleep(*this.timeout)));
            this.sleep.as_mut().as_pin_mut().unwrap()
        };

        // Error if the timeout has expired.
        if let Poll::Ready(()) = sleep_pinned.poll(cx) {
            return Poll::Ready(Some(Err(Box::new(TimeoutError(())))));
        }

        // Check for body data.
        let frame = ready!(this.body.poll_frame(cx));
        // A frame is ready. Reset the `Sleep`...
        this.sleep.set(None);

    pub fn new(timeout: Duration, body: B) -> Self {
        DeadlineBody {
            sleep: sleep(timeout),
            body,
        }
    }

tower-http = { version = "=0.7.1", features = ["timeout"] }
tower = { version = "=0.5.3", features = ["util"] }
tokio = { version = "=1.54.1", features = ["macros", "rt", "net", "io-util", "time"] }
hyper = { version = "1", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio", "service"] }
http = "1"
http-body-util = "0.1"
bytes = "1"

async fn hello(_: Request<hyper::body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  Ok(Response::new(Full::from("hello")))
}

// probe!(label, svc)：在 127.0.0.1:0 上 accept 一条连接，用 hyper http1 + TowerToHyperService 服务它；
// 客户端用裸 TcpStream 发 "GET / HTTP/1.1\r\nhost: x\r\nconnection: close\r\n\r\n"，
// 读到 EOF，打印 content-length / transfer-encoding 两行

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let t = Duration::from_secs(5);
  probe!("plain", tower::service_fn(hello));
  probe!("ResponseBodyTimeout", ResponseBodyTimeoutLayer::new(t).layer(tower::service_fn(hello)));
  probe!("ResponseBodyDeadline", ResponseBodyDeadlineLayer::new(t).layer(tower::service_fn(hello)));
}

plain                ["content-length: 5"]
ResponseBodyTimeout  ["transfer-encoding: chunked"]
ResponseBodyDeadline ["content-length: 5"]

tower-http = { version = "=0.7.1", features = ["timeout"] }
tower = { version = "=0.5.3", features = ["util"] }
tokio = { version = "=1.54.1", features = ["macros", "rt", "time", "test-util"] }
http = "1"
http-body = "1"
http-body-util = "0.1"
bytes = "1"
futures = "0.3"

type Body = UnsyncBoxBody<Bytes, BoxError>;

// 每隔 gaps[i] 毫秒发一个字节
fn trickle(gaps: Vec<u64>) -> Body {
  let s = futures::stream::unfold(gaps.into_iter(), |mut it| async move {
    let gap = it.next()?;
    sleep(Duration::from_millis(gap)).await;
    Some((Ok::<_, BoxError>(Frame::data(Bytes::from_static(b"x"))), it))
  });
  StreamBody::new(s).boxed_unsync()
}

// handler 读请求 body，报告结果和读完的时刻；出错时检查 e.is::<TimeoutError>()
async fn read_req<B>(req: Request<B>) -> Result<Response<Full<Bytes>>, Infallible>
where
  B: http_body::Body,
  B::Error: Into<BoxError>,
{ /* ... */ }

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let even = vec![50; 10];
  let gap = vec![50, 50, 150];
  println!("request body");
  for (name, gaps) in [("even 10x50ms", even.clone()), ("one 150ms gap", gap.clone())] {
    let svc = RequestBodyTimeoutLayer::new(Duration::from_millis(100)).layer(tower::service_fn(read_req::<_>));
    // oneshot(Request::new(trickle(gaps.clone())))，打印 handler 的回报
    let svc = RequestBodyDeadlineLayer::new(Duration::from_millis(300)).layer(tower::service_fn(read_req::<_>));
    // 同上
  }

  println!("response body, read after 200 ms");
  let ok = |_: Request<()>| async { Ok::<_, Infallible>(Response::new(Full::<Bytes>::from("hello"))) };
  let res = ResponseBodyTimeoutLayer::new(Duration::from_millis(100)).layer(tower::service_fn(ok)).oneshot(Request::new(())).await.unwrap();
  println!("  timeout  size_hint.exact={:?}", res.body().size_hint().exact());
  sleep(Duration::from_millis(200)).await;
  println!("  timeout  100ms -> {:?}", res.into_body().collect().await.map(|c| c.to_bytes()));
  // ResponseBodyDeadlineLayer 同样走一遍
}

request body
  timeout  100ms, even 10x50ms  -> read 10 bytes at 500 ms
  deadline 300ms, even 10x50ms  -> error (TimeoutError: true): data was not received within the designated timeout at 300 ms
  timeout  100ms, one 150ms gap -> error (TimeoutError: true): data was not received within the designated timeout at 200 ms
  deadline 300ms, one 150ms gap -> read 3 bytes at 250 ms
response body, read after 200 ms
  timeout  size_hint.exact=None
  timeout  100ms -> Ok(b"hello")
  deadline size_hint.exact=Some(5)
  deadline 100ms -> Err("data was not received within the designated timeout")