proxy: Use `impl Trait` to unbox some futures (#969)

Now that `impl Trait` is stable, we don't need to box as many futures. We still
need to box before spawning them on an executor, but the component futures no 
longer require their own boxes. 

Signed-off-by: Eliza Weisman <eliza@buoyant.io

proxy/src/connection.rs

@@ -101,13 +101,13 @@ impl BoundPort {
     // In the future it will also ensure that the connection is upgraded with
     // TLS when needed.
     pub fn listen_and_fold<T, F, Fut>(self, initial: T, f: F)
-        -> Box<Future<Item = (), Error = io::Error> + Send + 'static>
+        -> impl Future<Item = (), Error = io::Error> + Send + 'static
     where
         F: Fn(T, (Connection, SocketAddr)) -> Fut + Send + 'static,
         T: Send + 'static,
         Fut: IntoFuture<Item = T, Error = std::io::Error> + Send + 'static,
         <Fut as IntoFuture>::Future: Send, {
-        let fut = future::lazy(move || {
+        future::lazy(move || {
             // Create the TCP listener lazily, so that it's not bound to a
             // reactor until the future is run. This will avoid
             // `Handle::current()` creating a mew thread for the global
@@ -129,9 +129,7 @@ impl BoundPort {
                     f(b, (Connection::plain(socket), remote_addr))
                 })
         )
-        .map(|_| ());
+        .map(|_| ())
-
-        Box::new(fut)
     }
 }
 

proxy/src/lib.rs

@@ -345,7 +345,7 @@ fn serve<R, B, E, F, G>(
     sensors: telemetry::Sensors,
     get_orig_dst: G,
     drain_rx: drain::Watch,
-) -> Box<Future<Item = (), Error = io::Error> + Send + 'static>
+) -> impl Future<Item = (), Error = io::Error> + Send + 'static
 where
     B: tower_h2::Body + Default + Send + 'static,
     <B::Data as ::bytes::IntoBuf>::Buf: Send,
@@ -426,9 +426,9 @@ where
 
     // As soon as we get a shutdown signal, the listener
     // is canceled immediately.
-    Box::new(drain_rx.watch(accept_until, |accept| {
+    drain_rx.watch(accept_until, |accept| {
         accept.canceled = true;
-    }))
+    })
 }
 
 /// Can cancel a future by setting a flag.
@@ -459,7 +459,7 @@ where
 fn serve_control<N, B>(
     bound_port: BoundPort,
     new_service: N,
-) -> Box<Future<Item = (), Error = io::Error> + 'static>
+) -> impl Future<Item = (), Error = io::Error> + 'static
 where
     B: tower_h2::Body + Send + 'static,
     <B::Data as bytes::IntoBuf>::Buf: Send,

proxy/src/telemetry/control.rs

@@ -111,7 +111,7 @@ impl Control {
     }
 
     pub fn serve_metrics(&self, bound_port: connection::BoundPort)
-        -> Box<Future<Item = (), Error = io::Error>>
+        -> impl Future<Item = (), Error = io::Error>
     {
         use hyper;
         let service = self.metrics_service.clone();

proxy/src/transparency/server.rs

@@ -3,7 +3,7 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use futures::Future;
+use futures::{future::Either, Future};
 use http;
 use hyper;
 use indexmap::IndexSet;
@@ -139,7 +139,7 @@ where
             );
 
             DefaultExecutor::current()
-                .spawn(fut)
+                .spawn(Box::new(fut))
                 .expect("spawn TCP server task");
             return;
         }
@@ -152,13 +152,13 @@ where
         let drain_signal = self.drain_signal.clone();
         let fut = io.peek()
             .map_err(|e| debug!("peek error: {}", e))
-            .and_then(move |io| -> Box<Future<Item=(), Error=()> + Send> {
+            .and_then(move |io| {
                 if let Some(proto) = Protocol::detect(io.peeked()) {
-                    match proto {
+                    Either::A(match proto {
                         Protocol::Http1 => {
                             trace!("transparency detected HTTP/1");
 
-                            Box::new(new_service.new_service()
+                            let fut = new_service.new_service()
                                 .map_err(|_| ())
                                 .and_then(move |s| {
                                     let svc = HyperServerSvc::new(s, srv_ctx);
@@ -168,7 +168,8 @@ where
                                         })
                                         .map(|_| ())
                                         .map_err(|e| trace!("http1 server error: {:?}", e))
-                                }))
+                                });
+                            Either::A(fut)
                         },
                         Protocol::Http2 => {
                             trace!("transparency detected HTTP/2");
@@ -183,17 +184,17 @@ where
                                 })
                                 .map_err(|e| trace!("h2 server error: {:?}", e));
 
-                            Box::new(fut)
+                            Either::B(fut)
                         }
-                    }
+                    })
                 } else {
                     trace!("transparency did not detect protocol, treating as TCP");
-                    tcp_serve(
+                    Either::B(tcp_serve(
                         &tcp,
                         io,
                         srv_ctx,
                         drain_signal,
-                    )
+                    ))
                 }
             });
 
@@ -208,11 +209,11 @@ fn tcp_serve<T: AsyncRead + AsyncWrite + Send + 'static>(
     io: T,
     srv_ctx: Arc<ServerCtx>,
     drain_signal: drain::Watch,
-) -> Box<Future<Item=(), Error=()> + Send + 'static> {
+) -> impl Future<Item=(), Error=()> + Send + 'static {
     let fut = tcp.serve(io, srv_ctx);
 
     // There's nothing to do when drain is signaled, we just have to hope
     // the sockets finish soon. However, the drain signal still needs to
     // 'watch' the TCP future so that the process doesn't close early.
-    Box::new(drain_signal.watch(fut, |_| ()))
+    drain_signal.watch(fut, |_| ())
 }

proxy/src/transparency/tcp.rs

@@ -30,7 +30,7 @@ impl Proxy {
 
     /// Serve a TCP connection, trying to forward it to its destination.
     pub fn serve<T>(&self, tcp_in: T, srv_ctx: Arc<ServerCtx>)
-        -> Box<Future<Item=(), Error=()> + Send>
+        -> impl Future<Item=(), Error=()> + Send
     where
         T: AsyncRead + AsyncWrite + Send + 'static,
     {
@@ -51,7 +51,7 @@ impl Proxy {
                 "tcp accepted, no SO_ORIGINAL_DST to forward: remote={}",
                 srv_ctx.remote,
             );
-            return Box::new(future::ok(()));
+            return future::Either::B(future::ok(()));
         };
 
         let client_ctx = ClientCtx::new(
@@ -65,13 +65,12 @@ impl Proxy {
         );
         let connect = self.sensors.connect(c, &client_ctx);
 
-        let fut = connect.connect()
+        future::Either::A(connect.connect()
             .map_err(move |e| error!("tcp connect error to {}: {:?}", orig_dst, e))
             .and_then(move |tcp_out| {
                 Duplex::new(tcp_in, tcp_out)
                     .map_err(|e| error!("tcp duplex error: {}", e))
-            });
+            }))
-        Box::new(fut)
     }
 }