remove unpin from body types (#2152)

actix-http/src/body/body.rs

@@ -12,15 +12,19 @@ use crate::error::Error;
 use super::{BodySize, BodyStream, MessageBody, SizedStream};
 
 /// Represents various types of HTTP message body.
+// #[deprecated(since = "4.0.0", note = "Use body types directly.")]
 pub enum Body {
     /// Empty response. `Content-Length` header is not set.
     None,
+
     /// Zero sized response body. `Content-Length` header is set to `0`.
     Empty,
+
     /// Specific response body.
     Bytes(Bytes),
+
     /// Generic message body.
-    Message(Box<dyn MessageBody + Unpin>),
+    Message(Pin<Box<dyn MessageBody>>),
 }
 
 impl Body {
@@ -30,8 +34,8 @@ impl Body {
     }
 
     /// Create body from generic message body.
-    pub fn from_message<B: MessageBody + Unpin + 'static>(body: B) -> Body {
-        Body::Message(Box::new(body))
+    pub fn from_message<B: MessageBody + 'static>(body: B) -> Body {
+        Body::Message(Box::pin(body))
     }
 }
 
@@ -60,7 +64,7 @@ impl MessageBody for Body {
                     Poll::Ready(Some(Ok(mem::take(bin))))
                 }
             }
-            Body::Message(body) => Pin::new(&mut **body).poll_next(cx),
+            Body::Message(body) => body.as_mut().poll_next(cx),
         }
     }
 }
@@ -134,7 +138,7 @@ impl From<BytesMut> for Body {
 
 impl<S> From<SizedStream<S>> for Body
 where
-    S: Stream<Item = Result<Bytes, Error>> + Unpin + 'static,
+    S: Stream<Item = Result<Bytes, Error>> + 'static,
 {
     fn from(s: SizedStream<S>) -> Body {
         Body::from_message(s)
@@ -143,7 +147,7 @@ where
 
 impl<S, E> From<BodyStream<S>> for Body
 where
-    S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
+    S: Stream<Item = Result<Bytes, E>> + 'static,
     E: Into<Error> + 'static,
 {
     fn from(s: BodyStream<S>) -> Body {

actix-http/src/body/body_stream.rs

@@ -5,21 +5,25 @@ use std::{
 
 use bytes::Bytes;
 use futures_core::{ready, Stream};
+use pin_project_lite::pin_project;
 
 use crate::error::Error;
 
 use super::{BodySize, MessageBody};
 
-/// Streaming response wrapper.
-///
-/// Response does not contain `Content-Length` header and appropriate transfer encoding is used.
-pub struct BodyStream<S: Unpin> {
-    stream: S,
+pin_project! {
+    /// Streaming response wrapper.
+    ///
+    /// Response does not contain `Content-Length` header and appropriate transfer encoding is used.
+    pub struct BodyStream<S> {
+        #[pin]
+        stream: S,
+    }
 }
 
 impl<S, E> BodyStream<S>
 where
-    S: Stream<Item = Result<Bytes, E>> + Unpin,
+    S: Stream<Item = Result<Bytes, E>>,
     E: Into<Error>,
 {
     pub fn new(stream: S) -> Self {
@@ -29,7 +33,7 @@ where
 
 impl<S, E> MessageBody for BodyStream<S>
 where
-    S: Stream<Item = Result<Bytes, E>> + Unpin,
+    S: Stream<Item = Result<Bytes, E>>,
     E: Into<Error>,
 {
     fn size(&self) -> BodySize {
@@ -46,9 +50,9 @@ where
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Bytes, Error>>> {
         loop {
-            let stream = &mut self.as_mut().stream;
+            let stream = self.as_mut().project().stream;
 
-            let chunk = match ready!(Pin::new(stream).poll_next(cx)) {
+            let chunk = match ready!(stream.poll_next(cx)) {
                 Some(Ok(ref bytes)) if bytes.is_empty() => continue,
                 opt => opt.map(|res| res.map_err(Into::into)),
             };

actix-http/src/body/message_body.rs

@@ -52,6 +52,19 @@ impl<T: MessageBody + Unpin> MessageBody for Box<T> {
     }
 }
 
+impl<T: MessageBody> MessageBody for Pin<Box<T>> {
+    fn size(&self) -> BodySize {
+        self.as_ref().size()
+    }
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Bytes, Error>>> {
+        self.as_mut().poll_next(cx)
+    }
+}
+
 impl MessageBody for Bytes {
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)

actix-http/src/body/sized_stream.rs

@@ -5,23 +5,27 @@ use std::{
 
 use bytes::Bytes;
 use futures_core::{ready, Stream};
+use pin_project_lite::pin_project;
 
 use crate::error::Error;
 
 use super::{BodySize, MessageBody};
 
-/// Known sized streaming response wrapper.
-///
-/// This body implementation should be used if total size of stream is known. Data get sent as is
-/// without using transfer encoding.
-pub struct SizedStream<S: Unpin> {
-    size: u64,
-    stream: S,
+pin_project! {
+    /// Known sized streaming response wrapper.
+    ///
+    /// This body implementation should be used if total size of stream is known. Data get sent as is
+    /// without using transfer encoding.
+    pub struct SizedStream<S> {
+        size: u64,
+        #[pin]
+        stream: S,
+    }
 }
 
 impl<S> SizedStream<S>
 where
-    S: Stream<Item = Result<Bytes, Error>> + Unpin,
+    S: Stream<Item = Result<Bytes, Error>>,
 {
     pub fn new(size: u64, stream: S) -> Self {
         SizedStream { size, stream }
@@ -30,7 +34,7 @@ where
 
 impl<S> MessageBody for SizedStream<S>
 where
-    S: Stream<Item = Result<Bytes, Error>> + Unpin,
+    S: Stream<Item = Result<Bytes, Error>>,
 {
     fn size(&self) -> BodySize {
         BodySize::Sized(self.size as u64)
@@ -46,9 +50,9 @@ where
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Bytes, Error>>> {
         loop {
-            let stream = &mut self.as_mut().stream;
+            let stream = self.as_mut().project().stream;
 
-            let chunk = match ready!(Pin::new(stream).poll_next(cx)) {
+            let chunk = match ready!(stream.poll_next(cx)) {
                 Some(Ok(ref bytes)) if bytes.is_empty() => continue,
                 val => val,
             };