feat(all): add socket timeouts

Methods added to `Client` and `Server` to control read and write
timeouts of the underlying socket.

Keep-Alive is re-enabled by default on the server, with a default
timeout of 5 seconds.

BREAKING CHANGE: This adds 2 required methods to the `NetworkStream`
  trait, `set_read_timeout` and `set_write_timeout`. Any local
  implementations will need to add them.

Cargo.toml

@@ -47,5 +47,4 @@ env_logger = "*"
 default = ["ssl"]
 ssl = ["openssl", "cookie/secure"]
 serde-serialization = ["serde"]
-timeouts = []
+nightly = []
-nightly = ["timeouts"]

benches/client.rs

@@ -7,7 +7,6 @@ extern crate test;
 use std::fmt;
 use std::io::{self, Read, Write, Cursor};
 use std::net::SocketAddr;
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use hyper::net;
@@ -75,12 +74,10 @@ impl net::NetworkStream for MockStream {
     fn peer_addr(&mut self) -> io::Result<SocketAddr> {
         Ok("127.0.0.1:1337".parse().unwrap())
     }
-    #[cfg(feature = "timeouts")]
     fn set_read_timeout(&self, _: Option<Duration>) -> io::Result<()> {
         // can't time out
         Ok(())
     }
-    #[cfg(feature = "timeouts")]
     fn set_write_timeout(&self, _: Option<Duration>) -> io::Result<()> {
         // can't time out
         Ok(())

src/client/mod.rs

@@ -59,7 +59,6 @@ use std::default::Default;
 use std::io::{self, copy, Read};
 use std::iter::Extend;
 
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use url::UrlParser;
@@ -68,7 +67,7 @@ use url::ParseError as UrlError;
 use header::{Headers, Header, HeaderFormat};
 use header::{ContentLength, Location};
 use method::Method;
-use net::{NetworkConnector, NetworkStream, Fresh};
+use net::{NetworkConnector, NetworkStream};
 use {Url};
 use Error;
 
@@ -89,9 +88,7 @@ use http::h1::Http11Protocol;
 pub struct Client {
     protocol: Box<Protocol + Send + Sync>,
     redirect_policy: RedirectPolicy,
-    #[cfg(feature = "timeouts")]
     read_timeout: Option<Duration>,
-    #[cfg(feature = "timeouts")]
     write_timeout: Option<Duration>,
 }
 
@@ -113,16 +110,6 @@ impl Client {
         Client::with_protocol(Http11Protocol::with_connector(connector))
     }
 
-    #[cfg(not(feature = "timeouts"))]
-    /// Create a new client with a specific `Protocol`.
-    pub fn with_protocol<P: Protocol + Send + Sync + 'static>(protocol: P) -> Client {
-        Client {
-            protocol: Box::new(protocol),
-            redirect_policy: Default::default(),
-        }
-    }
-
-    #[cfg(feature = "timeouts")]
     /// Create a new client with a specific `Protocol`.
     pub fn with_protocol<P: Protocol + Send + Sync + 'static>(protocol: P) -> Client {
         Client {
@@ -139,13 +126,11 @@ impl Client {
     }
 
     /// Set the read timeout value for all requests.
-    #[cfg(feature = "timeouts")]
     pub fn set_read_timeout(&mut self, dur: Option<Duration>) {
         self.read_timeout = dur;
     }
 
     /// Set the write timeout value for all requests.
-    #[cfg(feature = "timeouts")]
     pub fn set_write_timeout(&mut self, dur: Option<Duration>) {
         self.write_timeout = dur;
     }
@@ -273,19 +258,8 @@ impl<'a> RequestBuilder<'a> {
             let mut req = try!(Request::with_message(method.clone(), url.clone(), message));
             headers.as_ref().map(|headers| req.headers_mut().extend(headers.iter()));
 
-            #[cfg(not(feature = "timeouts"))]
-            fn set_timeouts(_req: &mut Request<Fresh>, _client: &Client) -> ::Result<()> {
-                Ok(())
-            }
-
-            #[cfg(feature = "timeouts")]
-            fn set_timeouts(req: &mut Request<Fresh>, client: &Client) -> ::Result<()> {
             try!(req.set_write_timeout(client.write_timeout));
             try!(req.set_read_timeout(client.read_timeout));
-                Ok(())
-            }
-
-            try!(set_timeouts(&mut req, &client));
 
             match (can_have_body, body.as_ref()) {
                 (true, Some(body)) => match body.size() {

src/client/pool.rs

@@ -5,7 +5,6 @@ use std::io::{self, Read, Write};
 use std::net::{SocketAddr, Shutdown};
 use std::sync::{Arc, Mutex};
 
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use net::{NetworkConnector, NetworkStream, DefaultConnector};
@@ -176,13 +175,11 @@ impl<S: NetworkStream> NetworkStream for PooledStream<S> {
         self.inner.as_mut().unwrap().stream.peer_addr()
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.inner.as_ref().unwrap().stream.set_read_timeout(dur)
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.inner.as_ref().unwrap().stream.set_write_timeout(dur)

src/client/request.rs

@@ -2,7 +2,6 @@
 use std::marker::PhantomData;
 use std::io::{self, Write};
 
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use url::Url;
@@ -44,14 +43,12 @@ impl<W> Request<W> {
     pub fn method(&self) -> method::Method { self.method.clone() }
 
     /// Set the write timeout.
-    #[cfg(feature = "timeouts")]
     #[inline]
     pub fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.message.set_write_timeout(dur)
     }
 
     /// Set the read timeout.
-    #[cfg(feature = "timeouts")]
     #[inline]
     pub fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.message.set_read_timeout(dur)

src/http/h1.rs

@@ -4,7 +4,6 @@ use std::cmp::min;
 use std::fmt;
 use std::io::{self, Write, BufWriter, BufRead, Read};
 use std::net::Shutdown;
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use httparse;
@@ -341,13 +340,11 @@ impl HttpMessage for Http11Message {
         }
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.get_ref().set_read_timeout(dur)
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.get_ref().set_write_timeout(dur)

src/http/h2.rs

@@ -4,7 +4,6 @@ use std::io::{self, Write, Read, Cursor};
 use std::net::Shutdown;
 use std::ascii::AsciiExt;
 use std::mem;
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use http::{
@@ -404,13 +403,11 @@ impl<S> HttpMessage for Http2Message<S> where S: CloneableStream {
         true
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_read_timeout(&self, _dur: Option<Duration>) -> io::Result<()> {
         Ok(())
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_write_timeout(&self, _dur: Option<Duration>) -> io::Result<()> {
         Ok(())

src/http/message.rs

@@ -6,9 +6,7 @@ use std::fmt::Debug;
 use std::io::{Read, Write};
 use std::mem;
 
-#[cfg(feature = "timeouts")]
 use std::io;
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use typeable::Typeable;
@@ -65,10 +63,8 @@ pub trait HttpMessage: Write + Read + Send + Any + Typeable + Debug {
     /// the response body.
     fn get_incoming(&mut self) -> ::Result<ResponseHead>;
     /// Set the read timeout duration for this message.
-    #[cfg(feature = "timeouts")]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
     /// Set the write timeout duration for this message.
-    #[cfg(feature = "timeouts")]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
     /// Closes the underlying HTTP connection.
     fn close_connection(&mut self) -> ::Result<()>;

src/mock.rs

@@ -3,9 +3,7 @@ use std::io::{self, Read, Write, Cursor};
 use std::cell::RefCell;
 use std::net::{SocketAddr, Shutdown};
 use std::sync::{Arc, Mutex};
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
-#[cfg(feature = "timeouts")]
 use std::cell::Cell;
 
 use solicit::http::HttpScheme;
@@ -24,9 +22,7 @@ pub struct MockStream {
     pub is_closed: bool,
     pub error_on_write: bool,
     pub error_on_read: bool,
-    #[cfg(feature = "timeouts")]
     pub read_timeout: Cell<Option<Duration>>,
-    #[cfg(feature = "timeouts")]
     pub write_timeout: Cell<Option<Duration>>,
 }
 
@@ -45,7 +41,6 @@ impl MockStream {
         MockStream::with_responses(vec![input])
     }
 
-    #[cfg(feature = "timeouts")]
     pub fn with_responses(mut responses: Vec<&[u8]>) -> MockStream {
         MockStream {
             read: Cursor::new(responses.remove(0).to_vec()),
@@ -58,18 +53,6 @@ impl MockStream {
             write_timeout: Cell::new(None),
         }
     }
-
-    #[cfg(not(feature = "timeouts"))]
-    pub fn with_responses(mut responses: Vec<&[u8]>) -> MockStream {
-        MockStream {
-            read: Cursor::new(responses.remove(0).to_vec()),
-            next_reads: responses.into_iter().map(|arr| arr.to_vec()).collect(),
-            write: vec![],
-            is_closed: false,
-            error_on_write: false,
-            error_on_read: false,
-        }
-    }
 }
 
 impl Read for MockStream {
@@ -111,13 +94,11 @@ impl NetworkStream for MockStream {
         Ok("127.0.0.1:1337".parse().unwrap())
     }
 
-    #[cfg(feature = "timeouts")]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.read_timeout.set(dur);
         Ok(())
     }
 
-    #[cfg(feature = "timeouts")]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.write_timeout.set(dur);
         Ok(())
@@ -167,12 +148,10 @@ impl NetworkStream for CloneableMockStream {
         self.inner.lock().unwrap().peer_addr()
     }
 
-    #[cfg(feature = "timeouts")]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.inner.lock().unwrap().set_read_timeout(dur)
     }
 
-    #[cfg(feature = "timeouts")]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.inner.lock().unwrap().set_write_timeout(dur)
     }

src/net.rs

@@ -8,7 +8,6 @@ use std::mem;
 #[cfg(feature = "openssl")]
 pub use self::openssl::Openssl;
 
-#[cfg(feature = "timeouts")]
 use std::time::Duration;
 
 use typeable::Typeable;
@@ -53,11 +52,9 @@ pub trait NetworkStream: Read + Write + Any + Send + Typeable {
     fn peer_addr(&mut self) -> io::Result<SocketAddr>;
 
     /// Set the maximum time to wait for a read to complete.
-    #[cfg(feature = "timeouts")]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
 
     /// Set the maximum time to wait for a write to complete.
-    #[cfg(feature = "timeouts")]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
 
     /// This will be called when Stream should no longer be kept alive.
@@ -341,13 +338,11 @@ impl NetworkStream for HttpStream {
             self.0.peer_addr()
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.0.set_read_timeout(dur)
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         self.0.set_write_timeout(dur)
@@ -471,7 +466,6 @@ impl<S: NetworkStream> NetworkStream for HttpsStream<S> {
         }
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         match *self {
@@ -480,7 +474,6 @@ impl<S: NetworkStream> NetworkStream for HttpsStream<S> {
         }
     }
 
-    #[cfg(feature = "timeouts")]
     #[inline]
     fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
         match *self {
@@ -580,7 +573,6 @@ mod openssl {
     use std::net::{SocketAddr, Shutdown};
     use std::path::Path;
     use std::sync::Arc;
-    #[cfg(feature = "timeouts")]
     use std::time::Duration;
 
     use openssl::ssl::{Ssl, SslContext, SslStream, SslMethod, SSL_VERIFY_NONE};
@@ -660,13 +652,11 @@ mod openssl {
             self.get_mut().peer_addr()
         }
 
-        #[cfg(feature = "timeouts")]
         #[inline]
         fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
             self.get_ref().set_read_timeout(dur)
         }
 
-        #[cfg(feature = "timeouts")]
         #[inline]
         fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
             self.get_ref().set_write_timeout(dur)

src/server/mod.rs

@@ -147,13 +147,23 @@ pub struct Server<L = HttpListener> {
     timeouts: Timeouts,
 }
 
-#[derive(Clone, Copy, Default, Debug)]
+#[derive(Clone, Copy, Debug)]
 struct Timeouts {
     read: Option<Duration>,
     write: Option<Duration>,
     keep_alive: Option<Duration>,
 }
 
+impl Default for Timeouts {
+    fn default() -> Timeouts {
+        Timeouts {
+            read: None,
+            write: None,
+            keep_alive: Some(Duration::from_secs(5))
+        }
+    }
+}
+
 macro_rules! try_option(
     ($e:expr) => {{
         match $e {
@@ -169,28 +179,29 @@ impl<L: NetworkListener> Server<L> {
     pub fn new(listener: L) -> Server<L> {
         Server {
             listener: listener,
-            timeouts: Timeouts::default(),
+            timeouts: Timeouts::default()
         }
     }
 
-    /// Enables keep-alive for this server.
+    /// Controls keep-alive for this server.
     ///
     /// The timeout duration passed will be used to determine how long
     /// to keep the connection alive before dropping it.
     ///
-    /// **NOTE**: The timeout will only be used when the `timeouts` feature
+    /// Passing `None` will disable keep-alive.
-    /// is enabled for hyper, and rustc is 1.4 or greater.
+    ///
+    /// Default is enabled with a 5 second timeout.
     #[inline]
-    pub fn keep_alive(&mut self, timeout: Duration) {
+    pub fn keep_alive(&mut self, timeout: Option<Duration>) {
-        self.timeouts.keep_alive = Some(timeout);
+        self.timeouts.keep_alive = timeout;
     }
 
-    #[cfg(feature = "timeouts")]
+    /// Sets the read timeout for all Request reads.
     pub fn set_read_timeout(&mut self, dur: Option<Duration>) {
         self.timeouts.read = dur;
     }
 
-    #[cfg(feature = "timeouts")]
+    /// Sets the write timeout for all Response writes.
     pub fn set_write_timeout(&mut self, dur: Option<Duration>) {
         self.timeouts.write = dur;
     }
@@ -296,22 +307,10 @@ impl<H: Handler + 'static> Worker<H> {
         self.set_write_timeout(s, self.timeouts.write)
     }
 
-    #[cfg(not(feature = "timeouts"))]
-    fn set_write_timeout(&self, _s: &NetworkStream, _timeout: Option<Duration>) -> io::Result<()> {
-        Ok(())
-    }
-
-    #[cfg(feature = "timeouts")]
     fn set_write_timeout(&self, s: &NetworkStream, timeout: Option<Duration>) -> io::Result<()> {
         s.set_write_timeout(timeout)
     }
 
-    #[cfg(not(feature = "timeouts"))]
-    fn set_read_timeout(&self, _s: &NetworkStream, _timeout: Option<Duration>) -> io::Result<()> {
-        Ok(())
-    }
-
-    #[cfg(feature = "timeouts")]
     fn set_read_timeout(&self, s: &NetworkStream, timeout: Option<Duration>) -> io::Result<()> {
         s.set_read_timeout(timeout)
     }

src/server/request.rs

@@ -66,18 +66,11 @@ impl<'a, 'b: 'a> Request<'a, 'b> {
     }
 
     /// Set the read timeout of the underlying NetworkStream.
-    #[cfg(feature = "timeouts")]
     #[inline]
     pub fn set_read_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
         self.body.get_ref().get_ref().set_read_timeout(timeout)
     }
 
-    /// Set the read timeout of the underlying NetworkStream.
-    #[cfg(not(feature = "timeouts"))]
-    #[inline]
-    pub fn set_read_timeout(&self, _timeout: Option<Duration>) -> io::Result<()> {
-        Ok(())
-    }
     /// Get a reference to the underlying `NetworkStream`.
     #[inline]
     pub fn downcast_ref<T: NetworkStream>(&self) -> Option<&T> {