拼一个客户端栈

tower-http = { version = "=0.7.1", features = ["timeout"] }
tower = { version = "=0.5.3", features = ["util"] }
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http = "1"
http-body-util = "0.1"
bytes = "1"

use bytes::Bytes;
use http::{Request, StatusCode};
use http_body_util::Full;
use hyper_util::{client::legacy::Client, rt::TokioExecutor};
use std::time::Duration;
use tower::{Layer, Service};
use tower_http::timeout::TimeoutLayer;

fn is_service<S: Service<Request<Full<Bytes>>>>(_: &S) {}

fn main() {
  let client = Client::builder(TokioExecutor::new()).build_http::<Full<Bytes>>();
  let svc = TimeoutLayer::with_status_code(StatusCode::GATEWAY_TIMEOUT, Duration::from_millis(100)).layer(client);
  is_service(&svc);
}

error[E0277]: the trait bound `hyper::body::incoming::Incoming: Default` is not satisfied
   --> src/main.rs:14:14
    |
 14 |   is_service(&svc);
    |   ---------- ^^^^ the trait `Default` is not implemented for `hyper::body::incoming::Incoming`
    |   |
    |   required by a bound introduced by this call
    |
help: the trait `Service<http::Request<ReqBody>>` is conditionally implemented for `tower_http::timeout::Timeout<S>`
   --> tower-http-0.7.1/src/timeout/service.rs:99:1
    |
 99 | / impl<S, ReqBody, ResBody> Service<Request<ReqBody>> for Timeout<S>
100 | | where
101 | |     S: Service<Request<ReqBody>, Response = Response<ResBody>>,
102 | |     ResBody: Default,
    | |______________-------^
    |                |
    |                unsatisfied requirement introduced here: `hyper::body::incoming::Incoming: Default`

tower-http = { version = "=0.7.1", features = ["trace", "follow-redirect", "decompression-gzip", "set-header", "auth", "timeout", "map-response-body", "compression-gzip"] }
tower = { version = "=0.5.3", features = ["util"] }
axum = "=0.8.9"
hyper = "1"
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
tokio = { version = "=1.54.1", features = ["macros", "rt", "net", "time"] }
http = "1"
http-body-util = "0.1"
bytes = "1"
tracing = "0.1"

use axum::{extract::Request as AxReq, middleware::{from_fn, Next}, response::Redirect, routing::get, Router};
use bytes::Bytes;
use http::{header, HeaderValue, Request, Response, StatusCode};
use http_body_util::{BodyExt, Full};
use hyper_util::{client::legacy::Client, rt::TokioExecutor};
use std::time::Duration;
use tower::{ServiceBuilder, ServiceExt};
use tower_http::{
  auth::AddAuthorizationLayer,
  classify::{StatusInRangeAsFailures, StatusInRangeFailureClass},
  compression::CompressionLayer,
  decompression::DecompressionLayer,
  follow_redirect::FollowRedirectLayer,
  map_response_body::MapResponseBodyLayer,
  set_header::SetRequestHeaderLayer,
  timeout::TimeoutLayer,
  trace::TraceLayer,
};
use tracing::Span;

// 服务端：打印收到的每个请求，正文用 gzip 压缩
async fn log(req: AxReq, next: Next) -> axum::response::Response {
  let line = {
    let h = |n| req.headers().get(n).map(|v: &HeaderValue| v.to_str().unwrap().to_owned()).unwrap_or("-".into());
    format!("    server: {} {:<6} host={:<9} auth={:<13} ae={}", req.method(), req.uri().path(),
      h(header::HOST).split(':').next().unwrap().to_owned(), h(header::AUTHORIZATION), h(header::ACCEPT_ENCODING))
  };
  println!("{line}");
  next.run(req).await
}

