Repository: algesten/ureq
Pull request #127 (2020-08-05 05:41:37)

Add unittest for cookie handling on redirects.
Related to #113 


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..270cc9f
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,14 @@
+# 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 553da94..2849a94 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ureq"
-version = "1.3.0"
+version = "1.4.0"
 authors = ["Martin Algesten <martin@algesten.se>"]
 description = "Minimal HTTP request library"
 license = "MIT/Apache-2.0"
diff --git a/src/agent.rs b/src/agent.rs
index 4eaf6d1..b7a43ee 100644
--- a/src/agent.rs
+++ b/src/agent.rs
@@ -166,6 +166,38 @@ 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 optional_state = self.state.lock().unwrap();
+        if let Some(state) = optional_state.as_mut() {
+            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 optional_state = self.state.lock().unwrap();
+        if let Some(state) = optional_state.as_mut() {
+            state
+                .pool
+                .set_max_idle_connections_per_host(max_connections);
+        }
+    }
+
     /// 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.
@@ -256,20 +288,21 @@ 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");
         });
     }
@@ -309,7 +342,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..b6b188f 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),
@@ -50,8 +51,8 @@ pub(crate) struct SizedReader {
     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)
     }
 }
diff --git a/src/header.rs b/src/header.rs
index 719ccb9..bf3dfed 100644
--- a/src/header.rs
+++ b/src/header.rs
@@ -1,7 +1,8 @@
 use crate::error::Error;
+use std::fmt;
 use std::str::FromStr;
 
-#[derive(Clone)]
+#[derive(Clone, PartialEq)]
 /// Wrapper type for a header field.
 /// https://tools.ietf.org/html/rfc7230#section-3.2
 pub struct Header {
@@ -14,8 +15,8 @@ pub struct Header {
     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)
     }
 }
diff --git a/src/pool.rs b/src/pool.rs
index d9ca195..ae2dd3f 100644
--- a/src/pool.rs
+++ b/src/pool.rs
@@ -1,3 +1,4 @@
+use std::collections::hash_map::Entry;
 use std::collections::{HashMap, VecDeque};
 use std::io::{Read, Result as IoResult};
 
@@ -8,7 +9,8 @@ use crate::Proxy;
 use url::Url;
 
 pub const DEFAULT_HOST: &str = "localhost";
-const MAX_IDLE_CONNECTIONS: usize = 100;
+const DEFAULT_MAX_IDLE_CONNECTIONS: usize = 100;
+const DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST: usize = 1;
 
 /// Holder of recycled connections.
 ///
