hyper-boring: update to hyper 1.0.0

hyper-boring/CHANGELOG.md

@@ -1,7 +1,11 @@
 # Change Log
 
-## [Unreleased]
+## [v3.0.0]
 
+### Changed
+
+* Upgraded to hyper 1.0.0-rc.3.
+* Removed the `runtime` feature.
 
 ## [v2.1.2] - 2022-09-16
 

hyper-boring/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hyper-boring"
-version = "2.1.2"
+version = "3.0.0"
 authors = ["Steven Fackler <[email protected]>", "Ivan Nikulin <[email protected]>"]
 edition = "2018"
 description = "Hyper TLS support via BoringSSL"
@@ -11,23 +11,24 @@ readme = "README.md"
 exclude = ["test/*"]
 
 [features]
-default = ["runtime"]
-
-runtime = ["hyper/runtime"]
+default = []
 fips = ["tokio-boring/fips"]
 
 [dependencies]
 antidote = "1.0.0"
 http = "0.2"
-hyper = { version = "0.14", default-features = false, features = ["client"] }
+hyper = { version = "1.0.0-rc.3",  default-features = false }
+hyper-util = { git = "https://github.com/howardjohn/hyper-util", branch = "h2-timer-expose-exec", features = ["client"] }
 linked_hash_set = "0.1"
 once_cell = "1.0"
 boring = { version = ">=1.1.0,<3.0.0", path = "../boring" }
 tokio = "1"
 tokio-boring = { version = "2", path = "../tokio-boring" }
 tower-layer = "0.3"
+tower-service = "0.3"
 
 [dev-dependencies]
-hyper = { version = "0.14", features = ["full"] }
 tokio = { version = "1", features = ["full"] }
+hyper-util = { git = "https://github.com/howardjohn/hyper-util", branch = "h2-timer-expose-exec", features = ["full"] }
 futures = "0.3"
+http-body-util = "0.1.0-rc.2"

hyper-boring/src/lib.rs

@@ -9,11 +9,9 @@ use boring::ssl::{
     ConnectConfiguration, Ssl, SslConnector, SslConnectorBuilder, SslMethod, SslSessionCacheMode,
 };
 use http::uri::Scheme;
-use hyper::client::connect::{Connected, Connection};
-#[cfg(feature = "runtime")]
-use hyper::client::HttpConnector;
-use hyper::service::Service;
 use hyper::Uri;
+use hyper_util::client::connect::HttpConnector;
+use hyper_util::client::connect::{Connected, Connection};
 use once_cell::sync::OnceCell;
 use std::fmt::Debug;
 use std::future::Future;
@@ -26,6 +24,7 @@ use std::{error::Error, fmt};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio_boring::SslStream;
 use tower_layer::Layer;
+use tower_service::Service;
 
 mod cache;
 #[cfg(test)]
@@ -147,7 +146,6 @@ pub struct HttpsConnector<T> {
     inner: Inner,
 }
 
-#[cfg(feature = "runtime")]
 impl HttpsConnector<HttpConnector> {
     /// Creates a a new `HttpsConnector` using default settings.
     ///

hyper-boring/src/test.rs

@@ -1,28 +1,27 @@
 use super::*;
 use boring::ssl::{SslAcceptor, SslFiletype, SslMethod};
-use futures::StreamExt;
-use hyper::client::HttpConnector;
-use hyper::server::conn::Http;
+use http_body_util::{BodyExt, Empty, Full};
+use hyper::body::{Body, Bytes};
+use hyper::rt::Sleep;
 use hyper::{service, Response};
-use hyper::{Body, Client};
+use hyper_util::client::connect::{Connect, HttpConnector};
+use hyper_util::client::legacy::Client;
+use hyper_util::rt::TokioExecutor;
+use std::time::{Duration, Instant};
 use tokio::net::TcpListener;
 
 #[tokio::test]
-#[cfg(feature = "runtime")]
 async fn google() {
     let ssl = HttpsConnector::new().unwrap();
-    let client = Client::builder()
-        .pool_max_idle_per_host(0)
-        .build::<_, Body>(ssl);
+    let client = pooling_client::<_, Full<Bytes>>(ssl);
 
     for _ in 0..3 {
         let resp = client
             .get("https://www.google.com".parse().unwrap())
             .await
             .unwrap();
         assert!(resp.status().is_success(), "{}", resp.status());
-        let mut body = resp.into_body();
-        while body.next().await.transpose().unwrap().is_some() {}
+        resp.into_body().collect().await.unwrap();
     }
 }
 
@@ -46,11 +45,12 @@ async fn localhost() {
             let stream = listener.accept().await.unwrap().0;
             let stream = tokio_boring::accept(&acceptor, stream).await.unwrap();
 
-            let service =
-                service::service_fn(|_| async { Ok::<_, io::Error>(Response::new(Body::empty())) });
+            let service = service::service_fn(|_| async {
+                Ok::<_, io::Error>(Response::new(Empty::<Bytes>::new()))
+            });
 
-            Http::new()
-                .http1_keep_alive(false)
+            hyper::server::conn::http1::Builder::new()
+                .keep_alive(false)
                 .serve_connection(stream, service)
                 .await
                 .unwrap();
@@ -72,16 +72,15 @@ async fn localhost() {
     });
 
     let ssl = HttpsConnector::with_connector(connector, ssl).unwrap();
-    let client = Client::builder().build::<_, Body>(ssl);
+    let client = pooling_client::<_, Full<Bytes>>(ssl);
 
     for _ in 0..3 {
         let resp = client
             .get(format!("https://localhost:{}", port).parse().unwrap())
             .await
             .unwrap();
         assert!(resp.status().is_success(), "{}", resp.status());
-        let mut body = resp.into_body();
-        while body.next().await.transpose().unwrap().is_some() {}
+        resp.into_body().collect().await.unwrap();
     }
 }
 
