Repository: sagebind/isahc (rust)

Issue:
Convenience methods to consume response body
It would be nice to provide convenience methods for consuming a response body without storing it. This is a useful feature because if the user drops the response without reading the response body fully, the underlying connection cannot be reused and must be closed prematurely. While totally acceptable, this is not a good way to maintain high throughput since it prevents connection reuse.

The API could be something as simple as:

```rust
impl Body {
    pub fn consume(&mut self) -> io::Result<u64> {
        // ...
    }

    pub async fn consume_async(&mut self) -> io::Result<u64> {
        // ...
    }
}
```

Which would be essentially a convenience around `io::copy(&mut response.body_mut(), &mut io::sink())`.

Pull request:
Add consume API
Add `ReadResponseExt::consume` and `AsyncReadResponseExt::consume` methods for reading and discarding a response body as described in #257.

Also update future types used by `AsyncReadResponseExt` to be `Send` whenever `T` is `Send` as appropriate. Consolidate the newtype futures with a macro that handles conditional `Send` properly to reduce boilerplate and chance of mistakes.

Fixes #257.

Patch:
diff --git a/src/response.rs b/src/response.rs
index fa455b3..baef77b 100644
--- a/src/response.rs
+++ b/src/response.rs
@@ -92,6 +92,54 @@ impl<T> ResponseExt<T> for Response<T> {
 
 /// Provides extension methods for consuming HTTP response streams.
 pub trait ReadResponseExt<R: Read> {
+    /// Read any remaining bytes from the response body stream and discard them
+    /// until the end of the stream is reached. It is usually a good idea to
+    /// call this method before dropping a response if you know you haven't read
+    /// the entire response body.
+    ///
+    /// # Background
+    ///
+    /// By default, if a response stream is dropped before it has been
+    /// completely read from, then that HTTP connection will be terminated.
+    /// Depending on which version of HTTP is being used, this may require
+    /// closing the network connection to the server entirely. This can result
+    /// in sub-optimal performance for making multiple requests, as it prevents
+    /// Isahc from keeping the connection alive to be reused for subsequent
+    /// requests.
+    ///
+    /// If you are downloading a file on behalf of a user and have been
+    /// requested to cancel the operation, then this is probably what you want.
+    /// But if you are making many small API calls to a known server, then you
+    /// may want to call `consume()` before dropping the response, as reading a
+    /// few megabytes off a socket is usually more efficient in the long run
+    /// than taking a hit on connection reuse, and opening new connections can
+    /// be expensive.
+    ///
+    /// Note that in HTTP/2 and newer, it is not necessary to close the network
+    /// connection in order to interrupt the transfer of a particular response.
+    /// If you know that you will be using only HTTP/2 or newer, then calling
+    /// this method is probably unnecessary.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// use isahc::prelude::*;
+    ///
+    /// let mut response = isahc::get("https://example.org")?;
+    ///
+    /// println!("Status: {}", response.status());
+    /// println!("Headers: {:#?}", response.headers());
+    ///
+    /// // Read and discard the response body until the end.
+    /// response.consume()?;
+    /// # Ok::<(), isahc::Error>(())
+    /// ```
+    fn consume(&mut self) -> io::Result<()> {
+        self.copy_to(io::sink())?;
+
+        Ok(())
+    }
+
     /// Copy the response body into a writer.
     ///
     /// Returns the number of bytes that were written.
@@ -208,6 +256,51 @@ impl<R: Read> ReadResponseExt<R> for Response<R> {
 
 /// Provides extension methods for consuming asynchronous HTTP response streams.
 pub trait AsyncReadResponseExt<R: AsyncRead + Unpin> {
+    /// Read any remaining bytes from the response body stream and discard them
+    /// until the end of the stream is reached. It is usually a good idea to
+    /// call this method before dropping a response if you know you haven't read
+    /// the entire response body.
+    ///
+    /// # Background
+    ///
+    /// By default, if a response stream is dropped before it has been
+    /// completely read from, then that HTTP connection will be terminated.
+    /// Depending on which version of HTTP is being used, this may require
+    /// closing the network connection to the server entirely. This can result
+    /// in sub-optimal performance for making multiple requests, as it prevents
+    /// Isahc from keeping the connection alive to be reused for subsequent
+    /// requests.
+    ///
+    /// If you are downloading a file on behalf of a user and have been
+    /// requested to cancel the operation, then this is probably what you want.
+    /// But if you are making many small API calls to a known server, then you
+    /// may want to call `consume()` before dropping the response, as reading a
+    /// few megabytes off a socket is usually more efficient in the long run
+    /// than taking a hit on connection reuse, and opening new connections can
+    /// be expensive.
+    ///
+    /// Note that in HTTP/2 and newer, it is not necessary to close the network
+    /// connection in order to interrupt the transfer of a particular response.
+    /// If you know that you will be using only HTTP/2 or newer, then calling
+    /// this method is probably unnecessary.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// use isahc::prelude::*;
+    ///
+    /// # async fn run() -> Result<(), isahc::Error> {
+    /// let mut response = isahc::get_async("https://example.org").await?;
+    ///
+    /// println!("Status: {}", response.status());
+    /// println!("Headers: {:#?}", response.headers());
+    ///
+    /// // Read and discard the response body until the end.
+    /// response.consume().await?;
+    /// # Ok(()) }
+    /// ```
+    fn consume(&mut self) -> ConsumeFuture<'_, R>;
+
     /// Copy the response body into a writer asynchronously.
     ///
     /// Returns the number of bytes that were written.
@@ -257,6 +350,14 @@ pub trait AsyncReadResponseExt<R: AsyncRead + Unpin> {
 }
 
 impl<R: AsyncRead + Unpin> AsyncReadResponseExt<R> for Response<R> {
+    fn consume(&mut self) -> ConsumeFuture<'_, R> {
+        ConsumeFuture::new(async move {
+            futures_lite::io::copy(self.body_mut(), futures_lite::io::sink()).await?;
+
+            Ok(())
+        })
+    }
+
     fn copy_to<'a, W>(&'a mut self, writer: W) -> CopyFuture<'a, R, W>
     where
         W: AsyncWrite + Unpin + 'a,
@@ -271,6 +372,10 @@ impl<R: AsyncRead + Unpin> AsyncReadResponseExt<R> for Response<R> {
 }
 
 decl_future! {
+    /// A future which reads any remaining bytes from the response body stream
+    /// and discard them.
+    pub type ConsumeFuture<R> = impl Future<Output = io::Result<()>> + SendIf<R>;
+
     /// A future which copies all the response body bytes into a sink.
     pub type CopyFuture<R, W> = impl Future<Output = io::Result<u64>> + SendIf<R, W>;
 }


Test patch:
diff --git a/tests/response_body.rs b/tests/response_body.rs
index 02c0521..0655d67 100644
--- a/tests/response_body.rs
+++ b/tests/response_body.rs
@@ -1,6 +1,11 @@
+use futures_lite::{future::block_on, io::AsyncReadExt};
 use isahc::prelude::*;
+use std::{io, io::Read};
 use testserver::mock;
 
+#[macro_use]
+mod utils;
+
 #[test]
 fn simple_response_body() {
     let m = mock! {
@@ -74,8 +79,6 @@ fn dropping_client_does_not_abort_response_transfer() {
 // See issue #72.
 #[test]
 fn reading_from_response_body_after_eof_continues_to_return_eof() {
-    use std::{io, io::Read};
-
     let m = mock! {
         body: "hello world",
     };
@@ -92,3 +95,41 @@ fn reading_from_response_body_after_eof_continues_to_return_eof() {
         assert_eq!(body.read(&mut buf).unwrap(), 0);
     }
 }
+
+#[test]
+fn consume_unread_response_body() {
+    let body = "wow so large ".repeat(1000);
+
+    let m = {
+        let body = body.clone();
+        mock! {
+            body: body.clone(),
+        }
+    };
+
+    let mut response = isahc::get(m.url()).unwrap();
+    response.consume().unwrap();
+
+    let mut buf = [0; 8192];
+    assert_matches!(response.body_mut().read(&mut buf), Ok(0));
+}
+
+#[test]
+fn consume_unread_response_body_async() {
+    let body = "wow so large ".repeat(1000);
+
+    let m = {
+        let body = body.clone();
+        mock! {
+            body: body.clone(),
+        }
+    };
+
+    block_on(async move {
+        let mut response = isahc::get_async(m.url()).await.unwrap();
+        response.consume().await.unwrap();
+
+        let mut buf = [0; 8192];
+        assert_matches!(response.body_mut().read(&mut buf).await, Ok(0));
+    });
+}