Repository: algesten/ureq
Pull request #73 (2020-06-20 20:02:20)

Update documentation.
Synchronize goals section between README and rustdoc, and add two goals
(blocking API; no unsafe) that were mentioned elsewhere in the README.

Add error handling to examples for the module and for the Response
object.

And a section on synthetic errors to the top-level module documentation.

Related to #55


Patch:
diff --git a/README.md b/README.md
index f66ca1c..9a55d30 100644
--- a/README.md
+++ b/README.md
@@ -13,20 +13,21 @@
 #[macro_use]
 extern crate ureq;
 
-fn main() {
-
-    // sync post request of some json.
-    let resp = ureq::post("https://myapi.acme.com/ingest")
-        .set("X-My-Header", "Secret")
-        .send_json(json!({
-            "name": "martin",
-            "rust": true
-        }));
-
-    // .ok() tells if response is 200-299.
-    if resp.ok() {
-        // ...
-    }
+// sync post request of some json.
+let resp = ureq::post("https://myapi.example.com/ingest")
+    .set("X-My-Header", "Secret")
+    .send_json(json!({
+        "name": "martin",
+        "rust": true
+    }));
+
+// .ok() tells if response is 200-299.
+if resp.ok() {
+  println!("success: {}", resp.into_string()?);
+} else {
+  // This can include errors like failure to parse URL or connect timeout.
+  // They are treated as synthetic HTTP-level error statuses.
+  println!("error {}: {}", resp.status(), resp.into_string()?);
 }
 ```
 
@@ -77,7 +78,9 @@ You can control them when including `ureq` as a dependency.
 
   * Minimal dependency tree
   * Obvious API
-  * Convencience over correctness
+  * Blocking API
+  * Convenience over correctness
+  * No use of unsafe
 
 This library tries to provide a convenient request library with a minimal dependency
 tree and an obvious API. It is inspired by libraries like
@@ -104,8 +107,8 @@ for convenience over correctness, so the decision is left to the user.
 This library uses blocking socket reads and writes. When it was
 created, there wasn't any async/await support in rust, and for my own
 purposes, blocking IO was fine. At this point, one good reason to keep
-this library going is that it is blocking (the other is that it relies
-on very little use of unsafe).
+this library going is that it is blocking (the other is that it does not
+use unsafe).
 
 ## TODO
 
diff --git a/src/agent.rs b/src/agent.rs
index 3d5839e..ad8266e 100644
--- a/src/agent.rs
+++ b/src/agent.rs
@@ -249,11 +249,6 @@ impl Agent {
     pub fn patch(&self, path: &str) -> Request {
         self.request("PATCH", path)
     }
-
-    #[cfg(all(test, any(feature = "tls", feature = "native-tls")))]
-    pub(crate) fn state(&self) -> &Arc<Mutex<Option<AgentState>>> {
-        &self.state
-    }
 }
 
 pub(crate) fn basic_auth(user: &str, pass: &str) -> String {
diff --git a/src/lib.rs b/src/lib.rs
index e922c11..df3e1f2 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -6,23 +6,33 @@
 //!
 //! * Minimal dependency tree
 //! * Obvious API
+//! * Blocking API
+//! * Convenience over correctness
+//! * No use of unsafe
 //!
 //! ```
 //! // requires feature: `ureq = { version = "*", features = ["json"] }`
 //! # #[cfg(feature = "json")] {
 //! use ureq::json;
 //!
-//! // sync post request of some json.
-//! let resp = ureq::post("https://myapi.acme.com/ingest")
-//!     .set("X-My-Header", "Secret")
-//!     .send_json(json!({
-//!         "name": "martin",
-//!         "rust": true
-//!     }));
-//!
-//! // .ok() tells if response is 200-299.
-//! if resp.ok() {
-//!   // ....
+//! fn main() -> std::io::Result<()> {
+//!   // sync post request of some json.
+//!   let resp = ureq::post("https://myapi.example.com/ingest")
+//!       .set("X-My-Header", "Secret")
+//!       .send_json(json!({
+//!           "name": "martin",
+//!           "rust": true
+//!       }));
+//!
+//!   // .ok() tells if response is 200-299.
+//!   if resp.ok() {
+//!     println!("success: {}", resp.into_string()?);
+//!   } else {
+//!     // This can include errors like failure to parse URL or connect timeout.
+//!     // They are treated as synthetic HTTP-level error statuses.
+//!     println!("error {}: {}", resp.status(), resp.into_string()?);
+//!   }
+//!   Ok(())
 //! }
 //! # }
 //! ```
@@ -37,8 +47,10 @@
 //! which follows a build pattern. The builders are finished using:
 //!
 //! * [`.call()`](struct.Request.html#method.call) without a request body.
-//! * [`.send()`](struct.Request.html#method.send) with a request body as `Read`.
+//! * [`.send()`](struct.Request.html#method.send) with a request body as `Read` (chunked encoding).
 //! * [`.send_string()`](struct.Request.html#method.send_string) body as string.
+//! * [`.send_bytes()`](struct.Request.html#method.send_bytes) body as bytes.
+//! * [`.send_form()`](struct.Request.html#method.send_form) key-value pairs as application/x-www-form-urlencoded.
 //!
 //! # JSON
 //!
@@ -91,6 +103,20 @@
 //! we first check if the user has set a `; charset=<whatwg charset>` and attempt
 //! to encode the request body using that.
 //!
+//! # Synthetic errors
+//!
+//! Rather than exposing a custom error type through results, this library has opted for
+//! representing potential connection/TLS/etc errors as HTTP response codes. These invented codes
+//! are called "[synthetic](struct.Response.html#method.synthetic)."
+//!
+//! The idea is that from a library user's point of view the distinction of whether a failure
+//! originated in the remote server (500, 502) etc, or some transient network failure, the code
+//! path of handling that would most often be the same.
+//!
+//! As a result, reading from a Response may yield an error message generated by the ureq library.
+//! To handle these errors, use the
+//! [`response.synthetic_error()`](struct.Response.html#method.synthetic_error) method.
+//!
 
 mod agent;
 mod body;
@@ -181,12 +207,6 @@ pub fn patch(path: &str) -> Request {
     request("PATCH", path)
 }
 
-// Compilation error when both tls and native-tls features are enabled
-#[cfg(all(feature = "tls", feature = "native-tls"))]
-std::compile_error!(
-    "You have both the \"tls\" and \"native-tls\" features enabled on ureq. Please disable one of these features."
-);
-
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/src/pool.rs b/src/pool.rs
index 0b87dab..0fe8a56 100644
--- a/src/pool.rs
+++ b/src/pool.rs
@@ -29,7 +29,7 @@ impl ConnectionPool {
         self.recycle.remove(&PoolKey::new(url))
     }
 
-    #[cfg(all(test, any(feature = "tls", feature = "native-tls")))]
+    #[cfg(test)]
     pub fn len(&self) -> usize {
         self.recycle.len()
     }
diff --git a/src/request.rs b/src/request.rs
index 958208c..7429538 100644
--- a/src/request.rs
+++ b/src/request.rs
@@ -1,5 +1,6 @@
 use std::io::Read;
 use std::sync::{Arc, Mutex};
+use std::time;
 
 use lazy_static::lazy_static;
 use qstring::QString;
@@ -46,6 +47,7 @@ pub struct Request {
     pub(crate) timeout_connect: u64,
     pub(crate) timeout_read: u64,
     pub(crate) timeout_write: u64,
+    pub(crate) timeout: Option<time::Duration>,
     pub(crate) redirects: u32,
     pub(crate) proxy: Option<crate::proxy::Proxy>,
     #[cfg(feature = "tls")]
@@ -212,16 +214,19 @@ impl Request {
 
     /// Send data from a reader.
     ///
-    /// The `Content-Length` header is not set because we can't know the length of the reader.
+    /// This uses [chunked transfer encoding](https://tools.ietf.org/html/rfc7230#section-4.1).
+    /// The caller is responsible for setting the Transfer-Encoding: chunked header.
+    ///
+    /// The input from the reader is buffered into chunks of size 16,384, the max size of a TLS fragment.
     ///
     /// ```
     /// use std::io::Cursor;
     ///
-    /// let text = "Hello there!\n";
-    /// let read = Cursor::new(text.to_string().into_bytes());
+    /// let read = Cursor::new(vec![0x20; 100_000]);
     ///
-    /// let resp = ureq::post("/somewhere")
+    /// let resp = ureq::post("http://localhost/example-upload")
     ///     .set("Content-Type", "text/plain")
+    ///     .set("Transfer-Encoding", "chunked")
     ///     .send(read);
     /// ```
     pub fn send(&mut self, reader: impl Read + 'static) -> Response {
@@ -336,6 +341,8 @@ impl Request {
     }
 
     /// Timeout for the socket connection to be successful.
+    /// If both this and .timeout() are both set, .timeout_connect()
+    /// takes precedence.
     ///
     /// The default is `0`, which means a request can block forever.
     ///
@@ -351,6 +358,8 @@ impl Request {
     }
 
     /// Timeout for the individual reads of the socket.
+    /// If both this and .timeout() are both set, .timeout()
+    /// takes precedence.
     ///
     /// The default is `0`, which means it can block forever.
     ///
@@ -360,12 +369,15 @@ impl Request {
     ///     .call();
     /// println!("{:?}", r);
     /// ```
+    #[deprecated(note = "Please use the timeout() function instead")]
     pub fn timeout_read(&mut self, millis: u64) -> &mut Request {
         self.timeout_read = millis;
         self
     }
 
     /// Timeout for the individual writes to the socket.
+    /// If both this and .timeout() are both set, .timeout()
+    /// takes precedence.
     ///
     /// The default is `0`, which means it can block forever.
     ///
@@ -375,11 +387,32 @@ impl Request {
     ///     .call();
     /// println!("{:?}", r);
     /// ```
+    #[deprecated(note = "Please use the timeout() function instead")]
     pub fn timeout_write(&mut self, millis: u64) -> &mut Request {
         self.timeout_write = millis;
         self
     }
 
+    /// Timeout for the overall request, including DNS resolution, connection
+    /// time, redirects, and reading the response body. Slow DNS resolution
+    /// may cause a request to exceed the timeout, because the DNS request
+    /// cannot be interrupted with the available APIs.
+    ///
+    /// This takes precedence over .timeout_read() and .timeout_write(), but
+    /// not .timeout_connect().
+    ///
+    /// ```
+    /// // wait max 1 second for whole request to complete.
+    /// let r = ureq::get("/my_page")
+    ///     .timeout(std::time::Duration::from_secs(1))
+    ///     .call();
+    /// println!("{:?}", r);
+    /// ```
+    pub fn timeout(&mut self, timeout: time::Duration) -> &mut Request {
+        self.timeout = Some(timeout);
+        self
+    }
+
     /// Basic auth.
     ///
     /// These are the same
diff --git a/src/response.rs b/src/response.rs
index 50e2ab2..3e52ead 100644
--- a/src/response.rs
+++ b/src/response.rs
@@ -1,12 +1,13 @@
 use std::io::{Cursor, Error as IoError, ErrorKind, Read, Result as IoResult};
 use std::str::FromStr;
+use std::time::Instant;
 
 use chunked_transfer::Decoder as ChunkDecoder;
 
 use crate::error::Error;
 use crate::header::Header;
 use crate::pool::PoolReturnRead;
-use crate::stream::Stream;
+use crate::stream::{DeadlineStream, Stream};
 use crate::unit::Unit;
 
 #[cfg(feature = "json")]
@@ -28,8 +29,16 @@ pub const DEFAULT_CHARACTER_SET: &str = "utf-8";
 /// [`into_json_deserialize()`](#method.into_json_deserialize) or
 /// [`into_string()`](#method.into_string) consumes the response.
 ///
+/// All error handling, including URL parse errors and connection errors, is done by mapping onto
+/// [synthetic errors](#method.synthetic). Callers must check response.synthetic_error(),
+/// response.is_ok(), or response.error() before relying on the contents of the reader.
+///
 /// ```
 /// let response = ureq::get("https://www.google.com").call();
+/// if let Some(error) = response.synthetic_error() {
+///     eprintln!("{}", error);
+///     return;
+/// }
 ///
 /// // socket is still open and the response body has not been read.
 ///
@@ -46,6 +55,7 @@ pub struct Response {
     headers: Vec<Header>,
     unit: Option<Unit>,
     stream: Option<Stream>,
+    deadline: Option<Instant>,
 }
 
 /// index into status_line where we split: HTTP/1.1 200 OK
@@ -216,9 +226,11 @@ impl Response {
     /// Example:
     ///
     /// ```
+    /// # #[cfg(feature = "tls")] {
     /// let resp = ureq::get("https://www.google.com/").call();
     /// assert_eq!("text/html; charset=ISO-8859-1", resp.header("content-type").unwrap());
     /// assert_eq!("text/html", resp.content_type());
+    /// # }
     /// ```
     pub fn content_type(&self) -> &str {
         self.header("content-type")
@@ -236,9 +248,11 @@ impl Response {
     /// Example:
     ///
     /// ```
+    /// # #[cfg(feature = "tls")] {
     /// let resp = ureq::get("https://www.google.com/").call();
     /// assert_eq!("text/html; charset=ISO-8859-1", resp.header("content-type").unwrap());
     /// assert_eq!("ISO-8859-1", resp.charset());
+    /// # }
     /// ```
     pub fn charset(&self) -> &str {
         charset_from_content_type(self.header("content-type"))
@@ -255,6 +269,7 @@ impl Response {
     /// Example:
     ///
     /// ```
+    /// # #[cfg(feature = "tls")] {
     /// use std::io::Read;
     ///
     /// let resp =
@@ -270,10 +285,10 @@ impl Response {
     /// reader.read_to_end(&mut bytes);
     ///
     /// assert_eq!(bytes.len(), len);
+    /// # }
     /// ```
     pub fn into_reader(self) -> impl Read {
         //
-
         let is_http10 = self.http_version().eq_ignore_ascii_case("HTTP/1.0");
         let is_close = self
             .header("connection")
@@ -306,6 +321,8 @@ impl Response {
 
         let stream = self.stream.expect("No reader in response?!");
         let unit = self.unit;
+        let deadline = unit.as_ref().and_then(|u| u.deadline);
+        let stream = DeadlineStream::new(stream, deadline);
 
         match (use_chunked, limit_bytes) {
             (true, _) => {
@@ -328,6 +345,7 @@ impl Response {
     /// Example:
     ///
     /// ```
+    /// # #[cfg(feature = "tls")] {
     /// let resp =
     ///     ureq::get("https://ureq.s3.eu-central-1.amazonaws.com/hello_world.json")
     ///         .call();
@@ -335,6 +353,7 @@ impl Response {
     /// let text = resp.into_string().unwrap();
     ///
     /// assert!(text.contains("hello"));
+    /// # }
     /// ```
     ///
     /// ## Charset support
@@ -472,6 +491,7 @@ impl Response {
             headers,
             unit: None,
             stream: None,
+            deadline: None,
         })
     }
 
@@ -551,6 +571,9 @@ impl Into<Response> for Error {
 /// *Internal API*
 pub(crate) fn set_stream(resp: &mut Response, url: String, unit: Option<Unit>, stream: Stream) {
     resp.url = Some(url);
+    if let Some(unit) = &unit {
+        resp.deadline = unit.deadline;
+    }
     resp.unit = unit;
     resp.stream = Some(stream);
 }
@@ -586,7 +609,7 @@ struct LimitedRead<R> {
     position: usize,
 }
 
-impl<R> LimitedRead<R> {
+impl<R: Read> LimitedRead<R> {
     fn new(reader: R, limit: usize) -> Self {
         LimitedRead {
             reader,
@@ -617,7 +640,7 @@ impl<R: Read> Read for LimitedRead<R> {
     }
 }
 
-impl<R> From<LimitedRead<R>> for Stream
+impl<R: Read> From<LimitedRead<R>> for Stream
 where
     Stream: From<R>,
 {
diff --git a/src/stream.rs b/src/stream.rs
index b6b999e..f7b4088 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -35,6 +35,61 @@ pub enum Stream {
     Test(Box<dyn Read + Send>, Vec<u8>),
 }
 
+// DeadlineStream wraps a stream such that read() will return an error
+// after the provided deadline, and sets timeouts on the underlying
+// TcpStream to ensure read() doesn't block beyond the deadline.
+// When the From trait is used to turn a DeadlineStream back into a
+// Stream (by PoolReturningRead), the timeouts are removed.
+pub struct DeadlineStream {
+    stream: Stream,
+    deadline: Option<Instant>,
+}
+
+impl DeadlineStream {
+    pub(crate) fn new(stream: Stream, deadline: Option<Instant>) -> Self {
+        DeadlineStream { stream, deadline }
+    }
+}
+
+impl From<DeadlineStream> for Stream {
+    fn from(deadline_stream: DeadlineStream) -> Stream {
+        // Since we are turning this back into a regular, non-deadline Stream,
+        // remove any timeouts we set.
+        let stream = deadline_stream.stream;
+        if let Some(socket) = stream.socket() {
+            socket.set_read_timeout(None).unwrap();
+            socket.set_write_timeout(None).unwrap();
+        }
+        stream
+    }
+}
+
+impl Read for DeadlineStream {
+    fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+        if let Some(deadline) = self.deadline {
+            let timeout = time_until_deadline(deadline)?;
+            if let Some(socket) = self.stream.socket() {
+                socket.set_read_timeout(Some(timeout))?;
+                socket.set_write_timeout(Some(timeout))?;
+            }
+        }
+        self.stream.read(buf)
+    }
+}
+
+// If the deadline is in the future, return the remaining time until
+// then. Otherwise return a TimedOut error.
+fn time_until_deadline(deadline: Instant) -> IoResult<Duration> {
+    let now = Instant::now();
+    match now.checked_duration_since(deadline) {
+        Some(_) => Err(IoError::new(
+            ErrorKind::TimedOut,
+            "timed out reading response",
+        )),
+        None => Ok(deadline - now),
+    }
+}
+
 impl ::std::fmt::Debug for Stream {
     fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::result::Result<(), ::std::fmt::Error> {
         write!(
@@ -77,10 +132,9 @@ impl Stream {
     }
     // Return true if the server has closed this connection.
     pub(crate) fn server_closed(&self) -> IoResult<bool> {
-        match self {
-            Stream::Http(tcpstream) => Stream::serverclosed_stream(tcpstream),
-            Stream::Https(rustls_stream) => Stream::serverclosed_stream(&rustls_stream.sock),
-            _ => Ok(false),
+        match self.socket() {
+            Some(socket) => Stream::serverclosed_stream(socket),
+            None => Ok(false),
         }
     }
     pub fn is_poolable(&self) -> bool {
@@ -95,6 +149,15 @@ impl Stream {
         }
     }
 
+    pub(crate) fn socket(&self) -> Option<&TcpStream> {
+        match self {
+            Stream::Http(tcpstream) => Some(tcpstream),
+            #[cfg(feature = "tls")]
+            Stream::Https(rustls_stream) => Some(&rustls_stream.sock),
+            _ => None,
+        }
+    }
+
     #[cfg(test)]
     pub fn to_write_vec(&self) -> Vec<u8> {
         match self {
@@ -130,6 +193,7 @@ where
     }
 }
 
+#[cfg(all(feature = "tls", not(feature = "native-tls")))]
 fn read_https(
     stream: &mut StreamOwned<ClientSession, TcpStream>,
     buf: &mut [u8],
@@ -261,7 +325,13 @@ pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
 }
 
 pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<TcpStream, Error> {
-    //
+    let deadline: Option<Instant> = if unit.timeout_connect > 0 {
+        Instant::now().checked_add(Duration::from_millis(unit.timeout_connect))
+    } else {
+        unit.deadline
+    };
+
+    // TODO: Find a way to apply deadline to DNS lookup.
     let sock_addrs: Vec<SocketAddr> = match unit.proxy {
         Some(ref proxy) => format!("{}:{}", proxy.server, proxy.port),
         None => format!("{}:{}", hostname, port),
@@ -282,34 +352,24 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
 
     let mut any_err = None;
     let mut any_stream = None;
-    let mut timeout_connect = unit.timeout_connect;
-    let start_time = Instant::now();
-    let has_timeout = unit.timeout_connect > 0;
-
     // Find the first sock_addr that accepts a connection
     for sock_addr in sock_addrs {
-        // ensure connect timeout isn't hit overall.
-        if has_timeout {
-            let lapsed = (Instant::now() - start_time).as_millis() as u64;
-            if lapsed >= unit.timeout_connect {
-                any_err = Some(IoError::new(ErrorKind::TimedOut, "Didn't connect in time"));
-                break;
-            } else {
-                timeout_connect = unit.timeout_connect - lapsed;
-            }
-        }
+        // ensure connect timeout or overall timeout aren't yet hit.
+        let timeout = match deadline {
+            Some(deadline) => Some(time_until_deadline(deadline)?),
+            None => None,
+        };
 
         // connect with a configured timeout.
         let stream = if Some(Proto::SOCKS5) == proto {
             connect_socks5(
                 unit.proxy.to_owned().unwrap(),
-                timeout_connect,
+                deadline,
                 sock_addr,
                 hostname,
                 port,
             )
-        } else if has_timeout {
-            let timeout = Duration::from_millis(timeout_connect);
+        } else if let Some(timeout) = timeout {
             TcpStream::connect_timeout(&sock_addr, timeout)
         } else {
             TcpStream::connect(&sock_addr)
@@ -332,7 +392,11 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
 
     // rust's absurd api returns Err if we set 0.
     // Setting it to None will disable the native system timeout
-    if unit.timeout_read > 0 {
+    if let Some(deadline) = deadline {
+        stream
+            .set_read_timeout(Some(deadline - Instant::now()))
+            .ok();
+    } else if unit.timeout_read > 0 {
         stream
             .set_read_timeout(Some(Duration::from_millis(unit.timeout_read as u64)))
             .ok();
@@ -340,7 +404,11 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
         stream.set_read_timeout(None).ok();
     }
 
-    if unit.timeout_write > 0 {
+    if let Some(deadline) = deadline {
+        stream
+            .set_write_timeout(Some(deadline - Instant::now()))
+            .ok();
+    } else if unit.timeout_write > 0 {
         stream
             .set_write_timeout(Some(Duration::from_millis(unit.timeout_write as u64)))
             .ok();
@@ -399,7 +467,7 @@ fn socks5_local_nslookup(hostname: &str, port: u16) -> Result<TargetAddr, std::i
 #[cfg(feature = "socks-proxy")]
 fn connect_socks5(
     proxy: Proxy,
-    timeout_connect: u64,
+    deadline: Option<time::Instant>,
     proxy_addr: SocketAddr,
     host: &str,
     port: u16,
@@ -430,7 +498,7 @@ fn connect_socks5(
     // 1) In the event of a timeout, a thread may be left running in the background.
     // TODO: explore supporting timeouts upstream in Socks5Proxy.
     #[allow(clippy::mutex_atomic)]
-    let stream = if timeout_connect > 0 {
+    let stream = if let Some(deadline) = deadline {
         use std::sync::mpsc::channel;
         use std::sync::{Arc, Condvar, Mutex};
         use std::thread;
@@ -455,9 +523,7 @@ fn connect_socks5(
         let (lock, cvar) = &*master_signal;
         let done = lock.lock().unwrap();
 
-        let done_result = cvar
-            .wait_timeout(done, Duration::from_millis(timeout_connect))
-            .unwrap();
+        let done_result = cvar.wait_timeout(done, deadline - Instant::now()).unwrap();
         let done = done_result.0;
         if *done {
             rx.recv().unwrap()?
@@ -504,7 +570,7 @@ fn get_socks5_stream(
 #[cfg(not(feature = "socks-proxy"))]
 fn connect_socks5(
     _proxy: Proxy,
-    _timeout_connect: u64,
+    _deadline: Option<Instant>,
     _proxy_addr: SocketAddr,
     _hostname: &str,
     _port: u16,
diff --git a/src/unit.rs b/src/unit.rs
index 4828cdc..2cca994 100644
--- a/src/unit.rs
+++ b/src/unit.rs
@@ -1,5 +1,6 @@
 use std::io::{Result as IoResult, Write};
 use std::sync::{Arc, Mutex};
+use std::time;
 
 use qstring::QString;
 use url::Url;
@@ -10,11 +11,6 @@ use cookie::{Cookie, CookieJar};
 use crate::agent::AgentState;
 use crate::body::{self, Payload, SizedReader};
 use crate::header;
-#[cfg(any(
-    all(feature = "tls", not(feature = "native-tls")),
-    all(feature = "native-tls", not(feature = "tls")),
-))]
-use crate::stream::connect_https;
 use crate::stream::{self, connect_test, Stream};
 use crate::Proxy;
 use crate::{Error, Header, Request, Response};
@@ -38,6 +34,7 @@ pub(crate) struct Unit {
     pub timeout_connect: u64,
     pub timeout_read: u64,
     pub timeout_write: u64,
+    pub deadline: Option<time::Instant>,
     pub method: String,
     pub proxy: Option<Proxy>,
     #[cfg(feature = "tls")]
@@ -89,6 +86,14 @@ impl Unit {
             .cloned()
             .collect();
 
+        let deadline = match req.timeout {
+            None => None,
+            Some(timeout) => {
+                let now = time::Instant::now();
+                Some(now.checked_add(timeout).unwrap())
+            }
+        };
+
         Unit {
             agent: Arc::clone(&req.agent),
             url: url.clone(),
@@ -98,6 +103,7 @@ impl Unit {
             timeout_connect: req.timeout_connect,
             timeout_read: req.timeout_read,
             timeout_write: req.timeout_write,
+            deadline,
             method: req.method.clone(),
             proxy: req.proxy.clone(),
             #[cfg(feature = "tls")]
@@ -154,6 +160,7 @@ pub(crate) fn connect(
     let body_bytes_sent = body::send_body(body, unit.is_chunked, &mut stream)?;
 
     // start reading the response to process cookies and redirects.
+    let mut stream = stream::DeadlineStream::new(stream, unit.deadline);
     let mut resp = Response::from_read(&mut stream);
 
     if let Some(err) = resp.synthetic_error() {
@@ -208,8 +215,8 @@ pub(crate) fn connect(
     }
 
     // since it is not a redirect, or we're not following redirects,
-    // give away the incoming stream to the response object
-    crate::response::set_stream(&mut resp, unit.url.to_string(), Some(unit), stream);
+    // give away the incoming stream to the response object.
+    crate::response::set_stream(&mut resp, unit.url.to_string(), Some(unit), stream.into());
 
     // release the response
     Ok(resp)
@@ -291,11 +298,7 @@ fn connect_socket(unit: &Unit, use_pooled: bool) -> Result<(Stream, bool), Error
     }
     let stream = match unit.url.scheme() {
         "http" => stream::connect_http(&unit),
-        #[cfg(any(
-            all(feature = "tls", not(feature = "native-tls")),
-            all(feature = "native-tls", not(feature = "tls")),
-        ))]
-        "https" => connect_https(&unit),
+        "https" => stream::connect_https(&unit),
         "test" => connect_test(&unit),
         _ => Err(Error::UnknownScheme(unit.url.scheme().to_string())),
     };


Test patch:
diff --git a/src/test/agent_test.rs b/src/test/agent_test.rs
index 35e335a..55934b9 100644
--- a/src/test/agent_test.rs
+++ b/src/test/agent_test.rs
@@ -1,7 +1,9 @@
 use crate::test;
+use std::io::{BufRead, BufReader, Read, Write};
+use std::thread;
+use std::time::Duration;
 
 use super::super::*;
-use std::thread;
 
 #[test]
 fn agent_reuse_headers() {
@@ -57,8 +59,6 @@ fn agent_cookies() {
 // Start a test server on an available port, that times out idle connections at 2 seconds.
 // Return the port this server is listening on.
 fn start_idle_timeout_server() -> u16 {
-    use std::io::{BufRead, BufReader, Write};
-    use std::time::Duration;
     let listener = std::net::TcpListener::bind("localhost:0").unwrap();
     let port = listener.local_addr().unwrap().port();
     thread::spawn(move || {
@@ -88,9 +88,6 @@ fn start_idle_timeout_server() -> u16 {
 
 #[test]
 fn connection_reuse() {
-    use std::io::Read;
-    use std::time::Duration;
-
     let port = start_idle_timeout_server();
     let url = format!("http://localhost:{}", port);
     let agent = Agent::default().build();
@@ -102,7 +99,7 @@ fn connection_reuse() {
     resp.into_reader().read_to_end(&mut buf).unwrap();
 
     {
-        let mut guard_state = agent.state().lock().unwrap();
+        let mut guard_state = agent.state.lock().unwrap();
         let mut state = guard_state.take().unwrap();
         assert!(state.pool().len() > 0);
     }
diff --git a/src/test/mod.rs b/src/test/mod.rs
index 5d12af7..2baa08f 100644
--- a/src/test/mod.rs
+++ b/src/test/mod.rs
@@ -14,6 +14,7 @@ mod query_string;
 mod range;
 mod redirect;
 mod simple;
+mod timeout;
 
 type RequestHandler = dyn Fn(&Unit) -> Result<Stream, Error> + Send + 'static;
 
diff --git a/src/test/range.rs b/src/test/range.rs
index a73acee..9c4adbc 100644
--- a/src/test/range.rs
+++ b/src/test/range.rs
@@ -39,7 +39,7 @@ fn agent_pool() {
     assert_eq!(len, 1000);
 
     {
-        let mut lock = agent.state().lock().unwrap();
+        let mut lock = agent.state.lock().unwrap();
         let state = lock.as_mut().unwrap();
         let pool = state.pool();
         assert_eq!(pool.len(), 1);
diff --git a/src/test/timeout.rs b/src/test/timeout.rs
new file mode 100644
index 0000000..1ee33e7
--- /dev/null
+++ b/src/test/timeout.rs
@@ -0,0 +1,125 @@
+
+use crate::test;
+use std::io::{self, BufRead, BufReader, Read, Write};
+use std::net::TcpStream;
+use std::thread;
+use std::time::Duration;
+
+use super::super::*;
+
+// Send an HTTP response on the TcpStream at a rate of two bytes every 10
+// milliseconds, for a total of 600 bytes.
+fn dribble_body_respond(stream: &mut TcpStream) -> io::Result<()> {
+    let contents = [b'a'; 300];
+    let headers = format!(
+        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n",
+        contents.len() * 2
+    );
+    stream.write_all(headers.as_bytes())?;
+    for i in 0..contents.len() {
+        stream.write_all(&contents[i..i + 1])?;
+        stream.write_all(&[b'\n'; 1])?;
+        stream.flush()?;
+        thread::sleep(Duration::from_millis(10));
+    }
+    Ok(())
+}
+
+// Read a stream until reaching a blank line, in order to consume
+// request headers.
+fn read_headers(stream: &TcpStream) {
+    for line in BufReader::new(stream).lines() {
+        let line = match line {
+            Ok(x) => x,
+            Err(_) => return,
+        };
+        if line == "" {
+            break;
+        }
+    }
+}
+
+// Start a test server on an available port, that dribbles out a response at 1 write per 10ms.
+// Return the port this server is listening on.
+fn start_dribble_body_server() -> u16 {
+    let listener = std::net::TcpListener::bind("localhost:0").unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let dribble_handler = |mut stream: TcpStream| {
+        read_headers(&stream);
+        if let Err(e) = dribble_body_respond(&mut stream) {
+            eprintln!("sending dribble repsonse: {}", e);
+        }
+    };
+    thread::spawn(move || {
+        for stream in listener.incoming() {
+            thread::spawn(move || dribble_handler(stream.unwrap()));
+        }
+    });
+    port
+}
+
+fn get_and_expect_timeout(url: String) {
+    let agent = Agent::default().build();
+    let timeout = Duration::from_millis(500);
+    let resp = agent.get(&url).timeout(timeout).call();
+
+    let mut reader = resp.into_reader();
+    let mut bytes = vec![];
+    let result = reader.read_to_end(&mut bytes);
+
+    match result {
+        Err(io_error) => match io_error.kind() {
+            io::ErrorKind::WouldBlock => Ok(()),
+            io::ErrorKind::TimedOut => Ok(()),
+            _ => Err(format!("{:?}", io_error)),
+        },
+        Ok(_) => Err("successful response".to_string()),
+    }
+    .expect("expected timeout but got something else");
+}
+
+#[test]
+fn overall_timeout_during_body() {
+    let port = start_dribble_body_server();
+    let url = format!("http://localhost:{}/", port);
+
+    get_and_expect_timeout(url);
+}
+
+// Send HTTP headers on the TcpStream at a rate of one header every 100
+// milliseconds, for a total of 30 headers.
+fn dribble_headers_respond(stream: &mut TcpStream) -> io::Result<()> {
+    stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n")?;
+    for _ in 0..30 {
+        stream.write_all(b"a: b\n")?;
+        stream.flush()?;
+        thread::sleep(Duration::from_millis(100));
+    }
+    Ok(())
+}
+
+// Start a test server on an available port, that dribbles out response *headers* at 1 write per 10ms.
+// Return the port this server is listening on.
+fn start_dribble_headers_server() -> u16 {
+    let listener = std::net::TcpListener::bind("localhost:0").unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let dribble_handler = |mut stream: TcpStream| {
+        read_headers(&stream);
+        if let Err(e) = dribble_headers_respond(&mut stream) {
+            eprintln!("sending dribble repsonse: {}", e);
+        }
+    };
+    thread::spawn(move || {
+        for stream in listener.incoming() {
+            thread::spawn(move || dribble_handler(stream.unwrap()));
+        }
+    });
+    port
+}
+
+#[test]
+fn overall_timeout_during_headers() {
+    let port = start_dribble_headers_server();
+    let url = format!("http://localhost:{}/", port);
+    get_and_expect_timeout(url);
+}