Relax Service's response body type

Allow using more from the Tower ecosystem.

examples/Cargo.toml

@@ -50,6 +50,7 @@ warp = { version = "0.3", features = ["tls"] }
 http = "0.2.3"
 json-patch = "0.2.6"
 tower = { version = "0.4.6" }
+tower-http = { version = "0.1.0", features = ["trace", "decompression-gzip"] }
 hyper = { version = "0.14.2", features = ["client", "http1", "stream", "tcp"] }
 
 [[example]]

examples/custom_client.rs

@@ -9,6 +9,13 @@ use serde_json::json;
 use tokio_native_tls::TlsConnector;
 use tower::ServiceBuilder;
 
+use tower_http::{
+    decompression::DecompressionLayer,
+    trace::{DefaultMakeSpan, DefaultOnRequest, DefaultOnResponse, TraceLayer},
+    LatencyUnit,
+};
+use tracing::Level;
+
 use kube::{
     api::{Api, DeleteParams, ListParams, PostParams, ResourceExt, WatchEvent},
     service::SetBaseUriLayer,
@@ -17,13 +24,24 @@ use kube::{
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    std::env::set_var("RUST_LOG", "info,kube=debug");
+    std::env::set_var("RUST_LOG", "info,kube=debug,tower_http=trace");
     env_logger::init();
 
     let config = Config::infer().await?;
     let cluster_url = config.cluster_url.clone();
     let common = ServiceBuilder::new()
         .layer(SetBaseUriLayer::new(cluster_url))
+        .layer(DecompressionLayer::new())
+        .layer(
+            TraceLayer::new_for_http()
+                .make_span_with(DefaultMakeSpan::new().include_headers(true))
+                .on_request(DefaultOnRequest::new().level(Level::INFO))
+                .on_response(
+                    DefaultOnResponse::new()
+                        .level(Level::INFO)
+                        .latency_unit(LatencyUnit::Micros),
+                ),
+        )
         .into_inner();
     let mut http = HttpConnector::new();
     http.enforce_http(false);

kube/Cargo.toml

@@ -22,7 +22,7 @@ rustls-tls = ["rustls", "rustls-pemfile", "hyper-rustls", "webpki"]
 ws = ["client", "tokio-tungstenite", "rand", "kube-core/ws"]
 oauth = ["client", "tame-oauth"]
 gzip = ["client", "async-compression"]
-client = ["config", "__non_core", "hyper", "tower", "hyper-timeout", "pin-project", "chrono", "jsonpath_lib", "bytes", "futures", "tokio", "tokio-util", "either"]
+client = ["config", "__non_core", "hyper", "tower", "tower-http", "hyper-timeout", "pin-project", "chrono", "jsonpath_lib", "bytes", "futures", "tokio", "tokio-util", "either"]
 jsonpatch = ["kube-core/jsonpatch"]
 admission = ["kube-core/admission"]
 derive = ["kube-derive"]
@@ -60,11 +60,12 @@ kube-derive = { path = "../kube-derive", version = "^0.55.0", optional = true }
 kube-core = { path = "../kube-core", version = "^0.55.0"}
 jsonpath_lib = { version = "0.3.0", optional = true }
 tokio-util = { version = "0.6.0", optional = true, features = ["io", "codec"] }
-hyper = { version = "0.14.2", optional = true, features = ["client", "http1", "stream", "tcp"] }
+hyper = { version = "0.14.8", optional = true, features = ["client", "http1", "stream", "tcp"] }
 hyper-tls = { version = "0.5.0", optional = true }
 hyper-rustls = { version = "0.22.1", optional = true }
 tokio-tungstenite = { version = "0.14.0", optional = true }
 tower = { version = "0.4.6", optional = true, features = ["buffer", "util"] }
+tower-http = { version = "0.1.0", optional = true, features = ["map-response-body"] }
 async-compression = { version = "0.3.7", features = ["gzip", "tokio"], optional = true }
 hyper-timeout = {version = "0.4.1", optional = true }
 tame-oauth = { version = "0.4.7", features = ["gcp"], optional = true }

kube/src/client/body.rs

@@ -0,0 +1,32 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures::stream::Stream;
+use http_body::Body;
+use pin_project::pin_project;
+
+// Wrap `http_body::Body` to implement `Stream`.
+#[pin_project]
+pub struct IntoStream<B> {
+    #[pin]
+    body: B,
+}
+
+impl<B> IntoStream<B> {
+    pub(crate) fn new(body: B) -> Self {
+        Self { body }
+    }
+}
+
+impl<B> Stream for IntoStream<B>
+where
+    B: Body,
+{
+    type Item = Result<B::Data, B::Error>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.project().body.poll_data(cx)
+    }
+}

kube/src/client/mod.rs

@@ -26,7 +26,8 @@ use tokio_util::{
     codec::{FramedRead, LinesCodec, LinesCodecError},
     io::StreamReader,
 };
-use tower::{buffer::Buffer, util::BoxService, BoxError, Service, ServiceBuilder, ServiceExt};
+use tower::{buffer::Buffer, util::BoxService, BoxError, Layer, Service, ServiceBuilder, ServiceExt};
+use tower_http::map_response_body::MapResponseBodyLayer;
 
 #[cfg(feature = "gzip")]
 use crate::service::{accept_compressed, maybe_decompress};
@@ -37,6 +38,8 @@ use crate::{
     Config, Error, Result,
 };
 
+mod body;
+
 // Binary subprotocol v4. See `Client::connect`.
 #[cfg(feature = "ws")]
 const WS_PROTOCOL: &str = "v4.channel.k8s.io";
@@ -59,22 +62,32 @@ impl Client {
     /// Create and initialize a [`Client`] using the given `Service`.
     ///
     /// Use [`Client::try_from`](Self::try_from) to create with a [`Config`].
-    pub fn new<S, E: Into<BoxError>>(service: S) -> Self
+    pub fn new<S, B>(service: S) -> Self
     where
-        S: Service<Request<Body>, Response = Response<Body>, Error = E> + Send + 'static,
+        S: Service<Request<Body>, Response = Response<B>> + Send + 'static,
         S::Future: Send + 'static,
+        S::Error: Into<BoxError>,
+        B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
+        B::Error: std::error::Error + Send + Sync + 'static,
     {
         Self::new_with_default_ns(service, "default")
     }
 
     /// Create and initialize a [`Client`] using the given `Service` and the default namespace.
-    fn new_with_default_ns<S, E: Into<BoxError>, T: Into<String>>(service: S, default_ns: T) -> Self
+    fn new_with_default_ns<S, B, T: Into<String>>(service: S, default_ns: T) -> Self
     where
-        S: Service<Request<Body>, Response = Response<Body>, Error = E> + Send + 'static,
+        S: Service<Request<Body>, Response = Response<B>> + Send + 'static,
         S::Future: Send + 'static,
+        S::Error: Into<BoxError>,
+        B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
+        B::Error: std::error::Error + Send + Sync + 'static,
     {
+        // Transform response body to `hyper::Body` and use type erased error to avoid type parameters.
+        let service = MapResponseBodyLayer::new(|b| hyper::Body::wrap_stream(body::IntoStream::new(b)))
+            .layer(service)
+            .map_err(|e| e.into());
         Self {
-            inner: Buffer::new(BoxService::new(service.map_err(|e| e.into())), 1024),
+            inner: Buffer::new(BoxService::new(service), 1024),
             default_ns: default_ns.into(),
         }
     }