feat(ext): introduce extensions, starting with ConnectionInfo

- Adds `client::Builder::set_conn_info` to opt-in to having connection
  info added to `Response`s from clients.
- Adds `ext::ConnectionInfo` that allows querying types (like a
  `Response`) for connection info.

Closes #1402

examples/hello.rs

@@ -10,7 +10,6 @@ static PHRASE: &'static [u8] = b"Hello World!";
 
 fn main() {
     pretty_env_logger::init();
-
     let addr = ([127, 0, 0, 1], 3000).into();
 
     // new_service is run for each connection, creating a 'service'

src/client/connect.rs

@@ -7,6 +7,7 @@
 //! - The [`Connect`](Connect) trait and related types to build custom connectors.
 use std::error::Error as StdError;
 use std::mem;
+use std::net::SocketAddr;
 
 use bytes::{BufMut, BytesMut};
 use futures::Future;
@@ -42,10 +43,11 @@ pub struct Destination {
 ///
 /// This can be used to inform recipients about things like if ALPN
 /// was used, or if connected to an HTTP proxy.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct Connected {
     //alpn: Alpn,
     pub(super) is_proxied: bool,
+    pub(super) remote_addr: Option<SocketAddr>,
 }
 
 /*TODO: when HTTP1 Upgrades to H2 are added, this will be needed
@@ -234,8 +236,8 @@ impl Connected {
     /// Create new `Connected` type with empty metadata.
     pub fn new() -> Connected {
         Connected {
-            //alpn: Alpn::Http1,
             is_proxied: false,
+            remote_addr: None,
         }
     }
 
@@ -251,6 +253,14 @@ impl Connected {
         self
     }
 
+    /// Set the remote address of the connected transport.
+    ///
+    /// Default is `None`.
+    pub fn remote_addr(mut self, addr: SocketAddr) -> Connected {
+        self.remote_addr = Some(addr);
+        self
+    }
+
     /*
     /// Set that the connected transport negotiated HTTP/2 as it's
     /// next protocol.
@@ -640,16 +650,12 @@ mod http {
                         }
                     },
                     State::Resolving(ref mut future, local_addr) => {
-                        match try!(future.poll()) {
-                            Async::NotReady => return Ok(Async::NotReady),
-                            Async::Ready(addrs) => {
-                                state = State::Connecting(ConnectingTcp {
-                                    addrs: addrs,
-                                    local_addr: local_addr,
-                                    current: None,
-                                })
-                            }
-                        };
+                        let addrs = try_ready!(future.poll());
+                        state = State::Connecting(ConnectingTcp {
+                            addrs: addrs,
+                            local_addr: local_addr,
+                            current: None,
+                        });
                     },
                     State::Connecting(ref mut c) => {
                         let sock = try_ready!(c.poll(&self.handle));
@@ -659,8 +665,11 @@ mod http {
                         }
 
                         sock.set_nodelay(self.nodelay)?;
+                        let remote_addr = sock.peer_addr()?;
+                        let connected = Connected::new()
+                            .remote_addr(remote_addr);
 
-                        return Ok(Async::Ready((sock, Connected::new())));
+                        return Ok(Async::Ready((sock, connected)));
                     },
                     State::Error(ref mut e) => return Err(e.take().expect("polled more than once")),
                 }

src/client/mod.rs

@@ -92,7 +92,7 @@ use http::uri::Scheme;
 use body::{Body, Payload};
 use common::Exec;
 use common::lazy as hyper_lazy;
-use self::connect::{Connect, Destination};
+use self::connect::{Connect, Connected, Destination};
 use self::pool::{Pool, Poolable, Reservation};
 
 #[cfg(feature = "runtime")] pub use self::connect::HttpConnector;
@@ -113,6 +113,7 @@ pub struct Client<C, B = Body> {
     h1_title_case_headers: bool,
     pool: Pool<PoolClient<B>>,
     retry_canceled_requests: bool,
+    set_conn_info: bool,
     set_host: bool,
     ver: Ver,
 }
@@ -304,7 +305,7 @@ where C: Connect + Sync + 'static,
                                 })
                                 .map(move |tx| {
                                     pool.pooled(connecting, PoolClient {
-                                        is_proxied: connected.is_proxied,
+                                        conn_info: connected,
                                         tx: match ver {
                                             Ver::Http1 => PoolTx::Http1(tx),
                                             Ver::Http2 => PoolTx::Http2(tx.into_http2()),
@@ -380,14 +381,16 @@ where C: Connect + Sync + 'static,
                 }
             });
 
+        let set_conn_info = self.set_conn_info;
+
         let executor = self.executor.clone();
         let resp = race.and_then(move |mut pooled| {
             let conn_reused = pooled.is_reused();
             if ver == Ver::Http1 {
                 // CONNECT always sends origin-form, so check it first...
                 if req.method() == &Method::CONNECT {
                     authority_form(req.uri_mut());
-                } else if pooled.is_proxied {
+                } else if pooled.conn_info.is_proxied {
                     absolute_form(req.uri_mut());
                 } else {
                     origin_form(req.uri_mut());
@@ -401,6 +404,17 @@ where C: Connect + Sync + 'static,
 
             let fut = pooled.send_request_retryable(req);
 
+            let conn_info = pooled.conn_info.clone();
+            let fut = fut.map(move |mut res| {
+                if set_conn_info {
+                    let info = ::ext::ConnectionInfo {
+                        remote_addr: conn_info.remote_addr,
+                    };
+                    info.set(&mut res);
+                }
+                res
+            });
+
             // As of [email protected], there is a race condition in the mpsc
             // channel, such that sending when the receiver is closing can
             // result in the message being stuck inside the queue. It won't
@@ -498,6 +512,7 @@ impl<C, B> Clone for Client<C, B> {
             h1_title_case_headers: self.h1_title_case_headers,
             pool: self.pool.clone(),
             retry_canceled_requests: self.retry_canceled_requests,
+            set_conn_info: self.set_conn_info,
             set_host: self.set_host,
             ver: self.ver,
         }
@@ -584,7 +599,7 @@ where
 }
 
 struct PoolClient<B> {
-    is_proxied: bool,
+    conn_info: Connected,
     tx: PoolTx<B>,
 }
 
@@ -644,17 +659,17 @@ where
         match self.tx {
             PoolTx::Http1(tx) => {
                 Reservation::Unique(PoolClient {
-                    is_proxied: self.is_proxied,
+                    conn_info: self.conn_info,
                     tx: PoolTx::Http1(tx),
                 })
             },
             PoolTx::Http2(tx) => {
                 let b = PoolClient {
-                    is_proxied: self.is_proxied,
+                    conn_info: self.conn_info.clone(),
                     tx: PoolTx::Http2(tx.clone()),
                 };
                 let a = PoolClient {
-                    is_proxied: self.is_proxied,
+                    conn_info: self.conn_info,
                     tx: PoolTx::Http2(tx),
                 };
                 Reservation::Shared(a, b)
@@ -751,6 +766,7 @@ pub struct Builder {
     //TODO: make use of max_idle config
     max_idle: usize,
     retry_canceled_requests: bool,
+    set_conn_info: bool,
     set_host: bool,
     ver: Ver,
 }
@@ -765,6 +781,7 @@ impl Default for Builder {
             h1_title_case_headers: false,
             max_idle: 5,
             retry_canceled_requests: true,
+            set_conn_info: false,
             set_host: true,
             ver: Ver::Http1,
         }
@@ -851,6 +868,16 @@ impl Builder {
         self
     }
 
+
+    /// Set whether to automatically add [`ConnectionInfo`](::ext::ConnectionInfo)
+    /// to `Response`s.
+    ///
+    /// Default is `false`.
+    pub fn set_conn_info(&mut self, val: bool) -> &mut Self {
+        self.set_conn_info = val;
+        self
+    }
+
     /// Set whether to automatically add the `Host` header to requests.
     ///
     /// If true, and a request does not include a `Host` header, one will be
@@ -902,6 +929,7 @@ impl Builder {
             h1_title_case_headers: self.h1_title_case_headers,
             pool: Pool::new(self.keep_alive, self.keep_alive_timeout, &self.exec),
             retry_canceled_requests: self.retry_canceled_requests,
+            set_conn_info: self.set_conn_info,
             set_host: self.set_host,
             ver: self.ver,
         }

src/ext/conn_info.rs

@@ -0,0 +1,57 @@
+use std::net::SocketAddr;
+
+use super::Ext;
+
+/// dox
+#[derive(Debug)]
+pub struct ConnectionInfo {
+    pub(crate) remote_addr: Option<SocketAddr>,
+}
+
+// The private type that gets put into extensions()
+//
+// The reason for the public and private types is to, for now, prevent
+// a public API contract that crates could depend on. If the public type
+// were inserted into the Extensions directly, a user could depend on
+// `req.extensions().get::<ConnectionInfo>()`, and it's not clear that
+// we want this contract yet.
+#[derive(Copy, Clone, Default)]
+struct ConnInfo {
+    remote_addr: Option<SocketAddr>,
+}
+
+impl ConnectionInfo {
+    /// dox
+    pub fn get<E>(extend: &E) -> ConnectionInfo
+    where
+        E: Ext,
+    {
+        let info = extend
+            .ext()
+            .get::<ConnInfo>()
+            .map(|&info| info)
+            .unwrap_or_default();
+
+        ConnectionInfo {
+            remote_addr: info.remote_addr,
+        }
+    }
+
+    /// dox
+    pub(crate) fn set<E>(self, extend: &mut E)
+    where
+        E: Ext,
+    {
+        let info = ConnInfo {
+            remote_addr: self.remote_addr,
+        };
+
+        extend.ext_mut().insert(info);
+    }
+
+    /// dox
+    pub fn remote_addr(&self) -> Option<SocketAddr> {
+        self.remote_addr
+    }
+}
+

src/ext/mod.rs

@@ -0,0 +1,44 @@
+//! dox
+use http::{Extensions, Request, Response};
+
+use self::sealed::{Ext, Sealed};
+
+mod conn_info;
+
+pub use self::conn_info::ConnectionInfo;
+
+
+mod sealed {
+    use http::Extensions;
+
+    pub trait Sealed {
+        fn ext(&self) -> &Extensions;
+        fn ext_mut(&mut self) -> &mut Extensions;
+    }
+
+    pub trait Ext: Sealed {}
+}
+
+impl<B> Sealed for Request<B> {
+    fn ext(&self) -> &Extensions {
+        self.extensions()
+    }
+
+    fn ext_mut(&mut self) -> &mut Extensions {
+        self.extensions_mut()
+    }
+}
+
+impl<B> Ext for Request<B> {}
+
+impl<B> Sealed for Response<B> {
+    fn ext(&self) -> &Extensions {
+        self.extensions()
+    }
+
+    fn ext_mut(&mut self) -> &mut Extensions {
+        self.extensions_mut()
+    }
+}
+
+impl<B> Ext for Response<B> {}

src/lib.rs

@@ -58,6 +58,7 @@ mod mock;
 pub mod body;
 pub mod client;
 pub mod error;
+pub mod ext;
 mod headers;
 mod proto;
 pub mod server;