add associated error type to MessageBody (#2183)

actix-http/src/body/body.rs

@@ -1,16 +1,17 @@
 use std::{
     borrow::Cow,
+    error::Error as StdError,
     fmt, mem,
     pin::Pin,
     task::{Context, Poll},
 };
 
 use bytes::{Bytes, BytesMut};
-use futures_core::Stream;
+use futures_core::{ready, Stream};
 
 use crate::error::Error;
 
-use super::{BodySize, BodyStream, MessageBody, SizedStream};
+use super::{BodySize, BodyStream, MessageBody, MessageBodyMapErr, SizedStream};
 
 /// Represents various types of HTTP message body.
 // #[deprecated(since = "4.0.0", note = "Use body types directly.")]
@@ -25,7 +26,7 @@ pub enum Body {
     Bytes(Bytes),
 
     /// Generic message body.
-    Message(Pin<Box<dyn MessageBody>>),
+    Message(BoxAnyBody),
 }
 
 impl Body {
@@ -35,12 +36,18 @@ impl Body {
     }
 
     /// Create body from generic message body.
-    pub fn from_message<B: MessageBody + 'static>(body: B) -> Body {
-        Body::Message(Box::pin(body))
+    pub fn from_message<B>(body: B) -> Body
+    where
+        B: MessageBody + 'static,
+        B::Error: Into<Box<dyn StdError + 'static>>,
+    {
+        Self::Message(BoxAnyBody::from_body(body))
     }
 }
 
 impl MessageBody for Body {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         match self {
             Body::None => BodySize::None,
@@ -53,7 +60,7 @@ impl MessageBody for Body {
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         match self.get_mut() {
             Body::None => Poll::Ready(None),
             Body::Empty => Poll::Ready(None),
@@ -65,7 +72,13 @@ impl MessageBody for Body {
                     Poll::Ready(Some(Ok(mem::take(bin))))
                 }
             }
-            Body::Message(body) => body.as_mut().poll_next(cx),
+
+            // TODO: MSRV 1.51: poll_map_err
+            Body::Message(body) => match ready!(body.as_pin_mut().poll_next(cx)) {
+                Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
+                Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+                None => Poll::Ready(None),
+            },
         }
     }
 }
@@ -166,3 +179,51 @@ where
         Body::from_message(s)
     }
 }
+
+/// A boxed message body with boxed errors.
+pub struct BoxAnyBody(Pin<Box<dyn MessageBody<Error = Box<dyn StdError + 'static>>>>);
+
+impl BoxAnyBody {
+    /// Boxes a `MessageBody` and any errors it generates.
+    pub fn from_body<B>(body: B) -> Self
+    where
+        B: MessageBody + 'static,
+        B::Error: Into<Box<dyn StdError + 'static>>,
+    {
+        let body = MessageBodyMapErr::new(body, Into::into);
+        Self(Box::pin(body))
+    }
+
+    /// Returns a mutable pinned reference to the inner message body type.
+    pub fn as_pin_mut(
+        &mut self,
+    ) -> Pin<&mut (dyn MessageBody<Error = Box<dyn StdError + 'static>>)> {
+        self.0.as_mut()
+    }
+}
+
+impl fmt::Debug for BoxAnyBody {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("BoxAnyBody(dyn MessageBody)")
+    }
+}
+
+impl MessageBody for BoxAnyBody {
+    type Error = Error;
+
+    fn size(&self) -> BodySize {
+        self.0.size()
+    }
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
+        // TODO: MSRV 1.51: poll_map_err
+        match ready!(self.0.as_mut().poll_next(cx)) {
+            Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
+            Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+            None => Poll::Ready(None),
+        }
+    }
+}

actix-http/src/body/body_stream.rs

@@ -36,6 +36,8 @@ where
     S: Stream<Item = Result<Bytes, E>>,
     E: Into<Error>,
 {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Stream
     }
