Repository: algesten/ureq
Pull request #161 (2020-09-27 03:45:24)

Replace IoResult and IoError with io:: versions.


Patch:
diff --git a/src/body.rs b/src/body.rs
index b6b188f..1a99792 100644
--- a/src/body.rs
+++ b/src/body.rs
@@ -1,6 +1,6 @@
 use crate::stream::Stream;
 use std::fmt;
-use std::io::{copy, empty, Cursor, Read, Result as IoResult, Write};
+use std::io::{self, copy, empty, Cursor, Read, Write};
 
 #[cfg(feature = "charset")]
 use crate::response::DEFAULT_CHARACTER_SET;
@@ -43,11 +43,21 @@ 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>,
 }
 
@@ -58,7 +68,7 @@ impl fmt::Debug for SizedReader {
 }
 
 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 }
     }
 }
@@ -66,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 = {
@@ -79,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))
             }
         }
     }
@@ -109,7 +119,7 @@ const CHUNK_MAX_PAYLOAD_SIZE: usize = CHUNK_MAX_SIZE - CHUNK_HEADER_MAX_SIZE - C
 // 2) chunked_transfer's Encoder issues 4 separate write() per chunk. This is costly
 //    overhead. Instead, we do a single write() per chunk.
 // The measured benefit on a Linux machine is a 50% reduction in CPU usage on a https connection.
