feat(client): add Resolve, used by HttpConnector

This introduces a `Resolve` trait to describe asynchronous DNS
resolution. The `HttpConnector` can be configured with a resolver,
allowing a user to still use all the functionality of the
`HttpConnector`, while customizing the DNS resolution.

To prevent a breaking change, the `HttpConnector` has its `Resolve`
generic set by default to `GaiResolver`. This is same as the existing
resolver, which uses `getaddrinfo` inside a thread pool.

Closes #1517

src/client/connect/dns.rs

@@ -9,61 +9,99 @@ use std::sync::Arc;
 use futures::{Async, Future, Poll};
 use futures::future::{Executor, ExecuteError};
 use futures::sync::oneshot;
+use futures_cpupool::{Builder as CpuPoolBuilder};
 
 use self::sealed::GaiTask;
 
+/// Resolve a hostname to a set of IP addresses.
 pub trait Resolve {
+    /// The set of IP addresses to try to connect to.
     type Addrs: Iterator<Item=IpAddr>;
+    /// A Future of the resolved set of addresses.
     type Future: Future<Item=Self::Addrs, Error=io::Error>;
-    fn resolve(&self, host: &str) -> Self::Future;
+    /// Resolve a hostname.
+    fn resolve(&self, name: Name) -> Self::Future;
 }
 
-pub struct DefaultResolver {
-    inner: GaiResolver,
+/// A domain name to resolve into IP addresses.
+pub struct Name {
+    host: String,
 }
 
-pub struct DefaultAddrs;
-pub struct DefaultFuture;
-
+/// A resolver using blocking `getaddrinfo` calls in a threadpool.
+#[derive(Clone)]
 pub struct GaiResolver {
     executor: GaiExecutor,
 }
 
-pub struct GaiAddrs;
-pub struct GaiFuture;
+pub struct GaiAddrs {
+    inner: IpAddrs,
+}
 
-impl Resolve for DefaultResolver {
-    type Addrs = DefaultAddrs;
-    type Future = DefaultFuture;
+pub struct GaiFuture {
+    rx: oneshot::SpawnHandle<IpAddrs, io::Error>,
+}
 
-    fn resolve(&self, host: &str) -> Self::Future {
-        DefaultFuture
+impl Name {
+    pub(super) fn new(host: String) -> Name {
+        Name {
+            host,
+        }
     }
-}
 
-impl Future for DefaultFuture {
-    type Item = DefaultAddrs;
-    type Error = io::Error;
+    /// View the hostname as a string slice.
+    pub fn as_str(&self) -> &str {
+        &self.host
+    }
+}
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        unimplemented!("DefaultFuture::poll");
+impl fmt::Debug for Name {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        fmt::Debug::fmt(&self.host, f)
     }
 }
 
-impl Iterator for DefaultAddrs {
-    type Item = IpAddr;
+impl GaiResolver {
+    /// Construct a new `GaiResolver`.
+    ///
+    /// Takes number of DNS worker threads.
+    pub fn new(threads: usize) -> Self {
+        let pool = CpuPoolBuilder::new()
+            .name_prefix("hyper-dns")
+            .pool_size(threads)
+            .create();
+        GaiResolver::new_with_executor(pool)
+    }
 
-    fn next(&mut self) -> Option<Self::Item> {
-        unimplemented!("DefaultAddrs::next");
+    /// Construct a new `GaiResolver` with a shared thread pool executor.
+    ///
+    /// Takes an executor to run blocking `getaddrinfo` tasks on.
+    pub fn new_with_executor<E: 'static>(executor: E) -> Self
+    where
+        E: Executor<GaiTask> + Send + Sync,
+    {
+        GaiResolver {
+            executor: GaiExecutor(Arc::new(executor)),
+        }
     }
 }
 
 impl Resolve for GaiResolver {
     type Addrs = GaiAddrs;
     type Future = GaiFuture;
 
-    fn resolve(&self, host: &str) -> Self::Future {
-        GaiFuture
+    fn resolve(&self, name: Name) -> Self::Future {
+        let blocking = GaiBlocking::new(name.host);
+        let rx = oneshot::spawn(blocking, &self.executor);
+        GaiFuture {
+            rx,
+        }
+    }
+}
+
+impl fmt::Debug for GaiResolver {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.pad("GaiResolver")
     }
 }
 
@@ -72,15 +110,30 @@ impl Future for GaiFuture {
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        unimplemented!("GaiFuture::poll");
+        let addrs = try_ready!(self.rx.poll());
+        Ok(Async::Ready(GaiAddrs {
+            inner: addrs,
+        }))
+    }
+}
+
+impl fmt::Debug for GaiFuture {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.pad("GaiFuture")
     }
 }
 
 impl Iterator for GaiAddrs {
     type Item = IpAddr;
 
     fn next(&mut self) -> Option<Self::Item> {
-        unimplemented!("GaiAddrs::next");
+        self.inner.next().map(|sa| sa.ip())
+    }
+}
+
+impl fmt::Debug for GaiAddrs {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.pad("GaiAddrs")
     }
 }
 
@@ -96,12 +149,11 @@ impl Executor<oneshot::Execute<GaiBlocking>> for GaiExecutor {
 
 pub(super) struct GaiBlocking {
     host: String,
-    port: u16
 }
 
 impl GaiBlocking {
-    pub(super) fn new(host: String, port: u16) -> GaiBlocking {
-        GaiBlocking { host: host, port: port }
+    pub(super) fn new(host: String) -> GaiBlocking {
+        GaiBlocking { host }
     }
 }
 
@@ -110,8 +162,8 @@ impl Future for GaiBlocking {
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        debug!("resolving host={:?}, port={:?}", self.host, self.port);
-        (&*self.host, self.port).to_socket_addrs()
+        debug!("resolving host={:?}", self.host);
+        (&*self.host, 0).to_socket_addrs()
             .map(|i| Async::Ready(IpAddrs { iter: i }))
     }
 }
@@ -164,7 +216,7 @@ impl Iterator for IpAddrs {
 }
 
 // Make this Future unnameable outside of this crate.
-mod sealed {
+pub(super) mod sealed {
     use super::*;
     // Blocking task to be executed on a thread pool.
     pub struct GaiTask {