feat(client): change Connect trait into an alias for Service

The `Connect` trait is now essentially an alias for
`Service<Destination>`, with a blanket implementation as such, and is
sealed.

Closes #1902

BREAKING CHANGE: Any manual implementations of `Connect` must instead
  implement `tower::Service<Destination>`.

src/client/connect/http.rs

@@ -4,6 +4,7 @@ use std::error::Error as StdError;
 use std::io;
 use std::mem;
 use std::net::{IpAddr, SocketAddr};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use http::uri::Scheme;
@@ -29,16 +30,8 @@ type ConnectFuture = Pin<Box<dyn Future<Output = io::Result<TcpStream>> + Send>>
 /// transport information such as the remote socket address used.
 #[derive(Clone)]
 pub struct HttpConnector<R = GaiResolver> {
-    enforce_http: bool,
-    handle: Option<Handle>,
-    happy_eyeballs_timeout: Option<Duration>,
-    keep_alive_timeout: Option<Duration>,
-    local_address: Option<IpAddr>,
-    nodelay: bool,
+    config: Arc<Config>,
     resolver: R,
-    reuse_address: bool,
-    send_buffer_size: Option<usize>,
-    recv_buffer_size: Option<usize>,
 }
 
 /// Extra information about the transport when an HttpConnector is used.
@@ -74,6 +67,21 @@ pub struct HttpInfo {
     remote_addr: SocketAddr,
 }
 
