Repository: algesten/ureq
Pull request #519 (2022-05-19 09:12:24)

feat: Set multi query parameters
Add a `query_vec` function for setting multi query parameters. To avoid repeated deserialization and serialization. 

Signed-off-by: zu1k <i@zu1k.com>

algesten: Hi! Thanks for this! If we are to add this, I think the API should avoid `Vec`, and be:

```rs
 pub fn query_pairs(mut self, pairs: &[(&str, &str)]) -> Self {
```

or it could go all the way generic, but we generally have avoided "complicated" trait bounds in other contexts.

```rs
 pub fn query_pairs(mut self, pairs: impl IntoIter<(&str, &str)>) -> Self {
```

@jsha how do you feel about this API addition?
algesten: I think this looks good. Unless @jsha objects, we can merge it.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cf5b1a8..9abf815 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,15 @@
+# 2.5.0
+ * Add tcp no_delay option (#465)
+ * Rework public TLS traits
+ * Warn if requests aren't sent (#490)
+ * Fixes for returning stream to pool (#509)
+ * Avoid extra syscalls when content is buffered (#508)
+ * Remove dep on sync_wrapper (#514, #528)
+ * Error instead of panic on large deadlines (#517)
+ * Make ReadWrite trait simpler (used in bespoke TLS impls) (#530)
+ * Buffer short response bodies (#531)
+ * Update cookie/cookie_store dep
+
 # 2.4.0
 
  * Enable `gzip` feature by default (#455)
diff --git a/Cargo.toml b/Cargo.toml
index ca660ae..0b01876 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ureq"
-version = "2.4.0"
+version = "2.5.0"
 authors = ["Martin Algesten <martin@algesten.se>", "Jacob Hoffman-Andrews <ureq@hoffman-andrews.com>"]
 description = "Simple, safe HTTP client"
 license = "MIT/Apache-2.0"
@@ -50,7 +50,6 @@ serde = { version = "1", features = ["derive"] }
 env_logger = "0.9"
 rustls = { version = "0.20", features = ["dangerous_configuration"] }
 rustls-pemfile = { version = "1.0" }
-mbedtls = { version = "0.8.1" }
 
 [[example]]
 name = "cureq"
diff --git a/README.md b/README.md
index 5d97d5d..2df20f6 100644
--- a/README.md
+++ b/README.md
@@ -2,8 +2,27 @@
 
 # ureq
 
+<div align="center">
+ <!-- Version -->
+ <a href="https://crates.io/crates/ureq">
+   <img src="https://img.shields.io/crates/v/ureq.svg?style=flat-square"
+   alt="Crates.io version" />
+ </a>
+ <!-- Docs -->
+ <a href="https://docs.rs/ureq">
+   <img src="https://img.shields.io/badge/docs-latest-blue.svg?style=flat-square"
+     alt="docs.rs docs" />
+ </a>
+ <!-- Downloads -->
+ <a href="https://crates.io/crates/ureq">
+   <img src="https://img.shields.io/crates/d/ureq.svg?style=flat-square"
+     alt="Crates.io downloads" />
+ </a>
+</div>
+
 A simple, safe HTTP client.
 
+
 Ureq's first priority is being easy for you to use. It's great for
 anyone who wants a low-overhead HTTP client that just gets the job done. Works
 very well with HTTP APIs. Its features include cookies, JSON, HTTP proxies,
diff --git a/examples/custom-tls.rs b/examples/custom-tls.rs
index 43eba25..249c796 100644
--- a/examples/custom-tls.rs
+++ b/examples/custom-tls.rs
@@ -52,9 +52,6 @@ impl ReadWrite for CustomTlsStream {
     fn socket(&self) -> Option<&TcpStream> {
         self.0.socket()
     }
-    fn is_poolable(&self) -> bool {
-        self.0.is_poolable()
-    }
 }
 
 impl io::Read for CustomTlsStream {
diff --git a/examples/mbedtls/Cargo.toml b/examples/mbedtls/Cargo.toml
new file mode 100644
index 0000000..51abba1
--- /dev/null
+++ b/examples/mbedtls/Cargo.toml
@@ -0,0 +1,8 @@
+[package]
+name = "mbedtls-example"
+version = "0.0.1"
+edition = "2021"
+
+
+[dependencies]
+mbedtls = { version = "0.8.1" }
diff --git a/examples/mbedtls/README.md b/examples/mbedtls/README.md
new file mode 100644
index 0000000..0e493c0
--- /dev/null
+++ b/examples/mbedtls/README.md
@@ -0,0 +1,9 @@
+mbedtls-example
+===============
+
+To run this example you need to `cd` into the directory and use `cargo run`.
+
+```
+cd ureq/exampled/mbedtls
+cargo run
+```
diff --git a/examples/mbedtls/main.rs b/examples/mbedtls/src/main.rs
similarity index 100%
rename from examples/mbedtls/main.rs
rename to examples/mbedtls/src/main.rs
diff --git a/examples/mbedtls/mbedtls_connector.rs b/examples/mbedtls/src/mbedtls_connector.rs
similarity index 98%
rename from examples/mbedtls/mbedtls_connector.rs
rename to examples/mbedtls/src/mbedtls_connector.rs
index 941c6b4..856db7a 100644
--- a/examples/mbedtls/mbedtls_connector.rs
+++ b/examples/mbedtls/src/mbedtls_connector.rs
@@ -116,9 +116,6 @@ impl ReadWrite for MbedTlsStream {
     fn socket(&self) -> Option<&TcpStream> {
         None
     }
-    fn is_poolable(&self) -> bool {
-        true
-    }
 }
 
 impl io::Read for MbedTlsStream {
diff --git a/src/error.rs b/src/error.rs
index 3fec78b..9610fc7 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -328,7 +328,7 @@ impl Error {
 }
 
 /// One of the types of error the can occur when processing a Request.
-#[derive(Debug, PartialEq, Clone, Copy)]
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum ErrorKind {
     /// The url could not be understood.
     InvalidUrl,
diff --git a/src/header.rs b/src/header.rs
index db9d62e..ff8005e 100644
--- a/src/header.rs
+++ b/src/header.rs
@@ -62,7 +62,7 @@ impl fmt::Display for HeaderLine {
     }
 }
 
-#[derive(Clone, PartialEq)]
+#[derive(Clone, PartialEq, Eq)]
 /// Wrapper type for a header field.
 /// <https://tools.ietf.org/html/rfc7230#section-3.2>
 pub struct Header {
diff --git a/src/lib.rs b/src/lib.rs
index 87c91bc..ad8d813 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -6,9 +6,30 @@
 #![allow(clippy::match_like_matches_macro)]
 // we're not changing public api due to a lint.
 #![allow(clippy::upper_case_acronyms)]
-
+#![allow(clippy::result_large_err)]
+#![allow(clippy::only_used_in_recursion)]
+
+//!<div align="center">
+//!  <!-- Version -->
+//!  <a href="https://crates.io/crates/ureq">
+//!    <img src="https://img.shields.io/crates/v/ureq.svg?style=flat-square"
+//!    alt="Crates.io version" />
+//!  </a>
+//!  <!-- Docs -->
+//!  <a href="https://docs.rs/ureq">
+//!    <img src="https://img.shields.io/badge/docs-latest-blue.svg?style=flat-square"
+//!      alt="docs.rs docs" />
+//!  </a>
+//!  <!-- Downloads -->
+//!  <a href="https://crates.io/crates/ureq">
+//!    <img src="https://img.shields.io/crates/d/ureq.svg?style=flat-square"
+//!      alt="Crates.io downloads" />
+//!  </a>
+//!</div>
+//!
 //! A simple, safe HTTP client.
 //!
+//!
 //! Ureq's first priority is being easy for you to use. It's great for
 //! anyone who wants a low-overhead HTTP client that just gets the job done. Works
 //! very well with HTTP APIs. Its features include cookies, JSON, HTTP proxies,
diff --git a/src/ntls.rs b/src/ntls.rs
index 65b2c00..829f59b 100644
--- a/src/ntls.rs
+++ b/src/ntls.rs
@@ -31,7 +31,4 @@ impl ReadWrite for native_tls::TlsStream<Box<dyn ReadWrite>> {
     fn socket(&self) -> Option<&TcpStream> {
         self.get_ref().socket()
     }
-    fn is_poolable(&self) -> bool {
-        self.get_ref().is_poolable()
-    }
 }
diff --git a/src/pool.rs b/src/pool.rs
index 10e0c4d..812365d 100644
--- a/src/pool.rs
+++ b/src/pool.rs
@@ -248,10 +248,6 @@ impl<R: Read + Sized + Into<Stream>> PoolReturnRead<R> {
         if let Some(reader) = self.reader.take() {
             // bring back stream here to either go into pool or dealloc
             let mut stream = reader.into();
-            if !stream.is_poolable() {
-                // just let it deallocate
-                return Ok(());
-            }
 
             // ensure stream can be reused
             stream.reset()?;
@@ -306,8 +302,44 @@ impl<R: Read + Sized + Done + Into<Stream>> Read for PoolReturnRead<R> {
 
 #[cfg(test)]
 mod tests {
+    use std::io;
+
+    use crate::stream::{remote_addr_for_test, Stream};
+    use crate::ReadWrite;
+
     use super::*;
 
+    #[derive(Debug)]
+    struct NoopStream;
+
+    impl NoopStream {
+        fn stream() -> Stream {
+            Stream::new(NoopStream, remote_addr_for_test())
+        }
+    }
+
+    impl Read for NoopStream {
+        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+            Ok(buf.len())
+        }
+    }
+
+    impl std::io::Write for NoopStream {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl ReadWrite for NoopStream {
+        fn socket(&self) -> Option<&std::net::TcpStream> {
+            None
+        }
+    }
+
     #[test]
     fn poolkey_new() {
         // Test that PoolKey::new() does not panic on unrecognized schemes.
@@ -328,7 +360,7 @@ mod tests {
             proxy: None,
         });
         for key in poolkeys.clone() {
-            pool.add(&key, Stream::from_vec(vec![]))
+            pool.add(&key, NoopStream::stream());
         }
         assert_eq!(pool.len(), pool.max_idle_connections);
 
@@ -353,7 +385,7 @@ mod tests {
         };
 
         for _ in 0..pool.max_idle_connections_per_host * 2 {
-            pool.add(&poolkey, Stream::from_vec(vec![]))
+            pool.add(&poolkey, NoopStream::stream())
         }
         assert_eq!(pool.len(), pool.max_idle_connections_per_host);
 
@@ -372,12 +404,12 @@ mod tests {
         let url = Url::parse("zzz:///example.com").unwrap();
         let pool_key = PoolKey::new(&url, None);
 
-        pool.add(&pool_key, Stream::from_vec(vec![]));
+        pool.add(&pool_key, NoopStream::stream());
         assert_eq!(pool.len(), 1);
 
         let pool_key = PoolKey::new(&url, Some(Proxy::new("localhost:9999").unwrap()));
 
-        pool.add(&pool_key, Stream::from_vec(vec![]));
+        pool.add(&pool_key, NoopStream::stream());
         assert_eq!(pool.len(), 2);
 
         let pool_key = PoolKey::new(
@@ -385,7 +417,7 @@ mod tests {
             Some(Proxy::new("user:password@localhost:9999").unwrap()),
         );
 
-        pool.add(&pool_key, Stream::from_vec(vec![]));
+        pool.add(&pool_key, NoopStream::stream());
         assert_eq!(pool.len(), 3);
     }
 
@@ -396,10 +428,9 @@ mod tests {
         let url = Url::parse("https:///example.com").unwrap();
 
         let mut out_buf = [0u8; 500];
-        let long_vec = vec![0u8; 1000];
 
         let agent = Agent::new();
-        let stream = Stream::from_vec_poolable(long_vec);
+        let stream = NoopStream::stream();
         let limited_read = LimitedRead::new(stream, 500);
 
         let mut pool_return_read = PoolReturnRead::new(&agent, &url, limited_read);
@@ -408,4 +439,59 @@ mod tests {
 
         assert_eq!(agent.state.pool.len(), 1);
     }
+
+    // Test that a stream gets returned to the pool if it is gzip encoded and the gzip
+    // decoder reads the exact amount from a chunked stream, not past the 0. This
+    // happens because gzip has built-in knowledge of the length to read.
+    #[test]
+    #[cfg(feature = "gzip")]
+    fn read_exact_chunked_gzip() {
+        use crate::response::Compression;
+        use chunked_transfer::Decoder as ChunkDecoder;
+
+        let gz_body = vec![
+            b'E', b'\r', b'\n', // 14 first chunk
+            0x1F, 0x8B, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x03, 0xCB, 0x48, 0xCD, 0xC9,
+            b'\r', b'\n', //
+            b'E', b'\r', b'\n', // 14 second chunk
+            0xC9, 0x57, 0x28, 0xCF, 0x2F, 0xCA, 0x49, 0x51, 0xC8, 0x18, 0xBC, 0x6C, 0x00, 0xA5,
+            b'\r', b'\n', //
+            b'7', b'\r', b'\n', // 7 third chunk
+            0x5C, 0x7C, 0xEF, 0xA7, 0x00, 0x00, 0x00, //
+            b'\r', b'\n', //
+            // end
+            b'0', b'\r', b'\n', //
+            b'\r', b'\n', //
+        ];
+
+        println!("{:?}", gz_body);
+
+        impl ReadWrite for io::Cursor<Vec<u8>> {
+            fn socket(&self) -> Option<&std::net::TcpStream> {
+                None
+            }
+        }
+
+        impl From<io::Cursor<Vec<u8>>> for Stream {
+            fn from(c: io::Cursor<Vec<u8>>) -> Self {
+                Stream::new(c, "1.1.1.1:8080".parse().unwrap())
+            }
+        }
+
+        let agent = Agent::new();
+        let url = Url::parse("https://example.com").unwrap();
+
+        assert_eq!(agent.state.pool.len(), 0);
+
+        let chunked = ChunkDecoder::new(io::Cursor::new(gz_body));
+        let pool_return_read: Box<(dyn Read + Send + Sync + 'static)> =
+            Box::new(PoolReturnRead::new(&agent, &url, chunked));
+
+        let compression = Compression::Gzip;
+        let mut stream = compression.wrap_reader(pool_return_read);
+
+        io::copy(&mut stream, &mut io::sink()).unwrap();
+
+        assert_eq!(agent.state.pool.len(), 1);
+    }
 }
diff --git a/src/proxy.rs b/src/proxy.rs
index 3631055..ff94139 100644
--- a/src/proxy.rs
+++ b/src/proxy.rs
@@ -101,7 +101,7 @@ impl Proxy {
         };
 
         let remaining_parts = proxy_parts.next();
-        if remaining_parts == None {
+        if remaining_parts.is_none() {
             return Err(ErrorKind::InvalidProxyUrl.new());
         }
 
@@ -131,7 +131,7 @@ impl Proxy {
 
     pub(crate) fn connect<S: AsRef<str>>(&self, host: S, port: u16) -> String {
         let authorization = if self.use_authorization() {
-            let creds = base64::encode(&format!(
+            let creds = base64::encode(format!(
                 "{}:{}",
                 self.user.clone().unwrap_or_default(),
                 self.password.clone().unwrap_or_default()
diff --git a/src/request.rs b/src/request.rs
index ed2a987..1eda90d 100644
--- a/src/request.rs
+++ b/src/request.rs
@@ -3,12 +3,13 @@ use std::{fmt, time};
 
 use url::{form_urlencoded, ParseError, Url};
 
+use crate::agent::Agent;
 use crate::body::Payload;
+use crate::error::{Error, ErrorKind};
 use crate::header::{self, Header};
 use crate::middleware::MiddlewareNext;
 use crate::unit::{self, Unit};
 use crate::Response;
-use crate::{agent::Agent, error::Error};
 
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -123,7 +124,15 @@ impl Request {
             None => None,
             Some(timeout) => {
                 let now = time::Instant::now();
-                Some(now.checked_add(timeout).unwrap())
+                match now.checked_add(timeout) {
+                    Some(dl) => Some(dl),
+                    None => {
+                        return Err(Error::new(
+                            ErrorKind::Io,
+                            Some("Request deadline overflowed".to_string()),
+                        ))
+                    }
+                }
             }
         };
 
@@ -382,6 +391,43 @@ impl Request {
         self
     }
 
+    /// Set multi query parameters.
+    ///
+    /// For example, to set `?format=json&dest=/login`
+    ///
+    /// ```
+    /// # fn main() -> Result<(), ureq::Error> {
+    /// # ureq::is_test(true);
+    ///
+    /// let query = vec![
+    ///     ("format", "json"),
+    ///     ("dest", "/login"),
+    /// ];
+    ///
+    /// let resp = ureq::get("http://httpbin.org/get")
+    ///     .query_pairs(query)
+    ///     .call()?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn query_pairs<'a, P>(mut self, pairs: P) -> Self
+    where
+        P: IntoIterator<Item = (&'a str, &'a str)>,
+    {
+        if let Ok(mut url) = self.parse_url() {
+            {
+                let mut query_pairs = url.query_pairs_mut();
+                for (param, value) in pairs {
+                    query_pairs.append_pair(param, value);
+                }
+            }
+
+            // replace url
+            self.url = url.to_string();
+        }
+        self
+    }
+
     /// Returns the value of the request method. Something like `GET`, `POST`, `PUT` etc.
     ///
     /// ```
diff --git a/src/resolve.rs b/src/resolve.rs
index a8c368a..8b7087f 100644
--- a/src/resolve.rs
+++ b/src/resolve.rs
@@ -1,28 +1,28 @@
 use std::fmt;
-use std::io::Result as IoResult;
+use std::io;
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::sync::Arc;
 
 /// A custom resolver to override the default DNS behavior.
 pub trait Resolver: Send + Sync {
-    fn resolve(&self, netloc: &str) -> IoResult<Vec<SocketAddr>>;
+    fn resolve(&self, netloc: &str) -> io::Result<Vec<SocketAddr>>;
 }
 
 #[derive(Debug)]
 pub(crate) struct StdResolver;
 
 impl Resolver for StdResolver {
-    fn resolve(&self, netloc: &str) -> IoResult<Vec<SocketAddr>> {
+    fn resolve(&self, netloc: &str) -> io::Result<Vec<SocketAddr>> {
         ToSocketAddrs::to_socket_addrs(netloc).map(|iter| iter.collect())
     }
 }
 
 impl<F> Resolver for F
 where
-    F: Fn(&str) -> IoResult<Vec<SocketAddr>>,
+    F: Fn(&str) -> io::Result<Vec<SocketAddr>>,
     F: Send + Sync,
 {
-    fn resolve(&self, netloc: &str) -> IoResult<Vec<SocketAddr>> {
+    fn resolve(&self, netloc: &str) -> io::Result<Vec<SocketAddr>> {
         self(netloc)
     }
 }
diff --git a/src/response.rs b/src/response.rs
index 58c9830..cb655fe 100644
--- a/src/response.rs
+++ b/src/response.rs
@@ -1,16 +1,17 @@
-use std::io::{self, Read};
+use std::io::{self, Cursor, Read};
+use std::net::SocketAddr;
 use std::str::FromStr;
-use std::sync::Mutex;
 use std::{fmt, io::BufRead};
 
 use chunked_transfer::Decoder as ChunkDecoder;
+use log::debug;
 use url::Url;
 
 use crate::body::SizedReader;
 use crate::error::{Error, ErrorKind::BadStatus};
 use crate::header::{get_all_headers, get_header, Header, HeaderLine};
 use crate::pool::PoolReturnRead;
-use crate::stream::{DeadlineStream, Stream};
+use crate::stream::{DeadlineStream, ReadOnlyStream, Stream};
 use crate::unit::Unit;
 use crate::{stream, Agent, ErrorKind};
 
@@ -21,7 +22,7 @@ use serde::de::DeserializeOwned;
 use encoding_rs::Encoding;
 
 #[cfg(feature = "gzip")]
-use flate2::read::GzDecoder;
+use flate2::read::MultiGzDecoder;
 
 #[cfg(feature = "brotli")]
 use brotli_decompressor::Decompressor as BrotliDecoder;
@@ -68,8 +69,9 @@ pub struct Response {
     headers: Vec<Header>,
     // Boxed to avoid taking up too much size.
     unit: Box<Unit>,
-    // Boxed to avoid taking up too much size.
-    stream: Mutex<Box<Stream>>,
+    reader: Box<dyn Read + Send + Sync + 'static>,
+    /// The socket address of the server that sent the response.
+    remote_addr: SocketAddr,
     /// The redirect history of this response, if any. The history starts with
     /// the first response received and ends with the response immediately
     /// previous to this one.
@@ -224,6 +226,11 @@ impl Response {
         charset_from_content_type(self.header("content-type"))
     }
 
+    /// The socket address of the server that sent the response.
+    pub fn remote_addr(&self) -> SocketAddr {
+        self.remote_addr
+    }
+
     /// Turn this response into a `impl Read` of the body.
     ///
     /// 1. If `Transfer-Encoding: chunked`, the returned reader will unchunk it
@@ -260,7 +267,11 @@ impl Response {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn into_reader(self) -> impl Read + Send {
+    pub fn into_reader(self) -> Box<dyn Read + Send + Sync + 'static> {
+        self.reader
+    }
+
+    fn stream_to_reader(&self, stream: DeadlineStream) -> Box<dyn Read + Send + Sync + 'static> {
         //
         let is_http10 = self.http_version().eq_ignore_ascii_case("HTTP/1.0");
         let is_close = self
@@ -291,27 +302,49 @@ impl Response {
             self.length
         };
 
-        let stream = self.stream.into_inner().unwrap();
-        let unit = self.unit;
-        let result = stream.set_read_timeout(unit.agent.config.timeout_read);
+        let unit = &self.unit;
+        let inner = stream.inner_ref();
+        let result = inner.set_read_timeout(unit.agent.config.timeout_read);
         if let Err(e) = result {
-            return Box::new(ErrorReader(e)) as Box<dyn Read + Send>;
+            return Box::new(ErrorReader(e)) as Box<dyn Read + Send + Sync + 'static>;
         }
-        let deadline = unit.deadline;
-        let stream = DeadlineStream::new(*stream, deadline);
-
-        let body_reader: Box<dyn Read + Send> = match (use_chunked, limit_bytes) {
-            (true, _) => Box::new(PoolReturnRead::new(
-                &unit.agent,
-                &unit.url,
-                ChunkDecoder::new(stream),
-            )),
-            (false, Some(len)) => Box::new(PoolReturnRead::new(
-                &unit.agent,
-                &unit.url,
-                LimitedRead::new(stream, len),
-            )),
-            (false, None) => Box::new(stream),
+        let buffer_len = inner.buffer().len();
+
+        let body_reader: Box<dyn Read + Send + Sync> = match (use_chunked, limit_bytes) {
+            // Chunked responses have an unknown length, but do have an end of body
+            // marker. When we encounter the marker, we can return the underlying stream
+            // to the connection pool.
+            (true, _) => {
+                debug!("Chunked body in response");
+                Box::new(PoolReturnRead::new(
+                    &unit.agent,
+                    &unit.url,
+                    ChunkDecoder::new(stream),
+                ))
+            }
+            // Responses with a content-length header means we should limit the reading
+            // of the body to the number of bytes in the header. Once done, we can
+            // return the underlying stream to the connection pool.
+            (false, Some(len)) => {
+                let mut pooler =
+                    PoolReturnRead::new(&unit.agent, &unit.url, LimitedRead::new(stream, len));
+
+                if len <= buffer_len {
+                    debug!("Body entirely buffered (length: {})", len);
+                    let mut buf = vec![0; len];
+                    pooler
+                        .read_exact(&mut buf)
+                        .expect("failed to read exact buffer length from stream");
+                    Box::new(Cursor::new(buf))
+                } else {
+                    debug!("Streaming body until content-length: {}", len);
+                    Box::new(pooler)
+                }
+            }
+            (false, None) => {
+                debug!("Body of unknown size - read until socket close");
+                Box::new(stream)
+            }
         };
 
         match self.compression {
@@ -350,7 +383,7 @@ impl Response {
     /// Content-Type header, or the Content-Type header does not specify a charset, into_string()
     /// uses `utf-8`.
     ///
-    /// I.e. `Content-Length: text/plain; charset=iso-8859-1` would be decoded in latin-1.
+    /// I.e. `Content-Type: text/plain; charset=iso-8859-1` would be decoded in latin-1.
     ///
     pub fn into_string(self) -> io::Result<String> {
         #[cfg(feature = "charset")]
@@ -470,6 +503,7 @@ impl Response {
     ///
     /// assert_eq!(resp.status(), 401);
     pub(crate) fn do_from_stream(stream: Stream, unit: Unit) -> Result<Response, Error> {
+        let remote_addr = stream.remote_addr;
         //
         // HTTP/1.1 200 OK\r\n
         let mut stream = stream::DeadlineStream::new(stream, unit.deadline);
@@ -507,24 +541,21 @@ impl Response {
 
         let url = unit.url.clone();
 
-        Ok(Response {
+        let mut response = Response {
             url,
             status_line,
             index,
             status,
             headers,
             unit: Box::new(unit),
-            stream: Mutex::new(Box::new(stream.into())),
+            reader: Box::new(Cursor::new(vec![])),
+            remote_addr,
             history: vec![],
             length,
             compression,
-        })
-    }
-
-    #[cfg(test)]
-    pub fn into_written_bytes(self) -> Vec<u8> {
-        // Deliberately consume `self` so that any access to `self.stream` must be non-shared.
-        self.stream.into_inner().unwrap().written_bytes()
+        };
+        response.reader = response.stream_to_reader(stream);
+        Ok(response)
     }
 
     #[cfg(test)]
@@ -540,7 +571,7 @@ impl Response {
 }
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
-enum Compression {
+pub(crate) enum Compression {
     #[cfg(feature = "brotli")]
     Brotli,
     #[cfg(feature = "gzip")]
@@ -561,12 +592,15 @@ impl Compression {
 
     /// Wrap the raw reader with a decompressing reader
     #[allow(unused_variables)] // when no features enabled, reader is unused (unreachable)
-    fn wrap_reader(self, reader: Box<dyn Read + Send>) -> Box<dyn Read + Send> {
+    pub(crate) fn wrap_reader(
+        self,
+        reader: Box<dyn Read + Send + Sync + 'static>,
+    ) -> Box<dyn Read + Send + Sync + 'static> {
         match self {
             #[cfg(feature = "brotli")]
             Compression::Brotli => Box::new(BrotliDecoder::new(reader, 4096)),
             #[cfg(feature = "gzip")]
-            Compression::Gzip => Box::new(GzDecoder::new(reader)),
+            Compression::Gzip => Box::new(MultiGzDecoder::new(reader)),
         }
     }
 }
@@ -644,7 +678,8 @@ impl FromStr for Response {
     /// # }
     /// ```
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        let stream = Stream::from_vec(s.as_bytes().to_owned());
+        let remote_addr = "0.0.0.0:0".parse().unwrap();
+        let stream = Stream::new(ReadOnlyStream::new(s.into()), remote_addr);
         let request_url = "https://example.com".parse().unwrap();
         let request_reader = SizedReader {
             size: crate::body::BodySize::Empty,
@@ -775,9 +810,9 @@ pub(crate) fn charset_from_content_type(header: Option<&str>) -> &str {
     header
         .and_then(|header| {
             header.find(';').and_then(|semi| {
-                (&header[semi + 1..])
+                header[semi + 1..]
                     .find('=')
-                    .map(|equal| (&header[semi + equal + 2..]).trim())
+                    .map(|equal| header[semi + equal + 2..].trim())
             })
         })
         .unwrap_or(DEFAULT_CHARACTER_SET)
@@ -1005,7 +1040,10 @@ mod tests {
             OK",
         );
         let v = cow.to_vec();
-        let s = Stream::from_vec(v);
+        let s = Stream::new(
+            ReadOnlyStream::new(v),
+            crate::stream::remote_addr_for_test(),
+        );
         let request_url = "https://example.com".parse().unwrap();
         let request_reader = SizedReader {
             size: crate::body::BodySize::Empty,
diff --git a/src/rtls.rs b/src/rtls.rs
index a3888c6..30ec0a6 100644
--- a/src/rtls.rs
+++ b/src/rtls.rs
@@ -33,9 +33,6 @@ impl ReadWrite for RustlsStream {
     fn socket(&self) -> Option<&TcpStream> {
         self.0.get_ref().socket()
     }
-    fn is_poolable(&self) -> bool {
-        self.0.get_ref().is_poolable()
-    }
 }
 
 // TODO: After upgrading to rustls 0.20 or higher, we can remove these Read
diff --git a/src/stream.rs b/src/stream.rs
index 223dd9e..5f978a5 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -18,24 +18,14 @@ use crate::error::ErrorKind;
 use crate::unit::Unit;
 
 /// Trait for things implementing [std::io::Read] + [std::io::Write]. Used in [TlsConnector].
-pub trait ReadWrite: Read + Write + Send + fmt::Debug + 'static {
+pub trait ReadWrite: Read + Write + Send + Sync + fmt::Debug + 'static {
     fn socket(&self) -> Option<&TcpStream>;
-    fn is_poolable(&self) -> bool;
-
-    /// The bytes written to the stream as a Vec<u8>. This is used for tests only.
-    #[cfg(test)]
-    fn written_bytes(&self) -> Vec<u8> {
-        panic!("written_bytes on non Test stream");
-    }
 }
 
 impl ReadWrite for TcpStream {
     fn socket(&self) -> Option<&TcpStream> {
         Some(self)
     }
-    fn is_poolable(&self) -> bool {
-        true
-    }
 }
 
 pub trait TlsConnector: Send + Sync {
@@ -48,64 +38,21 @@ pub trait TlsConnector: Send + Sync {
 
 pub(crate) struct Stream {
     inner: BufReader<Box<dyn ReadWrite>>,
+    /// The remote address the stream is connected to.
+    pub(crate) remote_addr: SocketAddr,
 }
 
 impl<T: ReadWrite + ?Sized> ReadWrite for Box<T> {
     fn socket(&self) -> Option<&TcpStream> {
         ReadWrite::socket(self.as_ref())
     }
-    fn is_poolable(&self) -> bool {
-        ReadWrite::is_poolable(self.as_ref())
-    }
-    #[cfg(test)]
-    fn written_bytes(&self) -> Vec<u8> {
-        ReadWrite::written_bytes(self.as_ref())
-    }
-}
-
-struct TestStream(Box<dyn Read + Send + Sync>, Vec<u8>, bool);
-
-impl ReadWrite for TestStream {
-    fn is_poolable(&self) -> bool {
-        self.2
-    }
-    fn socket(&self) -> Option<&TcpStream> {
-        None
-    }
-
-    #[cfg(test)]
-    fn written_bytes(&self) -> Vec<u8> {
-        self.1.clone()
-    }
-}
-
-impl Read for TestStream {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        self.0.read(buf)
-    }
-}
-
-impl Write for TestStream {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.1.write(buf)
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        Ok(())
-    }
-}
-
-impl fmt::Debug for TestStream {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_tuple("TestStream").finish()
-    }
 }
 
 // DeadlineStream wraps a stream such that read() will return an error
 // after the provided deadline, and sets timeouts on the underlying
 // TcpStream to ensure read() doesn't block beyond the deadline.
 // When the From trait is used to turn a DeadlineStream back into a
-// Stream (by PoolReturningRead), the timeouts are removed.
+// Stream (by PoolReturnRead), the timeouts are removed.
 pub(crate) struct DeadlineStream {
     stream: Stream,
     deadline: Option<Instant>,
@@ -115,6 +62,10 @@ impl DeadlineStream {
     pub(crate) fn new(stream: Stream, deadline: Option<Instant>) -> Self {
         DeadlineStream { stream, deadline }
     }
+
+    pub(crate) fn inner_ref(&self) -> &Stream {
+        &self.stream
+    }
 }
 
 impl From<DeadlineStream> for Stream {
@@ -187,6 +138,37 @@ pub(crate) fn io_err_timeout(error: String) -> io::Error {
     io::Error::new(io::ErrorKind::TimedOut, error)
 }
 
+#[derive(Debug)]
+pub(crate) struct ReadOnlyStream(Cursor<Vec<u8>>);
+
+impl ReadOnlyStream {
+    pub(crate) fn new(v: Vec<u8>) -> Self {
+        Self(Cursor::new(v))
+    }
+}
+
+impl Read for ReadOnlyStream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.0.read(buf)
+    }
+}
+
+impl std::io::Write for ReadOnlyStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl ReadWrite for ReadOnlyStream {
+    fn socket(&self) -> Option<&std::net::TcpStream> {
+        None
+    }
+}
+
 impl fmt::Debug for Stream {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self.inner.get_ref().socket() {
@@ -197,9 +179,10 @@ impl fmt::Debug for Stream {
 }
 
 impl Stream {
-    fn new(t: impl ReadWrite) -> Stream {
+    pub(crate) fn new(t: impl ReadWrite, remote_addr: SocketAddr) -> Stream {
         Stream::logged_create(Stream {
             inner: BufReader::new(Box::new(t)),
+            remote_addr,
         })
     }
 
@@ -208,27 +191,8 @@ impl Stream {
         stream
     }
 
-    pub(crate) fn from_vec(v: Vec<u8>) -> Stream {
-        Stream::logged_create(Stream {
-            inner: BufReader::new(Box::new(TestStream(
-                Box::new(Cursor::new(v)),
-                vec![],
-                false,
-            ))),
-        })
-    }
-
-    #[cfg(test)]
-    pub(crate) fn from_vec_poolable(v: Vec<u8>) -> Stream {
-        Stream::logged_create(Stream {
-            inner: BufReader::new(Box::new(TestStream(Box::new(Cursor::new(v)), vec![], true))),
-        })
-    }
-
-    fn from_tcp_stream(t: TcpStream) -> Stream {
-        Stream::logged_create(Stream {
-            inner: BufReader::new(Box::new(t)),
-        })
+    pub(crate) fn buffer(&self) -> &[u8] {
+        self.inner.buffer()
     }
 
     // Check if the server has closed a stream by performing a one-byte
@@ -270,9 +234,6 @@ impl Stream {
             None => Ok(false),
         }
     }
-    pub fn is_poolable(&self) -> bool {
-        self.inner.get_ref().is_poolable()
-    }
 
     pub(crate) fn reset(&mut self) -> io::Result<()> {
         // When we are turning this back into a regular, non-deadline Stream,
@@ -296,11 +257,6 @@ impl Stream {
             Ok(())
         }
     }
-
-    #[cfg(test)]
-    pub fn written_bytes(&self) -> Vec<u8> {
-        self.inner.get_ref().written_bytes()
-    }
 }
 
 impl Read for Stream {
@@ -348,20 +304,25 @@ pub(crate) fn connect_http(unit: &Unit, hostname: &str) -> Result<Stream, Error>
     //
     let port = unit.url.port().unwrap_or(80);
 
-    connect_host(unit, hostname, port).map(Stream::from_tcp_stream)
+    connect_host(unit, hostname, port).map(|(t, r)| Stream::new(t, r))
 }
 
 pub(crate) fn connect_https(unit: &Unit, hostname: &str) -> Result<Stream, Error> {
     let port = unit.url.port().unwrap_or(443);
 
-    let sock = connect_host(unit, hostname, port)?;
+    let (sock, remote_addr) = connect_host(unit, hostname, port)?;
 
     let tls_conf = &unit.agent.config.tls_config;
     let https_stream = tls_conf.connect(hostname, Box::new(sock))?;
-    Ok(Stream::new(https_stream))
+    Ok(Stream::new(https_stream, remote_addr))
 }
 
-pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<TcpStream, Error> {
+/// If successful, returns a `TcpStream` and the remote address it is connected to.
+pub(crate) fn connect_host(
+    unit: &Unit,
+    hostname: &str,
+    port: u16,
+) -> Result<(TcpStream, SocketAddr), Error> {
     let connect_deadline: Option<Instant> =
         if let Some(timeout_connect) = unit.agent.config.timeout_connect {
             Instant::now().checked_add(timeout_connect)
@@ -388,7 +349,7 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
     let proto = proxy.as_ref().map(|proxy| proxy.proto);
 
     let mut any_err = None;
-    let mut any_stream = None;
+    let mut any_stream_and_addr = None;
     // Find the first sock_addr that accepts a connection
     for sock_addr in sock_addrs {
         // ensure connect timeout or overall timeout aren't yet hit.
@@ -400,7 +361,8 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
         debug!("connecting to {} at {}", netloc, &sock_addr);
 
         // connect with a configured timeout.
-        let stream = if None != proto && Some(Proto::HTTPConnect) != proto {
+        #[allow(clippy::unnecessary_unwrap)]
+        let stream = if proto.is_some() && Some(Proto::HTTPConnect) != proto {
             connect_socks(
                 unit,
                 proxy.clone().unwrap(),
@@ -413,19 +375,19 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
         } else if let Some(timeout) = timeout {
             TcpStream::connect_timeout(&sock_addr, timeout)
         } else {
-            TcpStream::connect(&sock_addr)
+            TcpStream::connect(sock_addr)
         };
 
         if let Ok(stream) = stream {
-            any_stream = Some(stream);
+            any_stream_and_addr = Some((stream, sock_addr));
             break;
         } else if let Err(err) = stream {
             any_err = Some(err);
         }
     }
 
-    let mut stream = if let Some(stream) = any_stream {
-        stream
+    let (mut stream, remote_addr) = if let Some(stream_and_addr) = any_stream_and_addr {
+        stream_and_addr
     } else if let Some(e) = any_err {
         return Err(ErrorKind::ConnectionFailed.msg("Connect error").src(e));
     } else {
@@ -466,7 +428,7 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
         }
     }
 
-    Ok(stream)
+    Ok((stream, remote_addr))
 }
 
 #[cfg(feature = "socks-proxy")]
@@ -652,6 +614,12 @@ pub(crate) fn connect_test(unit: &Unit) -> Result<Stream, Error> {
     Err(ErrorKind::UnknownScheme.msg(format!("unknown scheme '{}'", unit.url.scheme())))
 }
 
+#[cfg(test)]
+pub(crate) fn remote_addr_for_test() -> SocketAddr {
+    use std::net::{Ipv4Addr, SocketAddrV4};
+    SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 0).into()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -693,10 +661,6 @@ mod tests {
         fn socket(&self) -> Option<&TcpStream> {
             unimplemented!()
         }
-
-        fn is_poolable(&self) -> bool {
-            unimplemented!()
-        }
     }
 
     // Test that when a DeadlineStream wraps a Stream, and the user performs a series of
@@ -707,11 +671,11 @@ mod tests {
         let recorder = ReadRecorder {
             reads: reads.clone(),
         };
-        let stream = Stream::new(recorder);
+        let stream = Stream::new(recorder, remote_addr_for_test());
         let mut deadline_stream = DeadlineStream::new(stream, None);
         let mut buf = [0u8; 1];
         for _ in 0..8193 {
-            deadline_stream.read(&mut buf).unwrap();
+            let _ = deadline_stream.read(&mut buf).unwrap();
         }
         let reads = reads.lock().unwrap();
         assert_eq!(reads.len(), 2);
diff --git a/src/unit.rs b/src/unit.rs
index 6fce8bb..785a90e 100644
--- a/src/unit.rs
+++ b/src/unit.rs
@@ -87,7 +87,7 @@ impl Unit {
             if (!username.is_empty() || !password.is_empty())
                 && get_header(&headers, "authorization").is_none()
             {
-                let encoded = base64::encode(&format!("{}:{}", username, password));
+                let encoded = base64::encode(format!("{}:{}", username, password));
                 extra.push(Header::new("Authorization", &format!("Basic {}", encoded)));
             }
 


Test patch:
diff --git a/examples/smoke-test/main.rs b/examples/smoke-test/main.rs
index 4c7ea28..b19a79f 100644
--- a/examples/smoke-test/main.rs
+++ b/examples/smoke-test/main.rs
@@ -5,7 +5,6 @@ use std::time::Duration;
 use std::{env, error, fmt, result};
 
 use log::{error, info};
-use ureq;
 
 #[derive(Debug)]
 struct Oops(String);
@@ -77,7 +76,7 @@ fn get_many(urls: Vec<String>, simultaneous_fetches: usize) -> Result<()> {
 }
 
 fn main() -> Result<()> {
-    let args = env::args();
+    let mut args = env::args();
     if args.len() == 1 {
         println!(
             r##"Usage: {:#?} top-1m.csv
@@ -94,11 +93,11 @@ using 50 threads concurrently.
         return Ok(());
     }
     env_logger::init();
-    let file = std::fs::File::open(args.skip(1).next().unwrap())?;
+    let file = std::fs::File::open(args.nth(1).unwrap())?;
     let bufreader = BufReader::new(file);
     let mut urls = vec![];
     for line in bufreader.lines() {
-        let domain = line?.rsplit(",").next().unwrap().to_string();
+        let domain = line?.rsplit(',').next().unwrap().to_string();
         urls.push(format!("http://{}/", domain));
         urls.push(format!("https://{}/", domain));
         urls.push(format!("http://www.{}/", domain));
diff --git a/src/test/agent_test.rs b/src/test/agent_test.rs
index b7d254c..ea0abda 100644
--- a/src/test/agent_test.rs
+++ b/src/test/agent_test.rs
@@ -80,7 +80,7 @@ fn connection_reuse_with_408() {
     // pulls from the pool. If for some reason the timed-out
     // connection wasn't in the pool, we won't be testing what
     // we thought we were testing.
-    let resp = agent.post(&url).send_string("hello".into()).unwrap();
+    let resp = agent.post(&url).send_string("hello").unwrap();
     assert_eq!(resp.status(), 200);
 }
 
diff --git a/src/test/body_send.rs b/src/test/body_send.rs
index 76ae004..02192e2 100644
--- a/src/test/body_send.rs
+++ b/src/test/body_send.rs
@@ -1,79 +1,61 @@
-use crate::test;
+use crate::test::Recorder;
 
 use super::super::*;
 
 #[test]
 fn content_length_on_str() {
-    test::set_handler("/content_length_on_str", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = post("test://host/content_length_on_str")
+    let recorder = Recorder::register("/content_length_on_str");
+    post("test://host/content_length_on_str")
         .send_string("Hello World!!!")
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("\r\nContent-Length: 14\r\n"));
+    assert!(recorder.contains("\r\nContent-Length: 14\r\n"));
 }
 
 #[test]
 fn user_set_content_length_on_str() {
-    test::set_handler("/user_set_content_length_on_str", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = post("test://host/user_set_content_length_on_str")
+    let recorder = Recorder::register("/user_set_content_length_on_str");
+    post("test://host/user_set_content_length_on_str")
         .set("Content-Length", "12345")
         .send_string("Hello World!!!")
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("\r\nContent-Length: 12345\r\n"));
+    assert!(recorder.contains("\r\nContent-Length: 12345\r\n"));
 }
 
 #[test]
 #[cfg(feature = "json")]
 fn content_length_on_json() {
-    test::set_handler("/content_length_on_json", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
+    let recorder = Recorder::register("/content_length_on_json");
     let mut json = serde_json::Map::new();
     json.insert(
         "Hello".to_string(),
         serde_json::Value::String("World!!!".to_string()),
     );
-    let resp = post("test://host/content_length_on_json")
+    post("test://host/content_length_on_json")
         .send_json(serde_json::Value::Object(json))
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("\r\nContent-Length: 20\r\n"));
+    assert!(recorder.contains("\r\nContent-Length: 20\r\n"));
 }
 
 #[test]
 fn content_length_and_chunked() {
-    test::set_handler("/content_length_and_chunked", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = post("test://host/content_length_and_chunked")
+    let recorder = Recorder::register("/content_length_and_chunked");
+    post("test://host/content_length_and_chunked")
         .set("Transfer-Encoding", "chunked")
         .send_string("Hello World!!!")
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("Transfer-Encoding: chunked\r\n"));
-    assert!(!s.contains("\r\nContent-Length:\r\n"));
+    assert!(recorder.contains("Transfer-Encoding: chunked\r\n"));
+    assert!(!recorder.contains("\r\nContent-Length:\r\n"));
 }
 
 #[test]
 #[cfg(feature = "charset")]
 fn str_with_encoding() {
-    test::set_handler("/str_with_encoding", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = post("test://host/str_with_encoding")
+    let recorder = Recorder::register("/str_with_encoding");
+    post("test://host/str_with_encoding")
         .set("Content-Type", "text/plain; charset=iso-8859-1")
         .send_string("Hällo Wörld!!!")
         .unwrap();
-    let vec = resp.into_written_bytes();
+    let vec = recorder.to_vec();
     assert_eq!(
         &vec[vec.len() - 14..],
         //H  ä    l    l    o    _   W   ö    r    l    d    !   !   !
@@ -84,38 +66,30 @@ fn str_with_encoding() {
 #[test]
 #[cfg(feature = "json")]
 fn content_type_on_json() {
-    test::set_handler("/content_type_on_json", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
+    let recorder = Recorder::register("/content_type_on_json");
     let mut json = serde_json::Map::new();
     json.insert(
         "Hello".to_string(),
         serde_json::Value::String("World!!!".to_string()),
     );
-    let resp = post("test://host/content_type_on_json")
+    post("test://host/content_type_on_json")
         .send_json(serde_json::Value::Object(json))
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("\r\nContent-Type: application/json\r\n"));
+    assert!(recorder.contains("\r\nContent-Type: application/json\r\n"));
 }
 
 #[test]
 #[cfg(feature = "json")]
 fn content_type_not_overriden_on_json() {
-    test::set_handler("/content_type_not_overriden_on_json", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
+    let recorder = Recorder::register("/content_type_not_overriden_on_json");
     let mut json = serde_json::Map::new();
     json.insert(
         "Hello".to_string(),
         serde_json::Value::String("World!!!".to_string()),
     );
-    let resp = post("test://host/content_type_not_overriden_on_json")
+    post("test://host/content_type_not_overriden_on_json")
         .set("content-type", "text/plain")
         .send_json(serde_json::Value::Object(json))
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("\r\ncontent-type: text/plain\r\n"));
+    assert!(recorder.contains("\r\ncontent-type: text/plain\r\n"));
 }
diff --git a/src/test/mod.rs b/src/test/mod.rs
index 15f2bc0..eb0559a 100644
--- a/src/test/mod.rs
+++ b/src/test/mod.rs
@@ -1,9 +1,12 @@
 use crate::error::Error;
-use crate::stream::Stream;
+use crate::stream::{remote_addr_for_test, ReadOnlyStream, Stream};
 use crate::unit::Unit;
+use crate::ReadWrite;
 use once_cell::sync::Lazy;
 use std::collections::HashMap;
-use std::io::Write;
+use std::fmt;
+use std::io::{self, Cursor, Read, Write};
+use std::net::TcpStream;
 use std::sync::{Arc, Mutex};
 
 mod agent_test;
@@ -17,6 +20,7 @@ mod timeout;
 
 type RequestHandler = dyn Fn(&Unit) -> Result<Stream, Error> + Send + 'static;
 
+#[allow(clippy::type_complexity)]
 pub(crate) static TEST_HANDLERS: Lazy<Arc<Mutex<HashMap<String, Box<RequestHandler>>>>> =
     Lazy::new(|| Arc::new(Mutex::new(HashMap::new())));
 
@@ -48,7 +52,10 @@ pub(crate) fn make_response(
     }
     write!(&mut buf, "\r\n").ok();
     buf.append(&mut body);
-    Ok(Stream::from_vec(buf))
+    Ok(Stream::new(
+        ReadOnlyStream::new(buf),
+        remote_addr_for_test(),
+    ))
 }
 
 pub(crate) fn resolve_handler(unit: &Unit) -> Result<Stream, Error> {
@@ -66,3 +73,90 @@ pub(crate) fn resolve_handler(unit: &Unit) -> Result<Stream, Error> {
     drop(handlers);
     handler(unit)
 }
+
+#[derive(Default, Clone)]
+pub(crate) struct Recorder {
+    contents: Arc<Mutex<Vec<u8>>>,
+}
+
+impl Recorder {
+    fn register(path: &str) -> Self {
+        let recorder = Recorder::default();
+        let recorder2 = recorder.clone();
+        set_handler(path, move |_unit| Ok(recorder.stream()));
+        recorder2
+    }
+
+    #[cfg(feature = "charset")]
+    fn to_vec(self) -> Vec<u8> {
+        self.contents.lock().unwrap().clone()
+    }
+
+    fn contains(&self, s: &str) -> bool {
+        std::str::from_utf8(&self.contents.lock().unwrap())
+            .unwrap()
+            .contains(s)
+    }
+
+    fn stream(&self) -> Stream {
+        let cursor = Cursor::new(b"HTTP/1.1 200 OK\r\n\r\n");
+        Stream::new(
+            TestStream::new(cursor, self.clone()),
+            remote_addr_for_test(),
+        )
+    }
+}
+
+impl Write for Recorder {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.contents.lock().unwrap().write(buf)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+pub(crate) struct TestStream(
+    Box<dyn Read + Send + Sync>,
+    Box<dyn Write + Send + Sync>,
+    bool,
+);
+
+impl TestStream {
+    #[cfg(test)]
+    pub(crate) fn new(
+        response: impl Read + Send + Sync + 'static,
+        recorder: impl Write + Send + Sync + 'static,
+    ) -> Self {
+        Self(Box::new(response), Box::new(recorder), false)
+    }
+}
+
+impl ReadWrite for TestStream {
+    fn socket(&self) -> Option<&TcpStream> {
+        None
+    }
+}
+
+impl Read for TestStream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.0.read(buf)
+    }
+}
+
+impl Write for TestStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.1.write(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl fmt::Debug for TestStream {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_tuple("TestStream").finish()
+    }
+}
diff --git a/src/test/query_string.rs b/src/test/query_string.rs
index 4ab8f21..3722c1f 100644
--- a/src/test/query_string.rs
+++ b/src/test/query_string.rs
@@ -1,58 +1,37 @@
-use crate::test;
-
-use super::super::*;
+use super::Recorder;
+use crate::get;
 
 #[test]
 fn no_query_string() {
-    test::set_handler("/no_query_string", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = get("test://host/no_query_string").call().unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("GET /no_query_string HTTP/1.1"))
+    let recorder = Recorder::register("/no_query_string");
+    get("test://host/no_query_string").call().unwrap();
+    assert!(recorder.contains("GET /no_query_string HTTP/1.1"))
 }
 
 #[test]
 fn escaped_query_string() {
-    test::set_handler("/escaped_query_string", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = get("test://host/escaped_query_string")
+    let recorder = Recorder::register("/escaped_query_string");
+    get("test://host/escaped_query_string")
         .query("foo", "bar")
         .query("baz", "yo lo")
         .call()
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(
-        s.contains("GET /escaped_query_string?foo=bar&baz=yo+lo HTTP/1.1"),
-        "req: {}",
-        s
-    );
+    assert!(recorder.contains("GET /escaped_query_string?foo=bar&baz=yo+lo HTTP/1.1"));
 }
 
 #[test]
 fn query_in_path() {
-    test::set_handler("/query_in_path", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = get("test://host/query_in_path?foo=bar").call().unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("GET /query_in_path?foo=bar HTTP/1.1"))
+    let recorder = Recorder::register("/query_in_path");
+    get("test://host/query_in_path?foo=bar").call().unwrap();
+    assert!(recorder.contains("GET /query_in_path?foo=bar HTTP/1.1"))
 }
 
 #[test]
 fn query_in_path_and_req() {
-    test::set_handler("/query_in_path_and_req", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = get("test://host/query_in_path_and_req?foo=bar")
+    let recorder = Recorder::register("/query_in_path_and_req");
+    get("test://host/query_in_path_and_req?foo=bar")
         .query("baz", "1 2 3")
         .call()
         .unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("GET /query_in_path_and_req?foo=bar&baz=1+2+3 HTTP/1.1"))
+    assert!(recorder.contains("GET /query_in_path_and_req?foo=bar&baz=1+2+3 HTTP/1.1"));
 }
diff --git a/src/test/simple.rs b/src/test/simple.rs
index 3391c20..4ed4b03 100644
--- a/src/test/simple.rs
+++ b/src/test/simple.rs
@@ -1,7 +1,7 @@
 use crate::test;
 use std::io::Read;
 
-use super::super::*;
+use super::{super::*, Recorder};
 
 #[test]
 fn header_passing() {
@@ -116,13 +116,9 @@ fn body_as_reader() {
 
 #[test]
 fn escape_path() {
-    test::set_handler("/escape_path%20here", |_unit| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = get("test://host/escape_path here").call().unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("GET /escape_path%20here HTTP/1.1"))
+    let recorder = Recorder::register("/escape_path%20here");
+    get("test://host/escape_path here").call().unwrap();
+    assert!(recorder.contains("GET /escape_path%20here HTTP/1.1"))
 }
 
 #[test]
@@ -194,22 +190,14 @@ pub fn header_with_spaces_before_value() {
 
 #[test]
 pub fn host_no_port() {
-    test::set_handler("/host_no_port", |_| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = get("test://myhost/host_no_port").call().unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("\r\nHost: myhost\r\n"));
+    let recorder = Recorder::register("/host_no_port");
+    get("test://myhost/host_no_port").call().unwrap();
+    assert!(recorder.contains("\r\nHost: myhost\r\n"));
 }
 
 #[test]
 pub fn host_with_port() {
-    test::set_handler("/host_with_port", |_| {
-        test::make_response(200, "OK", vec![], vec![])
-    });
-    let resp = get("test://myhost:234/host_with_port").call().unwrap();
-    let vec = resp.into_written_bytes();
-    let s = String::from_utf8_lossy(&vec);
-    assert!(s.contains("\r\nHost: myhost:234\r\n"));
+    let recorder = Recorder::register("/host_with_port");
+    get("test://myhost:234/host_with_port").call().unwrap();
+    assert!(recorder.contains("\r\nHost: myhost:234\r\n"));
 }