add associated error type to MessageBody

actix-http/src/body/body.rs

@@ -1,5 +1,6 @@
 use std::{
     borrow::Cow,
+    error::Error as StdError,
     fmt, mem,
     pin::Pin,
     task::{Context, Poll},
@@ -10,7 +11,7 @@ use futures_core::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,
@@ -65,7 +72,7 @@ impl MessageBody for Body {
                     Poll::Ready(Some(Ok(mem::take(bin))))
                 }
             }
-            Body::Message(body) => body.as_mut().poll_next(cx),
+            Body::Message(body) => body.as_mut().poll_next(cx).map_err(Into::into),
         }
     }
 }
@@ -166,3 +173,45 @@ 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 {
+    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_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>>> {
+        self.0.as_mut().poll_next(cx).map_err(Into::into)
+    }
+}

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
     }

actix-http/src/body/message_body.rs

@@ -7,6 +7,7 @@ use std::{
 };
 
 use bytes::{Bytes, BytesMut};
+use pin_project_lite::pin_project;
 
 use crate::error::Error;
 
@@ -14,6 +15,8 @@ use super::BodySize;
 
 /// An interface for response bodies.
 pub trait MessageBody {
+    type Error;
+
     /// Body size hint.
     fn size(&self) -> BodySize;
 
@@ -21,14 +24,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 = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Empty
     }
@@ -36,12 +37,17 @@ 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<T> MessageBody for Box<T>
+where
+    T: MessageBody<Error = Error> + Unpin,
+{
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         self.as_ref().size()
     }
@@ -49,12 +55,14 @@ 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>> {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         self.as_ref().size()
     }
