Merge pull request #96 from CryZe/hyper-0.13.0

Implement Support for Hyper 0.13.0

Author: ctz

Cargo.toml

@@ -10,18 +10,18 @@ homepage = "https://github.com/ctz/hyper-rustls"
 repository = "https://github.com/ctz/hyper-rustls"
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.5.2"
 ct-logs = { version = "^0.6.0", optional = true }
-futures-util-preview = { version = "0.3.0-alpha.19" }
-hyper = { version = "0.13.0-alpha.4", default-features = false, features = ["unstable-stream"] }
+futures-util = "0.3.1"
+hyper = { version = "0.13.0", default-features = false, features = ["tcp"] }
 rustls = "0.16"
-tokio-io = { version="0.2.0-alpha.6" }
-tokio-rustls = "0.12.0-alpha.4"
+tokio = "0.2.4"
+tokio-rustls = "0.12.1"
 webpki = "^0.21.0"
 rustls-native-certs = { version = "^0.1.0", optional = true }
 
 [dev-dependencies]
-tokio = "0.2.0-alpha.6"
+tokio = { version = "0.2.4", features = ["io-std", "macros", "dns", "stream"] }
 
 [features]
 default = ["tokio-runtime"]

examples/client.rs

@@ -2,8 +2,7 @@
 //!
 //! First parameter is the mandatory URL to GET.
 //! Second parameter is an optional path to CA store.
-use futures_util::TryStreamExt;
-use hyper::{client, Body, Chunk, Uri};
+use hyper::{body::to_bytes, client, Body, Uri};
 use std::str::FromStr;
 use std::{env, fs, io};
 
@@ -74,8 +73,7 @@ async fn run_client() -> io::Result<()> {
         println!("Headers:\n{:#?}", res.headers());
 
         let body: Body = res.into_body();
-        let body: Chunk = body
-            .try_concat()
+        let body = to_bytes(body)
             .await
             .map_err(|e| error(format!("Could not get body: {:?}", e)))?;
         println!("Body:\n{}", String::from_utf8_lossy(&body));

examples/server.rs

@@ -6,17 +6,16 @@
 //! otherwise HTTP/1.1 will be used.
 use core::task::{Context, Poll};
 use futures_util::{
-    stream::{Stream, StreamExt},
-    try_future::TryFutureExt,
-    try_stream::TryStreamExt,
+    future::TryFutureExt,
+    stream::{Stream, StreamExt, TryStreamExt},
 };
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Request, Response, Server, StatusCode};
 use rustls::internal::pemfile;
 use std::pin::Pin;
-use std::{env, fs, io, sync};
 use std::vec::Vec;
-use tokio::net::tcp::{TcpListener, TcpStream};
+use std::{env, fs, io, sync};
+use tokio::net::{TcpListener, TcpStream};
 use tokio_rustls::server::TlsStream;
 use tokio_rustls::TlsAcceptor;
 
@@ -53,29 +52,26 @@ async fn run_server() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         cfg.set_single_cert(certs, key)
             .map_err(|e| error(format!("{}", e)))?;
         // Configure ALPN to accept HTTP/2, HTTP/1.1 in that order.
-        cfg.set_protocols(&[
-            b"h2".to_vec(),
-            b"http/1.1".to_vec(),
-        ]);
+        cfg.set_protocols(&[b"h2".to_vec(), b"http/1.1".to_vec()]);
         sync::Arc::new(cfg)
     };
 
     // Create a TCP listener via tokio.
-    let tcp = TcpListener::bind(&addr).await?;
+    let mut tcp = TcpListener::bind(&addr).await?;
     let tls_acceptor = TlsAcceptor::from(tls_cfg);
     // Prepare a long-running future stream to accept and serve cients.