+#[derive(Clone)]
+struct Config {
+    enforce_http: bool,
+    handle: Option<Handle>,
+    happy_eyeballs_timeout: Option<Duration>,
+    keep_alive_timeout: Option<Duration>,
+    local_address: Option<IpAddr>,
+    nodelay: bool,
+    reuse_address: bool,
+    send_buffer_size: Option<usize>,
+    recv_buffer_size: Option<usize>,
+}
+
+// ===== impl HttpConnector =====
+
 impl HttpConnector {
     /// Construct a new HttpConnector.
     ///
@@ -100,16 +108,18 @@ impl<R> HttpConnector<R> {
     /// Takes a `Resolve` to handle DNS lookups.
     pub fn new_with_resolver(resolver: R) -> HttpConnector<R> {
         HttpConnector {
-            enforce_http: true,
-            handle: None,
-            happy_eyeballs_timeout: Some(Duration::from_millis(300)),
-            keep_alive_timeout: None,
-            local_address: None,
-            nodelay: false,
+            config: Arc::new(Config {
+                enforce_http: true,
+                handle: None,
+                happy_eyeballs_timeout: Some(Duration::from_millis(300)),
+                keep_alive_timeout: None,
+                local_address: None,
+                nodelay: false,
+                reuse_address: false,
+                send_buffer_size: None,
+                recv_buffer_size: None,
+            }),
             resolver,
-            reuse_address: false,
-            send_buffer_size: None,
-            recv_buffer_size: None,
         }
     }
 
@@ -118,15 +128,15 @@ impl<R> HttpConnector<R> {
     /// Enabled by default.
     #[inline]
     pub fn enforce_http(&mut self, is_enforced: bool) {
-        self.enforce_http = is_enforced;
+        self.config_mut().enforce_http = is_enforced;
     }
 
     /// Set a handle to a `Reactor` to register connections to.
     ///
     /// If `None`, the implicit default reactor will be used.
     #[inline]
     pub fn set_reactor(&mut self, handle: Option<Handle>) {
-        self.handle = handle;
+        self.config_mut().handle = handle;
     }
 
     /// Set that all sockets have `SO_KEEPALIVE` set with the supplied duration.
@@ -136,27 +146,27 @@ impl<R> HttpConnector<R> {
     /// Default is `None`.
     #[inline]
     pub fn set_keepalive(&mut self, dur: Option<Duration>) {
-        self.keep_alive_timeout = dur;
+        self.config_mut().keep_alive_timeout = dur;
     }
 
     /// Set that all sockets have `SO_NODELAY` set to the supplied value `nodelay`.
     ///
     /// Default is `false`.
     #[inline]
     pub fn set_nodelay(&mut self, nodelay: bool) {
-        self.nodelay = nodelay;
+        self.config_mut().nodelay = nodelay;
     }
 
     /// Sets the value of the SO_SNDBUF option on the socket.
     #[inline]
     pub fn set_send_buffer_size(&mut self, size: Option<usize>) {
-        self.send_buffer_size = size;
+        self.config_mut().send_buffer_size = size;
     }
 
     /// Sets the value of the SO_RCVBUF option on the socket.
     #[inline]
     pub fn set_recv_buffer_size(&mut self, size: Option<usize>) {
-        self.recv_buffer_size = size;
+        self.config_mut().recv_buffer_size = size;
     }
 
     /// Set that all sockets are bound to the configured address before connection.
@@ -166,7 +176,7 @@ impl<R> HttpConnector<R> {
     /// Default is `None`.
     #[inline]
     pub fn set_local_address(&mut self, addr: Option<IpAddr>) {
-        self.local_address = addr;
+        self.config_mut().local_address = addr;
     }
 
     /// Set timeout for [RFC 6555 (Happy Eyeballs)][RFC 6555] algorithm.
@@ -183,17 +193,26 @@ impl<R> HttpConnector<R> {
     /// [RFC 6555]: https://tools.ietf.org/html/rfc6555
     #[inline]
     pub fn set_happy_eyeballs_timeout(&mut self, dur: Option<Duration>) {
-        self.happy_eyeballs_timeout = dur;
+        self.config_mut().happy_eyeballs_timeout = dur;
     }
 
     /// Set that all socket have `SO_REUSEADDR` set to the supplied value `reuse_address`.
     ///
     /// Default is `false`.
     #[inline]
     pub fn set_reuse_address(&mut self, reuse_address: bool) -> &mut Self {
-        self.reuse_address = reuse_address;
+        self.config_mut().reuse_address = reuse_address;
         self
     }
+
+    // private
+
+    fn config_mut(&mut self) -> &mut Config {
+        // If the are HttpConnector clones, this will clone the inner
+        // config. So mutating the config won't ever affect previous
+        // clones.
+        Arc::make_mut(&mut self.config)
+    }
 }
 
 // R: Debug required for now to allow adding it to debug output later...
@@ -204,50 +223,58 @@ impl<R: fmt::Debug> fmt::Debug for HttpConnector<R> {
     }
 }
 
-impl<R> Connect for HttpConnector<R>
+impl<R> tower_service::Service<Destination> for HttpConnector<R>
 where
     R: Resolve + Clone + Send + Sync,
     R::Future: Send,
 {
-    type Transport = TcpStream;
+    type Response = (TcpStream, Connected);
     type Error = io::Error;
     type Future = HttpConnecting<R>;
 
-    fn connect(&self, dst: Destination) -> Self::Future {
+    fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+        // For now, always ready.
+        // TODO: When `Resolve` becomes an alias for `Service`, check
+        // the resolver's readiness.
+        drop(cx);
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, dst: Destination) -> Self::Future {
         trace!(
             "Http::connect; scheme={}, host={}, port={:?}",
             dst.scheme(),
             dst.host(),
             dst.port(),
         );
 
-        if self.enforce_http {
+        if self.config.enforce_http {
             if dst.uri.scheme_part() != Some(&Scheme::HTTP) {
-                return invalid_url(InvalidUrl::NotHttp, &self.handle);
+                return invalid_url(InvalidUrl::NotHttp, &self.config.handle);
             }
         } else if dst.uri.scheme_part().is_none() {
-            return invalid_url(InvalidUrl::MissingScheme, &self.handle);
+            return invalid_url(InvalidUrl::MissingScheme, &self.config.handle);
         }
 
         let host = match dst.uri.host() {
             Some(s) => s,
-            None => return invalid_url(InvalidUrl::MissingAuthority, &self.handle),
+            None => return invalid_url(InvalidUrl::MissingAuthority, &self.config.handle),
         };
         let port = match dst.uri.port_part() {
             Some(port) => port.as_u16(),
             None => if dst.uri.scheme_part() == Some(&Scheme::HTTPS) { 443 } else { 80 },
         };
 
         HttpConnecting {
-            state: State::Lazy(self.resolver.clone(), host.into(), self.local_address),
-            handle: self.handle.clone(),
-            happy_eyeballs_timeout: self.happy_eyeballs_timeout,
-            keep_alive_timeout: self.keep_alive_timeout,
-            nodelay: self.nodelay,
+            state: State::Lazy(self.resolver.clone(), host.into(), self.config.local_address),
+            handle: self.config.handle.clone(),
+            happy_eyeballs_timeout: self.config.happy_eyeballs_timeout,
+            keep_alive_timeout: self.config.keep_alive_timeout,
+            nodelay: self.config.nodelay,
             port,
-            reuse_address: self.reuse_address,
-            send_buffer_size: self.send_buffer_size,
-            recv_buffer_size: self.recv_buffer_size,
+            reuse_address: self.config.reuse_address,
+            send_buffer_size: self.config.send_buffer_size,
+            recv_buffer_size: self.config.recv_buffer_size,
         }
     }
 }

src/client/connect/mod.rs

@@ -24,26 +24,49 @@ use crate::common::{Future, Unpin};
 /// A connector receives a [`Destination`](Destination) describing how a
 /// connection should be estabilished, and returns a `Future` of the
 /// ready connection.
-pub trait Connect: Send + Sync {
+///
+/// # Trait Alias
+///
+/// This is really just an *alias* for the `tower::Service` trait, with
+/// additional bounds set for convenience *inside* hyper. You don't actually
+/// implement this trait, but `tower::Service<Destination>` instead.
+// The `Sized` bound is to prevent creating `dyn Connect`, since they cannot
+// fit the `Connect` bounds because of the blanket impl for `Service`.
+pub trait Connect: sealed::Sealed + Sized {
     /// The connected IO Stream.
-    type Transport: AsyncRead + AsyncWrite + Unpin + Send + 'static;
+    type Transport: AsyncRead + AsyncWrite;
     /// An error occured when trying to connect.
     type Error: Into<Box<dyn StdError + Send + Sync>>;
     /// A Future that will resolve to the connected Transport.
-    type Future: Future<Output=Result<(Self::Transport, Connected), Self::Error>> + Unpin + Send;
-    /// Connect to a destination.
-    fn connect(&self, dst: Destination) -> Self::Future;
+    type Future: Future<Output=Result<(Self::Transport, Connected), Self::Error>>;
+    #[doc(hidden)]
+    fn connect(self, internal_only: sealed::Internal, dst: Destination) -> Self::Future;
 }
 
-impl<T: Connect + ?Sized> Connect for Box<T> {
-    type Transport = <T as Connect>::Transport;
-    type Error = <T as Connect>::Error;
-    type Future = <T as Connect>::Future;
-    fn connect(&self, dst: Destination) -> Self::Future {
-        <T as Connect>::connect(self, dst)
+impl<S, T> Connect for S
+where
+    S: tower_service::Service<Destination, Response=(T, Connected)> + Send,
+    S::Error: Into<Box<dyn StdError + Send + Sync>>,
+    S::Future: Unpin + Send,
+    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+{
+    type Transport = T;
+    type Error = S::Error;
+    type Future = crate::service::Oneshot<S, Destination>;
+    fn connect(self, _: sealed::Internal, dst: Destination) -> Self::Future {
+        crate::service::oneshot(self, dst)
     }
 }
 
+impl<S, T> sealed::Sealed for S
+where
+    S: tower_service::Service<Destination, Response=(T, Connected)> + Send,
+    S::Error: Into<Box<dyn StdError + Send + Sync>>,
+    S::Future: Unpin + Send,
+    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+{}
+
+
 /// A set of properties to describe where and how to try to connect.
 ///
 /// This type is passed an argument for the [`Connect`](Connect) trait.
@@ -398,6 +421,12 @@ where
     }
 }
 
+pub(super) mod sealed {
+    pub trait Sealed {}
+    #[allow(missing_debug_implementations)]
+    pub struct Internal;
+}
+
 #[cfg(test)]
 mod tests {
     use super::{Connected, Destination, TryFrom};

src/client/mod.rs

@@ -88,7 +88,7 @@ mod tests;
 pub struct Client<C, B = Body> {
     config: Config,
     conn_builder: conn::Builder,
-    connector: Arc<C>,
+    connector: C,
     pool: Pool<PoolClient<B>>,
 }
 
@@ -157,9 +157,9 @@ impl Client<(), Body> {
 }
 
 impl<C, B> Client<C, B>
-where C: Connect + Sync + 'static,
-      C::Transport: 'static,
-      C::Future: 'static,
+where C: Connect + Clone + Send + Sync + 'static,
+      C::Transport: Unpin + Send + 'static,
+      C::Future: Unpin + Send + 'static,
       B: Payload + Unpin + Send + 'static,
       B::Data: Send + Unpin,
 {
@@ -485,7 +485,7 @@ where C: Connect + Sync + 'static,
                     return Either::Right(future::err(canceled));
                 }
             };
-            Either::Left(connector.connect(dst)
+            Either::Left(connector.connect(connect::sealed::Internal, dst)
                 .map_err(crate::Error::new_connect)
                 .and_then(move |(io, connected)| {
                     // If ALPN is h2 and we aren't http2_only already,
@@ -543,7 +543,7 @@ where C: Connect + Sync + 'static,
     }
 }
 
-impl<C, B> Clone for Client<C, B> {
+impl<C: Clone, B> Clone for Client<C, B> {
     fn clone(&self) -> Client<C, B> {
         Client {
             config: self.config.clone(),
@@ -1034,15 +1034,13 @@ impl Builder {
     pub fn build<C, B>(&self, connector: C) -> Client<C, B>
     where
         C: Connect,
-        C::Transport: 'static,
-        C::Future: 'static,
         B: Payload + Send,
         B::Data: Send,
     {
         Client {
             config: self.client_config,
             conn_builder: self.conn_builder.clone(),
-            connector: Arc::new(connector),
+            connector,
             pool: Pool::new(self.pool_config, &self.conn_builder.exec),
         }
     }

src/service/mod.rs

@@ -31,7 +31,10 @@
 //! is called.
 
 mod make_service;
+mod oneshot;
 mod service;
 
 pub use self::make_service::{make_service_fn, MakeService, MakeServiceRef};
 pub use self::service::{service_fn, Service};
+
+pub(crate) use self::oneshot::{oneshot, Oneshot};