怎么读这一卷

git clone https://github.com/tower-rs/tower-http
cd tower-http
git checkout tower-http-0.7.1

tower-http = { version = "=0.7.1", features = ["trace", "metrics", "on-early-drop", "csrf"] }
tower = { version = "=0.5.3", features = ["util"] }
tokio = { version = "=1.54.1", features = ["macros", "rt"] }
http = "1"
http-body = "1"
http-body-util = "0.1"
bytes = "1"
futures = "0.3"
tracing = "0.1"

use bytes::Bytes;
use futures::stream;
use http::{Request, Response};
use http_body::Frame;
use http_body_util::{combinators::UnsyncBoxBody, BodyExt, Full, StreamBody};
use std::{convert::Infallible, time::Duration};
use tower::{BoxError, ServiceBuilder, ServiceExt};
use tower_http::{
  classify::ServerErrorsFailureClass,
  csrf::CsrfLayer,
  metrics::InFlightRequestsLayer,
  on_early_drop::{OnBodyDropFn, OnEarlyDropLayer},
  trace::TraceLayer,
};
use tracing::Span;

type Body = UnsyncBoxBody<Bytes, BoxError>;

// 最内层：/stream 回两个数据帧，其余回一个
async fn handler(req: Request<Full<Bytes>>) -> Result<Response<Body>, Infallible> {
  println!("    handler {} {}", req.method(), req.uri());
  let body: Body = if req.uri().path() == "/stream" {
    let frames = [Ok::<_, BoxError>(Frame::data(Bytes::from("one"))), Ok(Frame::data(Bytes::from("two")))];
    StreamBody::new(stream::iter(frames)).boxed_unsync()
  } else {
    Full::new(Bytes::from("hello")).map_err(|e| match e {}).boxed_unsync()
  };
  Ok(Response::new(body))
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let (in_flight, counter) = InFlightRequestsLayer::pair();
  let svc = ServiceBuilder::new()
    // 作用点：call 时 / 响应 future 完成时 / body 流动时，各有回调
    .layer(
      TraceLayer::new_for_http()
        .on_request(|_: &Request<Full<Bytes>>, _: &Span| println!("  [call]   trace on_request"))
        .on_response(|res: &Response<_>, _: Duration, _: &Span| println!("  [future] trace on_response {}", res.status()))
        .on_body_chunk(|c: &Bytes, _: Duration, _: &Span| println!("  [body]   trace on_body_chunk {:?}", c))
        .on_eos(|_: Option<&http::HeaderMap>, _: Duration, _: &Span| println!("  [body]   trace on_eos"))
        .on_failure(|c: ServerErrorsFailureClass, _: Duration, _: &Span| println!("  [结局]   trace on_failure {c}")),
    )
    // 结局：计数守卫在 body 里，body 被 drop 才减一
    .layer(in_flight)
    // 结局：body 没到 end-of-stream 就被 drop，回调
    .layer(OnEarlyDropLayer::builder().on_body_drop(OnBodyDropFn::new(|req: &Request<Full<Bytes>>| {
      let path = req.uri().path().to_owned();
      move |_: &http::response::Parts| move || println!("  [结局]   body of {path} dropped early")
    })))
    // 短路：跨站的 POST 不调内层，直接 403
    .layer(CsrfLayer::new())
    .service_fn(handler);

  // 1. 正常读完
  println!("GET /ok");
  let res = svc.clone().oneshot(Request::get("/ok").body(Full::default()).unwrap()).await.unwrap();
  println!("  in flight after response head: {}", counter.get());
  let body = res.into_body().collect().await.unwrap().to_bytes();
  println!("  client read {:?}; in flight: {}", body, counter.get());

  // 2. 短路
  println!("POST /form (Sec-Fetch-Site: cross-site)");
  let req = Request::post("/form").header("sec-fetch-site", "cross-site").body(Full::default()).unwrap();
  let res = svc.clone().oneshot(req).await.unwrap();
  drop(res);
  println!("  in flight: {}", counter.get());

  // 3. 读一帧就丢掉
  println!("GET /stream (read one frame, then drop)");
  let res = svc.clone().oneshot(Request::get("/stream").body(Full::default()).unwrap()).await.unwrap();
  let mut body = res.into_body();
  let first = body.frame().await.unwrap().unwrap();
  println!("  client read {:?}; in flight: {}", first.into_data().unwrap(), counter.get());
  drop(body);
  println!("  in flight: {}", counter.get());
}

GET /ok
  [call]   trace on_request
    handler GET /ok
  [future] trace on_response 200 OK
  in flight after response head: 1
  [body]   trace on_body_chunk b"hello"
  [body]   trace on_eos
  client read b"hello"; in flight: 0
POST /form (Sec-Fetch-Site: cross-site)
  [call]   trace on_request
  [future] trace on_response 403 Forbidden
  in flight: 0
GET /stream (read one frame, then drop)
  [call]   trace on_request
    handler GET /stream
  [future] trace on_response 200 OK
  [body]   trace on_body_chunk b"one"
  client read b"one"; in flight: 1
  [结局]   body of /stream dropped early
  in flight: 0