Introduce client::Background

src/client/background.rs

@@ -0,0 +1,64 @@
+use crate::body::LiftBody;
+use futures::{Future, Poll};
+use http_body::Body as HttpBody;
+use hyper::client::conn::Connection as HyperConnection;
+use log::error;
+use std::fmt::{self, Debug};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+/// Background task for a client connection.
+///
+/// This type is not used directly by a user of this library,
+/// but it can show up in trait bounds of generic types.
+pub struct Background<T, B>
+where
+    T: AsyncRead + AsyncWrite + Send + 'static,
+    B: HttpBody + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<crate::Error>,
+{
+    connection: HyperConnection<T, LiftBody<B>>,
+}
+
+impl<T, B> Background<T, B>
+where
+    T: AsyncRead + AsyncWrite + Send + 'static,
+    B: HttpBody + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<crate::Error>,
+{
+    pub(super) fn new(connection: HyperConnection<T, LiftBody<B>>) -> Self {
+        Background { connection }
+    }
+}
+
+impl<T, B> Future for Background<T, B>
+where
+    T: AsyncRead + AsyncWrite + Send + 'static,
+    B: HttpBody + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<crate::Error>,
+{
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<(), ()> {
+        self.connection.poll().map_err(|e| {
+            error!("error with hyper: {}", e);
+        })
+    }
+}
+
+impl<T, B> Debug for Background<T, B>
+where
+    T: Debug + AsyncRead + AsyncWrite + Send + 'static,
+    B: HttpBody + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<crate::Error>,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("Background")
+            .field("connection", &self.connection)
+            .finish()
+    }
+}

src/client/connect.rs

@@ -1,13 +1,11 @@
-use super::Connection;
+use super::{Background, Connection};
 use crate::body::LiftBody;
-use futures::{future::MapErr, try_ready, Async, Future, Poll};
+use futures::{try_ready, Async, Future, Poll};
 use http::Version;
 use http_body::Body as HttpBody;
 use http_connection::HttpConnection;
-use hyper::body::Payload;
-use hyper::client::conn::{Builder, Connection as HyperConnection, Handshake};
+use hyper::client::conn::{Builder, Handshake};
 use hyper::Error;
-use log::error;
 use std::fmt;
 use std::marker::PhantomData;
 use tokio_executor::{DefaultExecutor, TypedExecutor};
@@ -29,11 +27,12 @@ pub struct Connect<A, B, C, E> {
 }
 
 /// Executor that will spawn the background connection task.
-pub trait ConnectExecutor<T, B>:
-    TypedExecutor<MapErr<HyperConnection<T, B>, fn(hyper::Error) -> ()>>
+pub trait ConnectExecutor<T, B>: TypedExecutor<Background<T, B>>
 where
     T: AsyncRead + AsyncWrite + Send + 'static,
-    B: Payload + 'static,
+    B: HttpBody + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<crate::Error>,
 {
 }
 
@@ -75,8 +74,10 @@ pub enum ConnectError<T> {
 impl<E, T, B> ConnectExecutor<T, B> for E
 where
     T: AsyncRead + AsyncWrite + Send + 'static,
-    B: Payload + 'static,
-    E: TypedExecutor<MapErr<HyperConnection<T, B>, fn(hyper::Error) -> ()>>,
+    B: HttpBody + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<crate::Error>,
+    E: TypedExecutor<Background<T, B>>,
 {
 }
 
@@ -136,7 +137,7 @@ where
     B::Data: Send,
     B::Error: Into<crate::Error>,
     C::Connection: Send + 'static,
-    E: ConnectExecutor<C::Connection, LiftBody<B>> + Clone,
+    E: ConnectExecutor<C::Connection, B> + Clone,
 {
     type Response = Connection<B>;
     type Error = ConnectError<C::Error>;
@@ -170,7 +171,7 @@ where
     B::Data: Send,
     B::Error: Into<crate::Error>,
     C::Connection: Send + 'static,
-    E: ConnectExecutor<C::Connection, LiftBody<B>>,
+    E: ConnectExecutor<C::Connection, B>,
 {
     type Item = Connection<B>;
     type Error = ConnectError<C::Error>;
@@ -187,7 +188,7 @@ where
                     let (sender, conn) = try_ready!(fut.poll().map_err(ConnectError::Handshake));
 
                     self.exec
-                        .spawn(conn.map_err(|e| error!("error with hyper: {}", e)))
+                        .spawn(Background::new(conn))
                         .map_err(|_| ConnectError::SpawnError)?;
 
                     let connection = Connection::new(sender);

src/client/mod.rs

@@ -25,11 +25,13 @@
 //! [`Client`]: ./struct.Client.html
 //! [`hyper::Client`]: ../../hyper/struct.Client.html
 
+mod background;
 mod connect;
 mod connection;
 mod future;
 
-pub use self::connect::{Connect, ConnectError};
+pub use self::background::Background;
+pub use self::connect::{Connect, ConnectError, ConnectExecutor};
 pub use self::connection::Connection;
 use self::future::ResponseFuture;
 pub use hyper::client::conn::Builder;