@@ -48,7 +50,7 @@ where
     fn poll_next(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         loop {
             let stream = self.as_mut().project().stream;
 

actix-http/src/body/message_body.rs

@@ -1,12 +1,15 @@
 //! [`MessageBody`] trait and foreign implementations.
 
 use std::{
+    convert::Infallible,
     mem,
     pin::Pin,
     task::{Context, Poll},
 };
 
 use bytes::{Bytes, BytesMut};
+use futures_core::ready;
+use pin_project_lite::pin_project;
 
 use crate::error::Error;
 
@@ -14,6 +17,8 @@ use super::BodySize;
 
 /// An interface for response bodies.
 pub trait MessageBody {
+    type Error;
+
     /// Body size hint.
     fn size(&self) -> BodySize;
 
@@ -21,14 +26,12 @@ pub trait MessageBody {
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>>;
-
-    downcast_get_type_id!();
+    ) -> Poll<Option<Result<Bytes, Self::Error>>>;
 }
 
-downcast!(MessageBody);
-
 impl MessageBody for () {
+    type Error = Infallible;
+
     fn size(&self) -> BodySize {
         BodySize::Empty
     }
@@ -36,12 +39,18 @@ impl MessageBody for () {
     fn poll_next(
         self: Pin<&mut Self>,
         _: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         Poll::Ready(None)
     }
 }
 
-impl<T: MessageBody + Unpin> MessageBody for Box<T> {
+impl<B> MessageBody for Box<B>
+where
+    B: MessageBody + Unpin,
+    B::Error: Into<Error>,
+{
+    type Error = B::Error;
+
     fn size(&self) -> BodySize {
         self.as_ref().size()
     }
@@ -49,12 +58,18 @@ impl<T: MessageBody + Unpin> MessageBody for Box<T> {
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         Pin::new(self.get_mut().as_mut()).poll_next(cx)
     }
 }
 
-impl<T: MessageBody> MessageBody for Pin<Box<T>> {
+impl<B> MessageBody for Pin<Box<B>>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
+    type Error = B::Error;
+
     fn size(&self) -> BodySize {
         self.as_ref().size()
     }
@@ -62,12 +77,14 @@ impl<T: MessageBody> MessageBody for Pin<Box<T>> {
     fn poll_next(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         self.as_mut().poll_next(cx)
     }
 }
 
 impl MessageBody for Bytes {
+    type Error = Infallible;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -75,7 +92,7 @@ impl MessageBody for Bytes {
     fn poll_next(
         self: Pin<&mut Self>,
         _: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         if self.is_empty() {
             Poll::Ready(None)
         } else {
@@ -85,6 +102,8 @@ impl MessageBody for Bytes {
 }
 
 impl MessageBody for BytesMut {
+    type Error = Infallible;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -92,7 +111,7 @@ impl MessageBody for BytesMut {
     fn poll_next(
         self: Pin<&mut Self>,
         _: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         if self.is_empty() {
             Poll::Ready(None)
         } else {
@@ -102,6 +121,8 @@ impl MessageBody for BytesMut {
 }
 
 impl MessageBody for &'static str {
+    type Error = Infallible;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -109,7 +130,7 @@ impl MessageBody for &'static str {
     fn poll_next(
         self: Pin<&mut Self>,
         _: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         if self.is_empty() {
             Poll::Ready(None)
         } else {
@@ -121,6 +142,8 @@ impl MessageBody for &'static str {
 }
 
 impl MessageBody for Vec<u8> {
+    type Error = Infallible;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -128,7 +151,7 @@ impl MessageBody for Vec<u8> {
     fn poll_next(
         self: Pin<&mut Self>,
         _: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         if self.is_empty() {
             Poll::Ready(None)
         } else {
@@ -138,6 +161,8 @@ impl MessageBody for Vec<u8> {
 }
 
 impl MessageBody for String {
+    type Error = Infallible;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -145,7 +170,7 @@ impl MessageBody for String {
     fn poll_next(
         self: Pin<&mut Self>,
         _: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         if self.is_empty() {
             Poll::Ready(None)
         } else {
@@ -155,3 +180,53 @@ impl MessageBody for String {
         }
     }
 }
+
+pin_project! {
+    pub(crate) struct MessageBodyMapErr<B, F> {
+        #[pin]
+        body: B,
+        mapper: Option<F>,
+    }
+}
+
+impl<B, F, E> MessageBodyMapErr<B, F>
+where
+    B: MessageBody,
+    F: FnOnce(B::Error) -> E,
+{
+    pub(crate) fn new(body: B, mapper: F) -> Self {
+        Self {
+            body,
+            mapper: Some(mapper),
+        }
+    }
+}
+
+impl<B, F, E> MessageBody for MessageBodyMapErr<B, F>
+where
+    B: MessageBody,
+    F: FnOnce(B::Error) -> E,
+{
+    type Error = E;
+
+    fn size(&self) -> BodySize {
+        self.body.size()
+    }
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
+        let this = self.as_mut().project();
+
+        match ready!(this.body.poll_next(cx)) {
+            Some(Err(err)) => {
+                let f = self.as_mut().project().mapper.take().unwrap();
+                let mapped_err = (f)(err);
+                Poll::Ready(Some(Err(mapped_err)))
+            }
+            Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+            None => Poll::Ready(None),
+        }
+    }
+}

actix-http/src/body/mod.rs

@@ -15,9 +15,10 @@ mod response_body;
 mod size;
 mod sized_stream;
 
-pub use self::body::Body;
+pub use self::body::{Body, BoxAnyBody};
 pub use self::body_stream::BodyStream;
 pub use self::message_body::MessageBody;
+pub(crate) use self::message_body::MessageBodyMapErr;
 pub use self::response_body::ResponseBody;
 pub use self::size::BodySize;
 pub use self::sized_stream::SizedStream;
@@ -41,7 +42,7 @@ pub use self::sized_stream::SizedStream;
 /// assert_eq!(bytes, b"123"[..]);
 /// # }
 /// ```
-pub async fn to_bytes(body: impl MessageBody) -> Result<Bytes, crate::Error> {
+pub async fn to_bytes<B: MessageBody>(body: B) -> Result<Bytes, B::Error> {
     let cap = match body.size() {
         BodySize::None | BodySize::Empty | BodySize::Sized(0) => return Ok(Bytes::new()),
         BodySize::Sized(size) => size as usize,
@@ -237,10 +238,13 @@ mod tests {
         );
     }
 
+    // down-casting used to be done with a method on MessageBody trait
+    // test is kept to demonstrate equivalence of Any trait
     #[actix_rt::test]
     async fn test_body_casting() {
         let mut body = String::from("hello cast");
-        let resp_body: &mut dyn MessageBody = &mut body;
+        // let mut resp_body: &mut dyn MessageBody<Error = Error> = &mut body;
+        let resp_body: &mut dyn std::any::Any = &mut body;
         let body = resp_body.downcast_ref::<String>().unwrap();
         assert_eq!(body, "hello cast");
         let body = &mut resp_body.downcast_mut::<String>().unwrap();

actix-http/src/body/response_body.rs

@@ -5,7 +5,7 @@ use std::{
 };
 
 use bytes::Bytes;
-use futures_core::Stream;
+use futures_core::{ready, Stream};
 use pin_project::pin_project;
 
 use crate::error::Error;
@@ -43,7 +43,13 @@ impl<B: MessageBody> ResponseBody<B> {
     }
 }
 
-impl<B: MessageBody> MessageBody for ResponseBody<B> {
+impl<B> MessageBody for ResponseBody<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         match self {
             ResponseBody::Body(ref body) => body.size(),
@@ -54,12 +60,16 @@ impl<B: MessageBody> MessageBody for ResponseBody<B> {
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         Stream::poll_next(self, cx)
     }
 }
 
-impl<B: MessageBody> Stream for ResponseBody<B> {
+impl<B> Stream for ResponseBody<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
     type Item = Result<Bytes, Error>;
 
     fn poll_next(
@@ -67,7 +77,12 @@ impl<B: MessageBody> Stream for ResponseBody<B> {
         cx: &mut Context<'_>,
     ) -> Poll<Option<Self::Item>> {
         match self.project() {
-            ResponseBodyProj::Body(body) => body.poll_next(cx),
+            // TODO: MSRV 1.51: poll_map_err
+            ResponseBodyProj::Body(body) => match ready!(body.poll_next(cx)) {
+                Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
+                Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+                None => Poll::Ready(None),
+            },
             ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx),
         }
     }

actix-http/src/body/sized_stream.rs

@@ -36,6 +36,8 @@ impl<S> MessageBody for SizedStream<S>
 where
     S: Stream<Item = Result<Bytes, Error>>,
 {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.size as u64)
     }
@@ -48,7 +50,7 @@ where
     fn poll_next(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         loop {
             let stream = self.as_mut().project().stream;