@@ -109,11 +108,11 @@ async fn alpn_h2() {
         let stream = tokio_boring::accept(&acceptor, stream).await.unwrap();
         assert_eq!(stream.ssl().selected_alpn_protocol().unwrap(), b"h2");
 
-        let service =
-            service::service_fn(|_| async { Ok::<_, io::Error>(Response::new(Body::empty())) });
+        let service = service::service_fn(|_| async {
+            Ok::<_, io::Error>(Response::new(Empty::<Bytes>::new()))
+        });
 
-        Http::new()
-            .http2_only(true)
+        hyper::server::conn::http2::Builder::new(TokioExecutor::new())
             .serve_connection(stream, service)
             .await
             .unwrap();
@@ -127,13 +126,75 @@ async fn alpn_h2() {
     ssl.set_alpn_protos(b"\x02h2\x08http/1.1").unwrap();
 
     let ssl = HttpsConnector::with_connector(connector, ssl).unwrap();
-    let client = Client::builder().build::<_, Body>(ssl);
+    let client = pooling_client::<_, Full<Bytes>>(ssl);
 
     let resp = client
         .get(format!("https://localhost:{}", port).parse().unwrap())
         .await
         .unwrap();
     assert!(resp.status().is_success(), "{}", resp.status());
-    let mut body = resp.into_body();
-    while body.next().await.transpose().unwrap().is_some() {}
+    resp.into_body().collect().await.unwrap();
+}
+
+/// A Timer that uses the tokio runtime.
+#[derive(Clone, Debug)]
+pub struct TokioTimer;
+
+impl hyper::rt::Timer for TokioTimer {
+    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>> {
+        let s = tokio::time::sleep(duration);
+        let hs = TokioSleep { inner: Box::pin(s) };
+        Box::pin(hs)
+    }
+
+    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: Box::pin(tokio::time::sleep_until(deadline.into())),
+        })
+    }
+}
+
+struct TokioTimeout<T> {
+    inner: Pin<Box<tokio::time::Timeout<T>>>,
+}
+
+impl<T> Future for TokioTimeout<T>
+where
+    T: Future,
+{
+    type Output = Result<T::Output, tokio::time::error::Elapsed>;
+
+    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
+        self.inner.as_mut().poll(context)
+    }
+}
+
+// Use TokioSleep to get tokio::time::Sleep to implement Unpin.
+// see https://docs.rs/tokio/latest/tokio/time/struct.Sleep.html
+pub(crate) struct TokioSleep {
+    pub(crate) inner: Pin<Box<tokio::time::Sleep>>,
+}
+
+impl Future for TokioSleep {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.inner.as_mut().poll(cx)
+    }
+}
+
+// Use HasSleep to get tokio::time::Sleep to implement Unpin.
+// see https://docs.rs/tokio/latest/tokio/time/struct.Sleep.html
+
+impl Sleep for TokioSleep {}
+
+pub fn pooling_client<C, B>(connector: C) -> Client<C, B>
+where
+    C: Connect + Clone,
+    B: Body + Send,
+    B::Data: Send,
+{
+    Client::builder(TokioExecutor::new())
+        .timer(TokioTimer)
+        .build(connector)
 }