Hyper 0.13 support

Fixes #9

Author: hjr3

Cargo.toml

@@ -7,18 +7,18 @@ license = "MIT/Apache-2.0"
 documentation = "https://github.com/hjr3/hyper-timeout"
 homepage = "https://github.com/hjr3/hyper-timeout"
 repository = "https://github.com/hjr3/hyper-timeout"
+edition = "2018"
 
 [badges]
 travis-ci = { repository = "https://github.com/hjr3/hyper-timeout", branch = "master" }
 
 [dependencies]
-futures = "0.1"
-hyper = "0.12"
-tokio = "0.1"
+bytes = "0.5"
+hyper = { version = "0.13", default-features = false, features = ["tcp"] }
+tokio = "0.2"
 tokio-io = "0.1"
-tokio-io-timeout = "0.3"
-tokio-reactor = "0.1"
-tokio-service = "0.1"
+tokio-io-timeout = "0.4"
 
 [dev-dependencies]
-hyper-tls = "0.3"
+hyper-tls = "0.4"
+tokio = { version = "0.2", features = ["io-std", "macros"] }

README.md

@@ -19,21 +19,16 @@ There is a `TimeoutConnector` that implements the `hyper::Connect` trait. This c
 
 Hyper version compatibility:
 
-* The `0.1` release supports hyper 0.11.
+* The `master` branch will track on going development for hyper.
+* The `0.3` release supports hyper 0.13.
 * The `0.2` release supports hyper 0.12.
-* The `master` branch will track on going developer for hyper 0.13.
+* The `0.1` release supports hyper 0.11.
 