async fn serve() -> u16 {
  let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
  let port = listener.local_addr().unwrap().port();
  let app = Router::new()
    .route("/start", get(|| async { Redirect::to("/data") }))
    .route("/cross", get(move || async move { Redirect::to(&format!("http://localhost:{port}/data")) }))
    .route("/data", get(|| async { "hello from /data ".repeat(4) }))
    .route("/slow", get(|| async { tokio::time::sleep(Duration::from_millis(300)).await; "late" }))
    .layer(CompressionLayer::new())
    .layer(from_fn(log));
  tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
  port
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let port = serve().await;
  let client = Client::builder(TokioExecutor::new()).build_http::<Full<Bytes>>();

  // auth_outside = true：AddAuthorization 在 FollowRedirect 外面
  let stack = |auth_outside: bool| {
    ServiceBuilder::new()
      .layer(
        TraceLayer::new(StatusInRangeAsFailures::new_for_client_and_server_errors().into_make_classifier())
          .on_request(|req: &Request<Full<Bytes>>, _: &Span| println!("  trace: on_request {}", req.uri().path()))
          .on_response(|res: &Response<_>, _: Duration, _: &Span| println!("  trace: on_response {}", res.status()))
          .on_failure(|c: StatusInRangeFailureClass, _: Duration, _: &Span| println!("  trace: on_failure {c}")),
      )
      .option_layer(auth_outside.then(|| AddAuthorizationLayer::bearer("secret")))
      .layer(FollowRedirectLayer::new())
      .layer(DecompressionLayer::new())
      .layer(SetRequestHeaderLayer::overriding(header::USER_AGENT, HeaderValue::from_static("mini/0.1")))
      .option_layer((!auth_outside).then(|| AddAuthorizationLayer::bearer("secret")))
      .layer(TimeoutLayer::with_status_code(StatusCode::GATEWAY_TIMEOUT, Duration::from_millis(100)))
      // hyper 的 Incoming 不实现 Default，Timeout 造不出 504 的 body，先装箱
      .layer(MapResponseBodyLayer::new(BodyExt::boxed_unsync))
      .service(client.clone())
  };

  for (name, auth_outside, path) in [
    ("auth inside FollowRedirect", false, "/cross"),
    ("auth outside FollowRedirect", true, "/cross"),
    ("same origin", true, "/start"),
    ("timeout", true, "/slow"),
  ] {
    println!("[{name}] GET {path}");
    let req = Request::get(format!("http://127.0.0.1:{port}{path}")).body(Full::default()).unwrap();
    let res = stack(auth_outside).oneshot(req).await.unwrap();
    let (status, ce) = (res.status(), res.headers().get(header::CONTENT_ENCODING).cloned());
    let body = res.into_body().collect().await.unwrap().to_bytes();
    println!("  client: {status} content-encoding={ce:?} body={:?}", String::from_utf8_lossy(&body[..body.len().min(17)]));
  }
}

[auth inside FollowRedirect] GET /cross
  trace: on_request /cross
    server: GET /cross host=127.0.0.1 auth=Bearer secret ae=gzip
    server: GET /data  host=localhost auth=Bearer secret ae=gzip
  trace: on_response 200 OK
  client: 200 OK content-encoding=None body="hello from /data "
[auth outside FollowRedirect] GET /cross
  trace: on_request /cross
    server: GET /cross host=127.0.0.1 auth=Bearer secret ae=gzip
    server: GET /data  host=localhost auth=-             ae=gzip
  trace: on_response 200 OK
  client: 200 OK content-encoding=None body="hello from /data "
[same origin] GET /start
  trace: on_request /start
    server: GET /start host=127.0.0.1 auth=Bearer secret ae=gzip
    server: GET /data  host=127.0.0.1 auth=Bearer secret ae=gzip
  trace: on_response 200 OK
  client: 200 OK content-encoding=None body="hello from /data "
[timeout] GET /slow
  trace: on_request /slow
    server: GET /slow  host=127.0.0.1 auth=Bearer secret ae=gzip
  trace: on_response 504 Gateway Timeout
  trace: on_failure Status code: 504 Gateway Timeout
  client: 504 Gateway Timeout content-encoding=None body=""

    fn call(&mut self, mut req: Request<ReqBody>) -> Self::Future {
        req.headers_mut()
            .insert(http::header::AUTHORIZATION, self.value.clone());
        self.inner.call(req)
    }