共享内核：content_encoding 与 compression_utils

// This enum's variants are ordered from least to most preferred.
#[derive(Copy, Clone, Debug, Ord, PartialOrd, PartialEq, Eq)]
pub(crate) enum Encoding {
    #[allow(dead_code)]
    Identity,
    #[cfg(any(feature = "fs", feature = "compression-deflate"))]
    Deflate,
    #[cfg(any(feature = "fs", feature = "compression-gzip"))]
    Gzip,
    #[cfg(any(feature = "fs", feature = "compression-br"))]
    Brotli,
    #[cfg(any(feature = "fs", feature = "compression-zstd"))]
    Zstd,
}

            #[cfg(feature = "fs")]
            #[allow(unreachable_patterns)]
            (true, _) => {
                // This should never happen because the `AcceptEncoding` struct which is used to determine
                // `self.encoding` will only enable the different compression algorithms if the
                // corresponding crate feature has been enabled.
                // ...
                // To safeguard against refactors that changes this relationship or other bugs the
                // server will return an uncompressed response instead of panicking since that could
                // become a ddos attack vector.
                return Poll::Ready(Ok(Response::from_parts(
                    parts,
                    CompressionBody::new(BodyInner::identity(body)),
                )));
            }

        // convert `Body` into a `Stream`
        let stream = BodyIntoStream::new(body);

        // an adapter that converts the error type into `io::Error` while storing the actual error
        // `StreamReader` requires the error type is `io::Error`
        let stream = StreamErrorIntoIoError::<_, B::Error>::new(stream);

        // convert `Stream` into an `AsyncRead`
        let read = StreamReader::new(stream);

        // apply decorator to `AsyncRead` yielding another `AsyncRead`
        let read = M::apply(read, quality);

            Some(Err(err)) => {
                *this.error = Some(err);
                Poll::Ready(Some(Err(io::Error::from_raw_os_error(SENTINEL_ERROR_CODE))))
            }

                    if let Some(body_error) = body_error {
                        return Poll::Ready(Some(Err(body_error.into())));
                    } else if err.raw_os_error() == Some(SENTINEL_ERROR_CODE) {
                        // SENTINEL_ERROR_CODE only gets used when storing
                        // an underlying body error
                        unreachable!()
                    } else if *read_some_data {
                        return Poll::Ready(Some(Err(err.into())));
                    }

tower-http = { version = "=0.7.1", features = ["compression-gzip", "decompression-gzip"] }
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"

#[derive(Debug)]
struct DiskGone;
// impl Display（"disk gone"）与 std::error::Error 略

type Body = UnsyncBoxBody<Bytes, BoxError>;

fn origin_body(path: &str) -> Body {
  let data = |s: &'static str| Ok::<_, BoxError>(Frame::data(Bytes::from(s.repeat(20))));
  match path {
    "/ok" => {
      let mut t = HeaderMap::new();
      t.insert("grpc-status", "0".parse().unwrap());
      StreamBody::new(stream::iter(vec![data("hello "), data("world "), Ok(Frame::trailers(t))])).boxed_unsync()
    }
    _ => StreamBody::new(stream::iter(vec![data("hello "), Err(BoxError::from(DiskGone))])).boxed_unsync(),
  }
}

// read_all：逐帧读，记下每个数据帧的长度；出错时用 e.is::<DiskGone>() / e.is::<io::Error>() 分辨类型

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let server = ServiceBuilder::new()
    .layer(CompressionLayer::new())
    .service(service_fn(|req: Request<Full<Bytes>>| async move {
      Ok::<_, Infallible>(Response::new(origin_body(req.uri().path())))
    }));
  // 客户端：解压层包在「服务端」外面
  let client = ServiceBuilder::new().layer(DecompressionLayer::new()).service(server.clone());
  // 对 /ok、/err：先带 accept-encoding: gzip 直接问 server，再经 client 问一次
  // /bad：上游回 content-encoding: gzip 和明文 "not gzip at all"，经 DecompressionLayer 读
  // 最后用 accept-encoding: GZIP 问一次 server
  // ...
}

/ok
  compressed: content-encoding Some("gzip")
  compressed: data frames [48], trailers Some({"grpc-status": "0"}), end
  round trip: data frames [240], trailers Some({"grpc-status": "0"}), end
/err
  compressed: content-encoding Some("gzip")
  compressed: data frames [], then error (DiskGone): disk gone
  round trip: data frames [], then error (DiskGone): disk gone
/bad
  decompressed: data frames [], then error (io::Error): Invalid gzip header
accept-encoding: GZIP -> content-encoding Some("gzip")

        // poll any remaining frames, such as trailers
        let body = M::get_pin_mut(this.read).get_pin_mut().get_pin_mut();
        match ready!(body.poll_frame(cx)) {
            Some(Ok(frame)) if frame.is_trailers() => /* 原样转发 */,
            Some(Ok(frame)) => match frame.into_data() {
                Ok(data) if data.has_remaining() => Poll::Ready(Some(Err(
                    "there are extra bytes after body has been decompressed".into(),
                ))),
                Ok(mut data) => /* 空数据帧：转发，不结束 */,
                Err(frame) => /* 其他帧：转发 */,
            },
            Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
            None => Poll::Ready(None),
        }