@@ -62,12 +70,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 = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -75,7 +85,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 +95,8 @@ impl MessageBody for Bytes {
 }
 
 impl MessageBody for BytesMut {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -92,7 +104,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 +114,8 @@ impl MessageBody for BytesMut {
 }
 
 impl MessageBody for &'static str {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -109,7 +123,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 +135,8 @@ impl MessageBody for &'static str {
 }
 
 impl MessageBody for Vec<u8> {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -128,7 +144,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 +154,8 @@ impl MessageBody for Vec<u8> {
 }
 
 impl MessageBody for String {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         BodySize::Sized(self.len() as u64)
     }
@@ -145,7 +163,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 +173,48 @@ 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();
+
+        this.body.poll_next(cx).map_err(|err| {
+            let f = self.as_mut().project().mapper.take().unwrap();
+            (f)(err)
+        })
+    }
+}

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,
@@ -75,6 +76,7 @@ mod tests {
     use bytes::{Bytes, BytesMut};
 
     use super::*;
+    // use crate::Error;
 
     impl Body {
         pub(crate) fn get_ref(&self) -> &[u8] {
@@ -240,7 +242,8 @@ mod tests {
     #[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

@@ -43,7 +43,12 @@ impl<B: MessageBody> ResponseBody<B> {
     }
 }
 
-impl<B: MessageBody> MessageBody for ResponseBody<B> {
+impl<B> MessageBody for ResponseBody<B>
+where
+    B: MessageBody<Error = Error>,
+{
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         match self {
             ResponseBody::Body(ref body) => body.size(),
@@ -54,12 +59,15 @@ 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<Error = Error>,
+{
     type Item = Result<Bytes, Error>;
 
     fn poll_next(

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)
     }

actix-http/src/builder.rs

@@ -202,11 +202,13 @@ where
     /// Finish service configuration and create a HTTP service for HTTP/2 protocol.
     pub fn h2<F, B>(self, service: F) -> H2Service<T, S, B>
     where
-        B: MessageBody + 'static,
         F: IntoServiceFactory<S, Request>,
         S::Error: Into<Error> + 'static,
         S::InitError: fmt::Debug,
         S::Response: Into<Response<B>> + 'static,
+
+        B: MessageBody + 'static,
+        B: MessageBody<Error = Error>,
     {
         let cfg = ServiceConfig::new(
             self.keep_alive,

actix-http/src/client/connection.rs

@@ -12,10 +12,10 @@ use bytes::Bytes;
 use futures_core::future::LocalBoxFuture;
 use h2::client::SendRequest;
 
-use crate::body::MessageBody;
 use crate::h1::ClientCodec;
 use crate::message::{RequestHeadType, ResponseHead};
 use crate::payload::Payload;
+use crate::{body::MessageBody, Error};
 
 use super::error::SendRequestError;
 use super::pool::Acquired;
@@ -256,8 +256,8 @@ where
         body: RB,
     ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>
     where
-        RB: MessageBody + 'static,
         H: Into<RequestHeadType> + 'static,
+        RB: MessageBody<Error = Error> + 'static,
     {
         Box::pin(async move {
             match self {

actix-http/src/client/h1proto.rs

@@ -11,7 +11,6 @@ use bytes::{Bytes, BytesMut};
 use futures_core::{ready, Stream};
 use futures_util::SinkExt as _;
 
-use crate::error::PayloadError;
 use crate::h1;
 use crate::http::{
     header::{HeaderMap, IntoHeaderValue, EXPECT, HOST},
@@ -19,6 +18,7 @@ use crate::http::{
 };
 use crate::message::{RequestHeadType, ResponseHead};
 use crate::payload::Payload;
+use crate::{error::PayloadError, Error};
 
 use super::connection::{ConnectionIo, H1Connection};
 use super::error::{ConnectError, SendRequestError};
@@ -31,7 +31,7 @@ pub(crate) async fn send_request<Io, B>(
 ) -> Result<(ResponseHead, Payload), SendRequestError>
 where
     Io: ConnectionIo,
-    B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     // set request host header
     if !head.as_ref().headers.contains_key(HOST)
@@ -153,7 +153,7 @@ pub(crate) async fn send_body<Io, B>(
 ) -> Result<(), SendRequestError>
 where
     Io: ConnectionIo,
-    B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     actix_rt::pin!(body);
 

actix-http/src/client/h2proto.rs

@@ -9,14 +9,19 @@ use h2::{
 use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, TRANSFER_ENCODING};
 use http::{request::Request, Method, Version};
 
-use crate::body::{BodySize, MessageBody};
-use crate::header::HeaderMap;
-use crate::message::{RequestHeadType, ResponseHead};
-use crate::payload::Payload;
+use crate::{
+    body::{BodySize, MessageBody},
+    header::HeaderMap,
+    message::{RequestHeadType, ResponseHead},
+    payload::Payload,
+    Error,
+};
 
-use super::config::ConnectorConfig;
-use super::connection::{ConnectionIo, H2Connection};
-use super::error::SendRequestError;
+use super::{
+    config::ConnectorConfig,
+    connection::{ConnectionIo, H2Connection},
+    error::SendRequestError,
+};
 
 pub(crate) async fn send_request<Io, B>(
     mut io: H2Connection<Io>,
@@ -25,7 +30,7 @@ pub(crate) async fn send_request<Io, B>(
 ) -> Result<(ResponseHead, Payload), SendRequestError>
 where
     Io: ConnectionIo,
-    B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     trace!("Sending client request: {:?} {:?}", head, body.size());
 
@@ -125,10 +130,13 @@ where
     Ok((head, payload))
 }
 
-async fn send_body<B: MessageBody>(
+async fn send_body<B>(
     body: B,
     mut send: SendStream<Bytes>,
-) -> Result<(), SendRequestError> {
+) -> Result<(), SendRequestError>
+where
+    B: MessageBody<Error = Error>,
+{
     let mut buf = None;
     actix_rt::pin!(body);
     loop {

actix-http/src/encoding/encoder.rs

@@ -15,7 +15,7 @@ use futures_core::ready;
 use pin_project::pin_project;
 
 use crate::{
-    body::{Body, BodySize, MessageBody, ResponseBody},
+    body::{Body, BodySize, BoxAnyBody, MessageBody, ResponseBody},
     http::{
         header::{ContentEncoding, CONTENT_ENCODING},
         HeaderValue, StatusCode,
@@ -92,10 +92,12 @@ impl<B: MessageBody> Encoder<B> {
 enum EncoderBody<B> {
     Bytes(Bytes),
     Stream(#[pin] B),
-    BoxedStream(Pin<Box<dyn MessageBody>>),
+    BoxedStream(BoxAnyBody),
 }
 
-impl<B: MessageBody> MessageBody for EncoderBody<B> {
+impl<B: MessageBody<Error = Error>> MessageBody for EncoderBody<B> {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         match self {
             EncoderBody::Bytes(ref b) => b.size(),
@@ -117,12 +119,16 @@ impl<B: MessageBody> MessageBody for EncoderBody<B> {
                 }
             }
             EncoderBodyProj::Stream(b) => b.poll_next(cx),
-            EncoderBodyProj::BoxedStream(ref mut b) => b.as_mut().poll_next(cx),
+            EncoderBodyProj::BoxedStream(ref mut b) => {
+                b.as_mut().poll_next(cx).map_err(Into::into)
+            }
         }
     }
 }
 
-impl<B: MessageBody> MessageBody for Encoder<B> {
+impl<B: MessageBody<Error = Error>> MessageBody for Encoder<B> {
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         if self.encoder.is_none() {
             self.body.size()
@@ -154,7 +160,7 @@ impl<B: MessageBody> MessageBody for Encoder<B> {
                 }
             }
 
-            let result = ready!(this.body.as_mut().poll_next(cx));
+            let result = ready!(this.body.as_mut().poll_next(cx).map_err(Into::into));
 
             match result {
                 Some(Err(err)) => return Poll::Ready(Some(Err(err))),

actix-http/src/error.rs

@@ -2,6 +2,7 @@
 
 use std::{
     cell::RefCell,
+    error::Error as StdError,
     fmt,
     io::{self, Write as _},
     str::Utf8Error,
@@ -145,6 +146,8 @@ impl From<ResponseBuilder> for Error {
 #[display(fmt = "Unknown Error")]
 struct UnitError;
 
+impl ResponseError for Box<dyn StdError + 'static> {}
+
 /// Returns [`StatusCode::INTERNAL_SERVER_ERROR`] for [`UnitError`].
 impl ResponseError for UnitError {}
 

actix-http/src/h1/dispatcher.rs

@@ -51,9 +51,13 @@ pub struct Dispatcher<T, S, B, X, U>
 where
     S: Service<Request>,
     S::Error: Into<Error>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -69,9 +73,13 @@ enum DispatcherState<T, S, B, X, U>
 where
     S: Service<Request>,
     S::Error: Into<Error>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -84,9 +92,13 @@ struct InnerDispatcher<T, S, B, X, U>
 where
     S: Service<Request>,
     S::Error: Into<Error>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -122,7 +134,9 @@ enum State<S, B, X>
 where
     S: Service<Request>,
     X: Service<Request, Response = Request>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     None,
     ExpectCall(#[pin] X::Future),
@@ -133,8 +147,11 @@ where
 impl<S, B, X> State<S, B, X>
 where
     S: Service<Request>,
+
     X: Service<Request, Response = Request>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     fn is_empty(&self) -> bool {
         matches!(self, State::None)
@@ -150,12 +167,17 @@ enum PollResponse {
 impl<T, S, B, X, U> Dispatcher<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Error: Into<Error>,
     S::Response: Into<Response<B>>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -206,12 +228,17 @@ where
 impl<T, S, B, X, U> InnerDispatcher<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Error: Into<Error>,
     S::Response: Into<Response<B>>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -817,12 +844,17 @@ where
 impl<T, S, B, X, U> Future for Dispatcher<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Error: Into<Error>,
     S::Response: Into<Response<B>>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {

actix-http/src/h1/service.rs

@@ -64,11 +64,15 @@ where
     S::Error: Into<Error>,
     S::InitError: fmt::Debug,
     S::Response: Into<Response<B>>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
     X::Error: Into<Error>,
     X::InitError: fmt::Debug,
+
     U: ServiceFactory<(Request, Framed<TcpStream, Codec>), Config = (), Response = ()>,
     U::Future: 'static,
     U::Error: fmt::Display + Into<Error>,
@@ -109,11 +113,15 @@ mod openssl {
         S::Error: Into<Error>,
         S::InitError: fmt::Debug,
         S::Response: Into<Response<B>>,
+
         B: MessageBody,
+        B: MessageBody<Error = Error>,
+
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
         X::Error: Into<Error>,
         X::InitError: fmt::Debug,
+
         U: ServiceFactory<
             (Request, Framed<TlsStream<TcpStream>, Codec>),
             Config = (),
@@ -165,11 +173,15 @@ mod rustls {
         S::Error: Into<Error>,
         S::InitError: fmt::Debug,
         S::Response: Into<Response<B>>,
+
         B: MessageBody,
+        B: MessageBody<Error = Error>,
+
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
         X::Error: Into<Error>,
         X::InitError: fmt::Debug,
+
         U: ServiceFactory<
             (Request, Framed<TlsStream<TcpStream>, Codec>),
             Config = (),
@@ -253,16 +265,21 @@ impl<T, S, B, X, U> ServiceFactory<(T, Option<net::SocketAddr>)>
     for H1Service<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin + 'static,
+
     S: ServiceFactory<Request, Config = ()>,
     S::Future: 'static,
     S::Error: Into<Error>,
     S::Response: Into<Response<B>>,
     S::InitError: fmt::Debug,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
     X::Error: Into<Error>,
     X::InitError: fmt::Debug,
+
     U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
     U::Future: 'static,
     U::Error: fmt::Display + Into<Error>,
@@ -319,12 +336,17 @@ impl<T, S, B, X, U> Service<(T, Option<net::SocketAddr>)>
     for HttpServiceHandler<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Error: Into<Error>,
     S::Response: Into<Response<B>>,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display + Into<Error>,
 {

actix-http/src/h1/utils.rs

@@ -22,6 +22,7 @@ pub struct SendResponse<T, B> {
 impl<T, B> SendResponse<T, B>
 where
     B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     pub fn new(framed: Framed<T, Codec>, response: Response<B>) -> Self {
         let (res, body) = response.into_parts();
@@ -38,6 +39,7 @@ impl<T, B> Future for SendResponse<T, B>
 where
     T: AsyncRead + AsyncWrite + Unpin,
     B: MessageBody + Unpin,
+    B: MessageBody<Error = Error>,
 {
     type Output = Result<Framed<T, Codec>, Error>;
 

actix-http/src/h2/dispatcher.rs

@@ -73,7 +73,7 @@ where
     S::Error: Into<Error> + 'static,
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
-    B: MessageBody + 'static,
+    B: MessageBody<Error = Error> + 'static,
 {
     type Output = Result<(), DispatchError>;
 
@@ -216,7 +216,7 @@ where
     F: Future<Output = Result<I, E>>,
     E: Into<Error>,
     I: Into<Response<B>>,
-    B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     type Output = ();
 

actix-http/src/h2/service.rs

@@ -40,7 +40,9 @@ where
     S::Error: Into<Error> + 'static,
     S::Response: Into<Response<B>> + 'static,
     <S::Service as Service<Request>>::Future: 'static,
+
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
 {
     /// Create new `H2Service` instance with config.
     pub(crate) fn with_config<F: IntoServiceFactory<S, Request>>(
@@ -69,7 +71,9 @@ where
     S::Error: Into<Error> + 'static,
     S::Response: Into<Response<B>> + 'static,
     <S::Service as Service<Request>>::Future: 'static,
+
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
 {
     /// Create plain TCP based service
     pub fn tcp(
@@ -106,7 +110,9 @@ mod openssl {
         S::Error: Into<Error> + 'static,
         S::Response: Into<Response<B>> + 'static,
         <S::Service as Service<Request>>::Future: 'static,
+
         B: MessageBody + 'static,
+        B: MessageBody<Error = Error>,
     {
         /// Create OpenSSL based service
         pub fn openssl(
@@ -150,7 +156,9 @@ mod rustls {
         S::Error: Into<Error> + 'static,
         S::Response: Into<Response<B>> + 'static,
         <S::Service as Service<Request>>::Future: 'static,
+
         B: MessageBody + 'static,
+        B: MessageBody<Error = Error>,
     {
         /// Create Rustls based service
         pub fn rustls(
@@ -185,12 +193,15 @@ mod rustls {
 impl<T, S, B> ServiceFactory<(T, Option<net::SocketAddr>)> for H2Service<T, S, B>
 where
     T: AsyncRead + AsyncWrite + Unpin + 'static,
+
     S: ServiceFactory<Request, Config = ()>,
     S::Future: 'static,
     S::Error: Into<Error> + 'static,
     S::Response: Into<Response<B>> + 'static,
     <S::Service as Service<Request>>::Future: 'static,
+
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
 {
     type Response = ();
     type Error = DispatchError;
@@ -252,6 +263,7 @@ where
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
 {
     type Response = ();
     type Error = DispatchError;
@@ -315,7 +327,7 @@ where
     S::Error: Into<Error> + 'static,
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
-    B: MessageBody,
+    B: MessageBody<Error = Error>,
 {
     type Output = Result<(), DispatchError>;
 

actix-http/src/response.rs

@@ -242,7 +242,10 @@ impl<B> Response<B> {
     }
 }
 
-impl<B: MessageBody> fmt::Debug for Response<B> {
+impl<B> fmt::Debug for Response<B>
+where
+    B: MessageBody<Error = Error>,
+{
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let res = writeln!(
             f,

actix-http/src/service.rs

@@ -157,6 +157,7 @@ where
     <S::Service as Service<Request>>::Future: 'static,
 
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
 
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
@@ -208,6 +209,7 @@ mod openssl {
         <S::Service as Service<Request>>::Future: 'static,
 
         B: MessageBody + 'static,
+        B: MessageBody<Error = Error>,
 
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
@@ -275,6 +277,7 @@ mod rustls {
         <S::Service as Service<Request>>::Future: 'static,
 
         B: MessageBody + 'static,
+        B: MessageBody<Error = Error>,
 
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
@@ -339,6 +342,7 @@ where
     <S::Service as Service<Request>>::Future: 'static,
 
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
 
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
@@ -465,13 +469,18 @@ impl<T, S, B, X, U> Service<(T, Protocol, Option<net::SocketAddr>)>
     for HttpServiceHandler<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Error: Into<Error> + 'static,
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
+
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
     U::Error: fmt::Display + Into<Error>,
 {
@@ -522,13 +531,18 @@ where
 #[pin_project(project = StateProj)]
 enum State<T, S, B, X, U>
 where
+    T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Future: 'static,
     S::Error: Into<Error>,
-    T: AsyncRead + AsyncWrite + Unpin,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -549,13 +563,18 @@ where
 pub struct HttpServiceHandlerResponse<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Error: Into<Error> + 'static,
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
+
     B: MessageBody + 'static,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -566,13 +585,18 @@ where
 impl<T, S, B, X, U> Future for HttpServiceHandlerResponse<T, S, B, X, U>
 where
     T: AsyncRead + AsyncWrite + Unpin,
+
     S: Service<Request>,
     S::Error: Into<Error> + 'static,
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
+
     B: MessageBody,
+    B: MessageBody<Error = Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
     U::Error: fmt::Display,
 {