Repository: algesten/ureq
Pull request #86 (2020-06-23 18:47:18)

Set chunked Transfer-Encoding when using `send`
Related issue: #43 

Previously, using `.send()` on `Request` would require to set either the Transfer-Encoding or the Content-Length header.

In an effort to provide better ergonomics for this library and to avoid making users fall in a not-so obvious pitfall, the library should build a valid Request without asking the user to mess around with the headers.

This PR attempts to fix this issue: if the user use `.send()` to provide an unknown sized reader, the chunked Transfer-Encoding will be used. Of course, there are prior checks to ensure we do not override the user wish, like if the user already set a Content-Length or a Transfer-Encoding.

This PR also fix an edge case where the Content-Length would not be automatically set if the Transfer-Encoding is set but not chunked.

Deluvi: I have three remarks about this PR:

The Content-Length header is not set if the size of the body is 0 to not break the previous behavior. The HTTP standard does ask for a Content-Length header in some HTTP methods requests such as POST, PUT and PATCH. This might be better addressed on another PR.

[According to some RFC](https://tools.ietf.org/html/rfc7230#section-3.3.1):
> A sender MUST NOT send a Content-Length header field in any message that contains a Transfer-Encoding header field.

It does not make sense in the case where the Transfer-Encoding is something else than `chunked`. [According to the Mozilla wiki](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding#Directives), this restriction seems to only apply on the `chunked` encoding. On this PR, I took Mozilla side and automatically set the Content-Length if the size is known and if the Transfer-Encoding is not `chunked`.

Finally, I do not check if the user has set both the `chunked` Transfer-Encoding and the Content-Length headers, which is not allowed in the HTTP1.1 standard.
algesten: I've been thinking about this. There are basically three ways the user can provide a body:

1. A call without a body, GET, HEAD, etc.
2. Send a body with unknown size (results in chunked)
3. Send a body with known size (sets content-length)

The code did already not capture this intention because it is using an `Option<usize>` of value `None` covered both case 1 and 2. Changing case 1 to be represented as `Some(0)` feels wrong – it means we can't distinguish that case from say sending a 0 byte file. It's no worse than today's code, but it's a sideways move.

One intention of this PR, and a potential follow-up PR is to bring us closer to spec. In that spirit, regardless of how that next PR will look, I think we should introduce an enum that preserves the user intent so we can properly deal with the logic later.

Here's a suggestion, totally up for bike shedding:

```rust
pub enum BodySize {
  NoBody,
  Unknown,
  Known(u64),
}
```

I deliberately chose `u64` over `usize`.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..d4963e7
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,20 @@
+# 1.4.1
+
+ * Use buffer to avoid byte-by-byte parsing result in multiple syscalls.
+ * Allow pooling multiple connections per host.
+ * Put version in user agent "ureq/1.4.1".
+
+# 1.4.0
+
+  * Propagate WouldBlock in io:Error for Response::to_json.
+  * Merge multiple cookies into one header to be spec compliant.
+  * Allow setting TLSConnector for native-tls.
+  * Remove brackets against TLS lib when IPv6 addr is used as hostname.
+  * Include proxy in connection pool keys.
+  * Stop assuming localhost for URLs without host part.
+  * Error if body length is less than content-length.
+  * Validate header names.
+
+# 1.3.0
+
+  * Changelog start
diff --git a/Cargo.toml b/Cargo.toml
index ed99436..fe9ec80 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ureq"
-version = "1.2.0"
+version = "1.4.1"
 authors = ["Martin Algesten <martin@algesten.se>"]
 description = "Minimal HTTP request library"
 license = "MIT/Apache-2.0"
@@ -11,7 +11,8 @@ categories = ["web-programming::http-client"]
 edition = "2018"
 
 [package.metadata.docs.rs]
-all-features = true
+# Keep in sync with .github/workflows/test.yml
+features = [ "tls", "json", "charset", "cookies", "socks-proxy" ]
 
 [features]
 default = ["tls", "cookies"]
@@ -25,15 +26,15 @@ socks-proxy = ["socks"]
 [dependencies]
 base64 = "0.12"
 chunked_transfer = "1.2.0"
-cookie = { version = "0.13", features = ["percent-encode"], optional = true}
+cookie = { version = "0.14", features = ["percent-encode"], optional = true}
 lazy_static = "1"
 qstring = "0.7"
 url = "2"
 socks = { version = "0.3.2", optional = true }
-rustls = { version = "0.17", optional = true, features = [] }
+rustls = { version = "0.18", optional = true, features = [] }
 webpki = { version = "0.21", optional = true }
-webpki-roots = { version = "0.19", optional = true }
-rustls-native-certs = { version = "0.3", optional = true }
+webpki-roots = { version = "0.20", optional = true }
+rustls-native-certs = { version = "0.4", optional = true }
 serde = { version = "1", optional = true }
 serde_json = { version = "1", optional = true }
 encoding = { version = "0.2", optional = true }
@@ -41,3 +42,9 @@ native-tls = { version = "0.2", optional = true }
 
 [dev-dependencies]
 serde = { version = "1", features = ["derive"] }
+rayon = "1.3.0"
+rayon-core = "1.7.0"
+chrono = "0.4.11"
+
+[[example]]
+name = "smoke-test"
diff --git a/README.md b/README.md
index 9a55d30..008e1d9 100644
--- a/README.md
+++ b/README.md
@@ -9,14 +9,12 @@
 ## Usage
 
 ```rust
-// requires feature: `ureq = { version = "*", features = ["json"] }`
-#[macro_use]
-extern crate ureq;
-
 // sync post request of some json.
+// requires feature:
+// `ureq = { version = "*", features = ["json"] }`
 let resp = ureq::post("https://myapi.example.com/ingest")
     .set("X-My-Header", "Secret")
-    .send_json(json!({
+    .send_json(serde_json::json!({
         "name": "martin",
         "rust": true
     }));
@@ -25,8 +23,9 @@ let resp = ureq::post("https://myapi.example.com/ingest")
 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.
+  // 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()?);
 }
 ```
@@ -36,8 +35,7 @@ if resp.ok() {
 This crate is now 1.x.x. It signifies there will be no more breaking
 API changes (for better or worse). I personally use this code in
 production system reading data from AWS. Whether the quality is good
-enough for other use cases is a "YMMV". I know the Agent related code
-is rather  undertested and probably has issues.
+enough for other use cases is a "YMMV".
 
 ## ureq's future
 
@@ -69,7 +67,7 @@ You can control them when including `ureq` as a dependency.
 * `native-tls` enables https using the [`native-tls`](https://crates.io/crates/native-tls) crate. 
   NB: To make this work you currently need to use `default-features: false` to disable `tls`.
   We plan on fixing that.
-* `json` enables `response.into_json()` and `request.send_json()` serde json.
+* `json` enables `response.into_json()` and `request.send_json()` via serde_json.
 * `charset` enables interpreting the charset part of
   `Content-Type: text/plain; charset=iso-8859-1`. Without this, the library
   defaults to rust's built in `utf-8`.
diff --git a/src/agent.rs b/src/agent.rs
index 4eaf6d1..1aa5929 100644
--- a/src/agent.rs
+++ b/src/agent.rs
@@ -6,6 +6,7 @@ use std::sync::Mutex;
 use crate::header::{self, Header};
 use crate::pool::ConnectionPool;
 use crate::request::Request;
+use crate::resolve::ArcResolver;
 
 /// Agents keep state between requests.
 ///
@@ -40,28 +41,25 @@ pub struct Agent {
     /// Copied into each request of this agent.
     pub(crate) headers: Vec<Header>,
     /// Reused agent state for repeated requests from this agent.
-    pub(crate) state: Arc<Mutex<Option<AgentState>>>,
+    pub(crate) state: Arc<Mutex<AgentState>>,
 }
 
 /// Container of the state
 ///
 /// *Internal API*.
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub(crate) struct AgentState {
     /// Reused connections between requests.
     pub(crate) pool: ConnectionPool,
     /// Cookies saved between requests.
     #[cfg(feature = "cookie")]
     pub(crate) jar: CookieJar,
+    pub(crate) resolver: ArcResolver,
 }
 
 impl AgentState {
     fn new() -> Self {
-        AgentState {
-            pool: ConnectionPool::new(),
-            #[cfg(feature = "cookie")]
-            jar: CookieJar::new(),
-        }
+        Self::default()
     }
     pub fn pool(&mut self) -> &mut ConnectionPool {
         &mut self.pool
@@ -89,7 +87,7 @@ impl Agent {
     pub fn build(&self) -> Self {
         Agent {
             headers: self.headers.clone(),
-            state: Arc::new(Mutex::new(Some(AgentState::new()))),
+            state: Arc::new(Mutex::new(AgentState::new())),
         }
     }
 
@@ -166,6 +164,57 @@ impl Agent {
         Request::new(&self, method.into(), path.into())
     }
 
+    /// Sets the maximum number of connections allowed in the connection pool.
+    /// By default, this is set to 100. Setting this to zero would disable
+    /// connection pooling.
+    ///
+    /// ```
+    /// let agent = ureq::agent();
+    /// agent.set_max_pool_connections(200);
+    /// ```
+    pub fn set_max_pool_connections(&self, max_connections: usize) {
+        let mut state = self.state.lock().unwrap();
+        state.pool.set_max_idle_connections(max_connections);
+    }
+
+    /// Sets the maximum number of connections per host to keep in the
+    /// connection pool. By default, this is set to 1. Setting this to zero
+    /// would disable connection pooling.
+    ///
+    /// ```
+    /// let agent = ureq::agent();
+    /// agent.set_max_pool_connections_per_host(10);
+    /// ```
+    pub fn set_max_pool_connections_per_host(&self, max_connections: usize) {
+        let mut state = self.state.lock().unwrap();
+        state
+            .pool
+            .set_max_idle_connections_per_host(max_connections);
+    }
+
+    /// Configures a custom resolver to be used by this agent. By default,
+    /// address-resolution is done by std::net::ToSocketAddrs. This allows you
+    /// to override that resolution with your own alternative. Useful for
+    /// testing and special-cases like DNS-based load balancing.
+    ///
+    /// A `Fn(&str) -> io::Result<Vec<SocketAddr>>` is a valid resolver,
+    /// passing a closure is a simple way to override. Note that you might need
+    /// explicit type `&str` on the closure argument for type inference to
+    /// succeed.
+    /// ```
+    /// use std::net::ToSocketAddrs;
+    ///
+    /// let mut agent = ureq::agent();
+    /// agent.set_resolver(|addr: &str| match addr {
+    ///    "example.com" => Ok(vec![([127,0,0,1], 8096).into()]),
+    ///    addr => addr.to_socket_addrs().map(Iterator::collect),
+    /// });
+    /// ```
+    pub fn set_resolver(&mut self, resolver: impl crate::Resolver + 'static) -> &mut Self {
+        self.state.lock().unwrap().resolver = resolver.into();
+        self
+    }
+
     /// Gets a cookie in this agent by name. Cookies are available
     /// either by setting it in the agent, or by making requests
     /// that `Set-Cookie` in the agent.
@@ -180,10 +229,7 @@ impl Agent {
     #[cfg(feature = "cookie")]
     pub fn cookie(&self, name: &str) -> Option<Cookie<'static>> {
         let state = self.state.lock().unwrap();
-        state
-            .as_ref()
-            .and_then(|state| state.jar.get(name))
-            .cloned()
+        state.jar.get(name).cloned()
     }
 
     /// Set a cookie in this agent.
@@ -197,12 +243,7 @@ impl Agent {
     #[cfg(feature = "cookie")]
     pub fn set_cookie(&self, cookie: Cookie<'static>) {
         let mut state = self.state.lock().unwrap();
-        match state.as_mut() {
-            None => (),
-            Some(state) => {
-                state.jar.add_original(cookie);
-            }
-        }
+        state.jar.add_original(cookie);
     }
 
     /// Make a GET request from this agent.
@@ -256,20 +297,20 @@ pub(crate) fn basic_auth(user: &str, pass: &str) -> String {
         Some(idx) => &user[..idx],
         None => user,
     };
-    ::base64::encode(&format!("{}:{}", safe, pass))
+    base64::encode(&format!("{}:{}", safe, pass))
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::io::Read;
+    use std::thread;
 
     ///////////////////// AGENT TESTS //////////////////////////////
 
     #[test]
     fn agent_implements_send() {
         let mut agent = Agent::new();
-        ::std::thread::spawn(move || {
+        thread::spawn(move || {
             agent.set("Foo", "Bar");
         });
     }
@@ -277,6 +318,8 @@ mod tests {
     #[test]
     #[cfg(any(feature = "tls", feature = "native-tls"))]
     fn agent_pool() {
+        use std::io::Read;
+
         let agent = crate::agent();
         let url = "https://ureq.s3.eu-central-1.amazonaws.com/sherlock.txt";
         // req 1
@@ -288,8 +331,7 @@ mod tests {
         reader.read_to_end(&mut buf).unwrap();
 
         fn poolsize(agent: &Agent) -> usize {
-            let mut lock = agent.state.lock().unwrap();
-            let state = lock.as_mut().unwrap();
+            let mut state = agent.state.lock().unwrap();
             state.pool().len()
         }
         assert_eq!(poolsize(&agent), 1);
@@ -309,7 +351,7 @@ mod tests {
     fn request_implements_send() {
         let agent = Agent::new();
         let mut request = Request::new(&agent, "GET".to_string(), "/foo".to_string());
-        ::std::thread::spawn(move || {
+        thread::spawn(move || {
             request.set("Foo", "Bar");
         });
     }
diff --git a/src/body.rs b/src/body.rs
index f87763e..a7a2d97 100644
--- a/src/body.rs
+++ b/src/body.rs
@@ -1,4 +1,5 @@
 use crate::stream::Stream;
+use std::fmt;
 use std::io::{copy, empty, Cursor, Read, Result as IoResult, Write};
 
 #[cfg(feature = "charset")]
@@ -23,8 +24,8 @@ pub(crate) enum Payload {
     Bytes(Vec<u8>),
 }
 
-impl ::std::fmt::Debug for Payload {
-    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::result::Result<(), ::std::fmt::Error> {
+impl fmt::Debug for Payload {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
             Payload::Empty => write!(f, "Empty"),
             Payload::Text(t, _) => write!(f, "{}", t),
@@ -42,22 +43,32 @@ impl Default for Payload {
     }
 }
 
+/// The size of the body.
+///
+/// *Internal API*
+#[derive(Debug)]
+pub(crate) enum BodySize {
+    Empty,
+    Unknown,
+    Known(u64),
+}
+
 /// Payloads are turned into this type where we can hold both a size and the reader.
 ///
 /// *Internal API*
 pub(crate) struct SizedReader {
-    pub size: Option<usize>,
+    pub size: BodySize,
     pub reader: Box<dyn Read + 'static>,
 }
 
-impl ::std::fmt::Debug for SizedReader {
-    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::result::Result<(), ::std::fmt::Error> {
+impl fmt::Debug for SizedReader {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(f, "SizedReader[size={:?},reader]", self.size)
     }
 }
 
 impl SizedReader {
-    fn new(size: Option<usize>, reader: Box<dyn Read + 'static>) -> Self {
+    fn new(size: BodySize, reader: Box<dyn Read + 'static>) -> Self {
         SizedReader { size, reader }
     }
 }
@@ -65,7 +76,7 @@ impl SizedReader {
 impl Payload {
     pub fn into_read(self) -> SizedReader {
         match self {
-            Payload::Empty => SizedReader::new(None, Box::new(empty())),
+            Payload::Empty => SizedReader::new(BodySize::Empty, Box::new(empty())),
             Payload::Text(text, _charset) => {
                 #[cfg(feature = "charset")]
                 let bytes = {
@@ -78,20 +89,20 @@ impl Payload {
                 let bytes = text.into_bytes();
                 let len = bytes.len();
                 let cursor = Cursor::new(bytes);
-                SizedReader::new(Some(len), Box::new(cursor))
+                SizedReader::new(BodySize::Known(len as u64), Box::new(cursor))
             }
             #[cfg(feature = "json")]
             Payload::JSON(v) => {
                 let bytes = serde_json::to_vec(&v).expect("Bad JSON in payload");
                 let len = bytes.len();
                 let cursor = Cursor::new(bytes);
-                SizedReader::new(Some(len), Box::new(cursor))
+                SizedReader::new(BodySize::Known(len as u64), Box::new(cursor))
             }
-            Payload::Reader(read) => SizedReader::new(None, read),
+            Payload::Reader(read) => SizedReader::new(BodySize::Unknown, read),
             Payload::Bytes(bytes) => {
                 let len = bytes.len();
                 let cursor = Cursor::new(bytes);
-                SizedReader::new(Some(len), Box::new(cursor))
+                SizedReader::new(BodySize::Known(len as u64), Box::new(cursor))
             }
         }
     }
diff --git a/src/header.rs b/src/header.rs
index 056a4b8..bf3dfed 100644
--- a/src/header.rs
+++ b/src/header.rs
@@ -1,15 +1,22 @@
 use crate::error::Error;
+use std::fmt;
 use std::str::FromStr;
 
-#[derive(Clone)]
-/// Wrapper type for a header line.
+#[derive(Clone, PartialEq)]
+/// Wrapper type for a header field.
+/// https://tools.ietf.org/html/rfc7230#section-3.2
 pub struct Header {
+    // Line contains the unmodified bytes of single header field.
+    // It does not contain the final CRLF.
     line: String,
+    // Index is the position of the colon within the header field.
+    // Invariant: index > 0
+    // Invariant: index + 1 < line.len()
     index: usize,
 }
 
-impl ::std::fmt::Debug for Header {
-    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::result::Result<(), ::std::fmt::Error> {
+impl fmt::Debug for Header {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(f, "{}", self.line)
     }
 }
@@ -82,6 +89,29 @@ pub fn add_header(headers: &mut Vec<Header>, header: Header) {
     headers.push(header);
 }
 
+// https://tools.ietf.org/html/rfc7230#section-3.2.3
+// Each header field consists of a case-insensitive field name followed
+// by a colon (":"), optional leading whitespace, the field value, and
+// optional trailing whitespace.
+// field-name     = token
+// token = 1*tchar
+// tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*" / "+" / "-" / "." /
+// "^" / "_" / "`" / "|" / "~" / DIGIT / ALPHA
+fn valid_name(name: &str) -> bool {
+    name.len() > 0 && name.bytes().all(is_tchar)
+}
+
+#[inline]
+fn is_tchar(b: u8) -> bool {
+    match b {
+        b'!' | b'#' | b'$' | b'%' | b'&' => true,
+        b'\'' | b'*' | b'+' | b'-' | b'.' => true,
+        b'^' | b'_' | b'`' | b'|' | b'~' => true,
+        b if b.is_ascii_alphanumeric() => true,
+        _ => false,
+    }
+}
+
 impl FromStr for Header {
     type Err = Error;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
@@ -94,16 +124,43 @@ impl FromStr for Header {
             return Err(Error::BadHeader);
         }
 
-        // https://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2
-        // The field value MAY be preceded by any amount of LWS, though a single SP is preferred.
-        let value_from = &s[index..];
-        let voff = match value_from.find(|c: char| !c.is_whitespace()) {
-            Some(n) => n,
-            None => 0,
-        };
-
-        let index = index + voff;
+        if !valid_name(&line[0..index]) {
+            return Err(Error::BadHeader);
+        }
 
         Ok(Header { line, index })
     }
 }
+
+#[test]
+fn test_valid_name() {
+    assert!(valid_name("example"));
+    assert!(valid_name("Content-Type"));
+    assert!(valid_name("h-123456789"));
+    assert!(!valid_name("Content-Type:"));
+    assert!(!valid_name("Content-Type "));
+    assert!(!valid_name(" some-header"));
+    assert!(!valid_name("\"invalid\""));
+    assert!(!valid_name("Gödel"));
+}
+
+#[test]
+fn test_parse_invalid_name() {
+    let h = "Content-Type   :".parse::<Header>();
+    match h {
+        Err(Error::BadHeader) => {}
+        h => assert!(false, "expected BadHeader error, got {:?}", h),
+    }
+}
+
+#[test]
+fn empty_value() {
+    let h = "foo:".parse::<Header>().unwrap();
+    assert_eq!(h.value(), "");
+}
+
+#[test]
+fn value_with_whitespace() {
+    let h = "foo:      bar    ".parse::<Header>().unwrap();
+    assert_eq!(h.value(), "bar");
+}
diff --git a/src/lib.rs b/src/lib.rs
index df3e1f2..829657e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -47,7 +47,7 @@
 //! 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` (chunked encoding).
+//! * [`.send()`](struct.Request.html#method.send) with a request body as `Read` (chunked encoding support for non-known sized readers).
 //! * [`.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.
@@ -125,6 +125,7 @@ mod header;
 mod pool;
 mod proxy;
 mod request;
+mod resolve;
 mod response;
 mod stream;
 mod unit;
@@ -140,6 +141,7 @@ pub use crate::error::Error;
 pub use crate::header::Header;
 pub use crate::proxy::Proxy;
 pub use crate::request::Request;
+pub use crate::resolve::Resolver;
 pub use crate::response::Response;
 
 // re-export
diff --git a/src/pool.rs b/src/pool.rs
index f5db413..14b0da7 100644
--- a/src/pool.rs
+++ b/src/pool.rs
@@ -1,54 +1,230 @@
-use std::collections::HashMap;
+use std::collections::hash_map::Entry;
+use std::collections::{HashMap, VecDeque};
 use std::io::{Read, Result as IoResult};
 
 use crate::stream::Stream;
 use crate::unit::Unit;
+use crate::Proxy;
 
 use url::Url;
 
 pub const DEFAULT_HOST: &str = "localhost";
+const DEFAULT_MAX_IDLE_CONNECTIONS: usize = 100;
+const DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST: usize = 1;
 
 /// Holder of recycled connections.
 ///
+/// For each PoolKey (approximately hostname and port), there may be
+/// multiple connections stored in the `recycle` map. If so, they are stored in
+/// order from oldest at the front to freshest at the back.
+///
+/// The `lru` VecDeque is a companion struct to `recycle`, and is used to keep
+/// track of which connections to expire if the pool is full on the next insert.
+/// A given PoolKey can occur in lru multiple times. The first entry in lru for
+/// a key K represents the first entry in `recycle[K]`. The second entry in lru
+/// for `K` represents the second entry in `recycle[K]`, and so on. In other
+/// words, `lru` is ordered the same way as the VecDeque entries in `recycle`:
+/// oldest at the front, freshest at the back. This allows keeping track of which
+/// host should have its connection dropped next.
+///
+/// These invariants hold at the start and end of each method:
+///  - The length `lru` is equal to the sum of lengths of `recycle`'s VecDeques.
+///  - Each PoolKey exists the same number of times in `lru` as it has entries in `recycle`.
+///  - If there is an entry in `recycle`, it has at least one element.
+///  - The length of `lru` is less than or equal to max_idle_connections.
+///  - The length of recycle[K] is less than or equal to max_idle_connections_per_host.
+///
 /// *Internal API*
-#[derive(Default, Debug)]
+#[derive(Debug)]
 pub(crate) struct ConnectionPool {
     // the actual pooled connection. however only one per hostname:port.
-    recycle: HashMap<PoolKey, Stream>,
+    recycle: HashMap<PoolKey, VecDeque<Stream>>,
+    // This is used to keep track of which streams to expire when the
+    // pool reaches MAX_IDLE_CONNECTIONS. The corresponding PoolKeys for
+    // recently used Streams are added to the back of the queue;
+    // old streams are removed from the front.
+    lru: VecDeque<PoolKey>,
+    max_idle_connections: usize,
+    max_idle_connections_per_host: usize,
+}
+
+fn remove_first_match(list: &mut VecDeque<PoolKey>, key: &PoolKey) -> Option<PoolKey> {
+    match list.iter().position(|x| x == key) {
+        Some(i) => list.remove(i),
+        None => None,
+    }
+}
+
+fn remove_last_match(list: &mut VecDeque<PoolKey>, key: &PoolKey) -> Option<PoolKey> {
+    match list.iter().rposition(|x| x == key) {
+        Some(i) => list.remove(i),
+        None => None,
+    }
+}
+
+impl Default for ConnectionPool {
+    fn default() -> Self {
+        Self {
+            max_idle_connections: DEFAULT_MAX_IDLE_CONNECTIONS,
+            max_idle_connections_per_host: DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST,
+            recycle: HashMap::default(),
+            lru: VecDeque::default(),
+        }
+    }
 }
 
 impl ConnectionPool {
-    pub fn new() -> Self {
-        ConnectionPool {
-            ..Default::default()
+    pub fn set_max_idle_connections(&mut self, max_connections: usize) {
+        if self.max_idle_connections == max_connections {
+            return;
+        }
+
+        // Remove any extra connections if the number was decreased.
+        while self.lru.len() > max_connections {
+            self.remove_oldest();
         }
+        self.max_idle_connections = max_connections;
+    }
+
+    /// Return true if either of the max_* settings is 0, meaning we should do no work.
+    fn noop(&self) -> bool {
+        self.max_idle_connections == 0 || self.max_idle_connections_per_host == 0
+    }
+
+    pub fn set_max_idle_connections_per_host(&mut self, max_connections: usize) {
+        if self.max_idle_connections_per_host == max_connections {
+            return;
+        }
+
+        if max_connections == 0 {
+            // Clear the connection pool, caching is disabled.
+            self.lru.clear();
+            self.recycle.clear();
+            return;
+        }
+
+        // Remove any extra streams if the number was decreased.
+        for (key, val) in self.recycle.iter_mut() {
+            while val.len() > max_connections {
+                // Remove the oldest entry
+                val.pop_front();
+                remove_first_match(&mut self.lru, key)
+                    .expect("invariant failed: key in recycle but not in lru");
+            }
+        }
+        self.max_idle_connections_per_host = max_connections;
     }
 
     /// How the unit::connect tries to get a pooled connection.
-    pub fn try_get_connection(&mut self, url: &Url) -> Option<Stream> {
-        self.recycle.remove(&PoolKey::new(url))
+    pub fn try_get_connection(&mut self, url: &Url, proxy: &Option<Proxy>) -> Option<Stream> {
+        let key = PoolKey::new(url, proxy);
+        self.remove(&key)
+    }
+
+    fn remove(&mut self, key: &PoolKey) -> Option<Stream> {
+        match self.recycle.entry(key.clone()) {
+            Entry::Occupied(mut occupied_entry) => {
+                let streams = occupied_entry.get_mut();
+                // Take the newest stream.
+                let stream = streams.pop_back();
+                let stream = stream.expect("invariant failed: empty VecDeque in `recycle`");
+
+                if streams.len() == 0 {
+                    occupied_entry.remove();
+                }
+
+                // Remove the newest matching PoolKey from self.lru. That
+                // corresponds to the stream we just removed from `recycle`.
+                remove_last_match(&mut self.lru, &key)
+                    .expect("invariant failed: key in recycle but not in lru");
+
+                Some(stream)
+            }
+            Entry::Vacant(_) => None,
+        }
+    }
+
+    fn add(&mut self, key: PoolKey, stream: Stream) {
+        if self.noop() {
+            return;
+        }
+
+        match self.recycle.entry(key.clone()) {
+            Entry::Occupied(mut occupied_entry) => {
+                let streams = occupied_entry.get_mut();
+                streams.push_back(stream);
+                if streams.len() > self.max_idle_connections_per_host {
+                    // Remove the oldest entry
+                    streams.pop_front();
+                    remove_first_match(&mut self.lru, &key)
+                        .expect("invariant failed: key in recycle but not in lru");
+                }
+            }
+            Entry::Vacant(vacant_entry) => {
+                vacant_entry.insert(vec![stream].into());
+            }
+        }
+        self.lru.push_back(key);
+        if self.lru.len() > self.max_idle_connections {
+            self.remove_oldest()
+        }
+    }
+
+    /// Find the oldest stream in the pool. Remove its representation from lru,
+    /// and the stream itself from `recycle`. Drops the stream, which closes it.
+    fn remove_oldest(&mut self) {
+        assert!(!self.noop(), "remove_oldest called on Pool with max of 0");
+        let key = self.lru.pop_front();
+        let key = key.expect("tried to remove oldest but no entries found!");
+        match self.recycle.entry(key) {
+            Entry::Occupied(mut occupied_entry) => {
+                let streams = occupied_entry.get_mut();
+                streams
+                    .pop_front()
+                    .expect("invariant failed: key existed in recycle but no streams available");
+                if streams.len() == 0 {
+                    occupied_entry.remove();
+                }
+            }
+            Entry::Vacant(_) => panic!("invariant failed: key existed in lru but not in recycle"),
+        }
     }
 
     #[cfg(test)]
     pub fn len(&self) -> usize {
-        self.recycle.len()
+        self.lru.len()
     }
 }
 
-#[derive(Debug, PartialEq, Clone, Eq, Hash)]
+#[derive(PartialEq, Clone, Eq, Hash)]
 struct PoolKey {
     scheme: String,
     hostname: String,
     port: Option<u16>,
+    proxy: Option<Proxy>,
+}
+
+use std::fmt;
+
+impl fmt::Debug for PoolKey {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_fmt(format_args!(
+            "{}|{}|{}",
+            self.scheme,
+            self.hostname,
+            self.port.unwrap_or(0)
+        ))
+    }
 }
 
 impl PoolKey {
-    fn new(url: &Url) -> Self {
+    fn new(url: &Url, proxy: &Option<Proxy>) -> Self {
         let port = url.port_or_known_default();
         PoolKey {
             scheme: url.scheme().to_string(),
             hostname: url.host_str().unwrap_or("").to_string(),
             port,
+            proxy: proxy.clone(),
         }
     }
 }
@@ -56,7 +232,133 @@ impl PoolKey {
 #[test]
 fn poolkey_new() {
     // Test that PoolKey::new() does not panic on unrecognized schemes.
-    PoolKey::new(&Url::parse("zzz:///example.com").unwrap());
+    PoolKey::new(&Url::parse("zzz:///example.com").unwrap(), &None);
+}
+
+#[test]
+fn pool_connections_limit() {
+    // Test inserting connections with different keys into the pool,
+    // filling and draining it. The pool should evict earlier connections
+    // when the connection limit is reached.
+    let mut pool = ConnectionPool::default();
+    let hostnames = (0..DEFAULT_MAX_IDLE_CONNECTIONS * 2).map(|i| format!("{}.example", i));
+    let poolkeys = hostnames.map(|hostname| PoolKey {
+        scheme: "https".to_string(),
+        hostname,
+        port: Some(999),
+        proxy: None,
+    });
+    for key in poolkeys.clone() {
+        pool.add(key, Stream::Cursor(std::io::Cursor::new(vec![])));
+    }
+    assert_eq!(pool.len(), DEFAULT_MAX_IDLE_CONNECTIONS);
+
+    for key in poolkeys.skip(DEFAULT_MAX_IDLE_CONNECTIONS) {
+        let result = pool.remove(&key);
+        assert!(result.is_some(), "expected key was not in pool");
+    }
+    assert_eq!(pool.len(), 0)
+}
+
+#[test]
+fn pool_per_host_connections_limit() {
+    // Test inserting connections with the same key into the pool,
+    // filling and draining it. The pool should evict earlier connections
+    // when the per-host connection limit is reached.
+    let mut pool = ConnectionPool::default();
+    let poolkey = PoolKey {
+        scheme: "https".to_string(),
+        hostname: "example.com".to_string(),
+        port: Some(999),
+        proxy: None,
+    };
+
+    for _ in 0..pool.max_idle_connections_per_host * 2 {
+        pool.add(
+            poolkey.clone(),
+            Stream::Cursor(std::io::Cursor::new(vec![])),
+        );
+    }
+    assert_eq!(pool.len(), DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST);
+
+    for _ in 0..DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST {
+        let result = pool.remove(&poolkey);
+        assert!(result.is_some(), "expected key was not in pool");
+    }
+    assert_eq!(pool.len(), 0);
+}
+
+#[test]
+fn pool_update_connection_limit() {
+    let mut pool = ConnectionPool::default();
+    pool.set_max_idle_connections(50);
+
+    let hostnames = (0..pool.max_idle_connections).map(|i| format!("{}.example", i));
+    let poolkeys = hostnames.map(|hostname| PoolKey {
+        scheme: "https".to_string(),
+        hostname,
+        port: Some(999),
+        proxy: None,
+    });
+    for key in poolkeys.clone() {
+        pool.add(key, Stream::Cursor(std::io::Cursor::new(vec![])));
+    }
+    assert_eq!(pool.len(), 50);
+    pool.set_max_idle_connections(25);
+    assert_eq!(pool.len(), 25);
+}
+
+#[test]
+fn pool_update_per_host_connection_limit() {
+    let mut pool = ConnectionPool::default();
+    pool.set_max_idle_connections(50);
+    pool.set_max_idle_connections_per_host(50);
+
+    let poolkey = PoolKey {
+        scheme: "https".to_string(),
+        hostname: "example.com".to_string(),
+        port: Some(999),
+        proxy: None,
+    };
+
+    for _ in 0..50 {
+        pool.add(
+            poolkey.clone(),
+            Stream::Cursor(std::io::Cursor::new(vec![])),
+        );
+    }
+    assert_eq!(pool.len(), 50);
+    pool.set_max_idle_connections_per_host(25);
+    assert_eq!(pool.len(), 25);
+}
+
+#[test]
+fn pool_checks_proxy() {
+    // Test inserting different poolkeys with same address but different proxies.
+    // Each insertion should result in an additional entry in the pool.
+    let mut pool = ConnectionPool::default();
+    let url = Url::parse("zzz:///example.com").unwrap();
+
+    pool.add(
+        PoolKey::new(&url, &None),
+        Stream::Cursor(std::io::Cursor::new(vec![])),
+    );
+    assert_eq!(pool.len(), 1);
+
+    pool.add(
+        PoolKey::new(&url, &Some(Proxy::new("localhost:9999").unwrap())),
+        Stream::Cursor(std::io::Cursor::new(vec![])),
+    );
+    assert_eq!(pool.len(), 2);
+
+    pool.add(
+        PoolKey::new(
+            &url,
+            &Some(Proxy::new("user:password@localhost:9999").unwrap()),
+        ),
+        Stream::Cursor(std::io::Cursor::new(vec![])),
+    );
+    assert_eq!(pool.len(), 3);
 }
 
 /// Read wrapper that returns the stream to the pool once the
@@ -81,18 +383,16 @@ impl<R: Read + Sized + Into<Stream>> PoolReturnRead<R> {
     fn return_connection(&mut self) {
         // guard we only do this once.
         if let (Some(unit), Some(reader)) = (self.unit.take(), self.reader.take()) {
-            let state = &mut unit.agent.lock().unwrap();
+            let state = &mut unit.req.agent.lock().unwrap();
             // bring back stream here to either go into pool or dealloc
             let stream = reader.into();
-            if let Some(agent) = state.as_mut() {
-                if !stream.is_poolable() {
-                    // just let it deallocate
-                    return;
-                }
-                // insert back into pool
-                let key = PoolKey::new(&unit.url);
-                agent.pool().recycle.insert(key, stream);
+            if !stream.is_poolable() {
+                // just let it deallocate
+                return;
             }
+            // insert back into pool
+            let key = PoolKey::new(&unit.url, &unit.req.proxy);
+            state.pool().add(key, stream);
         }
     }
 
@@ -115,9 +415,3 @@ impl<R: Read + Sized + Into<Stream>> Read for PoolReturnRead<R> {
         Ok(amount)
     }
 }
-
-impl<R: Read + Sized + Into<Stream>> Drop for PoolReturnRead<R> {
-    fn drop(&mut self) {
-        self.return_connection();
-    }
-}
diff --git a/src/proxy.rs b/src/proxy.rs
index 9e06a37..692827e 100644
--- a/src/proxy.rs
+++ b/src/proxy.rs
@@ -1,14 +1,14 @@
 use crate::error::Error;
 
 /// Proxy protocol
-#[derive(Clone, Copy, Debug, PartialEq)]
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 pub enum Proto {
     HTTPConnect,
     SOCKS5,
 }
 
 /// Proxy server definition
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
 pub struct Proxy {
     pub(crate) server: String,
     pub(crate) port: u32,
diff --git a/src/request.rs b/src/request.rs
index 59e60b3..31478cc 100644
--- a/src/request.rs
+++ b/src/request.rs
@@ -1,15 +1,14 @@
+use std::fmt;
 use std::io::Read;
+use std::result::Result;
 use std::sync::{Arc, Mutex};
 use std::time;
 
-use lazy_static::lazy_static;
 use qstring::QString;
 use url::{form_urlencoded, Url};
 
-#[cfg(feature = "tls")]
-use std::fmt;
-
 use crate::agent::{self, Agent, AgentState};
+use crate::body::BodySize;
 use crate::body::{Payload, SizedReader};
 use crate::error::Error;
 use crate::header::{self, Header};
@@ -20,10 +19,6 @@ use crate::Response;
 #[cfg(feature = "json")]
 use super::SerdeValue;
 
-lazy_static! {
-    static ref URL_BASE: Url = Url::parse("http://localhost/").expect("Failed to parse URL_BASE");
-}
-
 /// Request instances are builders that creates a request.
 ///
 /// ```
@@ -35,11 +30,11 @@ lazy_static! {
 /// ```
 #[derive(Clone, Default)]
 pub struct Request {
-    pub(crate) agent: Arc<Mutex<Option<AgentState>>>,
+    pub(crate) agent: Arc<Mutex<AgentState>>,
 
     // via agent
     pub(crate) method: String,
-    path: String,
+    url: String,
 
     // from request itself
     pub(crate) headers: Vec<Header>,
@@ -52,10 +47,12 @@ pub struct Request {
     pub(crate) proxy: Option<crate::proxy::Proxy>,
     #[cfg(feature = "tls")]
     pub(crate) tls_config: Option<TLSClientConfig>,
+    #[cfg(all(feature = "native-tls", not(feature = "tls")))]
+    pub(crate) tls_connector: Option<TLSConnector>,
 }
 
-impl ::std::fmt::Debug for Request {
-    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::result::Result<(), ::std::fmt::Error> {
+impl fmt::Debug for Request {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         let (path, query) = self
             .to_url()
             .map(|u| {
@@ -72,11 +69,11 @@ impl ::std::fmt::Debug for Request {
 }
 
 impl Request {
-    pub(crate) fn new(agent: &Agent, method: String, path: String) -> Request {
+    pub(crate) fn new(agent: &Agent, method: String, url: String) -> Request {
         Request {
             agent: Arc::clone(&agent.state),
             method,
-            path,
+            url,
             headers: agent.headers.clone(),
             redirects: 5,
             ..Default::default()
@@ -214,8 +211,10 @@ impl Request {
 
     /// Send data from a 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.
+    /// If no Content-Length and Transfer-Encoding header has been set, it uses the [chunked transfer encoding](https://tools.ietf.org/html/rfc7230#section-4.1).
+    ///
+    /// The caller may set the Content-Length header to the expected byte size of the reader if is
+    /// known.
     ///
     /// The input from the reader is buffered into chunks of size 16,384, the max size of a TLS fragment.
     ///
@@ -226,7 +225,6 @@ impl Request {
     ///
     /// 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 {
@@ -260,7 +258,7 @@ impl Request {
     ///     .build();
     /// assert_eq!("foobar", req.header("x-api-Key").unwrap());
     /// ```
-    pub fn header<'a>(&self, name: &'a str) -> Option<&str> {
+    pub fn header(&self, name: &str) -> Option<&str> {
         header::get_header(&self.headers, name)
     }
 
@@ -288,7 +286,7 @@ impl Request {
     ///     .build();
     /// assert_eq!(true, req.has("x-api-Key"));
     /// ```
-    pub fn has<'a>(&self, name: &'a str) -> bool {
+    pub fn has(&self, name: &str) -> bool {
         header::has_header(&self.headers, name)
     }
 
@@ -304,7 +302,7 @@ impl Request {
     ///     "2.3.4.5",
     /// ]);
     /// ```
-    pub fn all<'a>(&self, name: &'a str) -> Vec<&str> {
+    pub fn all(&self, name: &str) -> Vec<&str> {
         header::get_all_headers(&self.headers, name)
     }
 
@@ -462,32 +460,6 @@ impl Request {
         self
     }
 
-    // pub fn retry(&self, times: u16) -> Request {
-    //     unimplemented!()
-    // }
-    // pub fn sortQuery(&self) -> Request {
-    //     unimplemented!()
-    // }
-    // pub fn sortQueryBy(&self, by: Box<Fn(&str, &str) -> usize>) -> Request {
-    //     unimplemented!()
-    // }
-    // pub fn ca<S>(&self, accept: S) -> Request
-    //     where S: Into<String> {
-    //     unimplemented!()
-    // }
-    // pub fn cert<S>(&self, accept: S) -> Request
-    //     where S: Into<String> {
-    //     unimplemented!()
-    // }
-    // pub fn key<S>(&self, accept: S) -> Request
-    //     where S: Into<String> {
-    //     unimplemented!()
-    // }
-    // pub fn pfx<S>(&self, accept: S) -> Request // TODO what type? u8?
-    //     where S: Into<String> {
-    //     unimplemented!()
-    // }
-
     /// Get the method this request is using.
     ///
     /// Example:
@@ -512,7 +484,7 @@ impl Request {
     /// assert_eq!(req.get_url(), "https://cool.server/innit");
     /// ```
     pub fn get_url(&self) -> &str {
-        &self.path
+        &self.url
     }
 
     /// Normalizes and returns the host that will be used for this request.
@@ -523,7 +495,7 @@ impl Request {
     ///     .build();
     /// assert_eq!(req1.get_host().unwrap(), "cool.server");
     ///
-    /// let req2 = ureq::post("/some/path")
+    /// let req2 = ureq::post("http://localhost/some/path")
     ///     .build();
     /// assert_eq!(req2.get_host().unwrap(), "localhost");
     /// ```
@@ -558,7 +530,7 @@ impl Request {
             .map(|u| unit::combine_query(&u, &self.query, true))
     }
 
-    /// The normalized path of this request.
+    /// The normalized url of this request.
     ///
     /// Example:
     /// ```
@@ -571,9 +543,7 @@ impl Request {
     }
 
     fn to_url(&self) -> Result<Url, Error> {
-        URL_BASE
-            .join(&self.path)
-            .map_err(|e| Error::BadUrl(format!("{}", e)))
+        Url::parse(&self.url).map_err(|e| Error::BadUrl(format!("{}", e)))
     }
 
     /// Set the proxy server to use for the connection.
@@ -606,6 +576,23 @@ impl Request {
         self
     }
 
+    /// Sets the TLS connector that will be used for the connection.
+    ///
+    /// Example:
+    /// ```
+    /// let tls_connector = std::sync::Arc::new(native_tls::TlsConnector::new().unwrap());
+    /// let req = ureq::post("https://cool.server")
+    ///     .set_tls_connector(tls_connector.clone());
+    /// ```
+    #[cfg(all(feature = "native-tls", not(feature = "tls")))]
+    pub fn set_tls_connector(
+        &mut self,
+        tls_connector: Arc<native_tls::TlsConnector>,
+    ) -> &mut Request {
+        self.tls_connector = Some(TLSConnector(tls_connector));
+        self
+    }
+
     // Returns true if this request, with the provided body, is retryable.
     pub(crate) fn is_retryable(&self, body: &SizedReader) -> bool {
         // Per https://tools.ietf.org/html/rfc7231#section-8.1.3
@@ -618,8 +605,14 @@ impl Request {
         // Sized bodies are retryable only if they are zero-length because of
         // coincidences of the current implementation - the function responsible
         // for retries doesn't have a way to replay a Payload.
-        let no_body = body.size.is_none() || body.size.unwrap() > 0;
-        idempotent && no_body
+        let retryable_body = match body.size {
+            BodySize::Unknown => false,
+            BodySize::Known(0) => true,
+            BodySize::Known(_) => false,
+            BodySize::Empty => true,
+        };
+
+        idempotent && retryable_body
     }
 }
 
@@ -633,3 +626,14 @@ impl fmt::Debug for TLSClientConfig {
         f.debug_struct("TLSClientConfig").finish()
     }
 }
+
+#[cfg(all(feature = "native-tls", not(feature = "tls")))]
+#[derive(Clone)]
+pub(crate) struct TLSConnector(pub(crate) Arc<native_tls::TlsConnector>);
+
+#[cfg(all(feature = "native-tls", not(feature = "tls")))]
+impl fmt::Debug for TLSConnector {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("TLSConnector").finish()
+    }
+}
diff --git a/src/resolve.rs b/src/resolve.rs
new file mode 100644
index 0000000..ca6497d
--- /dev/null
+++ b/src/resolve.rs
@@ -0,0 +1,59 @@
+use std::fmt;
+use std::io::Result as IoResult;
+use std::net::{SocketAddr, ToSocketAddrs};
+use std::sync::Arc;
+
+pub trait Resolver: Send + Sync {
+    fn resolve(&self, netloc: &str) -> IoResult<Vec<SocketAddr>>;
+}
+
+#[derive(Debug)]
+pub(crate) struct StdResolver;
+
+impl Resolver for StdResolver {
+    fn resolve(&self, netloc: &str) -> IoResult<Vec<SocketAddr>> {
+        ToSocketAddrs::to_socket_addrs(netloc).map(|iter| iter.collect())
+    }
+}
+
+impl<F> Resolver for F
+where
+    F: Fn(&str) -> IoResult<Vec<SocketAddr>>,
+    F: Send + Sync,
+{
+    fn resolve(&self, netloc: &str) -> IoResult<Vec<SocketAddr>> {
+        self(netloc)
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct ArcResolver(Arc<dyn Resolver>);
+
+impl<R> From<R> for ArcResolver
+where
+    R: Resolver + 'static,
+{
+    fn from(r: R) -> Self {
+        Self(Arc::new(r))
+    }
+}
+
+impl fmt::Debug for ArcResolver {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "ArcResolver(...)")
+    }
+}
+
+impl std::ops::Deref for ArcResolver {
+    type Target = dyn Resolver;
+
+    fn deref(&self) -> &Self::Target {
+        self.0.as_ref()
+    }
+}
+
+impl Default for ArcResolver {
+    fn default() -> Self {
+        StdResolver.into()
+    }
+}
diff --git a/src/response.rs b/src/response.rs
index 3e52ead..4e2d1c2 100644
--- a/src/response.rs
+++ b/src/response.rs
@@ -1,3 +1,4 @@
+use std::fmt;
 use std::io::{Cursor, Error as IoError, ErrorKind, Read, Result as IoResult};
 use std::str::FromStr;
 use std::time::Instant;
@@ -65,8 +66,8 @@ struct ResponseStatusIndex {
     response_code: usize,
 }
 
-impl ::std::fmt::Debug for Response {
-    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::result::Result<(), ::std::fmt::Error> {
+impl fmt::Debug for Response {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(
             f,
             "Response[status: {}, status_text: {}]",
@@ -122,7 +123,7 @@ impl Response {
     }
 
     /// The header corresponding header value for the give name, if any.
-    pub fn header<'a>(&self, name: &'a str) -> Option<&str> {
+    pub fn header(&self, name: &str) -> Option<&str> {
         self.headers
             .iter()
             .find(|h| h.is_name(name))
@@ -139,12 +140,12 @@ impl Response {
     }
 
     /// Tells if the response has the named header.
-    pub fn has<'a>(&self, name: &'a str) -> bool {
+    pub fn has(&self, name: &str) -> bool {
         self.header(name).is_some()
     }
 
     /// All headers corresponding values for the give name, or empty vector.
-    pub fn all<'a>(&self, name: &'a str) -> Vec<&str> {
+    pub fn all(&self, name: &str) -> Vec<&str> {
         self.headers
             .iter()
             .filter(|h| h.is_name(name))
@@ -400,8 +401,20 @@ impl Response {
     /// ```
     #[cfg(feature = "json")]
     pub fn into_json(self) -> IoResult<serde_json::Value> {
+        use crate::stream::io_err_timeout;
+        use std::error::Error;
+
         let reader = self.into_reader();
         serde_json::from_reader(reader).map_err(|e| {
+            // This is to unify TimedOut IoError in the API.
+            // We make a clone of the original error since serde_json::Error doesn't
+            // let us get the wrapped error instance back.
+            if let Some(ioe) = e.source().and_then(|s| s.downcast_ref::<IoError>()) {
+                if ioe.kind() == ErrorKind::TimedOut {
+                    return io_err_timeout(ioe.to_string());
+                }
+            }
+
             IoError::new(
                 ErrorKind::InvalidData,
                 format!("Failed to read JSON: {}", e),
@@ -581,14 +594,16 @@ pub(crate) fn set_stream(resp: &mut Response, url: String, unit: Option<Unit>, s
 fn read_next_line<R: Read>(reader: &mut R) -> IoResult<String> {
     let mut buf = Vec::new();
     let mut prev_byte_was_cr = false;
+    let mut one = [0_u8];
 
     loop {
-        let byte = reader.bytes().next();
+        let amt = reader.read(&mut one[..])?;
 
-        let byte = match byte {
-            Some(b) => b?,
-            None => return Err(IoError::new(ErrorKind::ConnectionAborted, "Unexpected EOF")),
-        };
+        if amt == 0 {
+            return Err(IoError::new(ErrorKind::ConnectionAborted, "Unexpected EOF"));
+        }
+
+        let byte = one[0];
 
         if byte == b'\n' && prev_byte_was_cr {
             buf.pop(); // removing the '\r'
@@ -631,6 +646,15 @@ impl<R: Read> Read for LimitedRead<R> {
             buf
         };
         match self.reader.read(from) {
+            // https://tools.ietf.org/html/rfc7230#page-33
+            // If the sender closes the connection or
+            // the recipient times out before the indicated number of octets are
+            // received, the recipient MUST consider the message to be
+            // incomplete and close the connection.
+            Ok(0) => Err(IoError::new(
+                ErrorKind::InvalidData,
+                "response body closed before all bytes were read",
+            )),
             Ok(amount) => {
                 self.position += amount;
                 Ok(amount)
@@ -640,6 +664,15 @@ impl<R: Read> Read for LimitedRead<R> {
     }
 }
 
+#[test]
+fn short_read() {
+    use std::io::Cursor;
+    let mut lr = LimitedRead::new(Cursor::new(vec![b'a'; 3]), 10);
+    let mut buf = vec![0; 1000];
+    let result = lr.read_to_end(&mut buf);
+    assert!(result.is_err());
+}
+
 impl<R: Read> From<LimitedRead<R>> for Stream
 where
     Stream: From<R>,
diff --git a/src/stream.rs b/src/stream.rs
index f7b4088..c51520f 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -1,7 +1,9 @@
-use std::io::{Cursor, Error as IoError, ErrorKind, Read, Result as IoResult, Write};
+use std::fmt;
+use std::io::{
+    BufRead, BufReader, Cursor, Error as IoError, ErrorKind, Read, Result as IoResult, Write,
+};
 use std::net::SocketAddr;
 use std::net::TcpStream;
-use std::net::ToSocketAddrs;
 use std::time::Duration;
 use std::time::Instant;
 
@@ -15,7 +17,7 @@ use rustls::StreamOwned;
 use socks::{TargetAddr, ToTargetAddr};
 
 #[cfg(feature = "native-tls")]
-use native_tls::{HandshakeError, TlsConnector, TlsStream};
+use native_tls::{HandshakeError, TlsStream};
 
 use crate::proxy::Proto;
 use crate::proxy::Proxy;
@@ -25,14 +27,14 @@ use crate::unit::Unit;
 
 #[allow(clippy::large_enum_variant)]
 pub enum Stream {
-    Http(TcpStream),
+    Http(BufReader<TcpStream>),
     #[cfg(all(feature = "tls", not(feature = "native-tls")))]
-    Https(rustls::StreamOwned<rustls::ClientSession, TcpStream>),
+    Https(BufReader<rustls::StreamOwned<rustls::ClientSession, TcpStream>>),
     #[cfg(all(feature = "native-tls", not(feature = "tls")))]
-    Https(TlsStream<TcpStream>),
+    Https(BufReader<TlsStream<TcpStream>>),
     Cursor(Cursor<Vec<u8>>),
     #[cfg(test)]
-    Test(Box<dyn Read + Send>, Vec<u8>),
+    Test(Box<dyn BufRead + Send>, Vec<u8>),
 }
 
 // DeadlineStream wraps a stream such that read() will return an error
@@ -73,7 +75,16 @@ impl Read for DeadlineStream {
                 socket.set_write_timeout(Some(timeout))?;
             }
         }
-        self.stream.read(buf)
+        self.stream.read(buf).map_err(|e| {
+            // On unix-y platforms set_read_timeout and set_write_timeout
+            // causes ErrorKind::WouldBlock instead of ErrorKind::TimedOut.
+            // Since the socket most definitely not set_nonblocking(true),
+            // we can safely normalize WouldBlock to TimedOut
+            if e.kind() == ErrorKind::WouldBlock {
+                return io_err_timeout("timed out reading response".to_string());
+            }
+            e
+        })
     }
 }
 
@@ -81,17 +92,18 @@ impl Read for DeadlineStream {
 // 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),
+    match deadline.checked_duration_since(now) {
+        None => Err(io_err_timeout("timed out reading response".to_string())),
+        Some(duration) => Ok(duration),
     }
 }
 
-impl ::std::fmt::Debug for Stream {
-    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::result::Result<(), ::std::fmt::Error> {
+pub(crate) fn io_err_timeout(error: String) -> IoError {
+    IoError::new(ErrorKind::TimedOut, error)
+}
+
+impl fmt::Debug for Stream {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(
             f,
             "Stream[{}]",
@@ -151,9 +163,9 @@ impl Stream {
 
     pub(crate) fn socket(&self) -> Option<&TcpStream> {
         match self {
-            Stream::Http(tcpstream) => Some(tcpstream),
+            Stream::Http(b) => Some(b.get_ref()),
             #[cfg(feature = "tls")]
-            Stream::Https(rustls_stream) => Some(&rustls_stream.sock),
+            Stream::Https(b) => Some(&b.get_ref().sock),
             _ => None,
         }
     }
@@ -183,6 +195,36 @@ impl Read for Stream {
     }
 }
 
+impl BufRead for Stream {
+    fn fill_buf(&mut self) -> IoResult<&[u8]> {
+        match self {
+            Stream::Http(r) => r.fill_buf(),
+            #[cfg(any(
+                all(feature = "tls", not(feature = "native-tls")),
+                all(feature = "native-tls", not(feature = "tls")),
+            ))]
+            Stream::Https(r) => r.fill_buf(),
+            Stream::Cursor(r) => r.fill_buf(),
+            #[cfg(test)]
+            Stream::Test(r, _) => r.fill_buf(),
+        }
+    }
+
+    fn consume(&mut self, amt: usize) {
+        match self {
+            Stream::Http(r) => r.consume(amt),
+            #[cfg(any(
+                all(feature = "tls", not(feature = "native-tls")),
+                all(feature = "native-tls", not(feature = "tls")),
+            ))]
+            Stream::Https(r) => r.consume(amt),
+            Stream::Cursor(r) => r.consume(amt),
+            #[cfg(test)]
+            Stream::Test(r, _) => r.consume(amt),
+        }
+    }
+}
+
 impl<R: Read> From<ChunkDecoder<R>> for Stream
 where
     R: Read,
@@ -195,7 +237,7 @@ where
 
 #[cfg(all(feature = "tls", not(feature = "native-tls")))]
 fn read_https(
-    stream: &mut StreamOwned<ClientSession, TcpStream>,
+    stream: &mut BufReader<StreamOwned<ClientSession, TcpStream>>,
     buf: &mut [u8],
 ) -> IoResult<usize> {
     match stream.read(buf) {
@@ -206,7 +248,7 @@ fn read_https(
 }
 
 #[cfg(all(feature = "native-tls", not(feature = "tls")))]
-fn read_https(stream: &mut TlsStream<TcpStream>, buf: &mut [u8]) -> IoResult<usize> {
+fn read_https(stream: &mut BufReader<TlsStream<TcpStream>>, buf: &mut [u8]) -> IoResult<usize> {
     match stream.read(buf) {
         Ok(size) => Ok(size),
         Err(ref e) if is_close_notify(e) => Ok(0),
@@ -233,12 +275,12 @@ fn is_close_notify(e: &std::io::Error) -> bool {
 impl Write for Stream {
     fn write(&mut self, buf: &[u8]) -> IoResult<usize> {
         match self {
-            Stream::Http(sock) => sock.write(buf),
+            Stream::Http(sock) => sock.get_mut().write(buf),
             #[cfg(any(
                 all(feature = "tls", not(feature = "native-tls")),
                 all(feature = "native-tls", not(feature = "tls")),
             ))]
-            Stream::Https(stream) => stream.write(buf),
+            Stream::Https(stream) => stream.get_mut().write(buf),
             Stream::Cursor(_) => panic!("Write to read only stream"),
             #[cfg(test)]
             Stream::Test(_, writer) => writer.write(buf),
@@ -246,12 +288,12 @@ impl Write for Stream {
     }
     fn flush(&mut self) -> IoResult<()> {
         match self {
-            Stream::Http(sock) => sock.flush(),
+            Stream::Http(sock) => sock.get_mut().flush(),
             #[cfg(any(
                 all(feature = "tls", not(feature = "native-tls")),
                 all(feature = "native-tls", not(feature = "tls")),
             ))]
-            Stream::Https(stream) => stream.flush(),
+            Stream::Https(stream) => stream.get_mut().flush(),
             Stream::Cursor(_) => panic!("Flush read only stream"),
             #[cfg(test)]
             Stream::Test(_, writer) => writer.flush(),
@@ -264,7 +306,9 @@ pub(crate) fn connect_http(unit: &Unit) -> Result<Stream, Error> {
     let hostname = unit.url.host_str().unwrap();
     let port = unit.url.port().unwrap_or(80);
 
-    connect_host(unit, hostname, port).map(Stream::Http)
+    connect_host(unit, hostname, port)
+        .map(BufReader::new)
+        .map(Stream::Http)
 }
 
 #[cfg(all(feature = "tls", feature = "native-certs"))]
@@ -298,53 +342,66 @@ pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
 
     let sni = webpki::DNSNameRef::try_from_ascii_str(hostname)
         .map_err(|err| Error::DnsFailed(err.to_string()))?;
-    let tls_conf: &Arc<rustls::ClientConfig> =
-        unit.tls_config.as_ref().map(|c| &c.0).unwrap_or(&*TLS_CONF);
+    let tls_conf: &Arc<rustls::ClientConfig> = unit
+        .req
+        .tls_config
+        .as_ref()
+        .map(|c| &c.0)
+        .unwrap_or(&*TLS_CONF);
     let sess = rustls::ClientSession::new(&tls_conf, sni);
 
     let sock = connect_host(unit, hostname, port)?;
 
     let stream = rustls::StreamOwned::new(sess, sock);
 
-    Ok(Stream::Https(stream))
+    Ok(Stream::Https(BufReader::new(stream)))
 }
 
 #[cfg(all(feature = "native-tls", not(feature = "tls")))]
 pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
+    use std::sync::Arc;
+
     let hostname = unit.url.host_str().unwrap();
     let port = unit.url.port().unwrap_or(443);
     let sock = connect_host(unit, hostname, port)?;
 
-    let tls_connector = TlsConnector::new().map_err(|e| Error::TlsError(e))?;
-    let stream = tls_connector.connect(hostname, sock).map_err(|e| match e {
-        HandshakeError::Failure(err) => Error::TlsError(err),
-        _ => Error::BadStatusRead,
-    })?;
-
-    Ok(Stream::Https(stream))
+    let tls_connector: Arc<native_tls::TlsConnector> = match &unit.req.tls_connector {
+        Some(connector) => connector.0.clone(),
+        None => Arc::new(native_tls::TlsConnector::new().map_err(|e| Error::TlsError(e))?),
+    };
+    let stream = tls_connector
+        .connect(&hostname.trim_matches(|c| c == '[' || c == ']'), sock)
+        .map_err(|e| match e {
+            HandshakeError::Failure(err) => Error::TlsError(err),
+            _ => Error::BadStatusRead,
+        })?;
+
+    Ok(Stream::Https(BufReader::new(stream)))
 }
 
 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))
+    let deadline: Option<Instant> = if unit.req.timeout_connect > 0 {
+        Instant::now().checked_add(Duration::from_millis(unit.req.timeout_connect))
     } else {
         unit.deadline
     };
-
-    // TODO: Find a way to apply deadline to DNS lookup.
-    let sock_addrs: Vec<SocketAddr> = match unit.proxy {
+  
+    let netloc = match unit.req.proxy {
         Some(ref proxy) => format!("{}:{}", proxy.server, proxy.port),
         None => format!("{}:{}", hostname, port),
-    }
-    .to_socket_addrs()
-    .map_err(|e| Error::DnsFailed(format!("{}", e)))?
-    .collect();
+    };
+
+    // TODO: Find a way to apply deadline to DNS lookup.
+    let sock_addrs = unit
+        .resolver()
+        .resolve(&netloc)
+        .map_err(|e| Error::DnsFailed(format!("{}", e)))?;
 
     if sock_addrs.is_empty() {
         return Err(Error::DnsFailed(format!("No ip address for {}", hostname)));
     }
 
-    let proto = if let Some(ref proxy) = unit.proxy {
+    let proto = if let Some(ref proxy) = unit.req.proxy {
         Some(proxy.proto)
     } else {
         None
@@ -363,7 +420,8 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
         // connect with a configured timeout.
         let stream = if Some(Proto::SOCKS5) == proto {
             connect_socks5(
-                unit.proxy.to_owned().unwrap(),
+                &unit,
+                unit.req.proxy.to_owned().unwrap(),
                 deadline,
                 sock_addr,
                 hostname,
@@ -394,11 +452,11 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
     // Setting it to None will disable the native system timeout
     if let Some(deadline) = deadline {
         stream
-            .set_read_timeout(Some(deadline - Instant::now()))
+            .set_read_timeout(Some(time_until_deadline(deadline)?))
             .ok();
-    } else if unit.timeout_read > 0 {
+    } else if unit.req.timeout_read > 0 {
         stream
-            .set_read_timeout(Some(Duration::from_millis(unit.timeout_read as u64)))
+            .set_read_timeout(Some(Duration::from_millis(unit.req.timeout_read as u64)))
             .ok();
     } else {
         stream.set_read_timeout(None).ok();
@@ -406,18 +464,18 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
 
     if let Some(deadline) = deadline {
         stream
-            .set_write_timeout(Some(deadline - Instant::now()))
+            .set_write_timeout(Some(time_until_deadline(deadline)?))
             .ok();
-    } else if unit.timeout_write > 0 {
+    } else if unit.req.timeout_write > 0 {
         stream
-            .set_write_timeout(Some(Duration::from_millis(unit.timeout_write as u64)))
+            .set_write_timeout(Some(Duration::from_millis(unit.req.timeout_write as u64)))
             .ok();
     } else {
         stream.set_write_timeout(None).ok();
     }
 
     if proto == Some(Proto::HTTPConnect) {
-        if let Some(ref proxy) = unit.proxy {
+        if let Some(ref proxy) = unit.req.proxy {
             write!(stream, "{}", proxy.connect(hostname, port)).unwrap();
             stream.flush()?;
 
@@ -440,11 +498,15 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
 }
 
 #[cfg(feature = "socks-proxy")]
-fn socks5_local_nslookup(hostname: &str, port: u16) -> Result<TargetAddr, std::io::Error> {
-    let addrs: Vec<SocketAddr> = format!("{}:{}", hostname, port)
-        .to_socket_addrs()
-        .map_err(|e| std::io::Error::new(ErrorKind::NotFound, format!("DNS failure: {}.", e)))?
-        .collect();
+fn socks5_local_nslookup(
+    unit: &Unit,
+    hostname: &str,
+    port: u16,
+) -> Result<TargetAddr, std::io::Error> {
+    let addrs: Vec<SocketAddr> = unit
+        .resolver()
+        .resolve(&format!("{}:{}", hostname, port))
+        .map_err(|e| std::io::Error::new(ErrorKind::NotFound, format!("DNS failure: {}.", e)))?;
 
     if addrs.is_empty() {
         return Err(std::io::Error::new(
@@ -466,8 +528,9 @@ fn socks5_local_nslookup(hostname: &str, port: u16) -> Result<TargetAddr, std::i
 
 #[cfg(feature = "socks-proxy")]
 fn connect_socks5(
+    unit: &Unit,
     proxy: Proxy,
-    deadline: Option<time::Instant>,
+    deadline: Option<Instant>,
     proxy_addr: SocketAddr,
     host: &str,
     port: u16,
@@ -477,7 +540,7 @@ fn connect_socks5(
     use std::str::FromStr;
 
     let host_addr = if Ipv4Addr::from_str(host).is_ok() || Ipv6Addr::from_str(host).is_ok() {
-        match socks5_local_nslookup(host, port) {
+        match socks5_local_nslookup(unit, host, port) {
             Ok(addr) => addr,
             Err(err) => return Err(err),
         }
@@ -523,18 +586,18 @@ fn connect_socks5(
         let (lock, cvar) = &*master_signal;
         let done = lock.lock().unwrap();
 
-        let done_result = cvar.wait_timeout(done, deadline - Instant::now()).unwrap();
+        let timeout_connect = time_until_deadline(deadline)?;
+        let done_result = cvar.wait_timeout(done, timeout_connect).unwrap();
         let done = done_result.0;
         if *done {
             rx.recv().unwrap()?
         } else {
-            return Err(std::io::Error::new(
-                ErrorKind::TimedOut,
-                format!(
-                    "SOCKS5 proxy: {}:{} timed out connecting after {}ms.",
-                    host, port, timeout_connect
-                ),
-            ));
+            return Err(io_err_timeout(format!(
+                "SOCKS5 proxy: {}:{} timed out connecting after {}ms.",
+                host,
+                port,
+                timeout_connect.as_millis()
+            )));
         }
     } else {
         get_socks5_stream(&proxy, &proxy_addr, host_addr)?
@@ -569,6 +632,7 @@ fn get_socks5_stream(
 
 #[cfg(not(feature = "socks-proxy"))]
 fn connect_socks5(
+    _unit: &Unit,
     _proxy: Proxy,
     _deadline: Option<Instant>,
     _proxy_addr: SocketAddr,
diff --git a/src/unit.rs b/src/unit.rs
index b6fd18b..6ca617b 100644
--- a/src/unit.rs
+++ b/src/unit.rs
@@ -1,5 +1,4 @@
 use std::io::{Result as IoResult, Write};
-use std::sync::{Arc, Mutex};
 use std::time;
 
 use qstring::QString;
@@ -9,36 +8,25 @@ use url::Url;
 use cookie::{Cookie, CookieJar};
 
 use crate::agent::AgentState;
-use crate::body::{self, Payload, SizedReader};
+use crate::body::{self, BodySize, Payload, SizedReader};
 use crate::header;
+use crate::resolve::ArcResolver;
 use crate::stream::{self, connect_test, Stream};
-use crate::Proxy;
 use crate::{Error, Header, Request, Response};
 
-#[cfg(feature = "tls")]
-use crate::request::TLSClientConfig;
-
 #[cfg(feature = "cookie")]
 use crate::pool::DEFAULT_HOST;
 
 /// It's a "unit of work". Maybe a bad name for it?
 ///
 /// *Internal API*
-#[derive(Debug)]
 pub(crate) struct Unit {
-    pub agent: Arc<Mutex<Option<AgentState>>>,
+    pub req: Request,
     pub url: Url,
     pub is_chunked: bool,
     pub query_string: String,
     pub headers: Vec<Header>,
-    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")]
-    pub tls_config: Option<TLSClientConfig>,
 }
 
 impl Unit {
@@ -47,16 +35,23 @@ impl Unit {
     pub(crate) fn new(req: &Request, url: &Url, mix_queries: bool, body: &SizedReader) -> Self {
         //
 
-        let is_chunked = req
+        let (is_transfer_encoding_set, mut is_chunked) = req
             .header("transfer-encoding")
             // if the user has set an encoding header, obey that.
-            .map(|enc| !enc.is_empty())
+            .map(|enc| {
+                let is_transfer_encoding_set = !enc.is_empty();
+                let last_encoding = enc.split(',').last();
+                let is_chunked = last_encoding
+                    .map(|last_enc| last_enc.trim() == "chunked")
+                    .unwrap_or(false);
+                (is_transfer_encoding_set, is_chunked)
+            })
             // otherwise, no chunking.
-            .unwrap_or(false);
+            .unwrap_or((false, false));
 
         let query_string = combine_query(&url, &req.query, mix_queries);
 
-        let cookie_headers: Vec<_> = extract_cookies(&req.agent, &url);
+        let cookie_header: Option<Header> = extract_cookies(&req.agent, &url);
 
         let extra_headers = {
             let mut extra = vec![];
@@ -64,8 +59,21 @@ impl Unit {
             // chunking and Content-Length headers are mutually exclusive
             // also don't write this if the user has set it themselves
             if !is_chunked && !req.has("content-length") {
-                if let Some(size) = body.size {
-                    extra.push(Header::new("Content-Length", &format!("{}", size)));
+                // if the payload is of known size (everything beside an unsized reader), set
+                // Content-Length,
+                // otherwise, use the chunked Transfer-Encoding (only if no other Transfer-Encoding
+                // has been set
+                match body.size {
+                    BodySize::Known(size) => {
+                        extra.push(Header::new("Content-Length", &format!("{}", size)))
+                    }
+                    BodySize::Unknown => {
+                        if !is_transfer_encoding_set {
+                            extra.push(Header::new("Transfer-Encoding", "chunked"));
+                            is_chunked = true;
+                        }
+                    }
+                    BodySize::Empty => {}
                 }
             }
 
@@ -81,7 +89,7 @@ impl Unit {
         let headers: Vec<_> = req
             .headers
             .iter()
-            .chain(cookie_headers.iter())
+            .chain(cookie_header.iter())
             .chain(extra_headers.iter())
             .cloned()
             .collect();
@@ -95,36 +103,33 @@ impl Unit {
         };
 
         Unit {
-            agent: Arc::clone(&req.agent),
+            req: req.clone(),
             url: url.clone(),
             is_chunked,
             query_string,
             headers,
-            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")]
-            tls_config: req.tls_config.clone(),
         }
     }
 
     pub fn is_head(&self) -> bool {
-        self.method.eq_ignore_ascii_case("head")
+        self.req.method.eq_ignore_ascii_case("head")
+    }
+
+    pub fn resolver(&self) -> ArcResolver {
+        self.req.agent.lock().unwrap().resolver.clone()
     }
 
     #[cfg(test)]
-    pub fn header<'a>(&self, name: &'a str) -> Option<&str> {
+    pub fn header(&self, name: &str) -> Option<&str> {
         header::get_header(&self.headers, name)
     }
     #[cfg(test)]
-    pub fn has<'a>(&self, name: &'a str) -> bool {
+    pub fn has(&self, name: &str) -> bool {
         header::has_header(&self.headers, name)
     }
     #[cfg(test)]
-    pub fn all<'a>(&self, name: &'a str) -> Vec<&str> {
+    pub fn all(&self, name: &str) -> Vec<&str> {
         header::get_all_headers(&self.headers, name)
     }
 }
@@ -170,9 +175,10 @@ pub(crate) fn connect(
     // sequence of requests if all of those requests have idempotent
     // methods.
     //
-    // We choose to retry only once. To do that, we rely on is_recycled,
-    // the "one connection per hostname" police of the ConnectionPool,
-    // and the fact that connections with errors are dropped.
+    // We choose to retry only requests that used a recycled connection
+    // from the ConnectionPool, since those are most likely to have
+    // reached a server-side timeout. Note that this means we may do
+    // up to N+1 total tries, where N is max_idle_connections_per_host.
     //
     // TODO: is_bad_status_read is too narrow since it covers only the
     // first line. It's also allowable to retry requests that hit a
@@ -210,8 +216,8 @@ pub(crate) fn connect(
                     let mut new_unit = Unit::new(req, &new_url, false, &empty);
                     // this is to follow how curl does it. POST, PUT etc change
                     // to GET on a redirect.
-                    new_unit.method = match &unit.method[..] {
-                        "GET" | "HEAD" => unit.method,
+                    new_unit.req.method = match &unit.req.method[..] {
+                        "GET" | "HEAD" => unit.req.method,
                         _ => "GET".into(),
                     };
                     return connect(req, new_unit, use_pooled, redirect_count + 1, empty, true);
@@ -232,51 +238,53 @@ pub(crate) fn connect(
 }
 
 #[cfg(feature = "cookie")]
-fn extract_cookies(state: &std::sync::Mutex<Option<AgentState>>, url: &Url) -> Vec<Header> {
+fn extract_cookies(state: &std::sync::Mutex<AgentState>, url: &Url) -> Option<Header> {
     let state = state.lock().unwrap();
     let is_secure = url.scheme().eq_ignore_ascii_case("https");
     let hostname = url.host_str().unwrap_or(DEFAULT_HOST).to_string();
 
-    match state.as_ref().map(|state| &state.jar) {
-        None => vec![],
-        Some(jar) => match_cookies(jar, &hostname, url.path(), is_secure),
-    }
+    match_cookies(&state.jar, &hostname, url.path(), is_secure)
 }
 
 #[cfg(not(feature = "cookie"))]
-fn extract_cookies(_state: &std::sync::Mutex<Option<AgentState>>, _url: &Url) -> Vec<Header> {
-    vec![]
+fn extract_cookies(_state: &std::sync::Mutex<AgentState>, _url: &Url) -> Option<Header> {
+    None
 }
 
 // TODO check so cookies can't be set for tld:s
 #[cfg(feature = "cookie")]
-fn match_cookies<'a>(jar: &'a CookieJar, domain: &str, path: &str, is_secure: bool) -> Vec<Header> {
-    jar.iter()
-        .filter(|c| {
-            // if there is a domain, it must be matched.
-            // if there is no domain, then ignore cookie
-            let domain_ok = c
-                .domain()
-                .map(|cdom| domain.contains(cdom))
-                .unwrap_or(false);
-            // a path must match the beginning of request path.
-            // no cookie path, we say is ok. is it?!
-            let path_ok = c
-                .path()
-                .map(|cpath| path.find(cpath).map(|pos| pos == 0).unwrap_or(false))
-                .unwrap_or(true);
-            // either the cookie isnt secure, or we're not doing a secure request.
-            let secure_ok = !c.secure().unwrap_or(false) || is_secure;
-
-            domain_ok && path_ok && secure_ok
-        })
-        .map(|c| {
-            let name = c.name().to_string();
-            let value = c.value().to_string();
-            let nameval = Cookie::new(name, value).encoded().to_string();
-            Header::new("Cookie", &nameval)
-        })
-        .collect()
+fn match_cookies(jar: &CookieJar, domain: &str, path: &str, is_secure: bool) -> Option<Header> {
+    Some(
+        jar.iter()
+            .filter(|c| {
+                // if there is a domain, it must be matched.
+                // if there is no domain, then ignore cookie
+                let domain_ok = c
+                    .domain()
+                    .map(|cdom| domain.contains(cdom))
+                    .unwrap_or(false);
+                // a path must match the beginning of request path.
+                // no cookie path, we say is ok. is it?!
+                let path_ok = c
+                    .path()
+                    .map(|cpath| path.find(cpath).map(|pos| pos == 0).unwrap_or(false))
+                    .unwrap_or(true);
+                // either the cookie isnt secure, or we're not doing a secure request.
+                let secure_ok = !c.secure().unwrap_or(false) || is_secure;
+
+                domain_ok && path_ok && secure_ok
+            })
+            .map(|c| {
+                let name = c.name().to_string();
+                let value = c.value().to_string();
+                let nameval = Cookie::new(name, value).encoded().to_string();
+                nameval
+            })
+            .collect::<Vec<_>>()
+            .join(";"),
+    )
+    .filter(|x| !x.is_empty())
+    .map(|s| Header::new("Cookie", &s))
 }
 
 /// Combine the query of the url and the query options set on the request object.
@@ -296,16 +304,14 @@ fn connect_socket(unit: &Unit, use_pooled: bool) -> Result<(Stream, bool), Error
         _ => return Err(Error::UnknownScheme(unit.url.scheme().to_string())),
     };
     if use_pooled {
-        let state = &mut unit.agent.lock().unwrap();
-        if let Some(agent) = state.as_mut() {
-            // The connection may have been closed by the server
-            // due to idle timeout while it was sitting in the pool.
-            // Loop until we find one that is still good or run out of connections.
-            while let Some(stream) = agent.pool.try_get_connection(&unit.url) {
-                let server_closed = stream.server_closed()?;
-                if !server_closed {
-                    return Ok((stream, true));
-                }
+        let state = &mut unit.req.agent.lock().unwrap();
+        // The connection may have been closed by the server
+        // due to idle timeout while it was sitting in the pool.
+        // Loop until we find one that is still good or run out of connections.
+        while let Some(stream) = state.pool.try_get_connection(&unit.url, &unit.req.proxy) {
+            let server_closed = stream.server_closed()?;
+            if !server_closed {
+                return Ok((stream, true));
             }
         }
     }
@@ -330,7 +336,7 @@ fn send_prelude(unit: &Unit, stream: &mut Stream, redir: bool) -> IoResult<()> {
     write!(
         prelude,
         "{} {}{} HTTP/1.1\r\n",
-        unit.method,
+        unit.req.method,
         unit.url.path(),
         &unit.query_string
     )?;
@@ -357,7 +363,11 @@ fn send_prelude(unit: &Unit, stream: &mut Stream, redir: bool) -> IoResult<()> {
         }
     }
     if !header::has_header(&unit.headers, "user-agent") {
-        write!(prelude, "User-Agent: ureq\r\n")?;
+        write!(
+            prelude,
+            "User-Agent: ureq/{}\r\n",
+            env!("CARGO_PKG_VERSION")
+        )?;
     }
     if !header::has_header(&unit.headers, "accept") {
         write!(prelude, "Accept: */*\r\n")?;
@@ -393,22 +403,56 @@ fn save_cookies(unit: &Unit, resp: &Response) {
     }
 
     // only lock if we know there is something to process
-    let state = &mut unit.agent.lock().unwrap();
-    if let Some(add_jar) = state.as_mut().map(|state| &mut state.jar) {
-        for raw_cookie in cookies.iter() {
-            let to_parse = if raw_cookie.to_lowercase().contains("domain=") {
-                (*raw_cookie).to_string()
-            } else {
-                let host = &unit.url.host_str().unwrap_or(DEFAULT_HOST).to_string();
-                format!("{}; Domain={}", raw_cookie, host)
-            };
-            match Cookie::parse_encoded(&to_parse[..]) {
-                Err(_) => (), // ignore unparseable cookies
-                Ok(cookie) => {
-                    let cookie = cookie.into_owned();
-                    add_jar.add(cookie)
-                }
+    let state = &mut unit.req.agent.lock().unwrap();
+    for raw_cookie in cookies.iter() {
+        let to_parse = if raw_cookie.to_lowercase().contains("domain=") {
+            (*raw_cookie).to_string()
+        } else {
+            let host = &unit.url.host_str().unwrap_or(DEFAULT_HOST).to_string();
+            format!("{}; Domain={}", raw_cookie, host)
+        };
+        match Cookie::parse_encoded(&to_parse[..]) {
+            Err(_) => (), // ignore unparseable cookies
+            Ok(cookie) => {
+                let cookie = cookie.into_owned();
+                state.jar.add(cookie)
             }
         }
     }
 }
+
+#[cfg(test)]
+#[cfg(feature = "cookies")]
+mod tests {
+    use super::*;
+
+    ///////////////////// COOKIE TESTS //////////////////////////////
+
+    #[test]
+    fn match_cookies_returns_nothing_when_no_cookies() {
+        let jar = CookieJar::new();
+
+        let result = match_cookies(&jar, "crates.io", "/", false);
+        assert_eq!(result, None);
+    }
+
+    #[test]
+    fn match_cookies_returns_one_header() {
+        let mut jar = CookieJar::new();
+        let cookie1 = Cookie::parse("cookie1=value1; Domain=crates.io").unwrap();
+        let cookie2 = Cookie::parse("cookie2=value2; Domain=crates.io").unwrap();
+        jar.add(cookie1);
+        jar.add(cookie2);
+
+        // There's no guarantee to the order in which cookies are defined.
+        // Ensure that they're either in one order or the other.
+        let result = match_cookies(&jar, "crates.io", "/", false);
+        let order1 = "cookie1=value1;cookie2=value2";
+        let order2 = "cookie2=value2;cookie1=value1";
+
+        assert!(
+            result == Some(Header::new("Cookie", order1))
+                || result == Some(Header::new("Cookie", order2))
+        );
+    }
+}


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 7324d90..8648ccc 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -3,16 +3,46 @@ on: [push, pull_request]
 name: CI
 
 jobs:
+  doc:
+    name: Docs
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@master
+      - name: Install Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          override: true
+      - name: Docs
+        uses: actions-rs/cargo@v1
+        with:
+          command: doc
+          # Keep in sync with Cargo.toml's [package.metadata.docs.rs]
+          args: --no-default-features --no-deps --features "tls json charset cookies socks-proxy"
   build_and_test:
-    name: Rust project
+    name: Test
     runs-on: ubuntu-latest
     strategy:
       matrix:
+        tls:
+          - ""
+          - tls
+          - native-tls
         feature:
+          - ""
+          - json
           - charset
           - cookies
+          - socks-proxy
+        what:
+          - "--doc"
+          - "--tests"
+        exclude:
+          - tls: ""
+            what: "--doc"
     env:
       RUST_BACKTRACE: "1"
+      RUSTFLAGS: "-D dead_code -D unused-variables -D unused"
     steps:
       - uses: actions/checkout@master
       - name: Install Rust
@@ -24,4 +54,4 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --no-default-features --features "json tls ${{ matrix.feature }}"
+          args: ${{ matrix.what }} --no-default-features --features "${{ matrix.tls }} ${{ matrix.feature }}"
diff --git a/examples/smoke-test/main.rs b/examples/smoke-test/main.rs
new file mode 100644
index 0000000..39770a7
--- /dev/null
+++ b/examples/smoke-test/main.rs
@@ -0,0 +1,107 @@
+use chrono::Local;
+use rayon::prelude::*;
+use rayon_core;
+
+use std::io::{self, BufRead, BufReader, Read};
+use std::iter::Iterator;
+use std::time::Duration;
+use std::{env, error, fmt, result};
+
+use ureq;
+
+#[derive(Debug)]
+struct Oops(String);
+
+impl From<io::Error> for Oops {
+    fn from(e: io::Error) -> Oops {
+        Oops(e.to_string())
+    }
+}
+
+impl From<&ureq::Error> for Oops {
+    fn from(e: &ureq::Error) -> Oops {
+        Oops(e.to_string())
+    }
+}
+
+impl From<rayon_core::ThreadPoolBuildError> for Oops {
+    fn from(e: rayon_core::ThreadPoolBuildError) -> Oops {
+        Oops(e.to_string())
+    }
+}
+
+impl error::Error for Oops {}
+
+impl fmt::Display for Oops {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+type Result<T> = result::Result<T, Oops>;
+
+fn get(agent: &ureq::Agent, url: &String) -> Result<Vec<u8>> {
+    let response = agent
+        .get(url)
+        .timeout_connect(5_000)
+        .timeout(Duration::from_secs(20))
+        .call();
+    if let Some(err) = response.synthetic_error() {
+        return Err(err.into());
+    }
+    let mut reader = response.into_reader();
+    let mut bytes = vec![];
+    reader.read_to_end(&mut bytes)?;
+    Ok(bytes)
+}
+
+fn get_and_write(agent: &ureq::Agent, url: &String) -> Result<()> {
+    println!("🕷️ {} {}", Local::now(), url);
+    match get(agent, url) {
+        Ok(_) => println!("✔️ {} {}", Local::now(), url),
+        Err(e) => println!("⚠️ {} {} {}", Local::now(), url, e),
+    }
+    Ok(())
+}
+
+fn get_many(urls: Vec<String>, simultaneous_fetches: usize) -> Result<()> {
+    let agent = ureq::Agent::default().build();
+    let pool = rayon::ThreadPoolBuilder::new()
+        .num_threads(simultaneous_fetches)
+        .build()?;
+    pool.scope(|_| {
+        urls.par_iter().map(|u| get_and_write(&agent, u)).count();
+    });
+    Ok(())
+}
+
+fn main() -> Result<()> {
+    let args = env::args();
+    if args.len() == 1 {
+        println!(
+            r##"Usage: {:#?} top-1m.csv
+        
+Where top-1m.csv is a simple, unquoted CSV containing two fields, a rank and
+a domain name. For instance you can get such a list from https://tranco-list.eu/.
+
+For each domain, this program will attempt to GET four URLs: The domain name
+name with HTTP and HTTPS, and with and without a www prefix. It will fetch
+using 50 threads concurrently.
+"##,
+            env::current_exe()?
+        );
+        return Ok(());
+    }
+    let file = std::fs::File::open(args.skip(1).next().unwrap())?;
+    let bufreader = BufReader::new(file);
+    let mut urls = vec![];
+    for line in bufreader.lines() {
+        let domain = line?.rsplit(",").next().unwrap().to_string();
+        urls.push(format!("http://{}/", domain));
+        urls.push(format!("https://{}/", domain));
+        urls.push(format!("http://www.{}/", domain));
+        urls.push(format!("https://www.{}/", domain));
+    }
+    get_many(urls, 50)?;
+    Ok(())
+}
diff --git a/src/test/agent_test.rs b/src/test/agent_test.rs
index 55934b9..97568b3 100644
--- a/src/test/agent_test.rs
+++ b/src/test/agent_test.rs
@@ -1,6 +1,9 @@
+#![allow(dead_code)]
+
 use crate::test;
-use std::io::{BufRead, BufReader, Read, Write};
-use std::thread;
+use crate::test::testserver::{read_headers, TestServer};
+use std::io::{self, Read, Write};
+use std::net::TcpStream;
 use std::time::Duration;
 
 use super::super::*;
@@ -56,51 +59,28 @@ fn agent_cookies() {
     agent.get("test://host/agent_cookies").call();
 }
 
-// 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 {
-    let listener = std::net::TcpListener::bind("localhost:0").unwrap();
-    let port = listener.local_addr().unwrap().port();
-    thread::spawn(move || {
-        for stream in listener.incoming() {
-            thread::spawn(move || {
-                let stream = stream.unwrap();
-                stream
-                    .set_read_timeout(Some(Duration::from_secs(2)))
-                    .unwrap();
-                let mut write_stream = stream.try_clone().unwrap();
-                for line in BufReader::new(stream).lines() {
-                    let line = match line {
-                        Ok(x) => x,
-                        Err(_) => return,
-                    };
-                    if line == "" {
-                        write_stream
-                            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nresponse")
-                            .unwrap();
-                    }
-                }
-            });
-        }
-    });
-    port
+// Handler that answers with a simple HTTP response, and times
+// out idle connections after 2 seconds.
+fn idle_timeout_handler(mut stream: TcpStream) -> io::Result<()> {
+    read_headers(&stream);
+    stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nresponse")?;
+    stream.set_read_timeout(Some(Duration::from_secs(2)))?;
+    Ok(())
 }
 
 #[test]
 fn connection_reuse() {
-    let port = start_idle_timeout_server();
-    let url = format!("http://localhost:{}", port);
+    let testserver = TestServer::new(idle_timeout_handler);
+    let url = format!("http://localhost:{}", testserver.port);
     let agent = Agent::default().build();
     let resp = agent.get(&url).call();
 
     // use up the connection so it gets returned to the pool
     assert_eq!(resp.status(), 200);
-    let mut buf = vec![];
-    resp.into_reader().read_to_end(&mut buf).unwrap();
+    resp.into_string().unwrap();
 
     {
-        let mut guard_state = agent.state.lock().unwrap();
-        let mut state = guard_state.take().unwrap();
+        let mut state = agent.state.lock().unwrap();
         assert!(state.pool().len() > 0);
     }
 
@@ -120,3 +100,132 @@ fn connection_reuse() {
     }
     assert_eq!(resp.status(), 200);
 }
+
+#[test]
+fn custom_resolver() {
+    use std::io::Read;
+    use std::net::TcpListener;
+
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+
+    let local_addr = listener.local_addr().unwrap();
+
+    let server = std::thread::spawn(move || {
+        let (mut client, _) = listener.accept().unwrap();
+        let mut buf = vec![0u8; 16];
+        let read = client.read(&mut buf).unwrap();
+        buf.truncate(read);
+        buf
+    });
+
+    crate::agent()
+        .set_resolver(move |_: &str| Ok(vec![local_addr]))
+        .get("http://cool.server/")
+        .call();
+
+    assert_eq!(&server.join().unwrap(), b"GET / HTTP/1.1\r\n");
+}
+
+#[cfg(feature = "cookie")]
+#[cfg(test)]
+fn cookie_and_redirect(mut stream: TcpStream) -> io::Result<()> {
+    let headers = read_headers(&stream);
+    match headers.path() {
+        "/first" => {
+            stream.write_all(b"HTTP/1.1 302 Found\r\n")?;
+            stream.write_all(b"Location: /second\r\n")?;
+            stream.write_all(b"Set-Cookie: first=true\r\n")?;
+            stream.write_all(b"Content-Length: 0\r\n\r\n")?;
+        }
+        "/second" => {
+            if headers
+                .headers()
+                .iter()
+                .find(|&x| x.contains("first=true"))
+                .is_none()
+            {
+                panic!("request did not contain cookie 'first'");
+            }
+            stream.write_all(b"HTTP/1.1 302 Found\r\n")?;
+            stream.write_all(b"Location: /third\r\n")?;
+            stream.write_all(b"Set-Cookie: second=true\r\n")?;
+            stream.write_all(b"Content-Length: 0\r\n\r\n")?;
+        }
+        "/third" => {
+            if headers
+                .headers()
+                .iter()
+                .find(|&x| x.contains("second=true"))
+                .is_none()
+            {
+                panic!("request did not contain cookie 'second'");
+            }
+            stream.write_all(b"HTTP/1.1 200 OK\r\n")?;
+            stream.write_all(b"Set-Cookie: third=true\r\n")?;
+            stream.write_all(b"Content-Length: 0\r\n\r\n")?;
+        }
+        _ => {}
+    }
+    Ok(())
+}
+
+#[cfg(feature = "cookie")]
+#[test]
+fn test_cookies_on_redirect() {
+    let testserver = TestServer::new(cookie_and_redirect);
+    let url = format!("http://localhost:{}/first", testserver.port);
+    let agent = Agent::default().build();
+    let resp = agent.post(&url).call();
+    if resp.error() {
+        panic!("error: {} {}", resp.status(), resp.into_string().unwrap());
+    }
+    assert!(agent.cookie("first").is_some());
+    assert!(agent.cookie("second").is_some());
+    assert!(agent.cookie("third").is_some());
+}
+
+#[test]
+fn dirty_streams_not_returned() -> io::Result<()> {
+    let testserver = TestServer::new(|mut stream: TcpStream| -> io::Result<()> {
+        read_headers(&stream);
+        stream.write_all(b"HTTP/1.1 200 OK\r\n")?;
+        stream.write_all(b"Transfer-Encoding: chunked\r\n")?;
+        stream.write_all(b"\r\n")?;
+        stream.write_all(b"5\r\n")?;
+        stream.write_all(b"corgi\r\n")?;
+        stream.write_all(b"8\r\n")?;
+        stream.write_all(b"dachsund\r\n")?;
+        stream.write_all(b"0\r\n")?;
+        stream.write_all(b"\r\n")?;
+        Ok(())
+    });
+    let url = format!("http://localhost:{}/", testserver.port);
+    let agent = Agent::default().build();
+    let resp = agent.get(&url).call();
+    if let Some(err) = resp.synthetic_error() {
+        panic!("resp failed: {:?}", err);
+    }
+    let resp_str = resp.into_string()?;
+    assert_eq!(resp_str, "corgidachsund");
+
+    // Now fetch it again, but only read part of the body.
+    let resp_to_be_dropped = agent.get(&url).call();
+    if let Some(err) = resp_to_be_dropped.synthetic_error() {
+        panic!("resp_to_be_dropped failed: {:?}", err);
+    }
+    let mut reader = resp_to_be_dropped.into_reader();
+
+    // Read 9 bytes of the response and then drop the reader.
+    let mut buf = [0_u8; 4];
+    let n = reader.read(&mut buf)?;
+    assert_ne!(n, 0, "early EOF");
+    assert_eq!(&buf, b"corg");
+    drop(reader);
+
+    let resp_to_succeed = agent.get(&url).call();
+    if let Some(err) = resp_to_succeed.synthetic_error() {
+        panic!("resp_to_succeed failed: {:?}", err);
+    }
+
+    Ok(())
+}
diff --git a/src/test/mod.rs b/src/test/mod.rs
index 2baa08f..801b7c9 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 testserver;
 mod timeout;
 
 type RequestHandler = dyn Fn(&Unit) -> Result<Stream, Error> + Send + 'static;
diff --git a/src/test/redirect.rs b/src/test/redirect.rs
index 705c672..03639e2 100644
--- a/src/test/redirect.rs
+++ b/src/test/redirect.rs
@@ -46,7 +46,7 @@ fn redirect_head() {
         test::make_response(302, "Go here", vec!["Location: /redirect_head2"], vec![])
     });
     test::set_handler("/redirect_head2", |unit| {
-        assert_eq!(unit.method, "HEAD");
+        assert_eq!(unit.req.method, "HEAD");
         test::make_response(200, "OK", vec!["x-foo: bar"], vec![])
     });
     let resp = head("test://host/redirect_head1").call();
@@ -62,7 +62,7 @@ fn redirect_get() {
         test::make_response(302, "Go here", vec!["Location: /redirect_get2"], vec![])
     });
     test::set_handler("/redirect_get2", |unit| {
-        assert_eq!(unit.method, "GET");
+        assert_eq!(unit.req.method, "GET");
         assert!(unit.has("Range"));
         assert_eq!(unit.header("Range").unwrap(), "bytes=10-50");
         test::make_response(200, "OK", vec!["x-foo: bar"], vec![])
@@ -82,7 +82,7 @@ fn redirect_post() {
         test::make_response(302, "Go here", vec!["Location: /redirect_post2"], vec![])
     });
     test::set_handler("/redirect_post2", |unit| {
-        assert_eq!(unit.method, "GET");
+        assert_eq!(unit.req.method, "GET");
         test::make_response(200, "OK", vec!["x-foo: bar"], vec![])
     });
     let resp = post("test://host/redirect_post1").call();
diff --git a/src/test/simple.rs b/src/test/simple.rs
index 0c4a508..fd7cc3c 100644
--- a/src/test/simple.rs
+++ b/src/test/simple.rs
@@ -122,7 +122,7 @@ fn escape_path() {
 
 #[test]
 fn request_debug() {
-    let req = get("/my/page")
+    let req = get("http://localhost/my/page")
         .set("Authorization", "abcdef")
         .set("Content-Length", "1234")
         .set("Content-Type", "application/json")
@@ -136,7 +136,7 @@ fn request_debug() {
          Content-Length: 1234, Content-Type: application/json])"
     );
 
-    let req = get("/my/page?q=z")
+    let req = get("http://localhost/my/page?q=z")
         .query("foo", "bar baz")
         .set("Authorization", "abcdef")
         .build();
diff --git a/src/test/testserver.rs b/src/test/testserver.rs
new file mode 100644
index 0000000..de17ca4
--- /dev/null
+++ b/src/test/testserver.rs
@@ -0,0 +1,73 @@
+use std::io::{self, BufRead, BufReader};
+use std::net::{TcpListener, TcpStream};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread;
+
+pub struct TestServer {
+    pub port: u16,
+    pub done: Arc<AtomicBool>,
+}
+
+pub struct TestHeaders(Vec<String>);
+
+impl TestHeaders {
+    // Return the path for a request, e.g. /foo from "GET /foo HTTP/1.1"
+    #[cfg(feature = "cookie")]
+    pub fn path(&self) -> &str {
+        if self.0.len() == 0 {
+            ""
+        } else {
+            &self.0[0].split(" ").nth(1).unwrap()
+        }
+    }
+
+    #[cfg(feature = "cookie")]
+    pub fn headers(&self) -> &[String] {
+        &self.0[1..]
+    }
+}
+
+// Read a stream until reaching a blank line, in order to consume
+// request headers.
+pub fn read_headers(stream: &TcpStream) -> TestHeaders {
+    let mut results = vec![];
+    for line in BufReader::new(stream).lines() {
+        match line {
+            Err(e) => panic!(e),
+            Ok(line) if line == "" => break,
+            Ok(line) => results.push(line),
+        };
+    }
+    TestHeaders(results)
+}
+
+impl TestServer {
+    pub fn new(handler: fn(TcpStream) -> io::Result<()>) -> Self {
+        let listener = TcpListener::bind("localhost:0").unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let done = Arc::new(AtomicBool::new(false));
+        let done_clone = done.clone();
+        thread::spawn(move || {
+            for stream in listener.incoming() {
+                thread::spawn(move || handler(stream.unwrap()));
+                if done.load(Ordering::Relaxed) {
+                    break;
+                }
+            }
+            println!("testserver on {} exiting", port);
+        });
+        TestServer {
+            port,
+            done: done_clone,
+        }
+    }
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        self.done.store(true, Ordering::Relaxed);
+        // Connect once to unblock the listen loop.
+        TcpStream::connect(format!("localhost:{}", self.port)).unwrap();
+    }
+}
diff --git a/src/test/timeout.rs b/src/test/timeout.rs
index 4e14df5..cb69055 100644
--- a/src/test/timeout.rs
+++ b/src/test/timeout.rs
@@ -1,4 +1,5 @@
-use std::io::{self, BufRead, BufReader, Read, Write};
+use crate::test::testserver::*;
+use std::io::{self, Write};
 use std::net::TcpStream;
 use std::thread;
 use std::time::Duration;
@@ -7,8 +8,8 @@ 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];
+fn dribble_body_respond(mut stream: TcpStream, contents: &[u8]) -> io::Result<()> {
+    read_headers(&stream);
     let headers = format!(
         "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n",
         contents.len() * 2
@@ -23,51 +24,13 @@ fn dribble_body_respond(stream: &mut TcpStream) -> io::Result<()> {
     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 {
+    match resp.into_string() {
         Err(io_error) => match io_error.kind() {
-            io::ErrorKind::WouldBlock => Ok(()),
             io::ErrorKind::TimedOut => Ok(()),
             _ => Err(format!("{:?}", io_error)),
         },
@@ -78,15 +41,15 @@ fn get_and_expect_timeout(url: String) {
 
 #[test]
 fn overall_timeout_during_body() {
-    let port = start_dribble_body_server();
-    let url = format!("http://localhost:{}/", port);
-
+    // Start a test server on an available port, that dribbles out a response at 1 write per 10ms.
+    let server = TestServer::new(|stream| dribble_body_respond(stream, &[b'a'; 300]));
+    let url = format!("http://localhost:{}/", server.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<()> {
+fn dribble_headers_respond(mut stream: 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")?;
@@ -96,28 +59,40 @@ fn dribble_headers_respond(stream: &mut TcpStream) -> io::Result<()> {
     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);
+    // Start a test server on an available port, that dribbles out a response at 1 write per 10ms.
+    let server = TestServer::new(dribble_headers_respond);
+    let url = format!("http://localhost:{}/", server.port);
     get_and_expect_timeout(url);
 }
+
+#[test]
+#[cfg(feature = "json")]
+fn overall_timeout_reading_json() {
+    // Start a test server on an available port, that dribbles out a response at 1 write per 10ms.
+    let server = TestServer::new(|stream| {
+        dribble_body_respond(
+            stream,
+            b"[1,1,1,1,1,1,1,1,1,1,1,1,1,
+        1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,
+        1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,
+        1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,
+        1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1]",
+        )
+    });
+    let url = format!("http://localhost:{}/", server.port);
+
+    let agent = Agent::default().build();
+    let timeout = Duration::from_millis(500);
+    let resp = agent.get(&url).timeout(timeout).call();
+
+    match resp.into_json() {
+        Ok(_) => Err("successful response".to_string()),
+        Err(e) => match e.kind() {
+            io::ErrorKind::TimedOut => Ok(()),
+            _ => Err(format!("Unexpected io::ErrorKind: {:?}", e)),
+        },
+    }
+    .expect("expected timeout but got something else");
+}