Adding upstream version 124.0.1.upstream/124.0.1

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

3 files changed, 1844 insertions, 0 deletions

diff --git a/third_party/rust/hyper/src/client/connect/dns.rs b/third_party/rust/hyper/src/client/connect/dns.rs
new file mode 100644
index 0000000000..e4465078b3
--- /dev/null
+++ b/third_party/rust/hyper/src/client/connect/dns.rs
@@ -0,0 +1,425 @@
+//! DNS Resolution used by the `HttpConnector`.
+//!
+//! This module contains:
+//!
+//! - A [`GaiResolver`](GaiResolver) that is the default resolver for the
+//!   `HttpConnector`.
+//! - The `Name` type used as an argument to custom resolvers.
+//!
+//! # Resolvers are `Service`s
+//!
+//! A resolver is just a
+//! `Service<Name, Response = impl Iterator<Item = SocketAddr>>`.
+//!
+//! A simple resolver that ignores the name and always returns a specific
+//! address:
+//!
+//! ```rust,ignore
+//! use std::{convert::Infallible, iter, net::SocketAddr};
+//!
+//! let resolver = tower::service_fn(|_name| async {
+//!     Ok::<_, Infallible>(iter::once(SocketAddr::from(([127, 0, 0, 1], 8080))))
+//! });
+//! ```
+use std::error::Error;
+use std::future::Future;
+use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs};
+use std::pin::Pin;
+use std::str::FromStr;
+use std::task::{self, Poll};
+use std::{fmt, io, vec};
+
+use tokio::task::JoinHandle;
+use tower_service::Service;
+use tracing::debug;
+
+pub(super) use self::sealed::Resolve;
+
+/// A domain name to resolve into IP addresses.
+#[derive(Clone, Hash, Eq, PartialEq)]
+pub struct Name {
+    host: Box<str>,
+}
+
+/// A resolver using blocking `getaddrinfo` calls in a threadpool.
+#[derive(Clone)]
+pub struct GaiResolver {
+    _priv: (),
+}
+
+/// An iterator of IP addresses returned from `getaddrinfo`.
+pub struct GaiAddrs {
+    inner: SocketAddrs,
+}
+
+/// A future to resolve a name returned by `GaiResolver`.
+pub struct GaiFuture {
+    inner: JoinHandle<Result<SocketAddrs, io::Error>>,
+}
+
+impl Name {
+    pub(super) fn new(host: Box<str>) -> Name {
+        Name { host }
+    }
+
+    /// View the hostname as a string slice.
+    pub fn as_str(&self) -> &str {
+        &self.host
+    }
+}
+
+impl fmt::Debug for Name {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Debug::fmt(&self.host, f)
+    }
+}
+
+impl fmt::Display for Name {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(&self.host, f)
+    }
+}
+
+impl FromStr for Name {
+    type Err = InvalidNameError;
+
+    fn from_str(host: &str) -> Result<Self, Self::Err> {
+        // Possibly add validation later
+        Ok(Name::new(host.into()))
+    }
+}
+
+/// Error indicating a given string was not a valid domain name.
+#[derive(Debug)]
+pub struct InvalidNameError(());
+
+impl fmt::Display for InvalidNameError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("Not a valid domain name")
+    }
+}
+
+impl Error for InvalidNameError {}
+
+impl GaiResolver {
+    /// Construct a new `GaiResolver`.
+    pub fn new() -> Self {
+        GaiResolver { _priv: () }
+    }
+}
+
+impl Service<Name> for GaiResolver {
+    type Response = GaiAddrs;
+    type Error = io::Error;
+    type Future = GaiFuture;
+
+    fn poll_ready(&mut self, _cx: &mut task::Context<'_>) -> Poll<Result<(), io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, name: Name) -> Self::Future {
+        let blocking = tokio::task::spawn_blocking(move || {
+            debug!("resolving host={:?}", name.host);
+            (&*name.host, 0)
+                .to_socket_addrs()
+                .map(|i| SocketAddrs { iter: i })
+        });
+
+        GaiFuture { inner: blocking }
+    }
+}
+
+impl fmt::Debug for GaiResolver {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("GaiResolver")
+    }
+}
+
+impl Future for GaiFuture {
+    type Output = Result<GaiAddrs, io::Error>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self.inner).poll(cx).map(|res| match res {
+            Ok(Ok(addrs)) => Ok(GaiAddrs { inner: addrs }),
+            Ok(Err(err)) => Err(err),
+            Err(join_err) => {
+                if join_err.is_cancelled() {
+                    Err(io::Error::new(io::ErrorKind::Interrupted, join_err))
+                } else {
+                    panic!("gai background task failed: {:?}", join_err)
+                }
+            }
+        })
+    }
+}
+
+impl fmt::Debug for GaiFuture {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("GaiFuture")
+    }
+}
+
+impl Drop for GaiFuture {
+    fn drop(&mut self) {
+        self.inner.abort();
+    }
+}
+
+impl Iterator for GaiAddrs {
+    type Item = SocketAddr;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.inner.next()
+    }
+}
+
+impl fmt::Debug for GaiAddrs {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("GaiAddrs")
+    }
+}
+
+pub(super) struct SocketAddrs {
+    iter: vec::IntoIter<SocketAddr>,
+}
+
+impl SocketAddrs {
+    pub(super) fn new(addrs: Vec<SocketAddr>) -> Self {
+        SocketAddrs {
+            iter: addrs.into_iter(),
+        }
+    }
+
+    pub(super) fn try_parse(host: &str, port: u16) -> Option<SocketAddrs> {
+        if let Ok(addr) = host.parse::<Ipv4Addr>() {
+            let addr = SocketAddrV4::new(addr, port);
+            return Some(SocketAddrs {
+                iter: vec![SocketAddr::V4(addr)].into_iter(),
+            });
+        }
+        if let Ok(addr) = host.parse::<Ipv6Addr>() {
+            let addr = SocketAddrV6::new(addr, port, 0, 0);
+            return Some(SocketAddrs {
+                iter: vec![SocketAddr::V6(addr)].into_iter(),
+            });
+        }
+        None
+    }
+
+    #[inline]
+    fn filter(self, predicate: impl FnMut(&SocketAddr) -> bool) -> SocketAddrs {
+        SocketAddrs::new(self.iter.filter(predicate).collect())
+    }
+
+    pub(super) fn split_by_preference(
+        self,
+        local_addr_ipv4: Option<Ipv4Addr>,
+        local_addr_ipv6: Option<Ipv6Addr>,
+    ) -> (SocketAddrs, SocketAddrs) {
+        match (local_addr_ipv4, local_addr_ipv6) {
+            (Some(_), None) => (self.filter(SocketAddr::is_ipv4), SocketAddrs::new(vec![])),
+            (None, Some(_)) => (self.filter(SocketAddr::is_ipv6), SocketAddrs::new(vec![])),
+            _ => {
+                let preferring_v6 = self
+                    .iter
+                    .as_slice()
+                    .first()
+                    .map(SocketAddr::is_ipv6)
+                    .unwrap_or(false);
+
+                let (preferred, fallback) = self
+                    .iter
+                    .partition::<Vec<_>, _>(|addr| addr.is_ipv6() == preferring_v6);
+
+                (SocketAddrs::new(preferred), SocketAddrs::new(fallback))
+            }
+        }
+    }
+
+    pub(super) fn is_empty(&self) -> bool {
+        self.iter.as_slice().is_empty()
+    }
+
+    pub(super) fn len(&self) -> usize {
+        self.iter.as_slice().len()
+    }
+}
+
+impl Iterator for SocketAddrs {
+    type Item = SocketAddr;
+    #[inline]
+    fn next(&mut self) -> Option<SocketAddr> {
+        self.iter.next()
+    }
+}
+
+/*
+/// A resolver using `getaddrinfo` calls via the `tokio_executor::threadpool::blocking` API.
+///
+/// Unlike the `GaiResolver` this will not spawn dedicated threads, but only works when running on the
+/// multi-threaded Tokio runtime.
+#[cfg(feature = "runtime")]
+#[derive(Clone, Debug)]
+pub struct TokioThreadpoolGaiResolver(());
+
+/// The future returned by `TokioThreadpoolGaiResolver`.
+#[cfg(feature = "runtime")]
+#[derive(Debug)]
+pub struct TokioThreadpoolGaiFuture {
+    name: Name,
+}
+
+#[cfg(feature = "runtime")]
+impl TokioThreadpoolGaiResolver {
+    /// Creates a new DNS resolver that will use tokio threadpool's blocking
+    /// feature.
+    ///
+    /// **Requires** its futures to be run on the threadpool runtime.
+    pub fn new() -> Self {
+        TokioThreadpoolGaiResolver(())
+    }
+}
+
+#[cfg(feature = "runtime")]
+impl Service<Name> for TokioThreadpoolGaiResolver {
+    type Response = GaiAddrs;
+    type Error = io::Error;
+    type Future = TokioThreadpoolGaiFuture;
+
+    fn poll_ready(&mut self, _cx: &mut task::Context<'_>) -> Poll<Result<(), io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, name: Name) -> Self::Future {
+        TokioThreadpoolGaiFuture { name }
+    }
+}
+
+#[cfg(feature = "runtime")]
+impl Future for TokioThreadpoolGaiFuture {
+    type Output = Result<GaiAddrs, io::Error>;
+
+    fn poll(self: Pin<&mut Self>, _cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+        match ready!(tokio_executor::threadpool::blocking(|| (
+            self.name.as_str(),
+            0
+        )
+            .to_socket_addrs()))
+        {
+            Ok(Ok(iter)) => Poll::Ready(Ok(GaiAddrs {
+                inner: IpAddrs { iter },
+            })),
+            Ok(Err(e)) => Poll::Ready(Err(e)),
+            // a BlockingError, meaning not on a tokio_executor::threadpool :(
+            Err(e) => Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
+        }
+    }
+}
+*/
+
+mod sealed {
+    use super::{SocketAddr, Name};
+    use crate::common::{task, Future, Poll};
+    use tower_service::Service;
+
+    // "Trait alias" for `Service<Name, Response = Addrs>`
+    pub trait Resolve {
+        type Addrs: Iterator<Item = SocketAddr>;
+        type Error: Into<Box<dyn std::error::Error + Send + Sync>>;
+        type Future: Future<Output = Result<Self::Addrs, Self::Error>>;
+
+        fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>>;
+        fn resolve(&mut self, name: Name) -> Self::Future;
+    }
+
+    impl<S> Resolve for S
+    where
+        S: Service<Name>,
+        S::Response: Iterator<Item = SocketAddr>,
+        S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+    {
+        type Addrs = S::Response;
+        type Error = S::Error;
+        type Future = S::Future;
+
+        fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Service::poll_ready(self, cx)
+        }
+
+        fn resolve(&mut self, name: Name) -> Self::Future {
+            Service::call(self, name)
+        }
+    }
+}
+
+pub(super) async fn resolve<R>(resolver: &mut R, name: Name) -> Result<R::Addrs, R::Error>
+where
+    R: Resolve,
+{
+    futures_util::future::poll_fn(|cx| resolver.poll_ready(cx)).await?;
+    resolver.resolve(name).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::{Ipv4Addr, Ipv6Addr};
+
+    #[test]
+    fn test_ip_addrs_split_by_preference() {
+        let ip_v4 = Ipv4Addr::new(127, 0, 0, 1);
+        let ip_v6 = Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 1);
+        let v4_addr = (ip_v4, 80).into();
+        let v6_addr = (ip_v6, 80).into();
+
+        let (mut preferred, mut fallback) = SocketAddrs {
+            iter: vec![v4_addr, v6_addr].into_iter(),
+        }
+        .split_by_preference(None, None);
+        assert!(preferred.next().unwrap().is_ipv4());
+        assert!(fallback.next().unwrap().is_ipv6());
+
+        let (mut preferred, mut fallback) = SocketAddrs {
+            iter: vec![v6_addr, v4_addr].into_iter(),
+        }
+        .split_by_preference(None, None);
+        assert!(preferred.next().unwrap().is_ipv6());
+        assert!(fallback.next().unwrap().is_ipv4());
+
+        let (mut preferred, mut fallback) = SocketAddrs {
+            iter: vec![v4_addr, v6_addr].into_iter(),
+        }
+        .split_by_preference(Some(ip_v4), Some(ip_v6));
+        assert!(preferred.next().unwrap().is_ipv4());
+        assert!(fallback.next().unwrap().is_ipv6());
+
+        let (mut preferred, mut fallback) = SocketAddrs {
+            iter: vec![v6_addr, v4_addr].into_iter(),
+        }
+        .split_by_preference(Some(ip_v4), Some(ip_v6));
+        assert!(preferred.next().unwrap().is_ipv6());
+        assert!(fallback.next().unwrap().is_ipv4());
+
+        let (mut preferred, fallback) = SocketAddrs {
+            iter: vec![v4_addr, v6_addr].into_iter(),
+        }
+        .split_by_preference(Some(ip_v4), None);
+        assert!(preferred.next().unwrap().is_ipv4());
+        assert!(fallback.is_empty());
+
+        let (mut preferred, fallback) = SocketAddrs {
+            iter: vec![v4_addr, v6_addr].into_iter(),
+        }
+        .split_by_preference(None, Some(ip_v6));
+        assert!(preferred.next().unwrap().is_ipv6());
+        assert!(fallback.is_empty());
+    }
+
+    #[test]
+    fn test_name_from_str() {
+        const DOMAIN: &str = "test.example.com";
+        let name = Name::from_str(DOMAIN).expect("Should be a valid domain");
+        assert_eq!(name.as_str(), DOMAIN);
+        assert_eq!(name.to_string(), DOMAIN);
+    }
+}
diff --git a/third_party/rust/hyper/src/client/connect/http.rs b/third_party/rust/hyper/src/client/connect/http.rs
new file mode 100644
index 0000000000..afe7b155eb
--- /dev/null
+++ b/third_party/rust/hyper/src/client/connect/http.rs
@@ -0,0 +1,1007 @@
+use std::error::Error as StdError;
+use std::fmt;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomData;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{self, Poll};
+use std::time::Duration;
+
+use futures_util::future::Either;
+use http::uri::{Scheme, Uri};
+use pin_project_lite::pin_project;
+use tokio::net::{TcpSocket, TcpStream};
+use tokio::time::Sleep;
+use tracing::{debug, trace, warn};
+
+use super::dns::{self, resolve, GaiResolver, Resolve};
+use super::{Connected, Connection};
+//#[cfg(feature = "runtime")] use super::dns::TokioThreadpoolGaiResolver;
+
+/// A connector for the `http` scheme.
+///
+/// Performs DNS resolution in a thread pool, and then connects over TCP.
+///
+/// # Note
+///
+/// Sets the [`HttpInfo`](HttpInfo) value on responses, which includes
+/// transport information such as the remote socket address used.
+#[cfg_attr(docsrs, doc(cfg(feature = "tcp")))]
+#[derive(Clone)]
+pub struct HttpConnector<R = GaiResolver> {
+    config: Arc<Config>,
+    resolver: R,
+}
+
+/// Extra information about the transport when an HttpConnector is used.
+///
+/// # Example
+///
+/// ```
+/// # async fn doc() -> hyper::Result<()> {
+/// use hyper::Uri;
+/// use hyper::client::{Client, connect::HttpInfo};
+///
+/// let client = Client::new();
+/// let uri = Uri::from_static("http://example.com");
+///
+/// let res = client.get(uri).await?;
+/// res
+///     .extensions()
+///     .get::<HttpInfo>()
+///     .map(|info| {
+///         println!("remote addr = {}", info.remote_addr());
+///     });
+/// # Ok(())
+/// # }
+/// ```
+///
+/// # Note
+///
+/// If a different connector is used besides [`HttpConnector`](HttpConnector),
+/// this value will not exist in the extensions. Consult that specific
+/// connector to see what "extra" information it might provide to responses.
+#[derive(Clone, Debug)]
+pub struct HttpInfo {
+    remote_addr: SocketAddr,
+    local_addr: SocketAddr,
+}
+
+#[derive(Clone)]
+struct Config {
+    connect_timeout: Option<Duration>,
+    enforce_http: bool,
+    happy_eyeballs_timeout: Option<Duration>,
+    keep_alive_timeout: Option<Duration>,
+    local_address_ipv4: Option<Ipv4Addr>,
+    local_address_ipv6: Option<Ipv6Addr>,
+    nodelay: bool,
+    reuse_address: bool,
+    send_buffer_size: Option<usize>,
+    recv_buffer_size: Option<usize>,
+}
+
+// ===== impl HttpConnector =====
+
+impl HttpConnector {
+    /// Construct a new HttpConnector.
+    pub fn new() -> HttpConnector {
+        HttpConnector::new_with_resolver(GaiResolver::new())
+    }
+}
+
+/*
+#[cfg(feature = "runtime")]
+impl HttpConnector<TokioThreadpoolGaiResolver> {
+    /// Construct a new HttpConnector using the `TokioThreadpoolGaiResolver`.
+    ///
+    /// This resolver **requires** the threadpool runtime to be used.
+    pub fn new_with_tokio_threadpool_resolver() -> Self {
+        HttpConnector::new_with_resolver(TokioThreadpoolGaiResolver::new())
+    }
+}
+*/
+
+impl<R> HttpConnector<R> {
+    /// Construct a new HttpConnector.
+    ///
+    /// Takes a [`Resolver`](crate::client::connect::dns#resolvers-are-services) to handle DNS lookups.
+    pub fn new_with_resolver(resolver: R) -> HttpConnector<R> {
+        HttpConnector {
+            config: Arc::new(Config {
+                connect_timeout: None,
+                enforce_http: true,
+                happy_eyeballs_timeout: Some(Duration::from_millis(300)),
+                keep_alive_timeout: None,
+                local_address_ipv4: None,
+                local_address_ipv6: None,
+                nodelay: false,
+                reuse_address: false,
+                send_buffer_size: None,
+                recv_buffer_size: None,
+            }),
+            resolver,
+        }
+    }
+
+    /// Option to enforce all `Uri`s have the `http` scheme.
+    ///
+    /// Enabled by default.
+    #[inline]
+    pub fn enforce_http(&mut self, is_enforced: bool) {
+        self.config_mut().enforce_http = is_enforced;
+    }
+
+    /// Set that all sockets have `SO_KEEPALIVE` set with the supplied duration.
+    ///
+    /// If `None`, the option will not be set.
+    ///
+    /// Default is `None`.
+    #[inline]
+    pub fn set_keepalive(&mut self, dur: Option<Duration>) {
+        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.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.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.config_mut().recv_buffer_size = size;
+    }
+
+    /// Set that all sockets are bound to the configured address before connection.
+    ///
+    /// If `None`, the sockets will not be bound.
+    ///
+    /// Default is `None`.
+    #[inline]
+    pub fn set_local_address(&mut self, addr: Option<IpAddr>) {
+        let (v4, v6) = match addr {
+            Some(IpAddr::V4(a)) => (Some(a), None),
+            Some(IpAddr::V6(a)) => (None, Some(a)),
+            _ => (None, None),
+        };
+
+        let cfg = self.config_mut();
+
+        cfg.local_address_ipv4 = v4;
+        cfg.local_address_ipv6 = v6;
+    }
+
+    /// Set that all sockets are bound to the configured IPv4 or IPv6 address (depending on host's
+    /// preferences) before connection.
+    #[inline]
+    pub fn set_local_addresses(&mut self, addr_ipv4: Ipv4Addr, addr_ipv6: Ipv6Addr) {
+        let cfg = self.config_mut();
+
+        cfg.local_address_ipv4 = Some(addr_ipv4);
+        cfg.local_address_ipv6 = Some(addr_ipv6);
+    }
+
+    /// Set the connect timeout.
+    ///
+    /// If a domain resolves to multiple IP addresses, the timeout will be
+    /// evenly divided across them.
+    ///
+    /// Default is `None`.
+    #[inline]
+    pub fn set_connect_timeout(&mut self, dur: Option<Duration>) {
+        self.config_mut().connect_timeout = dur;
+    }
+
+    /// Set timeout for [RFC 6555 (Happy Eyeballs)][RFC 6555] algorithm.
+    ///
+    /// If hostname resolves to both IPv4 and IPv6 addresses and connection
+    /// cannot be established using preferred address family before timeout
+    /// elapses, then connector will in parallel attempt connection using other
+    /// address family.
+    ///
+    /// If `None`, parallel connection attempts are disabled.
+    ///
+    /// Default is 300 milliseconds.
+    ///
+    /// [RFC 6555]: https://tools.ietf.org/html/rfc6555
+    #[inline]
+    pub fn set_happy_eyeballs_timeout(&mut self, dur: Option<Duration>) {
+        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.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)
+    }
+}
+
+static INVALID_NOT_HTTP: &str = "invalid URL, scheme is not http";
+static INVALID_MISSING_SCHEME: &str = "invalid URL, scheme is missing";
+static INVALID_MISSING_HOST: &str = "invalid URL, host is missing";
+
+// R: Debug required for now to allow adding it to debug output later...
+impl<R: fmt::Debug> fmt::Debug for HttpConnector<R> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("HttpConnector").finish()
+    }
+}
+
+impl<R> tower_service::Service<Uri> for HttpConnector<R>
+where
+    R: Resolve + Clone + Send + Sync + 'static,
+    R::Future: Send,
+{
+    type Response = TcpStream;
+    type Error = ConnectError;
+    type Future = HttpConnecting<R>;
+
+    fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+        ready!(self.resolver.poll_ready(cx)).map_err(ConnectError::dns)?;
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, dst: Uri) -> Self::Future {
+        let mut self_ = self.clone();
+        HttpConnecting {
+            fut: Box::pin(async move { self_.call_async(dst).await }),
+            _marker: PhantomData,
+        }
+    }
+}
+
+fn get_host_port<'u>(config: &Config, dst: &'u Uri) -> Result<(&'u str, u16), ConnectError> {
+    trace!(
+        "Http::connect; scheme={:?}, host={:?}, port={:?}",
+        dst.scheme(),
+        dst.host(),
+        dst.port(),
+    );
+
+    if config.enforce_http {
+        if dst.scheme() != Some(&Scheme::HTTP) {
+            return Err(ConnectError {
+                msg: INVALID_NOT_HTTP.into(),
+                cause: None,
+            });
+        }
+    } else if dst.scheme().is_none() {
+        return Err(ConnectError {
+            msg: INVALID_MISSING_SCHEME.into(),
+            cause: None,
+        });
+    }
+
+    let host = match dst.host() {
+        Some(s) => s,
+        None => {
+            return Err(ConnectError {
+                msg: INVALID_MISSING_HOST.into(),
+                cause: None,
+            })
+        }
+    };
+    let port = match dst.port() {
+        Some(port) => port.as_u16(),
+        None => {
+            if dst.scheme() == Some(&Scheme::HTTPS) {
+                443
+            } else {
+                80
+            }
+        }
+    };
+
+    Ok((host, port))
+}
+
+impl<R> HttpConnector<R>
+where
+    R: Resolve,
+{
+    async fn call_async(&mut self, dst: Uri) -> Result<TcpStream, ConnectError> {
+        let config = &self.config;
+
+        let (host, port) = get_host_port(config, &dst)?;
+        let host = host.trim_start_matches('[').trim_end_matches(']');
+
+        // If the host is already an IP addr (v4 or v6),
+        // skip resolving the dns and start connecting right away.
+        let addrs = if let Some(addrs) = dns::SocketAddrs::try_parse(host, port) {
+            addrs
+        } else {
+            let addrs = resolve(&mut self.resolver, dns::Name::new(host.into()))
+                .await
+                .map_err(ConnectError::dns)?;
+            let addrs = addrs
+                .map(|mut addr| {
+                    addr.set_port(port);
+                    addr
+                })
+                .collect();
+            dns::SocketAddrs::new(addrs)
+        };
+
+        let c = ConnectingTcp::new(addrs, config);
+
+        let sock = c.connect().await?;
+
+        if let Err(e) = sock.set_nodelay(config.nodelay) {
+            warn!("tcp set_nodelay error: {}", e);
+        }
+
+        Ok(sock)
+    }
+}
+
+impl Connection for TcpStream {
+    fn connected(&self) -> Connected {
+        let connected = Connected::new();
+        if let (Ok(remote_addr), Ok(local_addr)) = (self.peer_addr(), self.local_addr()) {
+            connected.extra(HttpInfo { remote_addr, local_addr })
+        } else {
+            connected
+        }
+    }
+}
+
+impl HttpInfo {
+    /// Get the remote address of the transport used.
+    pub fn remote_addr(&self) -> SocketAddr {
+        self.remote_addr
+    }
+
+    /// Get the local address of the transport used.
+    pub fn local_addr(&self) -> SocketAddr {
+        self.local_addr
+    }
+}
+
+pin_project! {
+    // Not publicly exported (so missing_docs doesn't trigger).
+    //
+    // We return this `Future` instead of the `Pin<Box<dyn Future>>` directly
+    // so that users don't rely on it fitting in a `Pin<Box<dyn Future>>` slot
+    // (and thus we can change the type in the future).
+    #[must_use = "futures do nothing unless polled"]
+    #[allow(missing_debug_implementations)]
+    pub struct HttpConnecting<R> {
+        #[pin]
+        fut: BoxConnecting,
+        _marker: PhantomData<R>,
+    }
+}
+
+type ConnectResult = Result<TcpStream, ConnectError>;
+type BoxConnecting = Pin<Box<dyn Future<Output = ConnectResult> + Send>>;
+
+impl<R: Resolve> Future for HttpConnecting<R> {
+    type Output = ConnectResult;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+        self.project().fut.poll(cx)
+    }
+}
+
+// Not publicly exported (so missing_docs doesn't trigger).
+pub struct ConnectError {
+    msg: Box<str>,
+    cause: Option<Box<dyn StdError + Send + Sync>>,
+}
+
+impl ConnectError {
+    fn new<S, E>(msg: S, cause: E) -> ConnectError
+    where
+        S: Into<Box<str>>,
+        E: Into<Box<dyn StdError + Send + Sync>>,
+    {
+        ConnectError {
+            msg: msg.into(),
+            cause: Some(cause.into()),
+        }
+    }
+
+    fn dns<E>(cause: E) -> ConnectError
+    where
+        E: Into<Box<dyn StdError + Send + Sync>>,
+    {
+        ConnectError::new("dns error", cause)
+    }
+
+    fn m<S, E>(msg: S) -> impl FnOnce(E) -> ConnectError
+    where
+        S: Into<Box<str>>,
+        E: Into<Box<dyn StdError + Send + Sync>>,
+    {
+        move |cause| ConnectError::new(msg, cause)
+    }
+}
+
+impl fmt::Debug for ConnectError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if let Some(ref cause) = self.cause {
+            f.debug_tuple("ConnectError")
+                .field(&self.msg)
+                .field(cause)
+                .finish()
+        } else {
+            self.msg.fmt(f)
+        }
+    }
+}
+
+impl fmt::Display for ConnectError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&self.msg)?;
+
+        if let Some(ref cause) = self.cause {
+            write!(f, ": {}", cause)?;
+        }
+
+        Ok(())
+    }
+}
+
+impl StdError for ConnectError {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        self.cause.as_ref().map(|e| &**e as _)
+    }
+}
+
+struct ConnectingTcp<'a> {
+    preferred: ConnectingTcpRemote,
+    fallback: Option<ConnectingTcpFallback>,
+    config: &'a Config,
+}
+
+impl<'a> ConnectingTcp<'a> {
+    fn new(remote_addrs: dns::SocketAddrs, config: &'a Config) -> Self {
+        if let Some(fallback_timeout) = config.happy_eyeballs_timeout {
+            let (preferred_addrs, fallback_addrs) = remote_addrs
+                .split_by_preference(config.local_address_ipv4, config.local_address_ipv6);
+            if fallback_addrs.is_empty() {
+                return ConnectingTcp {
+                    preferred: ConnectingTcpRemote::new(preferred_addrs, config.connect_timeout),
+                    fallback: None,
+                    config,
+                };
+            }
+
+            ConnectingTcp {
+                preferred: ConnectingTcpRemote::new(preferred_addrs, config.connect_timeout),
+                fallback: Some(ConnectingTcpFallback {
+                    delay: tokio::time::sleep(fallback_timeout),
+                    remote: ConnectingTcpRemote::new(fallback_addrs, config.connect_timeout),
+                }),
+                config,
+            }
+        } else {
+            ConnectingTcp {
+                preferred: ConnectingTcpRemote::new(remote_addrs, config.connect_timeout),
+                fallback: None,
+                config,
+            }
+        }
+    }
+}
+
+struct ConnectingTcpFallback {
+    delay: Sleep,
+    remote: ConnectingTcpRemote,
+}
+
+struct ConnectingTcpRemote {
+    addrs: dns::SocketAddrs,
+    connect_timeout: Option<Duration>,
+}
+
+impl ConnectingTcpRemote {
+    fn new(addrs: dns::SocketAddrs, connect_timeout: Option<Duration>) -> Self {
+        let connect_timeout = connect_timeout.map(|t| t / (addrs.len() as u32));
+
+        Self {
+            addrs,
+            connect_timeout,
+        }
+    }
+}
+
+impl ConnectingTcpRemote {
+    async fn connect(&mut self, config: &Config) -> Result<TcpStream, ConnectError> {
+        let mut err = None;
+        for addr in &mut self.addrs {
+            debug!("connecting to {}", addr);
+            match connect(&addr, config, self.connect_timeout)?.await {
+                Ok(tcp) => {
+                    debug!("connected to {}", addr);
+                    return Ok(tcp);
+                }
+                Err(e) => {
+                    trace!("connect error for {}: {:?}", addr, e);
+                    err = Some(e);
+                }
+            }
+        }
+
+        match err {
+            Some(e) => Err(e),
+            None => Err(ConnectError::new(
+                "tcp connect error",
+                std::io::Error::new(std::io::ErrorKind::NotConnected, "Network unreachable"),
+            )),
+        }
+    }
+}
+
+fn bind_local_address(
+    socket: &socket2::Socket,
+    dst_addr: &SocketAddr,
+    local_addr_ipv4: &Option<Ipv4Addr>,
+    local_addr_ipv6: &Option<Ipv6Addr>,
+) -> io::Result<()> {
+    match (*dst_addr, local_addr_ipv4, local_addr_ipv6) {
+        (SocketAddr::V4(_), Some(addr), _) => {
+            socket.bind(&SocketAddr::new(addr.clone().into(), 0).into())?;
+        }
+        (SocketAddr::V6(_), _, Some(addr)) => {
+            socket.bind(&SocketAddr::new(addr.clone().into(), 0).into())?;
+        }
+        _ => {
+            if cfg!(windows) {
+                // Windows requires a socket be bound before calling connect
+                let any: SocketAddr = match *dst_addr {
+                    SocketAddr::V4(_) => ([0, 0, 0, 0], 0).into(),
+                    SocketAddr::V6(_) => ([0, 0, 0, 0, 0, 0, 0, 0], 0).into(),
+                };
+                socket.bind(&any.into())?;
+            }
+        }
+    }
+
+    Ok(())
+}
+
+fn connect(
+    addr: &SocketAddr,
+    config: &Config,
+    connect_timeout: Option<Duration>,
+) -> Result<impl Future<Output = Result<TcpStream, ConnectError>>, ConnectError> {
+    // TODO(eliza): if Tokio's `TcpSocket` gains support for setting the
+    // keepalive timeout, it would be nice to use that instead of socket2,
+    // and avoid the unsafe `into_raw_fd`/`from_raw_fd` dance...
+    use socket2::{Domain, Protocol, Socket, TcpKeepalive, Type};
+    use std::convert::TryInto;
+
+    let domain = Domain::for_address(*addr);
+    let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))
+        .map_err(ConnectError::m("tcp open error"))?;
+
+    // When constructing a Tokio `TcpSocket` from a raw fd/socket, the user is
+    // responsible for ensuring O_NONBLOCK is set.
+    socket
+        .set_nonblocking(true)
+        .map_err(ConnectError::m("tcp set_nonblocking error"))?;
+
+    if let Some(dur) = config.keep_alive_timeout {
+        let conf = TcpKeepalive::new().with_time(dur);
+        if let Err(e) = socket.set_tcp_keepalive(&conf) {
+            warn!("tcp set_keepalive error: {}", e);
+        }
+    }
+
+    bind_local_address(
+        &socket,
+        addr,
+        &config.local_address_ipv4,
+        &config.local_address_ipv6,
+    )
+    .map_err(ConnectError::m("tcp bind local error"))?;
+
+    #[cfg(unix)]
+    let socket = unsafe {
+        // Safety: `from_raw_fd` is only safe to call if ownership of the raw
+        // file descriptor is transferred. Since we call `into_raw_fd` on the
+        // socket2 socket, it gives up ownership of the fd and will not close
+        // it, so this is safe.
+        use std::os::unix::io::{FromRawFd, IntoRawFd};
+        TcpSocket::from_raw_fd(socket.into_raw_fd())
+    };
+    #[cfg(windows)]
+    let socket = unsafe {
+        // Safety: `from_raw_socket` is only safe to call if ownership of the raw
+        // Windows SOCKET is transferred. Since we call `into_raw_socket` on the
+        // socket2 socket, it gives up ownership of the SOCKET and will not close
+        // it, so this is safe.
+        use std::os::windows::io::{FromRawSocket, IntoRawSocket};
+        TcpSocket::from_raw_socket(socket.into_raw_socket())
+    };
+
+    if config.reuse_address {
+        if let Err(e) = socket.set_reuseaddr(true) {
+            warn!("tcp set_reuse_address error: {}", e);
+        }
+    }
+
+    if let Some(size) = config.send_buffer_size {
+        if let Err(e) = socket.set_send_buffer_size(size.try_into().unwrap_or(std::u32::MAX)) {
+            warn!("tcp set_buffer_size error: {}", e);
+        }
+    }
+
+    if let Some(size) = config.recv_buffer_size {
+        if let Err(e) = socket.set_recv_buffer_size(size.try_into().unwrap_or(std::u32::MAX)) {
+            warn!("tcp set_recv_buffer_size error: {}", e);
+        }
+    }
+
+    let connect = socket.connect(*addr);
+    Ok(async move {
+        match connect_timeout {
+            Some(dur) => match tokio::time::timeout(dur, connect).await {
+                Ok(Ok(s)) => Ok(s),
+                Ok(Err(e)) => Err(e),
+                Err(e) => Err(io::Error::new(io::ErrorKind::TimedOut, e)),
+            },
+            None => connect.await,
+        }
+        .map_err(ConnectError::m("tcp connect error"))
+    })
+}
+
+impl ConnectingTcp<'_> {
+    async fn connect(mut self) -> Result<TcpStream, ConnectError> {
+        match self.fallback {
+            None => self.preferred.connect(self.config).await,
+            Some(mut fallback) => {
+                let preferred_fut = self.preferred.connect(self.config);
+                futures_util::pin_mut!(preferred_fut);
+
+                let fallback_fut = fallback.remote.connect(self.config);
+                futures_util::pin_mut!(fallback_fut);
+
+                let fallback_delay = fallback.delay;
+                futures_util::pin_mut!(fallback_delay);
+
+                let (result, future) =
+                    match futures_util::future::select(preferred_fut, fallback_delay).await {
+                        Either::Left((result, _fallback_delay)) => {
+                            (result, Either::Right(fallback_fut))
+                        }
+                        Either::Right(((), preferred_fut)) => {
+                            // Delay is done, start polling both the preferred and the fallback
+                            futures_util::future::select(preferred_fut, fallback_fut)
+                                .await
+                                .factor_first()
+                        }
+                    };
+
+                if result.is_err() {
+                    // Fallback to the remaining future (could be preferred or fallback)
+                    // if we get an error
+                    future.await
+                } else {
+                    result
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io;
+
+    use ::http::Uri;
+
+    use super::super::sealed::{Connect, ConnectSvc};
+    use super::{Config, ConnectError, HttpConnector};
+
+    async fn connect<C>(
+        connector: C,
+        dst: Uri,
+    ) -> Result<<C::_Svc as ConnectSvc>::Connection, <C::_Svc as ConnectSvc>::Error>
+    where
+        C: Connect,
+    {
+        connector.connect(super::super::sealed::Internal, dst).await
+    }
+
+    #[tokio::test]
+    async fn test_errors_enforce_http() {
+        let dst = "https://example.domain/foo/bar?baz".parse().unwrap();
+        let connector = HttpConnector::new();
+
+        let err = connect(connector, dst).await.unwrap_err();
+        assert_eq!(&*err.msg, super::INVALID_NOT_HTTP);
+    }
+
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
+    fn get_local_ips() -> (Option<std::net::Ipv4Addr>, Option<std::net::Ipv6Addr>) {
+        use std::net::{IpAddr, TcpListener};
+
+        let mut ip_v4 = None;
+        let mut ip_v6 = None;
+
+        let ips = pnet_datalink::interfaces()
+            .into_iter()
+            .flat_map(|i| i.ips.into_iter().map(|n| n.ip()));
+
+        for ip in ips {
+            match ip {
+                IpAddr::V4(ip) if TcpListener::bind((ip, 0)).is_ok() => ip_v4 = Some(ip),
+                IpAddr::V6(ip) if TcpListener::bind((ip, 0)).is_ok() => ip_v6 = Some(ip),
+                _ => (),
+            }
+
+            if ip_v4.is_some() && ip_v6.is_some() {
+                break;
+            }
+        }
+
+        (ip_v4, ip_v6)
+    }
+
+    #[tokio::test]
+    async fn test_errors_missing_scheme() {
+        let dst = "example.domain".parse().unwrap();
+        let mut connector = HttpConnector::new();
+        connector.enforce_http(false);
+
+        let err = connect(connector, dst).await.unwrap_err();
+        assert_eq!(&*err.msg, super::INVALID_MISSING_SCHEME);
+    }
+
+    // NOTE: pnet crate that we use in this test doesn't compile on Windows
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
+    #[tokio::test]
+    async fn local_address() {
+        use std::net::{IpAddr, TcpListener};
+        let _ = pretty_env_logger::try_init();
+
+        let (bind_ip_v4, bind_ip_v6) = get_local_ips();
+        let server4 = TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = server4.local_addr().unwrap().port();
+        let server6 = TcpListener::bind(&format!("[::1]:{}", port)).unwrap();
+
+        let assert_client_ip = |dst: String, server: TcpListener, expected_ip: IpAddr| async move {
+            let mut connector = HttpConnector::new();
+
+            match (bind_ip_v4, bind_ip_v6) {
+                (Some(v4), Some(v6)) => connector.set_local_addresses(v4, v6),
+                (Some(v4), None) => connector.set_local_address(Some(v4.into())),
+                (None, Some(v6)) => connector.set_local_address(Some(v6.into())),
+                _ => unreachable!(),
+            }
+
+            connect(connector, dst.parse().unwrap()).await.unwrap();
+
+            let (_, client_addr) = server.accept().unwrap();
+
+            assert_eq!(client_addr.ip(), expected_ip);
+        };
+
+        if let Some(ip) = bind_ip_v4 {
+            assert_client_ip(format!("http://127.0.0.1:{}", port), server4, ip.into()).await;
+        }
+
+        if let Some(ip) = bind_ip_v6 {
+            assert_client_ip(format!("http://[::1]:{}", port), server6, ip.into()).await;
+        }
+    }
+
+    #[test]
+    #[cfg_attr(not(feature = "__internal_happy_eyeballs_tests"), ignore)]
+    fn client_happy_eyeballs() {
+        use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, TcpListener};
+        use std::time::{Duration, Instant};
+
+        use super::dns;
+        use super::ConnectingTcp;
+
+        let _ = pretty_env_logger::try_init();
+        let server4 = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = server4.local_addr().unwrap();
+        let _server6 = TcpListener::bind(&format!("[::1]:{}", addr.port())).unwrap();
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+
+        let local_timeout = Duration::default();
+        let unreachable_v4_timeout = measure_connect(unreachable_ipv4_addr()).1;
+        let unreachable_v6_timeout = measure_connect(unreachable_ipv6_addr()).1;
+        let fallback_timeout = std::cmp::max(unreachable_v4_timeout, unreachable_v6_timeout)
+            + Duration::from_millis(250);
+
+        let scenarios = &[
+            // Fast primary, without fallback.
+            (&[local_ipv4_addr()][..], 4, local_timeout, false),
+            (&[local_ipv6_addr()][..], 6, local_timeout, false),
+            // Fast primary, with (unused) fallback.
+            (
+                &[local_ipv4_addr(), local_ipv6_addr()][..],
+                4,
+                local_timeout,
+                false,
+            ),
+            (
+                &[local_ipv6_addr(), local_ipv4_addr()][..],
+                6,
+                local_timeout,
+                false,
+            ),
+            // Unreachable + fast primary, without fallback.
+            (
+                &[unreachable_ipv4_addr(), local_ipv4_addr()][..],
+                4,
+                unreachable_v4_timeout,
+                false,
+            ),
+            (
+                &[unreachable_ipv6_addr(), local_ipv6_addr()][..],
+                6,
+                unreachable_v6_timeout,
+                false,
+            ),
+            // Unreachable + fast primary, with (unused) fallback.
+            (
+                &[
+                    unreachable_ipv4_addr(),
+                    local_ipv4_addr(),
+                    local_ipv6_addr(),
+                ][..],
+                4,
+                unreachable_v4_timeout,
+                false,
+            ),
+            (
+                &[
+                    unreachable_ipv6_addr(),
+                    local_ipv6_addr(),
+                    local_ipv4_addr(),
+                ][..],
+                6,
+                unreachable_v6_timeout,
+                true,
+            ),
+            // Slow primary, with (used) fallback.
+            (
+                &[slow_ipv4_addr(), local_ipv4_addr(), local_ipv6_addr()][..],
+                6,
+                fallback_timeout,
+                false,
+            ),
+            (
+                &[slow_ipv6_addr(), local_ipv6_addr(), local_ipv4_addr()][..],
+                4,
+                fallback_timeout,
+                true,
+            ),
+            // Slow primary, with (used) unreachable + fast fallback.
+            (
+                &[slow_ipv4_addr(), unreachable_ipv6_addr(), local_ipv6_addr()][..],
+                6,
+                fallback_timeout + unreachable_v6_timeout,
+                false,
+            ),
+            (
+                &[slow_ipv6_addr(), unreachable_ipv4_addr(), local_ipv4_addr()][..],
+                4,
+                fallback_timeout + unreachable_v4_timeout,
+                true,
+            ),
+        ];
+
+        // Scenarios for IPv6 -> IPv4 fallback require that host can access IPv6 network.
+        // Otherwise, connection to "slow" IPv6 address will error-out immediately.
+        let ipv6_accessible = measure_connect(slow_ipv6_addr()).0;
+
+        for &(hosts, family, timeout, needs_ipv6_access) in scenarios {
+            if needs_ipv6_access && !ipv6_accessible {
+                continue;
+            }
+
+            let (start, stream) = rt
+                .block_on(async move {
+                    let addrs = hosts
+                        .iter()
+                        .map(|host| (host.clone(), addr.port()).into())
+                        .collect();
+                    let cfg = Config {
+                        local_address_ipv4: None,
+                        local_address_ipv6: None,
+                        connect_timeout: None,
+                        keep_alive_timeout: None,
+                        happy_eyeballs_timeout: Some(fallback_timeout),
+                        nodelay: false,
+                        reuse_address: false,
+                        enforce_http: false,
+                        send_buffer_size: None,
+                        recv_buffer_size: None,
+                    };
+                    let connecting_tcp = ConnectingTcp::new(dns::SocketAddrs::new(addrs), &cfg);
+                    let start = Instant::now();
+                    Ok::<_, ConnectError>((start, ConnectingTcp::connect(connecting_tcp).await?))
+                })
+                .unwrap();
+            let res = if stream.peer_addr().unwrap().is_ipv4() {
+                4
+            } else {
+                6
+            };
+            let duration = start.elapsed();
+
+            // Allow actual duration to be +/- 150ms off.
+            let min_duration = if timeout >= Duration::from_millis(150) {
+                timeout - Duration::from_millis(150)
+            } else {
+                Duration::default()
+            };
+            let max_duration = timeout + Duration::from_millis(150);
+
+            assert_eq!(res, family);
+            assert!(duration >= min_duration);
+            assert!(duration <= max_duration);
+        }
+
+        fn local_ipv4_addr() -> IpAddr {
+            Ipv4Addr::new(127, 0, 0, 1).into()
+        }
+
+        fn local_ipv6_addr() -> IpAddr {
+            Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 1).into()
+        }
+
+        fn unreachable_ipv4_addr() -> IpAddr {
+            Ipv4Addr::new(127, 0, 0, 2).into()
+        }
+
+        fn unreachable_ipv6_addr() -> IpAddr {
+            Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 2).into()
+        }
+
+        fn slow_ipv4_addr() -> IpAddr {
+            // RFC 6890 reserved IPv4 address.
+            Ipv4Addr::new(198, 18, 0, 25).into()
+        }
+
+        fn slow_ipv6_addr() -> IpAddr {
+            // RFC 6890 reserved IPv6 address.
+            Ipv6Addr::new(2001, 2, 0, 0, 0, 0, 0, 254).into()
+        }
+
+        fn measure_connect(addr: IpAddr) -> (bool, Duration) {
+            let start = Instant::now();
+            let result =
+                std::net::TcpStream::connect_timeout(&(addr, 80).into(), Duration::from_secs(1));
+
+            let reachable = result.is_ok() || result.unwrap_err().kind() == io::ErrorKind::TimedOut;
+            let duration = start.elapsed();
+            (reachable, duration)
+        }
+    }
+}
diff --git a/third_party/rust/hyper/src/client/connect/mod.rs b/third_party/rust/hyper/src/client/connect/mod.rs
new file mode 100644
index 0000000000..862a0e65c1
--- /dev/null
+++ b/third_party/rust/hyper/src/client/connect/mod.rs
@@ -0,0 +1,412 @@
+//! Connectors used by the `Client`.
+//!
+//! This module contains:
+//!
+//! - A default [`HttpConnector`][] that does DNS resolution and establishes
+//!   connections over TCP.
+//! - Types to build custom connectors.
+//!
+//! # Connectors
+//!
+//! A "connector" is a [`Service`][] that takes a [`Uri`][] destination, and
+//! its `Response` is some type implementing [`AsyncRead`][], [`AsyncWrite`][],
+//! and [`Connection`][].
+//!
+//! ## Custom Connectors
+//!
+//! A simple connector that ignores the `Uri` destination and always returns
+//! a TCP connection to the same address could be written like this:
+//!
+//! ```rust,ignore
+//! let connector = tower::service_fn(|_dst| async {
+//!     tokio::net::TcpStream::connect("127.0.0.1:1337")
+//! })
+//! ```
+//!
+//! Or, fully written out:
+//!
+//! ```
+//! # #[cfg(feature = "runtime")]
+//! # mod rt {
+//! use std::{future::Future, net::SocketAddr, pin::Pin, task::{self, Poll}};
+//! use hyper::{service::Service, Uri};
+//! use tokio::net::TcpStream;
+//!
+//! #[derive(Clone)]
+//! struct LocalConnector;
+//!
+//! impl Service<Uri> for LocalConnector {
+//!     type Response = TcpStream;
+//!     type Error = std::io::Error;
+//!     // We can't "name" an `async` generated future.
+//!     type Future = Pin<Box<
+//!         dyn Future<Output = Result<Self::Response, Self::Error>> + Send
+//!     >>;
+//!
+//!     fn poll_ready(&mut self, _: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+//!         // This connector is always ready, but others might not be.
+//!         Poll::Ready(Ok(()))
+//!     }
+//!
+//!     fn call(&mut self, _: Uri) -> Self::Future {
+//!         Box::pin(TcpStream::connect(SocketAddr::from(([127, 0, 0, 1], 1337))))
+//!     }
+//! }
+//! # }
+//! ```
+//!
+//! It's worth noting that for `TcpStream`s, the [`HttpConnector`][] is a
+//! better starting place to extend from.
+//!
+//! Using either of the above connector examples, it can be used with the
+//! `Client` like this:
+//!
+//! ```
+//! # #[cfg(feature = "runtime")]
+//! # fn rt () {
+//! # let connector = hyper::client::HttpConnector::new();
+//! // let connector = ...
+//!
+//! let client = hyper::Client::builder()
+//!     .build::<_, hyper::Body>(connector);
+//! # }
+//! ```
+//!
+//!
+//! [`HttpConnector`]: HttpConnector
+//! [`Service`]: crate::service::Service
+//! [`Uri`]: ::http::Uri
+//! [`AsyncRead`]: tokio::io::AsyncRead
+//! [`AsyncWrite`]: tokio::io::AsyncWrite
+//! [`Connection`]: Connection
+use std::fmt;
+
+use ::http::Extensions;
+
+cfg_feature! {
+    #![feature = "tcp"]
+
+    pub use self::http::{HttpConnector, HttpInfo};
+
+    pub mod dns;
+    mod http;
+}
+
+cfg_feature! {
+    #![any(feature = "http1", feature = "http2")]
+
+    pub use self::sealed::Connect;
+}
+
+/// Describes a type returned by a connector.
+pub trait Connection {
+    /// Return metadata describing the connection.
+    fn connected(&self) -> Connected;
+}
+
+/// Extra information about the connected transport.
+///
+/// This can be used to inform recipients about things like if ALPN
+/// was used, or if connected to an HTTP proxy.
+#[derive(Debug)]
+pub struct Connected {
+    pub(super) alpn: Alpn,
+    pub(super) is_proxied: bool,
+    pub(super) extra: Option<Extra>,
+}
+
+pub(super) struct Extra(Box<dyn ExtraInner>);
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub(super) enum Alpn {
+    H2,
+    None,
+}
+
+impl Connected {
+    /// Create new `Connected` type with empty metadata.
+    pub fn new() -> Connected {
+        Connected {
+            alpn: Alpn::None,
+            is_proxied: false,
+            extra: None,
+        }
+    }
+
+    /// Set whether the connected transport is to an HTTP proxy.
+    ///
+    /// This setting will affect if HTTP/1 requests written on the transport
+    /// will have the request-target in absolute-form or origin-form:
+    ///
+    /// - When `proxy(false)`:
+    ///
+    /// ```http
+    /// GET /guide HTTP/1.1
+    /// ```
+    ///
+    /// - When `proxy(true)`:
+    ///
+    /// ```http
+    /// GET http://hyper.rs/guide HTTP/1.1
+    /// ```
+    ///
+    /// Default is `false`.
+    pub fn proxy(mut self, is_proxied: bool) -> Connected {
+        self.is_proxied = is_proxied;
+        self
+    }
+
+    /// Determines if the connected transport is to an HTTP proxy.
+    pub fn is_proxied(&self) -> bool {
+        self.is_proxied
+    }
+
+    /// Set extra connection information to be set in the extensions of every `Response`.
+    pub fn extra<T: Clone + Send + Sync + 'static>(mut self, extra: T) -> Connected {
+        if let Some(prev) = self.extra {
+            self.extra = Some(Extra(Box::new(ExtraChain(prev.0, extra))));
+        } else {
+            self.extra = Some(Extra(Box::new(ExtraEnvelope(extra))));
+        }
+        self
+    }
+
+    /// Copies the extra connection information into an `Extensions` map.
+    pub fn get_extras(&self, extensions: &mut Extensions) {
+        if let Some(extra) = &self.extra {
+            extra.set(extensions);
+        }
+    }
+
+    /// Set that the connected transport negotiated HTTP/2 as its next protocol.
+    pub fn negotiated_h2(mut self) -> Connected {
+        self.alpn = Alpn::H2;
+        self
+    }
+
+    /// Determines if the connected transport negotiated HTTP/2 as its next protocol.
+    pub fn is_negotiated_h2(&self) -> bool {
+        self.alpn == Alpn::H2
+    }
+
+    // Don't public expose that `Connected` is `Clone`, unsure if we want to
+    // keep that contract...
+    #[cfg(feature = "http2")]
+    pub(super) fn clone(&self) -> Connected {
+        Connected {
+            alpn: self.alpn.clone(),
+            is_proxied: self.is_proxied,
+            extra: self.extra.clone(),
+        }
+    }
+}
+
+// ===== impl Extra =====
+
+impl Extra {
+    pub(super) fn set(&self, res: &mut Extensions) {
+        self.0.set(res);
+    }
+}
+
+impl Clone for Extra {
+    fn clone(&self) -> Extra {
+        Extra(self.0.clone_box())
+    }
+}
+
+impl fmt::Debug for Extra {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Extra").finish()
+    }
+}
+
+trait ExtraInner: Send + Sync {
+    fn clone_box(&self) -> Box<dyn ExtraInner>;
+    fn set(&self, res: &mut Extensions);
+}
+
+// This indirection allows the `Connected` to have a type-erased "extra" value,
+// while that type still knows its inner extra type. This allows the correct
+// TypeId to be used when inserting into `res.extensions_mut()`.
+#[derive(Clone)]
+struct ExtraEnvelope<T>(T);
+
+impl<T> ExtraInner for ExtraEnvelope<T>
+where
+    T: Clone + Send + Sync + 'static,
+{
+    fn clone_box(&self) -> Box<dyn ExtraInner> {
+        Box::new(self.clone())
+    }
+
+    fn set(&self, res: &mut Extensions) {
+        res.insert(self.0.clone());
+    }
+}
+
+struct ExtraChain<T>(Box<dyn ExtraInner>, T);
+
+impl<T: Clone> Clone for ExtraChain<T> {
+    fn clone(&self) -> Self {
+        ExtraChain(self.0.clone_box(), self.1.clone())
+    }
+}
+
+impl<T> ExtraInner for ExtraChain<T>
+where
+    T: Clone + Send + Sync + 'static,
+{
+    fn clone_box(&self) -> Box<dyn ExtraInner> {
+        Box::new(self.clone())
+    }
+
+    fn set(&self, res: &mut Extensions) {
+        self.0.set(res);
+        res.insert(self.1.clone());
+    }
+}
+
+#[cfg(any(feature = "http1", feature = "http2"))]
+pub(super) mod sealed {
+    use std::error::Error as StdError;
+
+    use ::http::Uri;
+    use tokio::io::{AsyncRead, AsyncWrite};
+
+    use super::Connection;
+    use crate::common::{Future, Unpin};
+
+    /// Connect to a destination, returning an IO transport.
+    ///
+    /// A connector receives a [`Uri`](::http::Uri) and returns a `Future` of the
+    /// ready connection.
+    ///
+    /// # 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<Uri>` 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 + Sized {
+        #[doc(hidden)]
+        type _Svc: ConnectSvc;
+        #[doc(hidden)]
+        fn connect(self, internal_only: Internal, dst: Uri) -> <Self::_Svc as ConnectSvc>::Future;
+    }
+
+    pub trait ConnectSvc {
+        type Connection: AsyncRead + AsyncWrite + Connection + Unpin + Send + 'static;
+        type Error: Into<Box<dyn StdError + Send + Sync>>;
+        type Future: Future<Output = Result<Self::Connection, Self::Error>> + Unpin + Send + 'static;
+
+        fn connect(self, internal_only: Internal, dst: Uri) -> Self::Future;
+    }
+
+    impl<S, T> Connect for S
+    where
+        S: tower_service::Service<Uri, Response = T> + Send + 'static,
+        S::Error: Into<Box<dyn StdError + Send + Sync>>,
+        S::Future: Unpin + Send,
+        T: AsyncRead + AsyncWrite + Connection + Unpin + Send + 'static,
+    {
+        type _Svc = S;
+
+        fn connect(self, _: Internal, dst: Uri) -> crate::service::Oneshot<S, Uri> {
+            crate::service::oneshot(self, dst)
+        }
+    }
+
+    impl<S, T> ConnectSvc for S
+    where
+        S: tower_service::Service<Uri, Response = T> + Send + 'static,
+        S::Error: Into<Box<dyn StdError + Send + Sync>>,
+        S::Future: Unpin + Send,
+        T: AsyncRead + AsyncWrite + Connection + Unpin + Send + 'static,
+    {
+        type Connection = T;
+        type Error = S::Error;
+        type Future = crate::service::Oneshot<S, Uri>;
+
+        fn connect(self, _: Internal, dst: Uri) -> Self::Future {
+            crate::service::oneshot(self, dst)
+        }
+    }
+
+    impl<S, T> Sealed for S
+    where
+        S: tower_service::Service<Uri, Response = T> + Send,
+        S::Error: Into<Box<dyn StdError + Send + Sync>>,
+        S::Future: Unpin + Send,
+        T: AsyncRead + AsyncWrite + Connection + Unpin + Send + 'static,
+    {
+    }
+
+    pub trait Sealed {}
+    #[allow(missing_debug_implementations)]
+    pub struct Internal;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Connected;
+
+    #[derive(Clone, Debug, PartialEq)]
+    struct Ex1(usize);
+
+    #[derive(Clone, Debug, PartialEq)]
+    struct Ex2(&'static str);
+
+    #[derive(Clone, Debug, PartialEq)]
+    struct Ex3(&'static str);
+
+    #[test]
+    fn test_connected_extra() {
+        let c1 = Connected::new().extra(Ex1(41));
+
+        let mut ex = ::http::Extensions::new();
+
+        assert_eq!(ex.get::<Ex1>(), None);
+
+        c1.extra.as_ref().expect("c1 extra").set(&mut ex);
+
+        assert_eq!(ex.get::<Ex1>(), Some(&Ex1(41)));
+    }
+
+    #[test]
+    fn test_connected_extra_chain() {
+        // If a user composes connectors and at each stage, there's "extra"
+        // info to attach, it shouldn't override the previous extras.
+
+        let c1 = Connected::new()
+            .extra(Ex1(45))
+            .extra(Ex2("zoom"))
+            .extra(Ex3("pew pew"));
+
+        let mut ex1 = ::http::Extensions::new();
+
+        assert_eq!(ex1.get::<Ex1>(), None);
+        assert_eq!(ex1.get::<Ex2>(), None);
+        assert_eq!(ex1.get::<Ex3>(), None);
+
+        c1.extra.as_ref().expect("c1 extra").set(&mut ex1);
+
+        assert_eq!(ex1.get::<Ex1>(), Some(&Ex1(45)));
+        assert_eq!(ex1.get::<Ex2>(), Some(&Ex2("zoom")));
+        assert_eq!(ex1.get::<Ex3>(), Some(&Ex3("pew pew")));
+
+        // Just like extensions, inserting the same type overrides previous type.
+        let c2 = Connected::new()
+            .extra(Ex1(33))
+            .extra(Ex2("hiccup"))
+            .extra(Ex1(99));
+
+        let mut ex2 = ::http::Extensions::new();
+
+        c2.extra.as_ref().expect("c2 extra").set(&mut ex2);
+
+        assert_eq!(ex2.get::<Ex1>(), Some(&Ex1(99)));
+        assert_eq!(ex2.get::<Ex2>(), Some(&Ex2("hiccup")));
+    }
+}