-fn copy_chunked<R: Read, W: Write>(reader: &mut R, writer: &mut W) -> IoResult<u64> {
+fn copy_chunked<R: Read, W: Write>(reader: &mut R, writer: &mut W) -> io::Result<u64> {
     // The chunk layout is:
     // header:header_max_size | payload:max_payload_size | footer:footer_size
     let mut chunk = Vec::with_capacity(CHUNK_MAX_SIZE);
@@ -167,7 +177,7 @@ pub(crate) fn send_body(
     mut body: SizedReader,
     do_chunk: bool,
     stream: &mut Stream,
-) -> IoResult<()> {
+) -> io::Result<()> {
     if do_chunk {
         copy_chunked(&mut body.reader, stream)?;
     } else {
diff --git a/src/error.rs b/src/error.rs
index 9671a57..704ff76 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -1,5 +1,5 @@
 use std::fmt;
-use std::io::Error as IoError;
+use std::io;
 
 /// Errors that are translated to ["synthetic" responses](struct.Response.html#method.synthetic).
 #[derive(Debug)]
@@ -22,7 +22,7 @@ pub enum Error {
     /// A header line that couldn't be parsed. Synthetic error `500`.
     BadHeader,
     /// Some unspecified `std::io::Error`. Synthetic error `500`.
-    Io(IoError),
+    Io(io::Error),
     /// Proxy information was not properly formatted
     BadProxy,
     /// Proxy credentials were not properly formatted
@@ -110,8 +110,8 @@ impl Error {
     }
 }
 
-impl From<IoError> for Error {
-    fn from(err: IoError) -> Error {
+impl From<io::Error> for Error {
+    fn from(err: io::Error) -> Error {
         Error::Io(err)
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index de2129f..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.
diff --git a/src/pool.rs b/src/pool.rs
index 5c3482f..c984135 100644
--- a/src/pool.rs
+++ b/src/pool.rs
@@ -1,6 +1,6 @@
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, VecDeque};
-use std::io::{Read, Result as IoResult};
+use std::io::{self, Read};
 
 use crate::stream::Stream;
 use crate::unit::Unit;
@@ -8,7 +8,6 @@ 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;
 
@@ -79,13 +78,6 @@ impl ConnectionPool {
             return;
         }
 
-        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();
@@ -403,7 +395,7 @@ impl<R: Read + Sized + Into<Stream>> PoolReturnRead<R> {
         }
     }
 
-    fn do_read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+    fn do_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         match self.reader.as_mut() {
             None => Ok(0),
             Some(reader) => reader.read(buf),
@@ -412,7 +404,7 @@ impl<R: Read + Sized + Into<Stream>> PoolReturnRead<R> {
 }
 
 impl<R: Read + Sized + Into<Stream>> Read for PoolReturnRead<R> {
-    fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         let amount = self.do_read(buf)?;
         // only if the underlying reader is exhausted can we send a new
         // request to the same socket. hence, we only return it now.
@@ -422,9 +414,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/request.rs b/src/request.rs
index 4f1e6aa..85bce0c 100644
--- a/src/request.rs
+++ b/src/request.rs
@@ -8,10 +8,10 @@ use qstring::QString;
 use url::{form_urlencoded, Url};
 
 use crate::agent::{self, Agent, AgentState};
+use crate::body::BodySize;
 use crate::body::{Payload, SizedReader};
 use crate::error::Error;
 use crate::header::{self, Header};
-use crate::pool;
 use crate::unit::{self, Unit};
 use crate::Response;
 
@@ -210,8 +210,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.
     ///
@@ -222,7 +224,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 {
@@ -498,8 +499,13 @@ impl Request {
     /// assert_eq!(req2.get_host().unwrap(), "localhost");
     /// ```
     pub fn get_host(&self) -> Result<String, Error> {
-        self.to_url()
-            .map(|u| u.host_str().unwrap_or(pool::DEFAULT_HOST).to_string())
+        match self.to_url() {
+            Ok(u) => match u.host_str() {
+                Some(host) => Ok(host.to_string()),
+                None => Err(Error::BadUrl("No hostname in URL".into())),
+            },
+            Err(e) => Err(e),
+        }
     }
 
     /// Returns the scheme for this request.
@@ -603,8 +609,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
     }
 }
 
@@ -629,3 +641,13 @@ impl fmt::Debug for TLSConnector {
         f.debug_struct("TLSConnector").finish()
     }
 }
+
+#[test]
+fn no_hostname() {
+    let req = Request::new(
+        &Agent::default(),
+        "GET".to_string(),
+        "unix:/run/foo.socket".to_string(),
+    );
+    assert!(req.get_host().is_err());
+}
diff --git a/src/response.rs b/src/response.rs
index 4e2d1c2..3e7dcc5 100644
--- a/src/response.rs
+++ b/src/response.rs
@@ -1,5 +1,5 @@
 use std::fmt;
-use std::io::{Cursor, Error as IoError, ErrorKind, Read, Result as IoResult};
+use std::io::{self, Cursor, ErrorKind, Read};
 use std::str::FromStr;
 use std::time::Instant;
 
@@ -366,7 +366,7 @@ impl Response {
     ///
     /// I.e. `Content-Length: text/plain; charset=iso-8859-1` would be decoded in latin-1.
     ///
-    pub fn into_string(self) -> IoResult<String> {
+    pub fn into_string(self) -> io::Result<String> {
         #[cfg(feature = "charset")]
         {
             let encoding = encoding_from_whatwg_label(self.charset())
@@ -400,22 +400,22 @@ impl Response {
     /// assert_eq!(json["hello"], "world");
     /// ```
     #[cfg(feature = "json")]
-    pub fn into_json(self) -> IoResult<serde_json::Value> {
+    pub fn into_json(self) -> io::Result<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.
+            // This is to unify TimedOut io::Error 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 let Some(ioe) = e.source().and_then(|s| s.downcast_ref::<io::Error>()) {
                 if ioe.kind() == ErrorKind::TimedOut {
                     return io_err_timeout(ioe.to_string());
                 }
             }
 
-            IoError::new(
+            io::Error::new(
                 ErrorKind::InvalidData,
                 format!("Failed to read JSON: {}", e),
             )
@@ -445,10 +445,10 @@ impl Response {
     /// assert_eq!(json.hello, "world");
     /// ```
     #[cfg(feature = "json")]
-    pub fn into_json_deserialize<T: DeserializeOwned>(self) -> IoResult<T> {
+    pub fn into_json_deserialize<T: DeserializeOwned>(self) -> io::Result<T> {
         let reader = self.into_reader();
         serde_json::from_reader(reader).map_err(|e| {
-            IoError::new(
+            io::Error::new(
                 ErrorKind::InvalidData,
                 format!("Failed to read JSON: {}", e),
             )
@@ -591,7 +591,7 @@ pub(crate) fn set_stream(resp: &mut Response, url: String, unit: Option<Unit>, s
     resp.stream = Some(stream);
 }
 
-fn read_next_line<R: Read>(reader: &mut R) -> IoResult<String> {
+fn read_next_line<R: Read>(reader: &mut R) -> io::Result<String> {
     let mut buf = Vec::new();
     let mut prev_byte_was_cr = false;
     let mut one = [0_u8];
@@ -600,7 +600,10 @@ fn read_next_line<R: Read>(reader: &mut R) -> IoResult<String> {
         let amt = reader.read(&mut one[..])?;
 
         if amt == 0 {
-            return Err(IoError::new(ErrorKind::ConnectionAborted, "Unexpected EOF"));
+            return Err(io::Error::new(
+                ErrorKind::ConnectionAborted,
+                "Unexpected EOF",
+            ));
         }
 
         let byte = one[0];
@@ -608,7 +611,7 @@ fn read_next_line<R: Read>(reader: &mut R) -> IoResult<String> {
         if byte == b'\n' && prev_byte_was_cr {
             buf.pop(); // removing the '\r'
             return String::from_utf8(buf)
-                .map_err(|_| IoError::new(ErrorKind::InvalidInput, "Header is not in ASCII"));
+                .map_err(|_| io::Error::new(ErrorKind::InvalidInput, "Header is not in ASCII"));
         }
 
         prev_byte_was_cr = byte == b'\r';
@@ -635,7 +638,7 @@ impl<R: Read> LimitedRead<R> {
 }
 
 impl<R: Read> Read for LimitedRead<R> {
-    fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         let left = self.limit - self.position;
         if left == 0 {
             return Ok(0);
@@ -651,7 +654,7 @@ impl<R: Read> Read for LimitedRead<R> {
             // 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(
+            Ok(0) => Err(io::Error::new(
                 ErrorKind::InvalidData,
                 "response body closed before all bytes were read",
             )),
diff --git a/src/stream.rs b/src/stream.rs
index c51520f..5c79547 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -1,7 +1,5 @@
 use std::fmt;
-use std::io::{
-    BufRead, BufReader, Cursor, Error as IoError, ErrorKind, Read, Result as IoResult, Write,
-};
+use std::io::{self, BufRead, BufReader, Cursor, ErrorKind, Read, Write};
 use std::net::SocketAddr;
 use std::net::TcpStream;
 use std::time::Duration;
@@ -67,7 +65,7 @@ impl From<DeadlineStream> for Stream {
 }
 
 impl Read for DeadlineStream {
-    fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         if let Some(deadline) = self.deadline {
             let timeout = time_until_deadline(deadline)?;
             if let Some(socket) = self.stream.socket() {
@@ -90,7 +88,7 @@ impl Read for DeadlineStream {
 
 // If the deadline is in the future, return the remaining time until
 // then. Otherwise return a TimedOut error.
-fn time_until_deadline(deadline: Instant) -> IoResult<Duration> {
+fn time_until_deadline(deadline: Instant) -> io::Result<Duration> {
     let now = Instant::now();
     match deadline.checked_duration_since(now) {
         None => Err(io_err_timeout("timed out reading response".to_string())),
@@ -98,8 +96,8 @@ fn time_until_deadline(deadline: Instant) -> IoResult<Duration> {
     }
 }
 
-pub(crate) fn io_err_timeout(error: String) -> IoError {
-    IoError::new(ErrorKind::TimedOut, error)
+pub(crate) fn io_err_timeout(error: String) -> io::Error {
+    io::Error::new(ErrorKind::TimedOut, error)
 }
 
 impl fmt::Debug for Stream {
@@ -128,7 +126,7 @@ impl Stream {
     // connection: return true. If this returns WouldBlock (aka EAGAIN),
     // that means the connection is still open: return false. Otherwise
     // return an error.
-    fn serverclosed_stream(stream: &std::net::TcpStream) -> IoResult<bool> {
+    fn serverclosed_stream(stream: &std::net::TcpStream) -> io::Result<bool> {
         let mut buf = [0; 1];
         stream.set_nonblocking(true)?;
 
@@ -143,7 +141,7 @@ impl Stream {
         result
     }
     // Return true if the server has closed this connection.
-    pub(crate) fn server_closed(&self) -> IoResult<bool> {
+    pub(crate) fn server_closed(&self) -> io::Result<bool> {
         match self.socket() {
             Some(socket) => Stream::serverclosed_stream(socket),
             None => Ok(false),
@@ -180,7 +178,7 @@ impl Stream {
 }
 
 impl Read for Stream {
-    fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         match self {
             Stream::Http(sock) => sock.read(buf),
             #[cfg(any(
@@ -196,7 +194,7 @@ impl Read for Stream {
 }
 
 impl BufRead for Stream {
-    fn fill_buf(&mut self) -> IoResult<&[u8]> {
+    fn fill_buf(&mut self) -> io::Result<&[u8]> {
         match self {
             Stream::Http(r) => r.fill_buf(),
             #[cfg(any(
@@ -239,7 +237,7 @@ where
 fn read_https(
     stream: &mut BufReader<StreamOwned<ClientSession, TcpStream>>,
     buf: &mut [u8],
-) -> IoResult<usize> {
+) -> io::Result<usize> {
     match stream.read(buf) {
         Ok(size) => Ok(size),
         Err(ref e) if is_close_notify(e) => Ok(0),
@@ -248,7 +246,7 @@ fn read_https(
 }
 
 #[cfg(all(feature = "native-tls", not(feature = "tls")))]
-fn read_https(stream: &mut BufReader<TlsStream<TcpStream>>, buf: &mut [u8]) -> IoResult<usize> {
+fn read_https(stream: &mut BufReader<TlsStream<TcpStream>>, buf: &mut [u8]) -> io::Result<usize> {
     match stream.read(buf) {
         Ok(size) => Ok(size),
         Err(ref e) if is_close_notify(e) => Ok(0),
@@ -273,7 +271,7 @@ fn is_close_notify(e: &std::io::Error) -> bool {
 }
 
 impl Write for Stream {
-    fn write(&mut self, buf: &[u8]) -> IoResult<usize> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         match self {
             Stream::Http(sock) => sock.get_mut().write(buf),
             #[cfg(any(
@@ -286,7 +284,7 @@ impl Write for Stream {
             Stream::Test(_, writer) => writer.write(buf),
         }
     }
-    fn flush(&mut self) -> IoResult<()> {
+    fn flush(&mut self) -> io::Result<()> {
         match self {
             Stream::Http(sock) => sock.get_mut().flush(),
             #[cfg(any(
@@ -301,9 +299,8 @@ impl Write for Stream {
     }
 }
 
-pub(crate) fn connect_http(unit: &Unit) -> Result<Stream, Error> {
+pub(crate) fn connect_http(unit: &Unit, hostname: &str) -> Result<Stream, Error> {
     //
-    let hostname = unit.url.host_str().unwrap();
     let port = unit.url.port().unwrap_or(80);
 
     connect_host(unit, hostname, port)
@@ -325,7 +322,7 @@ fn configure_certs(config: &mut rustls::ClientConfig) {
 }
 
 #[cfg(all(feature = "tls", not(feature = "native-tls")))]
-pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
+pub(crate) fn connect_https(unit: &Unit, hostname: &str) -> Result<Stream, Error> {
     use lazy_static::lazy_static;
     use std::sync::Arc;
 
@@ -337,7 +334,6 @@ pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
         };
     }
 
-    let hostname = unit.url.host_str().unwrap();
     let port = unit.url.port().unwrap_or(443);
 
     let sni = webpki::DNSNameRef::try_from_ascii_str(hostname)
@@ -358,10 +354,9 @@ pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
 }
 
 #[cfg(all(feature = "native-tls", not(feature = "tls")))]
-pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
+pub(crate) fn connect_https(unit: &Unit, hostname: &str) -> 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)?;
 
@@ -657,6 +652,6 @@ pub(crate) fn connect_test(unit: &Unit) -> Result<Stream, Error> {
 }
 
 #[cfg(not(any(feature = "tls", feature = "native-tls")))]
-pub(crate) fn connect_https(unit: &Unit) -> Result<Stream, Error> {
+pub(crate) fn connect_https(unit: &Unit, _hostname: &str) -> Result<Stream, Error> {
     Err(Error::UnknownScheme(unit.url.scheme().to_string()))
 }
diff --git a/src/unit.rs b/src/unit.rs
index f3822f2..506bd83 100644
--- a/src/unit.rs
+++ b/src/unit.rs
@@ -1,4 +1,4 @@
-use std::io::{Result as IoResult, Write};
+use std::io::{self, Write};
 use std::time;
 
 use qstring::QString;
@@ -8,15 +8,12 @@ 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::{Error, Header, Request, Response};
 
-#[cfg(feature = "cookie")]
-use crate::pool::DEFAULT_HOST;
-
 /// It's a "unit of work". Maybe a bad name for it?
 ///
 /// *Internal API*
@@ -35,16 +32,25 @@ 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_header: Option<Header> = extract_cookies(&req.agent, &url);
+        let cookie_header: Option<Header> = url
+            .host_str()
+            .and_then(|host_str| extract_cookies(&req.agent, &url.scheme(), host_str, &url.path()));
 
         let extra_headers = {
             let mut extra = vec![];
@@ -52,8 +58,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 => {}
                 }
             }
 
@@ -125,8 +144,9 @@ pub(crate) fn connect(
 ) -> Result<Response, Error> {
     //
 
+    let host = req.get_host()?;
     // open socket
-    let (mut stream, is_recycled) = connect_socket(&unit, use_pooled)?;
+    let (mut stream, is_recycled) = connect_socket(&unit, &host, use_pooled)?;
 
     let send_result = send_prelude(&unit, &mut stream, redir);
 
@@ -218,16 +238,25 @@ pub(crate) fn connect(
 }
 
 #[cfg(feature = "cookie")]
-fn extract_cookies(state: &std::sync::Mutex<AgentState>, url: &Url) -> Option<Header> {
+fn extract_cookies(
+    state: &std::sync::Mutex<AgentState>,
+    scheme: &str,
+    host: &str,
+    path: &str,
+) -> 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();
+    let is_secure = scheme.eq_ignore_ascii_case("https");
 
-    match_cookies(&state.jar, &hostname, url.path(), is_secure)
+    match_cookies(&state.jar, host, path, is_secure)
 }
 
 #[cfg(not(feature = "cookie"))]
-fn extract_cookies(_state: &std::sync::Mutex<AgentState>, _url: &Url) -> Option<Header> {
+fn extract_cookies(
+    _state: &std::sync::Mutex<AgentState>,
+    _scheme: &str,
+    _host: &str,
+    _path: &str,
+) -> Option<Header> {
     None
 }
 
@@ -278,7 +307,7 @@ pub(crate) fn combine_query(url: &Url, query: &QString, mix_queries: bool) -> St
 }
 
 /// Connect the socket, either by using the pool or grab a new one.
-fn connect_socket(unit: &Unit, use_pooled: bool) -> Result<(Stream, bool), Error> {
+fn connect_socket(unit: &Unit, hostname: &str, use_pooled: bool) -> Result<(Stream, bool), Error> {
     match unit.url.scheme() {
         "http" | "https" | "test" => (),
         _ => return Err(Error::UnknownScheme(unit.url.scheme().to_string())),
@@ -296,8 +325,8 @@ fn connect_socket(unit: &Unit, use_pooled: bool) -> Result<(Stream, bool), Error
         }
     }
     let stream = match unit.url.scheme() {
-        "http" => stream::connect_http(&unit),
-        "https" => stream::connect_https(&unit),
+        "http" => stream::connect_http(&unit, hostname),
+        "https" => stream::connect_https(&unit, hostname),
         "test" => connect_test(&unit),
         _ => Err(Error::UnknownScheme(unit.url.scheme().to_string())),
     };
@@ -306,7 +335,7 @@ fn connect_socket(unit: &Unit, use_pooled: bool) -> Result<(Stream, bool), Error
 
 /// Send request line + headers (all up until the body).
 #[allow(clippy::write_with_newline)]
-fn send_prelude(unit: &Unit, stream: &mut Stream, redir: bool) -> IoResult<()> {
+fn send_prelude(unit: &Unit, stream: &mut Stream, redir: bool) -> io::Result<()> {
     //
 
     // build into a buffer and send in one go.
@@ -388,7 +417,7 @@ fn save_cookies(unit: &Unit, resp: &Response) {
         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();
+            let host = &unit.url.host_str().unwrap().to_string();
             format!("{}; Domain={}", raw_cookie, host)
         };
         match Cookie::parse_encoded(&to_parse[..]) {


Test patch:
diff --git a/src/test/agent_test.rs b/src/test/agent_test.rs
index ea107ec..97568b3 100644
--- a/src/test/agent_test.rs
+++ b/src/test/agent_test.rs
@@ -2,7 +2,7 @@
 
 use crate::test;
 use crate::test::testserver::{read_headers, TestServer};
-use std::io::{self, Write};
+use std::io::{self, Read, Write};
 use std::net::TcpStream;
 use std::time::Duration;
 
@@ -183,3 +183,49 @@ fn test_cookies_on_redirect() {
     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(())
+}