fix: properly check if connections can be recycled

Checks `poll_read`'s status directly, so that we can ensure
a socket read actually  happens, and then immediately continue
if we get `Poll::Pending`.

Related to #75

Author: Fishrock123

src/h1/mod.rs

@@ -19,6 +19,8 @@ cfg_if::cfg_if! {
 
 use super::{async_trait, Error, HttpClient, Request, Response};
 
+pub(crate) mod utils;
+
 mod tcp;
 #[cfg(any(feature = "native-tls", feature = "rustls"))]
 mod tls;

src/h1/tcp.rs

@@ -8,6 +8,8 @@ use deadpool::managed::{Manager, Object, RecycleResult};
 use futures::io::{AsyncRead, AsyncWrite};
 use futures::task::{Context, Poll};
 
+use super::utils::PollRead;
+
 #[derive(Clone, Debug)]
 pub(crate) struct TcpConnection {
     addr: SocketAddr,
@@ -63,7 +65,14 @@ impl Manager<TcpStream, std::io::Error> for TcpConnection {
 
     async fn recycle(&self, conn: &mut TcpStream) -> RecycleResult<std::io::Error> {
         let mut buf = [0; 4];
-        conn.peek(&mut buf[..]).await?;
+        match PollRead::new(conn, &mut buf).await {
+            Poll::Ready(Err(error)) => Err(error),
+            Poll::Ready(Ok(bytes)) if bytes == 0 => Err(std::io::Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                "connection appeared to be closed (EoF)",
+            )),
+            _ => Ok(()),
+        }?;
         Ok(())
     }
 }

src/h1/tls.rs

@@ -8,6 +8,8 @@ use deadpool::managed::{Manager, Object, RecycleResult};
 use futures::io::{AsyncRead, AsyncWrite};
 use futures::task::{Context, Poll};
 
+use super::utils::PollRead;
+
 cfg_if::cfg_if! {
     if #[cfg(feature = "rustls")] {
         use async_tls::client::TlsStream;
@@ -76,10 +78,15 @@ impl Manager<TlsStream<TcpStream>, Error> for TlsConnection {
 
     async fn recycle(&self, conn: &mut TlsStream<TcpStream>) -> RecycleResult<Error> {
         let mut buf = [0; 4];
-        conn.get_ref()
-            .peek(&mut buf[..])
-            .await
-            .map_err(Error::from)?;
+        match PollRead::new(conn, &mut buf).await {
+            Poll::Ready(Err(error)) => Err(error),
+            Poll::Ready(Ok(bytes)) if bytes == 0 => Err(std::io::Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                "connection appeared to be closed (EoF)",
+            )),
+            _ => Ok(()),
+        }
+        .map_err(Error::from)?;
         Ok(())
     }
 }

src/h1/utils.rs

@@ -0,0 +1,31 @@
+use std::io;
+use std::pin::Pin;
+
+use futures::future::Future;
+use futures::io::AsyncRead;
+use futures::task::{Context, Poll};
+
+/// Like the `futures::io::Read` future, but returning the underlying `futures::task::Poll`.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub(crate) struct PollRead<'a, R: ?Sized> {
+    reader: &'a mut R,
+    buf: &'a mut [u8],
+}
+
+impl<R: ?Sized + Unpin> Unpin for PollRead<'_, R> {}
+
+impl<'a, R: AsyncRead + ?Sized + Unpin> PollRead<'a, R> {
+    pub(super) fn new(reader: &'a mut R, buf: &'a mut [u8]) -> Self {
+        Self { reader, buf }
+    }
+}
+
+impl<R: AsyncRead + ?Sized + Unpin> Future for PollRead<'_, R> {
+    type Output = Poll<io::Result<usize>>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+        Poll::Ready(Pin::new(&mut this.reader).poll_read(cx, this.buf))
+    }
+}

tests/test.rs

@@ -135,3 +135,17 @@ SOFTWARE.
 
     Ok(())
 }
+
+#[atest]
+async fn keep_alive() {
+    let _mock_guard = mockito::mock("GET", "/report")
+        .with_status(200)
+        .expect_at_least(2)
+        .create();
+
+    let client = DefaultClient::new();
+    let url: Url = format!("{}/report", mockito::server_url()).parse().unwrap();
+    let req = Request::new(http_types::Method::Get, url);
+    client.send(req.clone()).await.unwrap();
+    client.send(req.clone()).await.unwrap();
+}