getting slightly further..

Signed-off-by: clux <[email protected]>

kube-client/Cargo.toml

@@ -23,7 +23,7 @@ ws = ["client", "tokio-tungstenite", "rand", "kube-core/ws", "tokio/macros"]
 oauth = ["client", "tame-oauth"]
 oidc = ["client", "form_urlencoded"]
 gzip = ["client", "tower-http/decompression-gzip"]
-client = ["config", "__non_core", "hyper", "http-body", "hyper-tls", "hyper-util", "tower", "tower-http", "hyper-timeout", "pin-project", "chrono", "jsonpath_lib", "bytes", "futures", "tokio", "tokio-util", "either"]
+client = ["config", "__non_core", "hyper", "http-body", "hyper-tls", "hyper-util", "http-body-util", "tower", "tower-http", "hyper-timeout", "pin-project", "chrono", "jsonpath_lib", "bytes", "futures", "tokio", "tokio-util", "either"]
 jsonpatch = ["kube-core/jsonpatch"]
 admission = ["kube-core/admission"]
 config = ["__non_core", "pem", "home"]
@@ -62,6 +62,7 @@ hyper = { version = "1.0.1", optional = true, features = ["client", "http1"] }
 hyper-rustls = { version = "0.24.0", optional = true }
 hyper-tls = { version = "0.5.0", optional = true }
 hyper-util = { version = "0.1.1", optional = true, features = ["client-legacy", "tokio"] }
+http-body-util = { version = "0.1.0", optional = true }
 hyper-socks2 = { version = "0.8.0", optional = true, default-features = false }
 tokio-tungstenite = { version = "0.20.0", optional = true }
 tower = { version = "0.4.13", optional = true, features = ["buffer", "filter", "util"] }

kube-client/src/client/builder.rs

@@ -1,10 +1,13 @@
 use bytes::Bytes;
 use http::{header::HeaderMap, Request, Response};
-use http_body::Body;
-use hyper::client::conn::http1::{Builder as HyperBuilder, Connection};
+use http_body_util::{combinators::BoxBody, BodyExt, StreamBody};
+use hyper::{
+    body::{Body, Incoming},
+    client::conn::http1::{Builder as HyperBuilder, Connection},
+};
 use hyper_timeout::TimeoutConnector;
 use hyper_tls::HttpsConnector;
-use hyper_util::client::legacy::connect::HttpConnector; //uh private?
+use hyper_util::client::legacy::connect::HttpConnector;
 pub use kube_core::response::Status;
 use std::time::Duration;
 use tokio::io::{AsyncRead, AsyncWrite};
@@ -18,8 +21,8 @@ use crate::{client::ConfigExt, Client, Config, Error, Result};
 
 /// HTTP body of a dynamic backing type.
 ///
-/// The suggested implementation type is [`Body`].
-pub type DynBody = dyn http_body::Body<Data = Bytes, Error = BoxError> + Send + Unpin;
+/// The suggested implementation type is [`hyper::Incoming`].
+pub type DynBody = dyn Body<Data = Bytes, Error = BoxError> + Send + Unpin;
 
 /// Builder for [`Client`] instances with customized [tower](`Service`) middleware.
 pub struct ClientBuilder<Svc> {
@@ -34,7 +37,7 @@ impl<Svc> ClientBuilder<Svc> {
     /// which provides a default stack as a starting point.
     pub fn new(service: Svc, default_namespace: impl Into<String>) -> Self
     where
-        Svc: Service<Request<dyn Body>>,
+        Svc: Service<Request<Incoming>>,
     {
         Self {
             service,
@@ -57,7 +60,7 @@ impl<Svc> ClientBuilder<Svc> {
     /// Build a [`Client`] instance with the current [`Service`] stack.
     pub fn build<B>(self) -> Client
     where
-        Svc: Service<Request<dyn Body>, Response = Response<B>> + Send + 'static,
+        Svc: Service<Request<Incoming>, Response = Response<B>> + Send + 'static,
         Svc::Future: Send + 'static,
         Svc::Error: Into<BoxError>,
         B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
@@ -67,7 +70,7 @@ impl<Svc> ClientBuilder<Svc> {
     }
 }
 
-pub type GenericService = BoxService<Request<dyn Body>, Response<Box<DynBody>>, BoxError>;
+pub type GenericService = BoxService<Request<Incoming>, Response<Box<DynBody>>, BoxError>;
 
 impl TryFrom<Config> for ClientBuilder<GenericService> {
     type Error = Error;

kube-client/src/client/config_ext.rs

@@ -6,8 +6,7 @@ use hyper_util::client::legacy::connect::HttpConnector;
 use secrecy::ExposeSecret;
 use tower::{filter::AsyncFilterLayer, util::Either};
 
-#[cfg(any(feature = "rustls-tls", feature = "openssl-tls"))]
-use super::tls;
+#[cfg(any(feature = "rustls-tls", feature = "openssl-tls"))] use super::tls;
 use super::{
     auth::Auth,
     middleware::{AddAuthorizationLayer, AuthLayer, BaseUriLayer, ExtraHeadersLayer},

kube-client/src/client/mod.rs

@@ -10,7 +10,9 @@
 use either::{Either, Left, Right};
 use futures::{self, AsyncBufRead, StreamExt, TryStream, TryStreamExt};
 use http::{self, Request, Response, StatusCode};
-use hyper::Body;
+use http_body::Body;
+use http_body_util::BodyStream;
+use hyper::body::Incoming;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1 as k8s_meta_v1;
 pub use kube_core::response::Status;
 use serde::de::DeserializeOwned;
@@ -65,7 +67,7 @@ pub use builder::{ClientBuilder, DynBody};
 pub struct Client {
     // - `Buffer` for cheap clone
     // - `BoxService` for dynamic response future type
-    inner: Buffer<BoxService<Request<Body>, Response<Body>, BoxError>, Request<Body>>,
+    inner: Buffer<BoxService<Request<Incoming>, Response<Incoming>, BoxError>, Request<Incoming>>,
     default_ns: String,
 }
 
@@ -99,15 +101,15 @@ impl Client {
     /// ```
     pub fn new<S, B, T>(service: S, default_namespace: T) -> Self
     where
-        S: Service<Request<Body>, Response = Response<B>> + Send + 'static,
+        S: Service<Request<Incoming>, Response = Response<B>> + Send + 'static,
         S::Future: Send + 'static,
         S::Error: Into<BoxError>,
         B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
         B::Error: Into<BoxError>,
         T: Into<String>,
     {
         // Transform response body to `hyper::Body` and use type erased error to avoid type parameters.
-        let service = MapResponseBodyLayer::new(|b: B| Body::wrap_stream(b.into_stream()))
+        let service = MapResponseBodyLayer::new(|b: B| BodyStream::new(b.into_stream()))
             .layer(service)
             .map_err(|e| e.into());
         Self {
@@ -141,7 +143,7 @@ impl Client {
     /// Perform a raw HTTP request against the API and return the raw response back.
     /// This method can be used to get raw access to the API which may be used to, for example,
     /// create a proxy server or application-level gateway between localhost and the API server.
-    pub async fn send(&self, request: Request<Body>) -> Result<Response<Body>> {
+    pub async fn send(&self, request: Request<Incoming>) -> Result<Response<Incoming>> {
         let mut svc = self.inner.clone();
         let res = svc
             .ready()