add helpers to box up futures and streams

Summary:
This was deprecated upstream, but is still useful to us.

Also add `nonsend` versions since those are still required in a couple of
places.

Reviewed By: jsgf

Differential Revision: D5817016

fbshipit-source-id: 3753fadcae666099a96b3b849e438c5c16c7232f

futures-ext/src/lib.rs

@@ -15,14 +15,14 @@ extern crate bytes;
 #[macro_use]
 extern crate futures;
 #[cfg(test)]
-extern crate tokio_core;
-extern crate tokio_io;
-#[cfg(test)]
 #[macro_use]
 extern crate quickcheck;
+#[cfg(test)]
+extern crate tokio_core;
+extern crate tokio_io;
 
-use futures::{Async, Future, IntoFuture, Poll, Sink, Stream};
 use bytes::Bytes;
+use futures::{Async, Future, IntoFuture, Poll, Sink, Stream};
 use tokio_io::AsyncRead;
 use tokio_io::codec::{Decoder, Encoder};
 
@@ -35,7 +35,7 @@ pub mod decode;
 pub mod encode;
 
 pub use frame::{FramedStream, ReadLeadingBuffer};
-pub use futures_ordered::{FuturesOrdered, futures_ordered};
+pub use futures_ordered::{futures_ordered, FuturesOrdered};
 pub use stream_wrappers::{BoxStreamWrapper, StreamWrapper, TakeWhile};
 
 /// Map `Item` and `Error` to `()`
@@ -66,6 +66,12 @@ where
     }
 }
 
+// Replacements for BoxFuture and BoxStream, deprecated in upstream futures-rs.
+pub type BoxFuture<T, E> = Box<Future<Item = T, Error = E> + Send>;
+pub type BoxFutureNonSend<T, E> = Box<Future<Item = T, Error = E>>;
+pub type BoxStream<T, E> = Box<Stream<Item = T, Error = E> + Send>;
+pub type BoxStreamNonSend<T, E> = Box<Stream<Item = T, Error = E>>;
+
 pub trait FutureExt: Future + Sized {
     /// Map a `Future` to have `Item=()` and `Error=()`. This is
     /// useful when a future is being used to drive a computation
@@ -74,6 +80,25 @@ pub trait FutureExt: Future + Sized {
     fn discard(self) -> Discard<Self> {
         Discard(self)
     }
+
+    /// Create a `Send`able boxed version of this `Future`.
+    #[inline]
+    fn boxify(self) -> BoxFuture<Self::Item, Self::Error>
+    where
+        Self: 'static + Send,
+    {
+        // TODO: (sid0) T21801845 rename to 'boxed' once gone from upstream.
+        Box::new(self)
+    }
+
+    /// Create a non-`Send`able boxed version of this `Future`.
+    #[inline]
+    fn boxify_nonsend(self) -> BoxFutureNonSend<Self::Item, Self::Error>
+    where
+        Self: 'static,
+    {
+        Box::new(self)
+    }
 }
 
 impl<T> FutureExt for T
@@ -131,6 +156,25 @@ pub trait StreamExt: Stream {
     {
         Enumerate::new(self)
     }
+
+    /// Create a `Send`able boxed version of this `Stream`.
+    #[inline]
+    fn boxify(self) -> BoxStream<Self::Item, Self::Error>
+    where
+        Self: 'static + Send + Sized,
+    {
+        // TODO: (sid0) T21801845 rename to 'boxed' once gone from upstream.
+        Box::new(self)
+    }
+
+    /// Create a non-`Send`able boxed version of this `Stream`.
+    #[inline]
+    fn boxify_nonsend(self) -> BoxStreamNonSend<Self::Item, Self::Error>
+    where
+        Self: 'static + Sized,
+    {
+        Box::new(self)
+    }
 }
 
 impl<T> StreamExt for T