Replace reqwest with hyper + tower

.circleci/config.yml

@@ -102,10 +102,10 @@ jobs:
       - run: cargo test --lib --all -j4
       - run: cargo test --doc --all -j4
       - run: cargo test -j4 -p examples
+      - run: cd examples && cargo build --examples
       - run: cd examples && cargo test --example crd_derive_no_schema --no-default-features --features=native-tls
-      - run: cd kube && cargo test --lib --no-default-features --features=rustls-tls
-      - run: cd kube && cargo test --lib --no-default-features --features=native-tls,ws,ws-native-tls
-      - run: cd kube && cargo test --lib --no-default-features --features=rustls-tls,ws,ws-rustls-tls
+      - run: cd kube && cargo test --lib --no-default-features --features=rustls-tls,ws,oauth
+      - run: cd kube && cargo test --lib --no-default-features --features=native-tls,ws,oauth
       - save_cache:
           paths:
             - /usr/local/cargo/registry

examples/Cargo.toml

@@ -13,9 +13,9 @@ edition = "2018"
 default = ["native-tls", "schema", "kubederive", "ws"]
 kubederive = ["kube/derive"] # by default import kube-derive with its default features
 schema = ["kube-derive/schema"] # crd_derive_no_schema shows how to opt out
-native-tls = ["reqwest/native-tls", "kube/native-tls", "kube-runtime/native-tls"]
-rustls-tls = ["reqwest/rustls-tls", "kube/rustls-tls", "kube-runtime/rustls-tls"]
-ws = ["kube/ws", "kube/ws-native-tls"]
+native-tls = ["kube/native-tls", "kube-runtime/native-tls"]
+rustls-tls = ["kube/rustls-tls", "kube-runtime/rustls-tls"]
+ws = ["kube/ws"]
 
 [dev-dependencies]
 anyhow = "1.0.37"
@@ -34,8 +34,6 @@ color-eyre = "0.5.10"
 snafu = { version = "0.6.10", features = ["futures"] }
 # Some Api::delete methods use Either
 either = "1.6.1"
-# Some configuration tweaking require reqwest atm
-reqwest = { version = "0.11.0", default-features = false, features = ["json", "gzip", "stream"] }
 schemars = "0.8.0"
 static_assertions = "1.1.0"
 
@@ -115,10 +113,6 @@ path = "pod_exec.rs"
 name = "pod_shell"
 path = "pod_shell.rs"
 
-[[example]]
-name = "proxy"
-path = "proxy.rs"
-
 [[example]]
 name = "pod_reflector"
 path = "pod_reflector.rs"

kube/Cargo.toml

@@ -17,18 +17,15 @@ edition = "2018"
 
 [features]
 default = ["native-tls"]
-native-tls = ["openssl", "reqwest/native-tls"]
-rustls-tls = ["rustls", "reqwest/rustls-tls"]
+native-tls = ["openssl", "hyper-tls", "tokio-native-tls"]
+rustls-tls = ["rustls", "hyper-rustls", "tokio-rustls"]
 derive = ["kube-derive"]
 jsonpatch = ["json-patch"]
-ws = ["async-tungstenite", "tokio-util"]
-# We can merge these with `native-tls`/`rustls-tls` when `dep?/feature` syntax becomes stable.
-# https://doc.rust-lang.org/nightly/cargo/reference/unstable.html#weak-dependency-features
-ws-native-tls = ["tokio-native-tls", "async-tungstenite/tokio-native-tls"]
-ws-rustls-tls = ["tokio-rustls", "async-tungstenite/tokio-rustls"]
+ws = ["tokio-tungstenite"]
+oauth = ["tame-oauth"]
 
 [package.metadata.docs.rs]
-features = ["derive", "ws", "ws-native-tls"]
+features = ["derive", "ws"]
 # Define the configuration attribute `docsrs`. Used to enable `doc_cfg` feature.
 rustdoc-args = ["--cfg", "docsrs"]
 
@@ -49,7 +46,7 @@ futures-util = "0.3.8"
 futures = "0.3.8"
 pem = "0.8.2"
 openssl = { version = "0.10.32", optional = true }
-rustls = { version = "0.19.0", optional = true }
+rustls = { version = "0.19.0", features = ["dangerous_configuration"], optional = true }
 tokio-native-tls = { version = "0.3.0", optional = true }
 tokio-rustls = { version = "0.22.0", optional = true }
 bytes = "1.0.0"
