apply robjtede patch + some tests

actix-http/src/body/boxed.rs

@@ -1,6 +1,6 @@
 use std::{
     error::Error as StdError,
-    fmt, mem,
+    fmt,
     pin::Pin,
     task::{Context, Poll},
 };
@@ -8,7 +8,6 @@ use std::{
 use bytes::Bytes;
 
 use super::{BodySize, MessageBody, MessageBodyMapErr};
-use crate::Error;
 
 /// A boxed message body with boxed errors.
 pub struct BoxBody(BoxBodyInner);
@@ -43,7 +42,6 @@ impl BoxBody {
 
     /// Returns a mutable pinned reference to the inner message body type.
     #[inline]
-    //pub fn as_pin_mut(&mut self) -> Pin<&mut (dyn MessageBody<Error = Box<dyn StdError>>)> {
     pub fn as_pin_mut(&mut self) -> Pin<&mut Self> {
         Pin::new(self)
     }
@@ -57,7 +55,7 @@ impl fmt::Debug for BoxBody {
 }
 
 impl MessageBody for BoxBody {
-    type Error = Error;
+    type Error = Box<dyn StdError>;
 
     #[inline]
     fn size(&self) -> BodySize {
@@ -75,8 +73,8 @@ impl MessageBody for BoxBody {
     ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         match &mut self.0 {
             BoxBodyInner::None => Poll::Ready(None),
-            BoxBodyInner::Bytes(bytes) => Poll::Ready(Some(Ok(mem::take(bytes)))),
+            BoxBodyInner::Bytes(bytes) => Pin::new(bytes).poll_next(cx).map_err(Into::into),
-            BoxBodyInner::Stream(stream) => Pin::new(stream).poll_next(cx).map_err(|err| Error::new_body().with_cause(err)),
+            BoxBodyInner::Stream(stream) => Pin::new(stream).poll_next(cx),
         }
     }
 

actix-http/src/body/message_body.rs

@@ -369,6 +369,7 @@ mod tests {
     use bytes::{Bytes, BytesMut};
 
     use super::*;
+    use crate::body::{BoxBody, EitherBody};
 
     macro_rules! assert_poll_next {
         ($pin:expr, $exp:expr) => {
@@ -470,19 +471,30 @@ mod tests {
         assert_poll_next!(pl, Bytes::from("test"));
     }
 
-    #[test]
+    #[actix_rt::test]
-    fn complete_body_combinators() {
+    async fn complete_body_combinators() {
-        use crate::body::{BoxBody, EitherBody};
+        let body = Bytes::from_static(b"test");
-
+        let body = BoxBody::new(body);
+        let body = EitherBody::<_, ()>::left(body);
+        let body = EitherBody::<(), _>::right(body);
+        // Do not support try_into_bytes:
+        // let body = Box::new(body);
+        // let body = Box::pin(body);
+
+        assert_eq!(body.try_into_bytes().unwrap(), Bytes::from("test"));
+    }
+
+    #[actix_rt::test]
+    async fn complete_body_combinators_poll() {
         let body = Bytes::from_static(b"test");
         let body = BoxBody::new(body);
         let body = EitherBody::<_, ()>::left(body);
         let body = EitherBody::<(), _>::right(body);
-        let body = Box::new(body);
-        let body = Box::pin(body);
         let mut body = body;
 
-        assert_eq!(body.try_into_bytes().unwrap(), b"test".as_ref());
+        assert_eq!(body.size(), BodySize::Sized(4));
+        assert_poll_next!(Pin::new(&mut body), Bytes::from("test"));
+        assert_poll_next_none!(Pin::new(&mut body));
     }
 
     // down-casting used to be done with a method on MessageBody trait

actix-http/src/encoding/encoder.rs

@@ -53,7 +53,7 @@ impl<B: MessageBody> Encoder<B> {
         }
     }
 
-    pub fn response(encoding: ContentEncoding, head: &mut ResponseHead, mut body: B) -> Self {
+    pub fn response(encoding: ContentEncoding, head: &mut ResponseHead, body: B) -> Self {
         let can_encode = !(head.headers().contains_key(&CONTENT_ENCODING)
             || head.status == StatusCode::SWITCHING_PROTOCOLS
             || head.status == StatusCode::NO_CONTENT
@@ -65,11 +65,9 @@ impl<B: MessageBody> Encoder<B> {
             return Self::none();
         }
 
-        let body = if body.is_complete_body() {
+        let body = match body.try_into_bytes() {
-            let body = body.take_complete_body();
+            Ok(body) => EncoderBody::Full { body },
-            EncoderBody::Full { body }
+            Err(body) => EncoderBody::Stream { body },
-        } else {
-            EncoderBody::Stream { body }
         };
 
         if can_encode {
@@ -133,21 +131,13 @@ where
         }
     }
 
-    fn is_complete_body(&self) -> bool {
+    fn try_into_bytes(self) -> Result<Bytes, Self>
+    where
+        Self: Sized,
+    {
         match self {
-            EncoderBody::None => true,
+            EncoderBody::Full { body } => Ok(body),
-            EncoderBody::Full { .. } => true,
+            _ => Err(self),
-            EncoderBody::Stream { .. } => false,
-        }
-    }
-
-    fn take_complete_body(&mut self) -> Bytes {
-        match self {
-            EncoderBody::None => Bytes::new(),
-            EncoderBody::Full { body } => body.take_complete_body(),
-            EncoderBody::Stream { .. } => {
-                panic!("EncoderBody::Stream variant cannot be taken")
-            }
         }
     }
 }
@@ -234,19 +224,20 @@ where
         }
     }
 
-    fn is_complete_body(&self) -> bool {
+    fn try_into_bytes(mut self) -> Result<Bytes, Self>
+    where
+        Self: Sized,
+    {
         if self.encoder.is_some() {
-            false
+            Err(self)
         } else {
-            self.body.is_complete_body()
+            match self.body.try_into_bytes() {
-        }
+                Ok(body) => Ok(body),
-    }
+                Err(body) => {
-
+                    self.body = body;
-    fn take_complete_body(&mut self) -> Bytes {
+                    Err(self)
-        if self.encoder.is_some() {
+                }
-            panic!("compressed body stream cannot be taken")
+            }
-        } else {
-            self.body.take_complete_body()
         }
     }
 }

actix-http/src/h1/dispatcher.rs

@@ -22,7 +22,7 @@ use crate::{
     config::ServiceConfig,
     error::{DispatchError, ParseError, PayloadError},
     service::HttpFlow,
-    Extensions, OnConnectData, Request, Response, StatusCode,
+    Error, Extensions, OnConnectData, Request, Response, StatusCode,
 };
 
 use super::{
@@ -458,7 +458,9 @@ where
                             }
 
                             Poll::Ready(Some(Err(err))) => {
-                                return Err(DispatchError::Service(err.into()))
+                                return Err(DispatchError::Service(
+                                    Error::new_body().with_cause(err).into(),
+                                ))
                             }
 
                             Poll::Pending => return Ok(PollResponse::DoNothing),

src/dev.rs

@@ -139,4 +139,12 @@ impl crate::body::MessageBody for AnyBody {
             AnyBody::Boxed { body } => body.as_pin_mut().poll_next(cx),
         }
     }
+
+    fn try_into_bytes(self) -> Result<crate::web::Bytes, Self> {
+        match self {
+            AnyBody::None => Ok(crate::web::Bytes::new()),
+            AnyBody::Full { body } => Ok(body),
+            _ => Err(self),
+        }
+    }
 }