Refactor!: Remove strict TokioIo response requirement from `hyper_b…

…oring::v1::HttpsConnector`

Closes cloudflare#295.

Signed-off-by: Paul Mabileau <[email protected]>

Cargo.toml

@@ -46,7 +46,7 @@ hyper_old = { package = "hyper", version = "0.14", default-features = false }
 linked_hash_set = "0.1"
 once_cell = "1.0"
 openssl-macros = "0.1.1"
-tower = "0.4"
+tower = { version = "0.4", default-features = false, features = ["util"] }
 tower-layer = "0.3"
 tower-service = "0.3"
 autocfg = "1.3.0"

hyper-boring/Cargo.toml

@@ -17,7 +17,7 @@ rustdoc-args = ["--cfg", "docsrs"]
 [features]
 default = ["runtime"]
 
-runtime = ["hyper_old/runtime"]
+runtime = ["hyper_old/runtime", "dep:tower"]
 
 # Use a FIPS-validated version of boringssl.
 fips = ["tokio-boring/fips"]
@@ -43,6 +43,7 @@ once_cell = { workspace = true }
 boring = { workspace = true }
 tokio = { workspace = true }
 tokio-boring = { workspace = true }
+tower = { workspace = true, optional = true }
 tower-layer = { workspace = true }
 tower-service = { workspace = true, optional = true }
 

hyper-boring/src/v1.rs

@@ -19,6 +19,11 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::{io, net};
 use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpStream;
+#[cfg(feature = "runtime")]
+use tower::util::MapResponse;
+#[cfg(feature = "runtime")]
+use tower::ServiceExt;
 use tower_layer::Layer;
 use tower_service::Service;
 
@@ -29,32 +34,41 @@ pub struct HttpsConnector<T> {
     inner: Inner,
 }
 
+/// Specialized version of [`HttpConnector`] with responses wrapped with
+/// [`TokioIo::new`] in order to bring back compatibility with Tokio traits.
+pub type TokioHttpConnector =
+    MapResponse<HttpConnector, fn(TokioIo<TcpStream>) -> TokioIo<TokioIo<TcpStream>>>;
+
 #[cfg(feature = "runtime")]
-impl HttpsConnector<HttpConnector> {
+impl HttpsConnector<TokioHttpConnector> {
     /// Creates a a new `HttpsConnector` using default settings.
     ///
     /// The Hyper `HttpConnector` is used to perform the TCP socket connection. ALPN is configured to support both
     /// HTTP/2 and HTTP/1.1.
     ///
     /// Requires the `runtime` Cargo feature.
-    pub fn new() -> Result<HttpsConnector<HttpConnector>, ErrorStack> {
+    pub fn new() -> Result<Self, ErrorStack> {
         let mut http = HttpConnector::new();
         http.enforce_http(false);
 
-        HttpsLayer::new().map(|l| l.layer(http))
+        HttpsLayer::new().map(|l| l.layer(http.map_response(TokioIo::new as _)))
     }
 }
 
 impl<S, T> HttpsConnector<S>
 where
-    S: Service<Uri, Response = TokioIo<T>> + Send,
+    S: Service<Uri, Response = T> + Send,
     S::Error: Into<Box<dyn Error + Send + Sync>>,
     S::Future: Unpin + Send + 'static,
     T: AsyncRead + AsyncWrite + Connection + Unpin + fmt::Debug + Sync + Send + 'static,
 {
     /// Creates a new `HttpsConnector`.
     ///
     /// The session cache configuration of `ssl` will be overwritten.
+    ///
+    /// If the provided service's response type does not fit the trait
+    /// requirements because it is closer to the Hyper ecosystem than the Tokio
+    /// one, wrapping your responses with [`TokioIo`] should work.
     pub fn with_connector(
         http: S,
         ssl: SslConnectorBuilder,
@@ -215,7 +229,7 @@ impl Inner {
 
 impl<T, S> Service<Uri> for HttpsConnector<S>
 where
-    S: Service<Uri, Response = TokioIo<T>> + Send,
+    S: Service<Uri, Response = T> + Send,
     S::Error: Into<Box<dyn Error + Send + Sync>>,
     S::Future: Unpin + Send + 'static,
     T: AsyncRead + AsyncWrite + Connection + Unpin + fmt::Debug + Sync + Send + 'static,
@@ -244,7 +258,7 @@ where
         let connect = self.http.call(uri);
 
         let f = async {
-            let conn = connect.await.map_err(Into::into)?.into_inner();
+            let conn = connect.await.map_err(Into::into)?;
 
             let (inner, uri) = match tls_setup {
                 Some((inner, uri)) => (inner, uri),

hyper-boring/tests/v1.rs

@@ -12,6 +12,7 @@ use hyper_util::rt::{TokioExecutor, TokioIo};
 use std::convert::Infallible;
 use std::{io, iter};
 use tokio::net::TcpListener;
+use tower::ServiceExt;
 
 #[tokio::test]
 async fn google() {
@@ -83,7 +84,7 @@ async fn localhost() {
         let _ = writeln!(&file, "{}", line);
     });
 
-    let ssl = HttpsConnector::with_connector(connector, ssl).unwrap();
+    let ssl = HttpsConnector::with_connector(connector.map_response(TokioIo::new), ssl).unwrap();
     let client = Client::builder(TokioExecutor::new()).build::<_, Empty<Bytes>>(ssl);
 
     for _ in 0..3 {
@@ -144,7 +145,8 @@ async fn alpn_h2() {
 
     ssl.set_ca_file("tests/test/root-ca.pem").unwrap();
 
-    let mut ssl = HttpsConnector::with_connector(connector, ssl).unwrap();
+    let mut ssl =
+        HttpsConnector::with_connector(connector.map_response(TokioIo::new), ssl).unwrap();
 
     ssl.set_ssl_callback(|ssl, _| ssl.set_alpn_protos(b"\x02h2\x08http/1.1"));