@@ -20,21 +22,71 @@ const MAX_IDLE_CONNECTIONS: usize = 100;
 #[derive(Default, 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,
 }
 
 impl ConnectionPool {
     pub fn new() -> Self {
         ConnectionPool {
+            max_idle_connections: DEFAULT_MAX_IDLE_CONNECTIONS,
+            max_idle_connections_per_host: DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST,
             ..Default::default()
         }
     }
 
+    pub fn set_max_idle_connections(&mut self, max_connections: usize) {
+        if self.max_idle_connections == max_connections {
+            return;
+        }
+        self.max_idle_connections = max_connections;
+
+        if max_connections == 0 {
+            // Clear the connection pool, caching is disabled.
+            self.lru.clear();
+            self.recycle.clear();
+            return;
+        }
+
+        // Remove any extra connections if the number was decreased.
+        while self.lru.len() > max_connections {
+            self.remove_oldest();
+        }
+    }
+
+    pub fn set_max_idle_connections_per_host(&mut self, max_connections: usize) {
+        if self.max_idle_connections_per_host == max_connections {
+            return;
+        }
+        self.max_idle_connections_per_host = max_connections;
+
+        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 {
+                val.pop_front();
+                let index = self
+                    .lru
+                    .iter()
+                    .position(|x| x == key)
+                    .expect("PoolKey not found in lru");
+                self.lru.remove(index);
+            }
+        }
+    }
+
     /// How the unit::connect tries to get a pooled connection.
     pub fn try_get_connection(&mut self, url: &Url, proxy: &Option<Proxy>) -> Option<Stream> {
         let key = PoolKey::new(url, proxy);
@@ -42,36 +94,85 @@ impl ConnectionPool {
     }
 
     fn remove(&mut self, key: &PoolKey) -> Option<Stream> {
-        if !self.recycle.contains_key(&key) {
-            return None;
+        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();
+                assert!(
+                    stream.is_some(),
+                    "key existed in recycle but no streams available"
+                );
+
+                if streams.len() == 0 {
+                    occupied_entry.remove();
+                }
+
+                // Remove the oldest matching PoolKey from self.lru.
+                // since this PoolKey was most recently used, removing the oldest
+                // PoolKey would delay other streams with this address from
+                // being removed.
+                self.remove_from_lru(key);
+
+                stream
+            }
+            Entry::Vacant(_) => None,
         }
-        let index = self.lru.iter().position(|k| k == key);
-        assert!(
-            index.is_some(),
-            "invariant failed: key existed in recycle but not lru"
-        );
-        self.lru.remove(index.unwrap());
-        self.recycle.remove(&key)
+    }
+
+    fn remove_from_lru(&mut self, key: &PoolKey) {
+        let index = self
+            .lru
+            .iter()
+            .position(|x| x == key)
+            .expect("PoolKey not found in lru");
+        self.lru.remove(index);
     }
 
     fn add(&mut self, key: PoolKey, stream: Stream) {
-        // If an entry with the same key already exists, remove it.
-        // The more recently used stream is likely to live longer.
-        self.remove(&key);
-        if self.recycle.len() + 1 > MAX_IDLE_CONNECTIONS {
-            self.remove_oldest();
+        if self.max_idle_connections == 0 || self.max_idle_connections_per_host == 0 {
+            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 {
+                    streams.pop_front();
+                    self.remove_from_lru(&key);
+                }
+            }
+            Entry::Vacant(vacant_entry) => {
+                let mut new_deque = VecDeque::new();
+                new_deque.push_back(stream);
+                vacant_entry.insert(new_deque);
+            }
+        }
+        self.lru.push_back(key);
+        if self.lru.len() > self.max_idle_connections {
+            self.remove_oldest()
         }
-        self.lru.push_back(key.clone());
-        self.recycle.insert(key, stream);
     }
 
     fn remove_oldest(&mut self) {
         if let Some(key) = self.lru.pop_front() {
-            let removed = self.recycle.remove(&key);
-            assert!(
-                removed.is_some(),
-                "invariant failed: key existed in lru but not in recycle"
-            );
+            match self.recycle.entry(key) {
+                Entry::Occupied(mut occupied_entry) => {
+                    let streams = occupied_entry.get_mut();
+                    let removed_stream = streams.pop_front();
+                    assert!(
+                        removed_stream.is_some(),
+                        "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")
+                }
+            }
         } else {
             panic!("tried to remove oldest but no entries found!");
         }
@@ -79,7 +180,7 @@ impl ConnectionPool {
 
     #[cfg(test)]
     pub fn len(&self) -> usize {
-        self.recycle.len()
+        self.lru.len()
     }
 }
 
@@ -123,10 +224,12 @@ fn poolkey_new() {
 }
 
 #[test]
-fn pool_size_limit() {
-    assert_eq!(MAX_IDLE_CONNECTIONS, 100);
+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::new();
-    let hostnames = (0..200).map(|i| format!("{}.example", i));
+    let hostnames = (0..DEFAULT_MAX_IDLE_CONNECTIONS * 2).map(|i| format!("{}.example", i));
     let poolkeys = hostnames.map(|hostname| PoolKey {
         scheme: "https".to_string(),
         hostname,
@@ -136,22 +239,49 @@ fn pool_size_limit() {
     for key in poolkeys.clone() {
         pool.add(key, Stream::Cursor(std::io::Cursor::new(vec![])));
     }
-    assert_eq!(pool.len(), 100);
+    assert_eq!(pool.len(), DEFAULT_MAX_IDLE_CONNECTIONS);
 
-    for key in poolkeys.skip(100) {
+    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_duplicates_limit() {
-    // Test inserting duplicates into the pool, and subsequently
-    // filling and draining it. The duplicates should evict earlier
-    // entries with the same key.
-    assert_eq!(MAX_IDLE_CONNECTIONS, 100);
+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::new();
-    let hostnames = (0..100).map(|i| format!("{}.example", i));
+    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::new();
+    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,
@@ -159,15 +289,35 @@ fn pool_duplicates_limit() {
         proxy: None,
     });
     for key in poolkeys.clone() {
-        pool.add(key.clone(), Stream::Cursor(std::io::Cursor::new(vec![])));
         pool.add(key, Stream::Cursor(std::io::Cursor::new(vec![])));
     }
-    assert_eq!(pool.len(), 100);
+    assert_eq!(pool.len(), 50);
+    pool.set_max_idle_connections(25);
+    assert_eq!(pool.len(), 25);
+}
 
-    for key in poolkeys {
-        let result = pool.remove(&key);
-        assert!(result.is_some(), "expected key was not in pool");
+#[test]
+fn pool_update_per_host_connection_limit() {
+    let mut pool = ConnectionPool::new();
+    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]
diff --git a/src/request.rs b/src/request.rs
index 3e72d3c..8a8b44e 100644
--- a/src/request.rs
+++ b/src/request.rs
@@ -1,16 +1,12 @@
+use std::fmt;
 use std::io::Read;
+use std::result::Result;
 use std::sync::{Arc, Mutex};
 use std::time;
 
 use qstring::QString;
 use url::{form_urlencoded, Url};
 
-#[cfg(feature = "tls")]
-use std::fmt;
-
-#[cfg(all(feature = "native-tls", not(feature = "tls")))]
-use std::fmt;
-
 use crate::agent::{self, Agent, AgentState};
 use crate::body::{Payload, SizedReader};
 use crate::error::Error;
@@ -54,8 +50,8 @@ pub struct Request {
     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| {
@@ -462,32 +458,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:
@@ -608,12 +578,15 @@ impl Request {
     ///
     /// Example:
     /// ```
-    /// let tls_connector = std::sync::Arc::new(native_tls::TlsConnector::new());
+    /// 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 {
+    pub fn set_tls_connector(
+        &mut self,
+        tls_connector: Arc<native_tls::TlsConnector>,
+    ) -> &mut Request {
         self.tls_connector = Some(TLSConnector(tls_connector));
         self
     }
diff --git a/src/response.rs b/src/response.rs
index 3611567..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: {}]",
@@ -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'
diff --git a/src/stream.rs b/src/stream.rs
index 042821e..b0f8309 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -1,4 +1,7 @@
-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;
@@ -25,14 +28,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 +76,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
+        })
     }
 }
 
@@ -82,16 +94,17 @@ impl Read for DeadlineStream {
 fn time_until_deadline(deadline: Instant) -> IoResult<Duration> {
     let now = Instant::now();
     match deadline.checked_duration_since(now) {
-        None => Err(IoError::new(
-            ErrorKind::TimedOut,
-            "timed out reading response",
-        )),
+        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 +164,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 +196,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 +238,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 +249,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 +276,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 +289,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 +307,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"))]
@@ -306,7 +351,7 @@ pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
 
     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")))]
@@ -328,7 +373,7 @@ pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
             _ => Error::BadStatusRead,
         })?;
 
-    Ok(Stream::Https(stream))
+    Ok(Stream::Https(BufReader::new(stream)))
 }
 
 pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<TcpStream, Error> {
@@ -531,20 +576,17 @@ fn connect_socks5(
         let done = lock.lock().unwrap();
 
         let timeout_connect = time_until_deadline(deadline)?;
-        let done_result = cvar
-            .wait_timeout(done, timeout_connect)
-            .unwrap();
+        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.as_millis()
-                ),
-            ));
+            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)?
diff --git a/src/unit.rs b/src/unit.rs
index 964214c..37530e8 100644
--- a/src/unit.rs
+++ b/src/unit.rs
@@ -61,7 +61,7 @@ impl Unit {
 
         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![];
@@ -86,7 +86,7 @@ impl Unit {
         let headers: Vec<_> = req
             .headers
             .iter()
-            .chain(cookie_headers.iter())
+            .chain(cookie_header.iter())
             .chain(extra_headers.iter())
             .cloned()
             .collect();
@@ -239,51 +239,56 @@ 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<Option<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),
-    }
+    state
+        .as_ref()
+        .map(|state| &state.jar)
+        .and_then(|jar| match_cookies(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<Option<AgentState>>, _url: &Url) -> Option<Header> {
+    None
 }
 
 // TODO check so cookies can't be set for tld:s
 #[cfg(feature = "cookie")]
-fn match_cookies(jar: &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.
@@ -419,3 +424,39 @@ fn save_cookies(unit: &Unit, resp: &Response) {
         }
     }
 }
+
+#[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/src/test/agent_test.rs b/src/test/agent_test.rs
index 66356cd..58bdd31 100644
--- a/src/test/agent_test.rs
+++ b/src/test/agent_test.rs
@@ -99,3 +99,50 @@ fn connection_reuse() {
     }
     assert_eq!(resp.status(), 200);
 }
+
+#[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("Cookie: first")).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("Cookie: first")).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());
+}
diff --git a/src/test/testserver.rs b/src/test/testserver.rs
index 7c563a1..45b70a8 100644
--- a/src/test/testserver.rs
+++ b/src/test/testserver.rs
@@ -9,14 +9,35 @@ pub struct TestServer {
     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"
+    pub fn path(&self) -> &str {
+        if self.0.len() == 0 {
+            ""
+        } else {
+            &self.0[0].split(" ").nth(1).unwrap()
+        }
+    }
+
+    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) {
+pub fn read_headers(stream: &TcpStream) -> TestHeaders {
+    let mut results = vec![];
     for line in BufReader::new(stream).lines() {
-        if line.unwrap() == "" {
-            break;
-        }
+        match line {
+            Err(e) => panic!(e),
+            Ok(line) if line == "" => break,
+            Ok(line) => results.push(line),
+        };
     }
+    TestHeaders(results)
 }
 
 impl TestServer {
diff --git a/src/test/timeout.rs b/src/test/timeout.rs
index 0ccafca..cb69055 100644
--- a/src/test/timeout.rs
+++ b/src/test/timeout.rs
@@ -8,9 +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(mut stream: TcpStream) -> io::Result<()> {
+fn dribble_body_respond(mut stream: TcpStream, contents: &[u8]) -> io::Result<()> {
     read_headers(&stream);
-    let contents = [b'a'; 300];
     let headers = format!(
         "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n",
         contents.len() * 2
@@ -32,7 +31,6 @@ fn get_and_expect_timeout(url: String) {
 
     match resp.into_string() {
         Err(io_error) => match io_error.kind() {
-            io::ErrorKind::WouldBlock => Ok(()),
             io::ErrorKind::TimedOut => Ok(()),
             _ => Err(format!("{:?}", io_error)),
         },
@@ -44,7 +42,7 @@ fn get_and_expect_timeout(url: String) {
 #[test]
 fn overall_timeout_during_body() {
     // Start a test server on an available port, that dribbles out a response at 1 write per 10ms.
-    let server = TestServer::new(dribble_body_respond);
+    let server = TestServer::new(|stream| dribble_body_respond(stream, &[b'a'; 300]));
     let url = format!("http://localhost:{}/", server.port);
     get_and_expect_timeout(url);
 }
@@ -68,3 +66,33 @@ fn overall_timeout_during_headers() {
     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");
+}