-First, (assuming you are using hyper 0.12) add this to your `Cargo.toml`:
+First, (assuming you are using hyper 0.13) add this to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-hyper-timeout = "0.2"
-```
-
-Next, add this to your crate:
-
-```rust
-extern crate hyper_timeout;
+hyper-timeout = "0.3"
 ```
 
 See the [client example](./examples/client.rs) for a working example.

examples/client.rs

@@ -1,61 +1,46 @@
-extern crate futures;
-extern crate hyper;
-extern crate hyper_tls;
-extern crate hyper_timeout;
-
 use std::env;
-use std::io::{self, Write};
 use std::time::Duration;
 
-use futures::Future;
-use futures::stream::Stream;
-
-use hyper::{rt, Client};
+use hyper::{Client, body::HttpBody as _};
+use tokio::io::{self, AsyncWriteExt as _};
 
 //use hyper::client::HttpConnector;
 use hyper_tls::HttpsConnector;
 
 use hyper_timeout::TimeoutConnector;
 
-fn main() {
-
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let url = match env::args().nth(1) {
         Some(url) => url,
         None => {
             println!("Usage: client <url>");
             println!("Example: client https://example.com");
-            return;
+            return Ok(());
         }
     };
 
     let url = url.parse::<hyper::Uri>().unwrap();
 
-    rt::run(rt::lazy(|| {
-        // This example uses `HttpsConnector`, but you can also use hyper `HttpConnector`
-        //let connector = HttpConnector::new(1);
-        let https = HttpsConnector::new(1).unwrap();
-        let mut connector = TimeoutConnector::new(https);
-        connector.set_connect_timeout(Some(Duration::from_secs(5)));
-        connector.set_read_timeout(Some(Duration::from_secs(5)));
-        connector.set_write_timeout(Some(Duration::from_secs(5)));
-        let client = Client::builder().build::<_, hyper::Body>(connector);
-
-        client.get(url).and_then(|res| {
-            println!("Response: {}", res.status());
-
-            res
-                .into_body()
-                // Body is a stream, so as each chunk arrives...
-                .for_each(|chunk| {
-                    io::stdout()
-                        .write_all(&chunk)
-                        .map_err(|e| {
-                            panic!("example expects stdout is open, error={}", e)
-                        })
-                })
-        })
-        .map_err(|err| {
-            println!("Error: {}", err);
-        })
-    }));
+    // This example uses `HttpsConnector`, but you can also use hyper `HttpConnector`
+    //let http = HttpConnector::new();
+    let https = HttpsConnector::new();
+    let mut connector = TimeoutConnector::new(https);
+    connector.set_connect_timeout(Some(Duration::from_secs(5)));
+    connector.set_read_timeout(Some(Duration::from_secs(5)));
+    connector.set_write_timeout(Some(Duration::from_secs(5)));
+    let client = Client::builder().build::<_, hyper::Body>(connector);
+
+    let mut res = client.get(url).await?;
+
+    println!("Status: {}", res.status());
+    println!("Headers:\n{:#?}", res.headers());
+
+    while let Some(chunk) = res.body_mut().data().await {
+        let chunk = chunk?;
+        io::stdout()
+            .write_all(&chunk)
+            .await?
+    }
+    Ok(())
 }

src/lib.rs

@@ -1,22 +1,24 @@
-extern crate futures;
-extern crate tokio;
-extern crate tokio_io;
-extern crate tokio_service;
-extern crate tokio_io_timeout;
-extern crate hyper;
-
-use std::time::Duration;
+use std::future::Future;
 use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::Duration;
 
-use futures::Future;
-
-use tokio::timer::Timeout;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::time::timeout;
 use tokio_io_timeout::TimeoutStream;
 
-use hyper::client::connect::{Connect, Connected, Destination};
+use hyper::{service::Service, Uri};
+use hyper::client::connect::{Connect, Connected, Connection};
+
+mod stream;
+
+use stream::TimeoutConnectorStream;
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
 
 /// A connector that enforces as connection timeout
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct TimeoutConnector<T> {
     /// A connector implementing the `Connect` trait
     connector: T,
@@ -40,42 +42,57 @@ impl<T: Connect> TimeoutConnector<T> {
     }
 }
 
-impl<T: Connect> Connect for TimeoutConnector<T>
+impl<T> Service<Uri> for TimeoutConnector<T>
 where
-    T: Connect<Error = io::Error> + 'static,
-    T::Future: 'static,
+    T: Service<Uri>,
+    T::Response: AsyncRead + AsyncWrite + Send + Unpin,
+    T::Future: Send + 'static,
+    T::Error: Into<BoxError>,
 {
-    type Transport = TimeoutStream<T::Transport>;
-    type Error = T::Error;
-    type Future = Box<Future<Item = (Self::Transport, Connected), Error = Self::Error> + Send>;
-
-    fn connect(&self, dst: Destination) -> Self::Future {
+    type Response = TimeoutConnectorStream<T::Response>;
+    type Error = BoxError;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        match self.connector.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 read_timeout = self.read_timeout.clone();
         let write_timeout = self.write_timeout.clone();
-        let connecting = self.connector.connect(dst);
+        let connecting = self.connector.call(dst);
 
         if self.connect_timeout.is_none() {
-            return Box::new(connecting.map(move |(io, c)| {
-                let mut tm = TimeoutStream::new(io);
+            let fut = async move {
+                let io = connecting.await.map_err(Into::into)?;
+
+                let mut tm = TimeoutConnectorStream::new(TimeoutStream::new(io));
                 tm.set_read_timeout(read_timeout);
                 tm.set_write_timeout(write_timeout);
-                (tm, c)
-            }));
+                Ok(tm)
+            };
+
+            return Box::pin(fut);
         }
 
         let connect_timeout = self.connect_timeout.expect("Connect timeout should be set");
-        let timeout = Timeout::new(connecting, connect_timeout);
+        let timeout = timeout(connect_timeout, connecting);
 
-        Box::new(timeout.then(move |res| match res {
-            Ok((io, c)) => {
-                let mut tm = TimeoutStream::new(io);
-                tm.set_read_timeout(read_timeout);
-                tm.set_write_timeout(write_timeout);
-                Ok((tm, c))
-            }
-            Err(e) => Err(io::Error::new(io::ErrorKind::TimedOut, e)),
-        }))
+        let fut = async move {
+            let connecting = timeout.await.map_err(|e| io::Error::new(io::ErrorKind::TimedOut, e))?;
+            let io = connecting.map_err(Into::into)?;
+
+            let mut tm = TimeoutConnectorStream::new(TimeoutStream::new(io));
+            tm.set_read_timeout(read_timeout);
+            tm.set_write_timeout(write_timeout);
+            Ok(tm)
+        };
+
+        Box::pin(fut)
     }
 }
 
@@ -105,32 +122,35 @@ impl<T> TimeoutConnector<T> {
     }
 }
 
+impl<T> Connection for TimeoutConnector<T> {
+    fn connected(&self) -> Connected {
+        Connected::new()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::error::Error;
     use std::io;
     use std::time::Duration;
-    use futures::future;
-    use tokio::runtime::current_thread::Runtime;
+
     use hyper::Client;
     use hyper::client::HttpConnector;
+
     use super::TimeoutConnector;
 
-    #[test]
-    fn test_timeout_connector() {
-        let mut rt = Runtime::new().unwrap();
-        let res = rt.block_on(future::lazy(|| {
-            // 10.255.255.1 is a not a routable IP address
-            let url = "http://10.255.255.1".parse().unwrap();
+    #[tokio::test]
+    async fn test_timeout_connector() {
+        // 10.255.255.1 is a not a routable IP address
+        let url = "http://10.255.255.1".parse().unwrap();
 
-            let http = HttpConnector::new(1);
-            let mut connector = TimeoutConnector::new(http);
-            connector.set_connect_timeout(Some(Duration::from_millis(1)));
+        let http = HttpConnector::new();
+        let mut connector = TimeoutConnector::new(http);
+        connector.set_connect_timeout(Some(Duration::from_millis(1)));
 
-            let client = Client::builder().build::<_, hyper::Body>(connector);
+        let client = Client::builder().build::<_, hyper::Body>(connector);
 
-            client.get(url)
-        }));
+        let res = client.get(url).await;
 
         match res {
             Ok(_) => panic!("Expected a timeout"),
@@ -144,21 +164,18 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_read_timeout() {
-        let mut rt = Runtime::new().unwrap();
-        let res = rt.block_on(future::lazy(|| {
-            let url = "http://example.com".parse().unwrap();
+    #[tokio::test]
+    async fn test_read_timeout() {
+        let url = "http://example.com".parse().unwrap();
 
-            let http = HttpConnector::new(1);
-            let mut connector = TimeoutConnector::new(http);
-            // A 1 ms read timeout should be so short that we trigger a timeout error
-            connector.set_read_timeout(Some(Duration::from_millis(1)));
+        let http = HttpConnector::new();
+        let mut connector = TimeoutConnector::new(http);
+        // A 1 ms read timeout should be so short that we trigger a timeout error
+        connector.set_read_timeout(Some(Duration::from_millis(1)));
 
-            let client = Client::builder().build::<_, hyper::Body>(connector);
+        let client = Client::builder().build::<_, hyper::Body>(connector);
 
-            client.get(url)
-        }));
+        let res = client.get(url).await;
 
         match res {
             Ok(_) => panic!("Expected a timeout"),