-    let incoming_tls_stream: Pin<Box<dyn Stream<Item = Result<TlsStream<TcpStream>, io::Error>>>> =
-        tcp.incoming()
-            .map_err(|e| error(format!("Incoming failed: {:?}", e)))
-            .and_then(move |s| {
-                tls_acceptor.accept(s).map_err(|e| {
-                    println!("[!] Voluntary server halt due to client-connection error...");
-                    // Errors could be handled here, instead of server aborting.
-                    // Ok(None)
-                    error(format!("TLS Error: {:?}", e))
-                })
+    let incoming_tls_stream = tcp
+        .incoming()
+        .map_err(|e| error(format!("Incoming failed: {:?}", e)))
+        .and_then(move |s| {
+            tls_acceptor.accept(s).map_err(|e| {
+                println!("[!] Voluntary server halt due to client-connection error...");
+                // Errors could be handled here, instead of server aborting.
+                // Ok(None)
+                error(format!("TLS Error: {:?}", e))
             })
-            .boxed();
+        })
+        .boxed();
 
     let service = make_service_fn(|_| async { Ok::<_, io::Error>(service_fn(echo)) });
     let server = Server::builder(HyperAcceptor {
@@ -89,11 +85,11 @@ async fn run_server() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     Ok(())
 }
 
-struct HyperAcceptor {
-    acceptor: Pin<Box<dyn Stream<Item = Result<TlsStream<TcpStream>, io::Error>>>>,
+struct HyperAcceptor<'a> {
+    acceptor: Pin<Box<dyn Stream<Item = Result<TlsStream<TcpStream>, io::Error>> + 'a>>,
 }
 
-impl hyper::server::accept::Accept for HyperAcceptor {
+impl hyper::server::accept::Accept for HyperAcceptor<'_> {
     type Conn = TlsStream<TcpStream>;
     type Error = io::Error;
 

src/connector.rs

@@ -1,17 +1,21 @@
 use futures_util::FutureExt;
-use hyper::client::connect::{self, Connect};
 #[cfg(feature = "tokio-runtime")]
-use hyper::client::HttpConnector;
-use rustls::{ClientConfig, Session};
+use hyper::client::connect::HttpConnector;
+use hyper::{client::connect::Connection, service::Service, Uri};
+use rustls::ClientConfig;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};
 use std::{fmt, io};
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_rustls::TlsConnector;
 use webpki::DNSNameRef;
 
 use crate::stream::MaybeHttpsStream;
 
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
 /// A Connector for the `https` scheme.
 #[derive(Clone)]
 pub struct HttpsConnector<T> {
@@ -28,8 +32,8 @@ impl HttpsConnector<HttpConnector> {
         let mut http = HttpConnector::new();
         http.enforce_http(false);
         let mut config = ClientConfig::new();
-        config.root_store = rustls_native_certs::load_native_certs()
-            .expect("cannot access native cert store");
+        config.root_store =
+            rustls_native_certs::load_native_certs().expect("cannot access native cert store");
         config.ct_logs = Some(&ct_logs::LOGS);
         HttpsConnector {
             http,
@@ -69,59 +73,55 @@ impl<T> From<(T, Arc<ClientConfig>)> for HttpsConnector<T> {
     }
 }
 
-impl<T> Connect for HttpsConnector<T>
+impl<T> Service<Uri> for HttpsConnector<T>
 where
-    T: Connect<Error = io::Error>,
-    T::Transport: 'static,
-    T::Future: 'static,
+    T: Service<Uri>,
+    T::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    T::Future: Send + 'static,
+    T::Error: Into<BoxError>,
 {
-    type Transport = MaybeHttpsStream<T::Transport>;
-    type Error = io::Error;
+    type Response = MaybeHttpsStream<T::Response>;
+    type Error = BoxError;
 
     #[allow(clippy::type_complexity)]
-    type Future = Pin<
-        Box<
-            dyn Future<
-                    Output = Result<
-                        (MaybeHttpsStream<T::Transport>, connect::Connected),
-                        io::Error,
-                    >,
-                > + Send,
-        >,
-    >;
-
-    fn connect(&self, dst: connect::Destination) -> Self::Future {
-        let is_https = dst.scheme() == "https";
+    type Future =
+        Pin<Box<dyn Future<Output = Result<MaybeHttpsStream<T::Response>, BoxError>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        match self.http.poll_ready(cx) {
+            Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
+            Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+
+    fn call(&mut self, dst: Uri) -> Self::Future {
+        let is_https = dst.scheme_str() == Some("https");
 
         if !is_https {
-            let connecting_future = self.http.connect(dst);
+            let connecting_future = self.http.call(dst);
 
             let f = async move {
-                let (tcp, conn) = connecting_future.await?;
+                let tcp = connecting_future.await.map_err(Into::into)?;
 
-                Ok((MaybeHttpsStream::Http(tcp), conn))
+                Ok(MaybeHttpsStream::Http(tcp))
             };
             f.boxed()
         } else {
             let cfg = self.tls_config.clone();
-            let hostname = dst.host().to_string();
-            let connecting_future = self.http.connect(dst);
+            let hostname = dst.host().unwrap_or_default().to_string();
+            let connecting_future = self.http.call(dst);
 
             let f = async move {
-                let (tcp, conn) = connecting_future.await?;
+                let tcp = connecting_future.await.map_err(Into::into)?;
                 let connector = TlsConnector::from(cfg);
                 let dnsname = DNSNameRef::try_from_ascii_str(&hostname)
                     .map_err(|_| io::Error::new(io::ErrorKind::Other, "invalid dnsname"))?;
                 let tls = connector
                     .connect(dnsname, tcp)
                     .await
                     .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
-                let connected = if tls.get_ref().1.get_alpn_protocol() == Some(b"h2") {
-                    conn.negotiated_h2()
-                } else {
-                    conn
-                };
-                Ok((MaybeHttpsStream::Https(tls), connected))
+                Ok(MaybeHttpsStream::Https(tls))
             };
             f.boxed()
         }

src/stream.rs

@@ -1,10 +1,14 @@
 // Copied from hyperium/hyper-tls#62e3376/src/stream.rs
 use std::fmt;
 use std::io;
+use std::mem::MaybeUninit;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use tokio_io::{AsyncRead, AsyncWrite};
+use hyper::client::connect::{Connected, Connection};
+
+use rustls::Session;
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_rustls::client::TlsStream;
 
 /// A stream that might be protected with TLS.
@@ -15,6 +19,22 @@ pub enum MaybeHttpsStream<T> {
     Https(TlsStream<T>),
 }
 
+impl<T: AsyncRead + AsyncWrite + Connection + Unpin> Connection for MaybeHttpsStream<T> {
+    fn connected(&self) -> Connected {
+        match self {
+            MaybeHttpsStream::Http(s) => s.connected(),
+            MaybeHttpsStream::Https(s) => {
+                let (tcp, tls) = s.get_ref();
+                if tls.get_alpn_protocol() == Some(b"h2") {
+                    tcp.connected().negotiated_h2()
+                } else {
+                    tcp.connected()
+                }
+            }
+        }
+    }
+}
+
 impl<T: fmt::Debug> fmt::Debug for MaybeHttpsStream<T> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match *self {
@@ -38,7 +58,7 @@ impl<T> From<TlsStream<T>> for MaybeHttpsStream<T> {
 
 impl<T: AsyncRead + AsyncWrite + Unpin> AsyncRead for MaybeHttpsStream<T> {
     #[inline]
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
         match self {
             MaybeHttpsStream::Http(s) => s.prepare_uninitialized_buffer(buf),
             MaybeHttpsStream::Https(s) => s.prepare_uninitialized_buffer(buf),
@@ -86,4 +106,4 @@ impl<T: AsyncWrite + AsyncRead + Unpin> AsyncWrite for MaybeHttpsStream<T> {
             MaybeHttpsStream::Https(s) => Pin::new(s).poll_shutdown(cx),
         }
     }
-}
+}