merge with msg-body

actix-http/src/body.rs

@@ -379,7 +379,7 @@ where
 {
     pub fn new(stream: S) -> Self {
         BodyStream {
-            stream,
+            stream: Box::pin(stream),
             _t: PhantomData,
         }
     }
@@ -416,8 +416,7 @@ where
 #[pin_project]
 pub struct SizedStream<S: Unpin> {
     size: u64,
-    #[pin]
+    stream: Pin<Box<S>>,
-    stream: S,
 }
 
 impl<S> SizedStream<S>
@@ -425,7 +424,7 @@ where
     S: Stream<Item = Result<Bytes, Error>> + Unpin,
 {
     pub fn new(size: u64, stream: S) -> Self {
-        SizedStream { size, stream }
+        SizedStream { size, stream: Box::pin(stream) }
     }
 }
 

tests/test_weird_poll.rs

@@ -0,0 +1,26 @@
+// Regression test for #/1321
+
+use futures::task::{noop_waker, Context};
+use futures::stream::once;
+use actix_http::body::{MessageBody, BodyStream};
+use bytes::Bytes;
+
+#[test]
+fn weird_poll() {
+    let (sender, receiver) = futures::channel::oneshot::channel();
+    let mut body_stream = Ok(BodyStream::new(once(async {
+        let x = Box::new(0);
+        let y = &x;
+        receiver.await.unwrap();
+        let _z = **y;
+        Ok::<_, ()>(Bytes::new())
+    })));
+
+    let waker = noop_waker();
+    let mut context = Context::from_waker(&waker);
+
+    let _ = body_stream.as_mut().unwrap().poll_next(&mut context);
+    sender.send(()).unwrap();
+    let _ = std::mem::replace(&mut body_stream, Err([0; 32])).unwrap().poll_next(&mut context);
+}
+