Future::boxed and Stream::box should prevent double boxing

Fixes #511

src/future/mod.rs

@@ -308,7 +308,26 @@ pub trait Future {
     fn boxed(self) -> BoxFuture<Self::Item, Self::Error>
         where Self: Sized + Send + 'static
     {
-        ::std::boxed::Box::new(self)
+        use std::any::TypeId;
+        use std::mem;
+        use std::ptr;
+        use std::boxed;
+
+        // Prevent double boxing
+        if TypeId::of::<Self>() == TypeId::of::<BoxFuture<Self::Item, Self::Error>>() {
+            assert!(mem::size_of::<Self>() == mem::size_of::<BoxFuture<Self::Item, Self::Error>>());
+            unsafe {
+                let mut r: BoxFuture<Self::Item, Self::Error> = mem::uninitialized();
+                ptr::copy_nonoverlapping(
+                    &self as *const Self,
+                    &mut r as *mut BoxFuture<Self::Item, Self::Error> as *mut u8 as *mut Self,
+                    1);
+                mem::forget(self);
+                r
+            }
+        } else {
+            boxed::Box::new(self)
+        }
     }
 
     /// Map this future's result to a different type, returning a new future of

src/stream/mod.rs

@@ -250,7 +250,26 @@ pub trait Stream {
     fn boxed(self) -> BoxStream<Self::Item, Self::Error>
         where Self: Sized + Send + 'static,
     {
-        ::std::boxed::Box::new(self)
+        use std::any::TypeId;
+        use std::mem;
+        use std::ptr;
+        use std::boxed;
+
+        // Prevent double boxing
+        if TypeId::of::<Self>() == TypeId::of::<BoxStream<Self::Item, Self::Error>>() {
+            assert!(mem::size_of::<Self>() == mem::size_of::<BoxStream<Self::Item, Self::Error>>());
+            unsafe {
+                let mut r: BoxStream<Self::Item, Self::Error> = mem::uninitialized();
+                ptr::copy_nonoverlapping(
+                    &self as *const Self,
+                    &mut r as *mut BoxStream<Self::Item, Self::Error> as *mut u8 as *mut Self,
+                    1);
+                mem::forget(self);
+                r
+            }
+        } else {
+            boxed::Box::new(self)
+        }
     }
 
     /// Converts this stream into a `Future`.

tests/boxed.rs

@@ -0,0 +1,60 @@
+extern crate futures;
+
+use futures::Async;
+use futures::Poll;
+use futures::future::Future;
+use futures::stream::Stream;
+
+
+#[test]
+fn future_boxed_prevents_double_boxing() {
+    struct MyFuture {
+        r: &'static str,
+    }
+
+    impl Future for MyFuture {
+        type Item = &'static str;
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            Ok(Async::Ready(self.r))
+        }
+    }
+
+    let f = MyFuture { r: "I'm ready" };
+    let f = f.boxed();
+    let ptr = f.as_ref() as *const Future<Item=_, Error=_>;
+    let f = f.boxed();
+    let f = f.boxed();
+    let mut f = f.boxed();
+    assert_eq!(f.as_ref() as *const Future<Item=_, Error=_>, ptr);
+    assert_eq!(Ok(Async::Ready("I'm ready")), f.poll());
+}
+
+#[test]
+fn stream_boxed_prevents_double_boxing() {
+    struct MyStream {
+        i: u32,
+    }
+
+    impl Stream for MyStream {
+        type Item = u32;
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+            self.i += 1;
+            Ok(Async::Ready(Some(self.i)))
+        }
+    }
+
+    let s = MyStream { i: 0 };
+    let s = s.boxed();
+    let ptr = s.as_ref() as *const Stream<Item=_, Error=_>;
+    let s = s.boxed();
+    let s = s.boxed();
+    let mut s = s.boxed();
+    assert_eq!(s.as_ref() as *const Stream<Item=_, Error=_>, ptr);
+    assert_eq!(Ok(Async::Ready(Some(1))), s.poll());
+    assert_eq!(Ok(Async::Ready(Some(2))), s.poll());
+    assert_eq!(Ok(Async::Ready(Some(3))), s.poll());
+}