@@ -58,19 +55,17 @@ tokio = { version = "1.0.1", features = ["time", "signal", "sync"] }
 static_assertions = "1.1.0"
 kube-derive = { path = "../kube-derive", version = "^0.48.0", optional = true }
 jsonpath_lib = "0.2.6"
-tokio-util = { version = "0.6.0", optional = true, features = ["io"] }
+tokio-util = { version = "0.6.0", features = ["io", "codec"] }
 json-patch = { version = "0.2.6", optional = true }
-
-[dependencies.async-tungstenite]
-version = "0.11.0"
-default-features = false
-features = ["tokio-runtime"]
-optional = true
-
-[dependencies.reqwest]
-version = "0.11.0"
-default-features = false
-features = ["json", "gzip", "stream"]
+hyper = { version = "0.14.2", 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.13.0", optional = true }
+tower = { version = "0.4.4", features = ["buffer", "util"] }
+async-compression = { version = "0.3.7", features = ["gzip", "tokio"] }
+hyper-timeout = "0.4.1"
+tame-oauth = { version = "0.4.7", features = ["gcp"], optional = true }
+pin-project = "1.0.4"
 
 [dependencies.k8s-openapi]
 version = "0.11.0"
@@ -81,6 +76,8 @@ features = []
 tempfile = "3.1.0"
 tokio = { version = "1.0.1", features = ["full"] }
 schemars = "0.8.0"
+tokio-test = "0.4.0"
+tower-test = "0.4.0"
 
 [dev-dependencies.k8s-openapi]
 version = "0.11.0"

kube/src/api/remote_command.rs

@@ -7,13 +7,13 @@ use std::{
 
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::Status;
 
-use async_tungstenite::{tokio::ConnectStream, tungstenite as ws, WebSocketStream};
 use futures::{
     future::Either::{Left, Right},
     SinkExt, StreamExt,
 };
 use futures_util::future::select;
 use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, DuplexStream};
+use tokio_tungstenite::{tungstenite as ws, WebSocketStream};
 
 use super::AttachParams;
 
@@ -46,7 +46,7 @@ pub struct AttachedProcess {
 }
 
 impl AttachedProcess {
-    pub(crate) fn new(stream: WebSocketStream<ConnectStream>, ap: &AttachParams) -> Self {
+    pub(crate) fn new(stream: WebSocketStream<hyper::upgrade::Upgraded>, ap: &AttachParams) -> Self {
         // To simplify the implementation, always create a pipe for stdin.
         // The caller does not have access to it unless they had requested.
         let (stdin_writer, stdin_reader) = tokio::io::duplex(ap.max_stdin_buf_size.unwrap_or(MAX_BUF_SIZE));
@@ -165,7 +165,7 @@ const STATUS_CHANNEL: u8 = 3;
 // const RESIZE_CHANNEL: u8 = 4;
 
 async fn start_message_loop(
-    stream: WebSocketStream<ConnectStream>,
+    stream: WebSocketStream<hyper::upgrade::Upgraded>,
     stdin: impl AsyncRead + Unpin,
     mut stdout: Option<impl AsyncWrite + Unpin>,
     mut stderr: Option<impl AsyncWrite + Unpin>,

kube/src/api/subresource.rs

@@ -389,15 +389,15 @@ impl AttachParams {
 #[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
 impl Resource {
     /// Attach to a pod
-    pub fn attach(&self, name: &str, ap: &AttachParams) -> Result<http::Request<()>> {
+    pub fn attach(&self, name: &str, ap: &AttachParams) -> Result<http::Request<Vec<u8>>> {
         ap.validate()?;
 
         let base_url = self.make_url() + "/" + name + "/" + "attach?";
         let mut qp = url::form_urlencoded::Serializer::new(base_url);
         ap.append_to_url_serializer(&mut qp);
 
         let req = http::Request::get(qp.finish());
-        req.body(()).map_err(Error::HttpError)
+        req.body(vec![]).map_err(Error::HttpError)
     }
 }
 
@@ -446,7 +446,7 @@ where
 #[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
 impl Resource {
     /// Execute command in a pod
-    pub fn exec<I, T>(&self, name: &str, command: I, ap: &AttachParams) -> Result<http::Request<()>>
+    pub fn exec<I, T>(&self, name: &str, command: I, ap: &AttachParams) -> Result<http::Request<Vec<u8>>>
     where
         I: IntoIterator<Item = T>,
         T: Into<String>,
@@ -462,7 +462,7 @@ impl Resource {
         }
 
         let req = http::Request::get(qp.finish());
-        req.body(()).map_err(Error::HttpError)
+        req.body(vec![]).map_err(Error::HttpError)
     }
 }