Merge branch 'master' into fix-response-to-error-conversion

.github/workflows/ci.yml

@@ -86,7 +86,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: -v --workspace --all-features --no-fail-fast -- --nocapture
+          args: --workspace --all-features --no-fail-fast -- --nocapture
             --skip=test_h2_content_length
             --skip=test_reading_deflate_encoding_large_random_rustls
 

CHANGES.md

@@ -1,8 +1,21 @@
 # Changes
 
 ## Unreleased - 2021-xx-xx
+### Added
+* `HttpServer::worker_max_blocking_threads` for setting block thread pool. [#2200]
+
 ### Changed
+* `ServiceResponse::error_response` now uses body type of `Body`. [#2201]
+* `ServiceResponse::checked_expr` now returns a `Result`. [#2201]
 * Update `language-tags` to `0.3`.
+* `ServiceResponse::take_body`. [#2201]
+* `ServiceResponse::map_body` closure receives and returns `B` instead of `ResponseBody<B>` types. [#2201]
+
+### Removed
+* `HttpResponse::take_body` and old `HttpResponse::into_body` method that casted body type. [#2201]
+
+[#2200]: https://github.com/actix/actix-web/pull/2200
+[#2201]: https://github.com/actix/actix-web/pull/2201
 
 
 ## 4.0.0-beta.6 - 2021-04-17

actix-files/src/files.rs

@@ -66,6 +66,7 @@ impl Clone for Files {
         }
     }
 }
+
 impl Files {
     /// Create new `Files` instance for a specified base directory.
     ///
@@ -83,7 +84,7 @@ impl Files {
     ///
     /// `Files` utilizes the existing Tokio thread-pool for blocking filesystem operations.
     /// The number of running threads is adjusted over time as needed, up to a maximum of 512 times
-    /// the number of server [workers](HttpServer::workers), by default.
+    /// the number of server [workers](actix_web::HttpServer::workers), by default.
     pub fn new<T: Into<PathBuf>>(mount_path: &str, serve_from: T) -> Files {
         let orig_dir = serve_from.into();
         let dir = match orig_dir.canonicalize() {

actix-files/src/service.rs

@@ -96,8 +96,7 @@ impl Service<ServiceRequest> for FilesService {
                     return Box::pin(ok(req.into_response(
                         HttpResponse::Found()
                             .insert_header((header::LOCATION, redirect_to))
-                            .body("")
+                            .finish(),
-                            .into_body(),
                     )));
                 }
 

actix-http/CHANGES.md

@@ -2,24 +2,38 @@
 
 ## Unreleased - 2021-xx-xx
 ### Added
+* `BoxAnyBody`: a boxed message body with boxed errors. [#2183]
 * Re-export `http` crate's `Error` type as `error::HttpError`. [#2171]
 * Re-export `StatusCode`, `Method`, `Version` and `Uri` at the crate root. [#2171]
 * Re-export `ContentEncoding` and `ConnectionType` at the crate root. [#2171]
+* `Response::into_body` that consumes response and returns body type. [#2201]
+* `impl Default` for `Response`. [#2201]
 
 ### Changed
+* The `MessageBody` trait now has an associated `Error` type. [#2183]
+* Places in `Response` where `ResponseBody<B>` was received or returned now simply use `B`. [#2201]
 * `header` mod is now public. [#2171]
 * `uri` mod is now public. [#2171]
 * Update `language-tags` to `0.3`.
-* Reduce the level from `error` to `debug` for the log line that is emitted when a `500 Internal Server Error` is built using `HttpResponse::from_error`. [#2196]
+* Reduce the level from `error` to `debug` for the log line that is emitted when a `500 Internal Server Error` is built using `HttpResponse::from_error`. [#2201]
+* `ResponseBuilder::message_body` now returns a `Result`. [#2201]
 
 ### Removed
 * Stop re-exporting `http` crate's `HeaderMap` types in addition to ours. [#2171]
+* Down-casting for `MessageBody` types. [#2183]
+* `error::Result` alias. [#2201]
+* Error field from `Response` and `Response::error`. [#2205]
+* `impl Future` for `Response`. [#2201]
+* `Response::take_body` and old `Response::into_body` method that casted body type. [#2201]
 
 ### Fixed
 * Converting an `HttpResponse` to an `Error` return the underlying `Error` if `HttpResponse` was built using `HttpResponse::from_error`.
 
 [#2171]: https://github.com/actix/actix-web/pull/2171
+[#2183]: https://github.com/actix/actix-web/pull/2183
 [#2196]: https://github.com/actix/actix-web/pull/2196
+[#2201]: https://github.com/actix/actix-web/pull/2201
+[#2205]: https://github.com/actix/actix-web/pull/2205
 
 
 ## 3.0.0-beta.6 - 2021-04-17

actix-http/examples/echo.rs

@@ -1,4 +1,4 @@
-use std::{env, io};
+use std::io;
 
 use actix_http::{http::StatusCode, Error, HttpService, Request, Response};
 use actix_server::Server;
@@ -9,8 +9,7 @@ use log::info;
 
 #[actix_rt::main]
 async fn main() -> io::Result<()> {
-    env::set_var("RUST_LOG", "echo=info");
+    env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
-    env_logger::init();
 
     Server::build()
         .bind("echo", "127.0.0.1:8080", || {

actix-http/examples/echo2.rs

@@ -1,4 +1,4 @@
-use std::{env, io};
+use std::io;
 
 use actix_http::{body::Body, http::HeaderValue, http::StatusCode};
 use actix_http::{Error, HttpService, Request, Response};
@@ -21,8 +21,7 @@ async fn handle_request(mut req: Request) -> Result<Response<Body>, Error> {
 
 #[actix_rt::main]
 async fn main() -> io::Result<()> {
-    env::set_var("RUST_LOG", "echo=info");
+    env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
-    env_logger::init();
 
     Server::build()
         .bind("echo", "127.0.0.1:8080", || {

actix-http/examples/hello-world.rs

@@ -1,4 +1,4 @@
-use std::{env, io};
+use std::io;
 
 use actix_http::{http::StatusCode, HttpService, Response};
 use actix_server::Server;
@@ -8,8 +8,7 @@ use log::info;
 
 #[actix_rt::main]
 async fn main() -> io::Result<()> {
-    env::set_var("RUST_LOG", "hello_world=info");
+    env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
-    env_logger::init();
 
     Server::build()
         .bind("hello-world", "127.0.0.1:8080", || {

actix-http/examples/ws.rs

@@ -4,7 +4,7 @@
 extern crate tls_rustls as rustls;
 
 use std::{
-    env, io,
+    io,
     pin::Pin,
     task::{Context, Poll},
     time::Duration,
@@ -20,8 +20,7 @@ use futures_core::{ready, Stream};
 
 #[actix_rt::main]
 async fn main() -> io::Result<()> {
-    env::set_var("RUST_LOG", "actix=info,h2_ws=info");
+    env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
-    env_logger::init();
 
     Server::build()
         .bind("tcp", ("127.0.0.1", 8080), || {
@@ -41,7 +40,7 @@ async fn handler(req: Request) -> Result<Response<BodyStream<Heartbeat>>, Error>
     // handshake will always fail under HTTP/2
 
     log::info!("responding");
-    Ok(res.message_body(BodyStream::new(Heartbeat::new(ws::Codec::new()))))
+    Ok(res.message_body(BodyStream::new(Heartbeat::new(ws::Codec::new())))?)
 }
 
 struct Heartbeat {

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 {
+    pub fn from_message<B>(body: B) -> Body
-        Body::Message(Box::pin(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>>>;
+    ) -> Poll<Option<Result<Bytes, Self::Error>>>;
-
-    downcast_get_type_id!();
 }
 
-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,
@@ -85,15 +86,6 @@ mod tests {
         }
     }
 
-    impl ResponseBody<Body> {
-        pub(crate) fn get_ref(&self) -> &[u8] {
-            match *self {
-                ResponseBody::Body(ref b) => b.get_ref(),
-                ResponseBody::Other(ref b) => b.get_ref(),
-            }
-        }
-    }
-
     #[actix_rt::test]
     async fn test_static_str() {
         assert_eq!(Body::from("").size(), BodySize::Sized(0));
@@ -237,10 +229,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;
 

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::Error: Into<Error>,
     {
         let cfg = ServiceConfig::new(
             self.keep_alive,
@@ -223,11 +225,13 @@ where
     /// Finish service configuration and create `HttpService` instance.
     pub fn finish<F, B>(self, service: F) -> HttpService<T, S, B, X, U>
     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::Error: Into<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,9 @@ where
         body: RB,
     ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>
     where
-        RB: MessageBody + 'static,
         H: Into<RequestHeadType> + 'static,
+        RB: MessageBody + 'static,
+        RB::Error: Into<Error>,
     {
         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};
@@ -32,6 +32,7 @@ pub(crate) async fn send_request<Io, B>(
 where
     Io: ConnectionIo,
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     // set request host header
     if !head.as_ref().headers.contains_key(HOST)
@@ -154,6 +155,7 @@ pub(crate) async fn send_body<Io, B>(
 where
     Io: ConnectionIo,
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     actix_rt::pin!(body);
 
@@ -161,9 +163,10 @@ where
     while !eof {
         while !eof && !framed.as_ref().is_write_buf_full() {
             match poll_fn(|cx| body.as_mut().poll_next(cx)).await {
-                Some(result) => {
+                Some(Ok(chunk)) => {
-                    framed.as_mut().write(h1::Message::Chunk(Some(result?)))?;
+                    framed.as_mut().write(h1::Message::Chunk(Some(chunk)))?;
                 }
+                Some(Err(err)) => return Err(err.into().into()),
                 None => {
                     eof = true;
                     framed.as_mut().write(h1::Message::Chunk(None))?;

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::{
-use crate::header::HeaderMap;
+    body::{BodySize, MessageBody},
-use crate::message::{RequestHeadType, ResponseHead};
+    header::HeaderMap,
-use crate::payload::Payload;
+    message::{RequestHeadType, ResponseHead},
+    payload::Payload,
+    Error,
+};
 
-use super::config::ConnectorConfig;
+use super::{
-use super::connection::{ConnectionIo, H2Connection};
+    config::ConnectorConfig,
-use super::error::SendRequestError;
+    connection::{ConnectionIo, H2Connection},
+    error::SendRequestError,
+};
 
 pub(crate) async fn send_request<Io, B>(
     mut io: H2Connection<Io>,
@@ -26,6 +31,7 @@ pub(crate) async fn send_request<Io, B>(
 where
     Io: ConnectionIo,
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     trace!("Sending client request: {:?} {:?}", head, body.size());
 
@@ -125,10 +131,14 @@ 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,
+    B::Error: Into<Error>,
+{
     let mut buf = None;
     actix_rt::pin!(body);
     loop {
@@ -138,7 +148,7 @@ async fn send_body<B: MessageBody>(
                     send.reserve_capacity(b.len());
                     buf = Some(b);
                 }
-                Some(Err(e)) => return Err(e.into()),
+                Some(Err(e)) => return Err(e.into().into()),
                 None => {
                     if let Err(e) = send.send_data(Bytes::new(), true) {
                         return Err(e.into());

actix-http/src/encoding/encoder.rs

@@ -1,6 +1,7 @@
 //! Stream encoders.
 
 use std::{
+    error::Error as StdError,
     future::Future,
     io::{self, Write as _},
     pin::Pin,
@@ -10,12 +11,13 @@ use std::{
 use actix_rt::task::{spawn_blocking, JoinHandle};
 use brotli2::write::BrotliEncoder;
 use bytes::Bytes;
+use derive_more::Display;
 use flate2::write::{GzEncoder, ZlibEncoder};
 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 +94,16 @@ 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 for EncoderBody<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
+    type Error = EncoderError<B::Error>;
+
     fn size(&self) -> BodySize {
         match self {
             EncoderBody::Bytes(ref b) => b.size(),
@@ -107,7 +115,7 @@ impl<B: MessageBody> MessageBody for EncoderBody<B> {
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         match self.project() {
             EncoderBodyProj::Bytes(b) => {
                 if b.is_empty() {
@@ -116,13 +124,32 @@ impl<B: MessageBody> MessageBody for EncoderBody<B> {
                     Poll::Ready(Some(Ok(std::mem::take(b))))
                 }
             }
-            EncoderBodyProj::Stream(b) => b.poll_next(cx),
+            // TODO: MSRV 1.51: poll_map_err
-            EncoderBodyProj::BoxedStream(ref mut b) => b.as_mut().poll_next(cx),
+            EncoderBodyProj::Stream(b) => match ready!(b.poll_next(cx)) {
+                Some(Err(err)) => Poll::Ready(Some(Err(EncoderError::Body(err)))),
+                Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+                None => Poll::Ready(None),
+            },
+            EncoderBodyProj::BoxedStream(ref mut b) => {
+                match ready!(b.as_pin_mut().poll_next(cx)) {
+                    Some(Err(err)) => {
+                        Poll::Ready(Some(Err(EncoderError::Boxed(err.into()))))
+                    }
+                    Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+                    None => Poll::Ready(None),
+                }
+            }
         }
     }
 }
 
-impl<B: MessageBody> MessageBody for Encoder<B> {
+impl<B> MessageBody for Encoder<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
+    type Error = EncoderError<B::Error>;
+
     fn size(&self) -> BodySize {
         if self.encoder.is_none() {
             self.body.size()
@@ -134,7 +161,7 @@ impl<B: MessageBody> MessageBody for Encoder<B> {
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         let mut this = self.project();
         loop {
             if *this.eof {
@@ -142,8 +169,9 @@ impl<B: MessageBody> MessageBody for Encoder<B> {
             }
 
             if let Some(ref mut fut) = this.fut {
-                let mut encoder =
+                let mut encoder = ready!(Pin::new(fut).poll(cx))
-                    ready!(Pin::new(fut).poll(cx)).map_err(|_| BlockingError)??;
+                    .map_err(|_| EncoderError::Blocking(BlockingError))?
+                    .map_err(EncoderError::Io)?;
 
                 let chunk = encoder.take();
                 *this.encoder = Some(encoder);
@@ -162,7 +190,7 @@ impl<B: MessageBody> MessageBody for Encoder<B> {
                 Some(Ok(chunk)) => {
                     if let Some(mut encoder) = this.encoder.take() {
                         if chunk.len() < MAX_CHUNK_SIZE_ENCODE_IN_PLACE {
-                            encoder.write(&chunk)?;
+                            encoder.write(&chunk).map_err(EncoderError::Io)?;
                             let chunk = encoder.take();
                             *this.encoder = Some(encoder);
 
@@ -182,7 +210,7 @@ impl<B: MessageBody> MessageBody for Encoder<B> {
 
                 None => {
                     if let Some(encoder) = this.encoder.take() {
-                        let chunk = encoder.finish()?;
+                        let chunk = encoder.finish().map_err(EncoderError::Io)?;
                         if chunk.is_empty() {
                             return Poll::Ready(None);
                         } else {
@@ -281,3 +309,36 @@ impl ContentEncoder {
         }
     }
 }
+
+#[derive(Debug, Display)]
+#[non_exhaustive]
+pub enum EncoderError<E> {
+    #[display(fmt = "body")]
+    Body(E),
+
+    #[display(fmt = "boxed")]
+    Boxed(Error),
+
+    #[display(fmt = "blocking")]
+    Blocking(BlockingError),
+
+    #[display(fmt = "io")]
+    Io(io::Error),
+}
+
+impl<E: StdError> StdError for EncoderError<E> {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        None
+    }
+}
+
+impl<E: Into<Error>> From<EncoderError<E>> for Error {
+    fn from(err: EncoderError<E>) -> Self {
+        match err {
+            EncoderError::Body(err) => err.into(),
+            EncoderError::Boxed(err) => err,
+            EncoderError::Blocking(err) => err.into(),
+            EncoderError::Io(err) => err.into(),
+        }
+    }
+}

actix-http/src/error.rs

@@ -2,6 +2,7 @@
 
 use std::{
     cell::RefCell,
+    error::Error as StdError,
     fmt,
     io::{self, Write as _},
     str::Utf8Error,
@@ -17,12 +18,6 @@ use crate::{body::Body, helpers::Writer, Response, ResponseBuilder};
 
 pub use http::Error as HttpError;
 
-/// A specialized [`std::result::Result`] for Actix Web operations.
-///
-/// This typedef is generally used to avoid writing out `actix_http::error::Error` directly and is
-/// otherwise a direct mapping to `Result`.
-pub type Result<T, E = Error> = std::result::Result<T, E>;
-
 /// General purpose actix web error.
 ///
 /// An actix web error is used to carry errors from `std::error`
@@ -105,8 +100,7 @@ impl From<()> for Error {
 
 impl From<std::convert::Infallible> for Error {
     fn from(_: std::convert::Infallible) -> Self {
-        // `std::convert::Infallible` indicates an error
+        // hint that an error that will never happen
-        // that will never happen
         unreachable!()
     }
 }
@@ -149,6 +143,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 {}
 
@@ -472,9 +468,8 @@ impl ResponseError for ContentTypeError {
 ///
 /// ```
 /// # use std::io;
-/// # use actix_http::*;
+/// # use actix_http::{error, Request};
-///
+/// fn index(req: Request) -> Result<&'static str, actix_http::Error> {
-/// fn index(req: Request) -> Result<&'static str> {
 ///     Err(error::ErrorBadRequest(io::Error::new(io::ErrorKind::Other, "error")))
 /// }
 /// ```

actix-http/src/h1/dispatcher.rs

@@ -17,7 +17,7 @@ use futures_core::ready;
 use log::{error, trace};
 use pin_project::pin_project;
 
-use crate::body::{Body, BodySize, MessageBody, ResponseBody};
+use crate::body::{Body, BodySize, MessageBody};
 use crate::config::ServiceConfig;
 use crate::error::{DispatchError, Error};
 use crate::error::{ParseError, PayloadError};
@@ -51,9 +51,13 @@ pub struct Dispatcher<T, S, B, X, U>
 where
     S: Service<Request>,
     S::Error: Into<Error>,
+
     B: MessageBody,
+    B::Error: Into<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::Error: Into<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::Error: Into<Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -122,19 +134,25 @@ enum State<S, B, X>
 where
     S: Service<Request>,
     X: Service<Request, Response = Request>,
+
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     None,
     ExpectCall(#[pin] X::Future),
     ServiceCall(#[pin] S::Future),
-    SendPayload(#[pin] ResponseBody<B>),
+    SendPayload(#[pin] B),
+    SendErrorPayload(#[pin] Body),
 }
 
 impl<S, B, X> State<S, B, X>
 where
     S: Service<Request>,
+
     X: Service<Request, Response = Request>,
+
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     fn is_empty(&self) -> bool {
         matches!(self, State::None)
@@ -150,12 +168,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::Error: Into<Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -206,12 +229,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::Error: Into<Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -268,11 +296,11 @@ where
         io.poll_flush(cx)
     }
 
-    fn send_response(
+    fn send_response_inner(
         self: Pin<&mut Self>,
         message: Response<()>,
-        body: ResponseBody<B>,
+        body: &impl MessageBody,
-    ) -> Result<(), DispatchError> {
+    ) -> Result<BodySize, DispatchError> {
         let size = body.size();
         let mut this = self.project();
         this.codec
@@ -285,10 +313,35 @@ where
             })?;
 
         this.flags.set(Flags::KEEPALIVE, this.codec.keepalive());
-        match size {
+
-            BodySize::None | BodySize::Empty => this.state.set(State::None),
+        Ok(size)
-            _ => this.state.set(State::SendPayload(body)),
+    }
+
+    fn send_response(
+        mut self: Pin<&mut Self>,
+        message: Response<()>,
+        body: B,
+    ) -> Result<(), DispatchError> {
+        let size = self.as_mut().send_response_inner(message, &body)?;
+        let state = match size {
+            BodySize::None | BodySize::Empty => State::None,
+            _ => State::SendPayload(body),
         };
+        self.project().state.set(state);
+        Ok(())
+    }
+
+    fn send_error_response(
+        mut self: Pin<&mut Self>,
+        message: Response<()>,
+        body: Body,
+    ) -> Result<(), DispatchError> {
+        let size = self.as_mut().send_response_inner(message, &body)?;
+        let state = match size {
+            BodySize::None | BodySize::Empty => State::None,
+            _ => State::SendErrorPayload(body),
+        };
+        self.project().state.set(state);
         Ok(())
     }
 
@@ -326,8 +379,7 @@ where
                         // send_response would update InnerDispatcher state to SendPayload or
                         // None(If response body is empty).
                         // continue loop to poll it.
-                        self.as_mut()
+                        self.as_mut().send_error_response(res, Body::Empty)?;
-                            .send_response(res, ResponseBody::Other(Body::Empty))?;
                     }
 
                     // return with upgrade request and poll it exclusively.
@@ -349,7 +401,7 @@ where
                     Poll::Ready(Err(err)) => {
                         let res = Response::from_error(err.into());
                         let (res, body) = res.replace_body(());
-                        self.as_mut().send_response(res, body.into_body())?;
+                        self.as_mut().send_error_response(res, body)?;
                     }
 
                     // service call pending and could be waiting for more chunk messages.
@@ -365,6 +417,41 @@ where
                 },
 
                 StateProj::SendPayload(mut stream) => {
+                    // keep populate writer buffer until buffer size limit hit,
+                    // get blocked or finished.
+                    while this.write_buf.len() < super::payload::MAX_BUFFER_SIZE {
+                        match stream.as_mut().poll_next(cx) {
+                            Poll::Ready(Some(Ok(item))) => {
+                                this.codec.encode(
+                                    Message::Chunk(Some(item)),
+                                    &mut this.write_buf,
+                                )?;
+                            }
+
+                            Poll::Ready(None) => {
+                                this.codec
+                                    .encode(Message::Chunk(None), &mut this.write_buf)?;
+                                // payload stream finished.
+                                // set state to None and handle next message
+                                this.state.set(State::None);
+                                continue 'res;
+                            }
+
+                            Poll::Ready(Some(Err(err))) => {
+                                return Err(DispatchError::Service(err.into()))
+                            }
+
+                            Poll::Pending => return Ok(PollResponse::DoNothing),
+                        }
+                    }
+                    // buffer is beyond max size.
+                    // return and try to write the whole buffer to io stream.
+                    return Ok(PollResponse::DrainWriteBuf);
+                }
+
+                StateProj::SendErrorPayload(mut stream) => {
+                    // TODO: de-dupe impl with SendPayload
+
                     // keep populate writer buffer until buffer size limit hit,
                     // get blocked or finished.
                     while this.write_buf.len() < super::payload::MAX_BUFFER_SIZE {
@@ -406,12 +493,14 @@ where
                         let fut = this.flow.service.call(req);
                         this.state.set(State::ServiceCall(fut));
                     }
+
                     // send expect error as response
                     Poll::Ready(Err(err)) => {
                         let res = Response::from_error(err.into());
                         let (res, body) = res.replace_body(());
-                        self.as_mut().send_response(res, body.into_body())?;
+                        self.as_mut().send_error_response(res, body)?;
                     }
+
                     // expect must be solved before progress can be made.
                     Poll::Pending => return Ok(PollResponse::DoNothing),
                 },
@@ -459,7 +548,7 @@ where
                         Poll::Ready(Err(err)) => {
                             let res = Response::from_error(err.into());
                             let (res, body) = res.replace_body(());
-                            return self.send_response(res, body.into_body());
+                            return self.send_error_response(res, body);
                         }
                     }
                 }
@@ -479,7 +568,7 @@ where
                         Poll::Ready(Err(err)) => {
                             let res = Response::from_error(err.into());
                             let (res, body) = res.replace_body(());
-                            self.send_response(res, body.into_body())
+                            self.send_error_response(res, body)
                         }
                     };
                 }
@@ -599,8 +688,10 @@ where
                     }
                     // Requests overflow buffer size should be responded with 431
                     this.messages.push_back(DispatcherMessage::Error(
-                        Response::new(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE)
+                        Response::with_body(
-                            .drop_body(),
+                            StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE,
+                            (),
+                        ),
                     ));
                     this.flags.insert(Flags::READ_DISCONNECT);
                     *this.error = Some(ParseError::TooLarge.into());
@@ -679,10 +770,9 @@ where
                             } else {
                                 // timeout on first request (slow request) return 408
                                 trace!("Slow request timeout");
-                                let _ = self.as_mut().send_response(
+                                let _ = self.as_mut().send_error_response(
-                                    Response::new(StatusCode::REQUEST_TIMEOUT)
+                                    Response::with_body(StatusCode::REQUEST_TIMEOUT, ()),
-                                        .drop_body(),
+                                    Body::Empty,
-                                    ResponseBody::Other(Body::Empty),
                                 );
                                 this = self.project();
                                 this.flags.insert(Flags::STARTED | Flags::SHUTDOWN);
@@ -817,12 +907,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::Error: Into<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/encoder.rs

@@ -630,8 +630,7 @@ mod tests {
     async fn test_no_content_length() {
         let mut bytes = BytesMut::with_capacity(2048);
 
-        let mut res: Response<()> =
+        let mut res = Response::with_body(StatusCode::SWITCHING_PROTOCOLS, ());
-            Response::new(StatusCode::SWITCHING_PROTOCOLS).into_body::<()>();
         res.headers_mut().insert(DATE, HeaderValue::from_static(""));
         res.headers_mut()
             .insert(CONTENT_LENGTH, HeaderValue::from_static("0"));

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::Error: Into<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::Error: Into<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::Error: Into<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::Error: Into<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::Error: Into<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

@@ -4,7 +4,7 @@ use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
 
-use crate::body::{BodySize, MessageBody, ResponseBody};
+use crate::body::{BodySize, MessageBody};
 use crate::error::Error;
 use crate::h1::{Codec, Message};
 use crate::response::Response;
@@ -14,7 +14,7 @@ use crate::response::Response;
 pub struct SendResponse<T, B> {
     res: Option<Message<(Response<()>, BodySize)>>,
     #[pin]
-    body: Option<ResponseBody<B>>,
+    body: Option<B>,
     #[pin]
     framed: Option<Framed<T, Codec>>,
 }
@@ -22,6 +22,7 @@ pub struct SendResponse<T, B> {
 impl<T, B> SendResponse<T, B>
 where
     B: MessageBody,
+    B::Error: Into<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::Error: Into<Error>,
 {
     type Output = Result<Framed<T, Codec>, Error>;
 
@@ -60,7 +62,18 @@ where
                         .unwrap()
                         .is_write_buf_full()
                 {
-                    match this.body.as_mut().as_pin_mut().unwrap().poll_next(cx)? {
+                    let next =
+                        // TODO: MSRV 1.51: poll_map_err
+                        match this.body.as_mut().as_pin_mut().unwrap().poll_next(cx) {
+                            Poll::Ready(Some(Ok(item))) => Poll::Ready(Some(item)),
+                            Poll::Ready(Some(Err(err))) => {
+                                return Poll::Ready(Err(err.into()))
+                            }
+                            Poll::Ready(None) => Poll::Ready(None),
+                            Poll::Pending => Poll::Pending,
+                        };
+
+                    match next {
                         Poll::Ready(item) => {
                             // body is done when item is None
                             body_done = item.is_none();

actix-http/src/h2/dispatcher.rs

@@ -12,7 +12,7 @@ use h2::{
 use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING};
 use log::{error, trace};
 
-use crate::body::{BodySize, MessageBody, ResponseBody};
+use crate::body::{Body, BodySize, MessageBody};
 use crate::config::ServiceConfig;
 use crate::error::{DispatchError, Error};
 use crate::message::ResponseHead;
@@ -69,11 +69,14 @@ 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> + 'static,
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
+
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     type Output = Result<(), DispatchError>;
 
@@ -132,7 +135,8 @@ struct ServiceResponse<F, I, E, B> {
 #[pin_project::pin_project(project = ServiceResponseStateProj)]
 enum ServiceResponseState<F, B> {
     ServiceCall(#[pin] F, Option<SendResponse<Bytes>>),
-    SendPayload(SendStream<Bytes>, #[pin] ResponseBody<B>),
+    SendPayload(SendStream<Bytes>, #[pin] B),
+    SendErrorPayload(SendStream<Bytes>, #[pin] Body),
 }
 
 impl<F, I, E, B> ServiceResponse<F, I, E, B>
@@ -140,7 +144,9 @@ where
     F: Future<Output = Result<I, E>>,
     E: Into<Error>,
     I: Into<Response<B>>,
+
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     fn prepare_response(
         &self,
@@ -216,7 +222,9 @@ where
     F: Future<Output = Result<I, E>>,
     E: Into<Error>,
     I: Into<Response<B>>,
+
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     type Output = ();
 
@@ -273,9 +281,8 @@ where
                         if size.is_eof() {
                             Poll::Ready(())
                         } else {
-                            this.state.set(ServiceResponseState::SendPayload(
+                            this.state.set(ServiceResponseState::SendErrorPayload(
-                                stream,
+                                stream, body,
-                                body.into_body(),
                             ));
                             self.poll(cx)
                         }
@@ -324,8 +331,65 @@ where
                                 *this.buffer = Some(chunk);
                             }
 
+                            Some(Err(err)) => {
+                                error!(
+                                    "Response payload stream error: {:?}",
+                                    err.into()
+                                );
+
+                                return Poll::Ready(());
+                            }
+                        },
+                    }
+                }
+            }
+
+            ServiceResponseStateProj::SendErrorPayload(ref mut stream, ref mut body) => {
+                // TODO: de-dupe impl with SendPayload
+
+                loop {
+                    match this.buffer {
+                        Some(ref mut buffer) => match ready!(stream.poll_capacity(cx)) {
+                            None => return Poll::Ready(()),
+
+                            Some(Ok(cap)) => {
+                                let len = buffer.len();
+                                let bytes = buffer.split_to(cmp::min(cap, len));
+
+                                if let Err(e) = stream.send_data(bytes, false) {
+                                    warn!("{:?}", e);
+                                    return Poll::Ready(());
+                                } else if !buffer.is_empty() {
+                                    let cap = cmp::min(buffer.len(), CHUNK_SIZE);
+                                    stream.reserve_capacity(cap);
+                                } else {
+                                    this.buffer.take();
+                                }
+                            }
+
                             Some(Err(e)) => {
-                                error!("Response payload stream error: {:?}", e);
+                                warn!("{:?}", e);
+                                return Poll::Ready(());
+                            }
+                        },
+
+                        None => match ready!(body.as_mut().poll_next(cx)) {
+                            None => {
+                                if let Err(e) = stream.send_data(Bytes::new(), true) {
+                                    warn!("{:?}", e);
+                                }
+                                return Poll::Ready(());
+                            }
+
+                            Some(Ok(chunk)) => {
+                                stream
+                                    .reserve_capacity(cmp::min(chunk.len(), CHUNK_SIZE));
+                                *this.buffer = Some(chunk);
+                            }
+
+                            Some(Err(err)) => {
+                                error!("Response payload stream error: {:?}", err);
+
                                 return Poll::Ready(());
                             }
                         },

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::Error: Into<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::Error: Into<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::Error: Into<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::Error: Into<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::Error: Into<Error>,
 {
     type Response = ();
     type Error = DispatchError;
@@ -252,6 +263,7 @@ where
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     type Response = ();
     type Error = DispatchError;
@@ -316,6 +328,7 @@ where
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
     B: MessageBody,
+    B::Error: Into<Error>,
 {
     type Output = Result<(), DispatchError>;
 

actix-http/src/header/shared/charset.rs

@@ -104,7 +104,7 @@ impl Display for Charset {
 impl FromStr for Charset {
     type Err = crate::Error;
 
-    fn from_str(s: &str) -> crate::Result<Charset> {
+    fn from_str(s: &str) -> Result<Charset, crate::Error> {
         Ok(match s.to_ascii_uppercase().as_ref() {
             "US-ASCII" => Us_Ascii,
             "ISO-8859-1" => Iso_8859_1,

actix-http/src/lib.rs

@@ -54,7 +54,7 @@ pub mod ws;
 
 pub use self::builder::HttpServiceBuilder;
 pub use self::config::{KeepAlive, ServiceConfig};
-pub use self::error::{Error, ResponseError, Result};
+pub use self::error::{Error, ResponseError};
 pub use self::extensions::Extensions;
 pub use self::header::ContentEncoding;
 pub use self::http_message::HttpMessage;

actix-http/src/macros.rs

@@ -15,8 +15,15 @@ macro_rules! downcast_get_type_id {
         /// making it impossible for safe code to construct outside of
         /// this module. This ensures that safe code cannot violate
         /// type-safety by implementing this method.
+        ///
+        /// We also take `PrivateHelper` as a parameter, to ensure that
+        /// safe code cannot obtain a `PrivateHelper` instance by
+        /// delegating to an existing implementation of `__private_get_type_id__`
         #[doc(hidden)]
-        fn __private_get_type_id__(&self) -> (std::any::TypeId, PrivateHelper)
+        fn __private_get_type_id__(
+            &self,
+            _: PrivateHelper,
+        ) -> (std::any::TypeId, PrivateHelper)
         where
             Self: 'static,
         {
@@ -39,7 +46,9 @@ macro_rules! downcast {
         impl dyn $name + 'static {
             /// Downcasts generic body to a specific type.
             pub fn downcast_ref<T: $name + 'static>(&self) -> Option<&T> {
-                if self.__private_get_type_id__().0 == std::any::TypeId::of::<T>() {
+                if self.__private_get_type_id__(PrivateHelper(())).0
+                    == std::any::TypeId::of::<T>()
+                {
                     // SAFETY: external crates cannot override the default
                     // implementation of `__private_get_type_id__`, since
                     // it requires returning a private type. We can therefore
@@ -53,7 +62,9 @@ macro_rules! downcast {
 
             /// Downcasts a generic body to a mutable specific type.
             pub fn downcast_mut<T: $name + 'static>(&mut self) -> Option<&mut T> {
-                if self.__private_get_type_id__().0 == std::any::TypeId::of::<T>() {
+                if self.__private_get_type_id__(PrivateHelper(())).0
+                    == std::any::TypeId::of::<T>()
+                {
                     // SAFETY: external crates cannot override the default
                     // implementation of `__private_get_type_id__`, since
                     // it requires returning a private type. We can therefore

actix-http/src/message.rs

@@ -293,14 +293,14 @@ impl ResponseHead {
         }
     }
 
-    #[inline]
     /// Check if keep-alive is enabled
+    #[inline]
     pub fn keep_alive(&self) -> bool {
         self.connection_type() == ConnectionType::KeepAlive
     }
 
-    #[inline]
     /// Check upgrade status of this message
+    #[inline]
     pub fn upgrade(&self) -> bool {
         self.connection_type() == ConnectionType::Upgrade
     }
@@ -389,12 +389,6 @@ impl BoxedResponseHead {
     pub fn new(status: StatusCode) -> Self {
         RESPONSE_POOL.with(|p| p.get_message(status))
     }
-
-    pub(crate) fn take(&mut self) -> Self {
-        BoxedResponseHead {
-            head: self.head.take(),
-        }
-    }
 }
 
 impl std::ops::Deref for BoxedResponseHead {

actix-http/src/response.rs

@@ -2,17 +2,13 @@
 
 use std::{
     cell::{Ref, RefMut},
-    fmt,
+    fmt, str,
-    future::Future,
-    pin::Pin,
-    str,
-    task::{Context, Poll},
 };
 
 use bytes::{Bytes, BytesMut};
 
 use crate::{
-    body::{Body, MessageBody, ResponseBody},
+    body::{Body, MessageBody},
     error::Error,
     extensions::Extensions,
     http::{HeaderMap, StatusCode},
@@ -23,22 +19,20 @@ use crate::{
 /// An HTTP response.
 pub struct Response<B> {
     pub(crate) head: BoxedResponseHead,
-    pub(crate) body: ResponseBody<B>,
+    pub(crate) body: B,
-    pub(crate) error: Option<Error>,
 }
 
 impl Response<Body> {
-    /// Constructs a response
+    /// Constructs a new response with default body.
     #[inline]
     pub fn new(status: StatusCode) -> Response<Body> {
         Response {
             head: BoxedResponseHead::new(status),
-            body: ResponseBody::Body(Body::Empty),
+            body: Body::Empty,
-            error: None,
         }
     }
 
-    /// Create HTTP response builder with specific status.
+    /// Constructs a new response builder.
     #[inline]
     pub fn build(status: StatusCode) -> ResponseBuilder {
         ResponseBuilder::new(status)
@@ -47,25 +41,25 @@ impl Response<Body> {
     // just a couple frequently used shortcuts
     // this list should not grow larger than a few
 
-    /// Creates a new response with status 200 OK.
+    /// Constructs a new response with status 200 OK.
     #[inline]
     pub fn ok() -> Response<Body> {
         Response::new(StatusCode::OK)
     }
 
-    /// Creates a new response with status 400 Bad Request.
+    /// Constructs a new response with status 400 Bad Request.
     #[inline]
     pub fn bad_request() -> Response<Body> {
         Response::new(StatusCode::BAD_REQUEST)
     }
 
-    /// Creates a new response with status 404 Not Found.
+    /// Constructs a new response with status 404 Not Found.
     #[inline]
     pub fn not_found() -> Response<Body> {
         Response::new(StatusCode::NOT_FOUND)
     }
 
-    /// Creates a new response with status 500 Internal Server Error.
+    /// Constructs a new response with status 500 Internal Server Error.
     #[inline]
     pub fn internal_server_error() -> Response<Body> {
         Response::new(StatusCode::INTERNAL_SERVER_ERROR)
@@ -73,176 +67,149 @@ impl Response<Body> {
 
     // end shortcuts
 
-    /// Constructs an error response
+    /// Constructs a new response from an error.
     #[inline]
     pub fn from_error(error: Error) -> Response<Body> {
-        let mut resp = error.as_response_error().error_response();
+        let resp = error.as_response_error().error_response();
         if resp.head.status == StatusCode::INTERNAL_SERVER_ERROR {
             debug!("Internal Server Error: {:?}", error);
         }
-        resp.error = Some(error);
         resp
     }
-
-    /// Convert response to response with body
-    pub fn into_body<B>(self) -> Response<B> {
-        let b = match self.body {
-            ResponseBody::Body(b) => b,
-            ResponseBody::Other(b) => b,
-        };
-        Response {
-            head: self.head,
-            error: self.error,
-            body: ResponseBody::Other(b),
-        }
-    }
 }
 
 impl<B> Response<B> {
-    /// Constructs a response with body
+    /// Constructs a new response with given body.
     #[inline]
     pub fn with_body(status: StatusCode, body: B) -> Response<B> {
         Response {
             head: BoxedResponseHead::new(status),
-            body: ResponseBody::Body(body),
+            body: body,
-            error: None,
         }
     }
 
+    /// Returns a reference to the head of this response.
     #[inline]
-    /// Http message part of the response
     pub fn head(&self) -> &ResponseHead {
         &*self.head
     }
 
+    /// Returns a mutable reference to the head of this response.
     #[inline]
-    /// Mutable reference to a HTTP message part of the response
     pub fn head_mut(&mut self) -> &mut ResponseHead {
         &mut *self.head
     }
 
-    /// The source `error` for this response
+    /// Returns the status code of this response.
-    #[inline]
-    pub fn error(&self) -> Option<&Error> {
-        self.error.as_ref()
-    }
-
-    /// Get the response status code
     #[inline]
     pub fn status(&self) -> StatusCode {
         self.head.status
     }
 
-    /// Set the `StatusCode` for this response
+    /// Returns a mutable reference the status code of this response.
     #[inline]
     pub fn status_mut(&mut self) -> &mut StatusCode {
         &mut self.head.status
     }
 
-    /// Get the headers from the response
+    /// Returns a reference to response headers.
     #[inline]
     pub fn headers(&self) -> &HeaderMap {
         &self.head.headers
     }
 
-    /// Get a mutable reference to the headers
+    /// Returns a mutable reference to response headers.
     #[inline]
     pub fn headers_mut(&mut self) -> &mut HeaderMap {
         &mut self.head.headers
     }
 
-    /// Connection upgrade status
+    /// Returns true if connection upgrade is enabled.
     #[inline]
     pub fn upgrade(&self) -> bool {
         self.head.upgrade()
     }
 
-    /// Keep-alive status for this connection
+    /// Returns true if keep-alive is enabled.
     pub fn keep_alive(&self) -> bool {
         self.head.keep_alive()
     }
 
-    /// Responses extensions
+    /// Returns a reference to the extensions of this response.
     #[inline]
     pub fn extensions(&self) -> Ref<'_, Extensions> {
         self.head.extensions.borrow()
     }
 
-    /// Mutable reference to a the response's extensions
+    /// Returns a mutable reference to the extensions of this response.
     #[inline]
     pub fn extensions_mut(&mut self) -> RefMut<'_, Extensions> {
         self.head.extensions.borrow_mut()
     }
 
-    /// Get body of this response
+    /// Returns a reference to the body of this response.
     #[inline]
-    pub fn body(&self) -> &ResponseBody<B> {
+    pub fn body(&self) -> &B {
         &self.body
     }
 
-    /// Set a body
+    /// Sets new body.
     pub fn set_body<B2>(self, body: B2) -> Response<B2> {
         Response {
             head: self.head,
-            body: ResponseBody::Body(body),
+            body,
-            error: None,
         }
     }
 
-    /// Split response and body
+    /// Drops body and returns new response.
-    pub fn into_parts(self) -> (Response<()>, ResponseBody<B>) {
-        (
-            Response {
-                head: self.head,
-                body: ResponseBody::Body(()),
-                error: self.error,
-            },
-            self.body,
-        )
-    }
-
-    /// Drop request's body
     pub fn drop_body(self) -> Response<()> {
-        Response {
+        self.set_body(())
-            head: self.head,
-            body: ResponseBody::Body(()),
-            error: None,
-        }
     }
 
-    /// Set a body and return previous body value
+    /// Sets new body, returning new response and previous body value.
-    pub(crate) fn replace_body<B2>(self, body: B2) -> (Response<B2>, ResponseBody<B>) {
+    pub(crate) fn replace_body<B2>(self, body: B2) -> (Response<B2>, B) {
         (
             Response {
                 head: self.head,
-                body: ResponseBody::Body(body),
+                body,
-                error: self.error,
             },
             self.body,
         )
     }
 
-    /// Set a body and return previous body value
+    /// Returns split head and body.
+    ///
+    /// # Implementation Notes
+    /// Due to internal performance optimisations, the first element of the returned tuple is a
+    /// `Response` as well but only contains the head of the response this was called on.
+    pub fn into_parts(self) -> (Response<()>, B) {
+        self.replace_body(())
+    }
+
+    /// Returns new response with mapped body.
     pub fn map_body<F, B2>(mut self, f: F) -> Response<B2>
     where
-        F: FnOnce(&mut ResponseHead, ResponseBody<B>) -> ResponseBody<B2>,
+        F: FnOnce(&mut ResponseHead, B) -> B2,
     {
         let body = f(&mut self.head, self.body);
 
         Response {
-            body,
             head: self.head,
-            error: self.error,
+            body,
         }
     }
 
-    /// Extract response body
+    /// Returns body, consuming this response.
-    pub fn take_body(&mut self) -> ResponseBody<B> {
+    pub fn into_body(self) -> B {
-        self.body.take_body()
+        self.body
     }
 }
 
-impl<B: MessageBody> fmt::Debug for Response<B> {
+impl<B> fmt::Debug for Response<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let res = writeln!(
             f,
@@ -260,19 +227,13 @@ impl<B: MessageBody> fmt::Debug for Response<B> {
     }
 }
 
-impl<B: Unpin> Future for Response<B> {
+impl<B: Default> Default for Response<B> {
-    type Output = Result<Response<B>, Error>;
+    #[inline]
-
+    fn default() -> Response<B> {
-    fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
+        Response::with_body(StatusCode::default(), B::default())
-        Poll::Ready(Ok(Response {
-            head: self.head.take(),
-            body: self.body.take_body(),
-            error: self.error.take(),
-        }))
     }
 }
 
-/// Helper converters
 impl<I: Into<Response<Body>>, E: Into<Error>> From<Result<I, E>> for Response<Body> {
     fn from(res: Result<I, E>) -> Self {
         match res {

actix-http/src/response_builder.rs

@@ -13,7 +13,7 @@ use bytes::Bytes;
 use futures_core::Stream;
 
 use crate::{
-    body::{Body, BodyStream, ResponseBody},
+    body::{Body, BodyStream},
     error::{Error, HttpError},
     header::{self, IntoHeaderPair, IntoHeaderValue},
     message::{BoxedResponseHead, ConnectionType, ResponseHead},
@@ -38,10 +38,11 @@ use crate::{
 ///     .body("1234");
 ///
 /// assert_eq!(res.status(), StatusCode::OK);
-/// assert_eq!(body::to_bytes(res.take_body()).await.unwrap(), &b"1234"[..]);
 ///
 /// assert!(res.headers().contains_key("server"));
 /// assert_eq!(res.headers().get_all("set-cookie").count(), 2);
+///
+/// assert_eq!(body::to_bytes(res.into_body()).await.unwrap(), &b"1234"[..]);
 /// # })
 /// ```
 pub struct ResponseBuilder {
@@ -236,23 +237,19 @@ impl ResponseBuilder {
     #[inline]
     pub fn body<B: Into<Body>>(&mut self, body: B) -> Response<Body> {
         self.message_body(body.into())
+            .unwrap_or_else(Response::from_error)
     }
 
     /// Generate response with a body.
     ///
     /// This `ResponseBuilder` will be left in a useless state.
-    pub fn message_body<B>(&mut self, body: B) -> Response<B> {
+    pub fn message_body<B>(&mut self, body: B) -> Result<Response<B>, Error> {
-        if let Some(e) = self.err.take() {
+        if let Some(err) = self.err.take() {
-            return Response::from(Error::from(e)).into_body();
+            return Err(err.into());
         }
 
-        let response = self.head.take().expect("cannot reuse response builder");
+        let head = self.head.take().expect("cannot reuse response builder");
-
+        Ok(Response { head, body })
-        Response {
-            head: response,
-            body: ResponseBody::Body(body),
-            error: None,
-        }
     }
 
     /// Generate response with a streaming body.

actix-http/src/service.rs

@@ -59,6 +59,7 @@ where
     S::Response: Into<Response<B>> + 'static,
     <S::Service as Service<Request>>::Future: 'static,
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     /// Create new `HttpService` instance.
     pub fn new<F: IntoServiceFactory<S, Request>>(service: F) -> Self {
@@ -157,6 +158,7 @@ where
     <S::Service as Service<Request>>::Future: 'static,
 
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
@@ -208,6 +210,7 @@ mod openssl {
         <S::Service as Service<Request>>::Future: 'static,
 
         B: MessageBody + 'static,
+        B::Error: Into<Error>,
 
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
@@ -275,6 +278,7 @@ mod rustls {
         <S::Service as Service<Request>>::Future: 'static,
 
         B: MessageBody + 'static,
+        B::Error: Into<Error>,
 
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
@@ -339,6 +343,7 @@ where
     <S::Service as Service<Request>>::Future: 'static,
 
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
@@ -465,13 +470,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::Error: Into<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 +532,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::Error: Into<Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -549,13 +564,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,
+    B::Error: Into<Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
     U::Error: fmt::Display,
 {
@@ -566,13 +586,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 + 'static,
+    B::Error: Into<Error>,
+
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,
+
     U: Service<(Request, Framed<T, h1::Codec>), Response = ()>,
     U::Error: fmt::Display,
 {

actix-http/tests/test_ws.rs

@@ -52,7 +52,7 @@ where
 
     fn call(&self, (req, mut framed): (Request, Framed<T, h1::Codec>)) -> Self::Future {
         let fut = async move {
-            let res = ws::handshake(req.head()).unwrap().message_body(());
+            let res = ws::handshake(req.head()).unwrap().message_body(()).unwrap();
 
             framed
                 .send((res, body::BodySize::None).into())

actix-test/src/lib.rs

@@ -86,6 +86,7 @@ where
     S::Response: Into<Response<B>> + 'static,
     <S::Service as Service<Request>>::Future: 'static,
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     start_with(TestServerConfig::default(), factory)
 }
@@ -125,6 +126,7 @@ where
     S::Response: Into<Response<B>> + 'static,
     <S::Service as Service<Request>>::Future: 'static,
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     let (tx, rx) = mpsc::channel();
 

src/app_service.rs

@@ -166,8 +166,7 @@ impl AppInitServiceState {
         Rc::new(AppInitServiceState {
             rmap,
             config,
-            // TODO: AppConfig can be used to pass user defined HttpRequestPool
+            // TODO: AppConfig can be used to pass user defined HttpRequestPool capacity.
-            // capacity.
             pool: HttpRequestPool::default(),
         })
     }

src/error.rs

@@ -9,6 +9,11 @@ use url::ParseError as UrlParseError;
 
 use crate::http::StatusCode;
 
+/// A convenience [`Result`](std::result::Result) for Actix Web operations.
+///
+/// This type alias is generally used to avoid writing out `actix_http::Error` directly.
+pub type Result<T, E = actix_http::Error> = std::result::Result<T, E>;
+
 /// Errors which can occur when attempting to generate resource uri.
 #[derive(Debug, PartialEq, Display, Error, From)]
 #[non_exhaustive]
@@ -26,7 +31,6 @@ pub enum UrlGenerationError {
     ParseError(UrlParseError),
 }
 
-/// `InternalServerError` for `UrlGeneratorError`
 impl ResponseError for UrlGenerationError {}
 
 /// A set of errors that can occur during parsing urlencoded payloads
@@ -70,7 +74,6 @@ pub enum UrlencodedError {
     Payload(PayloadError),
 }
 
-/// Return `BadRequest` for `UrlencodedError`
 impl ResponseError for UrlencodedError {
     fn status_code(&self) -> StatusCode {
         match self {
@@ -149,7 +152,6 @@ pub enum QueryPayloadError {
     Deserialize(serde::de::value::Error),
 }
 
-/// Return `BadRequest` for `QueryPayloadError`
 impl ResponseError for QueryPayloadError {
     fn status_code(&self) -> StatusCode {
         StatusCode::BAD_REQUEST
@@ -177,7 +179,6 @@ pub enum ReadlinesError {
     ContentTypeError(ContentTypeError),
 }
 
-/// Return `BadRequest` for `ReadlinesError`
 impl ResponseError for ReadlinesError {
     fn status_code(&self) -> StatusCode {
         match *self {

src/lib.rs

@@ -97,7 +97,7 @@ pub(crate) mod types;
 pub mod web;
 
 pub use actix_http::Response as BaseHttpResponse;
-pub use actix_http::{body, Error, HttpMessage, ResponseError, Result};
+pub use actix_http::{body, Error, HttpMessage, ResponseError};
 #[doc(inline)]
 pub use actix_rt as rt;
 pub use actix_web_codegen::*;
@@ -105,6 +105,7 @@ pub use actix_web_codegen::*;
 pub use cookie;
 
 pub use crate::app::App;
+pub use crate::error::Result;
 pub use crate::extract::FromRequest;
 pub use crate::request::HttpRequest;
 pub use crate::resource::Resource;

src/middleware/compat.rs

@@ -1,12 +1,13 @@
 //! For middleware documentation, see [`Compat`].
 
 use std::{
+    error::Error as StdError,
     future::Future,
     pin::Pin,
     task::{Context, Poll},
 };
 
-use actix_http::body::{Body, MessageBody, ResponseBody};
+use actix_http::body::{Body, MessageBody};
 use actix_service::{Service, Transform};
 use futures_core::{future::LocalBoxFuture, ready};
 
@@ -113,9 +114,13 @@ pub trait MapServiceResponseBody {
     fn map_body(self) -> ServiceResponse;
 }
 
-impl<B: MessageBody + Unpin + 'static> MapServiceResponseBody for ServiceResponse<B> {
+impl<B> MapServiceResponseBody for ServiceResponse<B>
+where
+    B: MessageBody + Unpin + 'static,
+    B::Error: Into<Box<dyn StdError + 'static>>,
+{
     fn map_body(self) -> ServiceResponse {
-        self.map_body(|_, body| ResponseBody::Other(Body::from_message(body)))
+        self.map_body(|_, body| Body::from_message(body))
     }
 }
 

src/middleware/compress.rs

@@ -10,7 +10,7 @@ use std::{
 };
 
 use actix_http::{
-    body::MessageBody,
+    body::{MessageBody, ResponseBody},
     encoding::Encoder,
     http::header::{ContentEncoding, ACCEPT_ENCODING},
     Error,
@@ -59,7 +59,7 @@ where
     B: MessageBody,
     S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
 {
-    type Response = ServiceResponse<Encoder<B>>;
+    type Response = ServiceResponse<ResponseBody<Encoder<B>>>;
     type Error = Error;
     type Transform = CompressMiddleware<S>;
     type InitError = ();
@@ -83,7 +83,7 @@ where
     B: MessageBody,
     S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
 {
-    type Response = ServiceResponse<Encoder<B>>;
+    type Response = ServiceResponse<ResponseBody<Encoder<B>>>;
     type Error = Error;
     type Future = CompressResponse<S, B>;
 
@@ -127,7 +127,7 @@ where
     B: MessageBody,
     S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
 {
-    type Output = Result<ServiceResponse<Encoder<B>>, Error>;
+    type Output = Result<ServiceResponse<ResponseBody<Encoder<B>>>, Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
@@ -140,9 +140,9 @@ where
                     *this.encoding
                 };
 
-                Poll::Ready(Ok(
+                Poll::Ready(Ok(resp.map_body(move |head, body| {
-                    resp.map_body(move |head, body| Encoder::response(enc, head, body))
+                    Encoder::response(enc, head, ResponseBody::Body(body))
-                ))
+                })))
             }
             Err(e) => Poll::Ready(Err(e)),
         }

src/middleware/logger.rs

@@ -21,11 +21,10 @@ use regex::{Regex, RegexSet};
 use time::OffsetDateTime;
 
 use crate::{
-    dev::{BodySize, MessageBody, ResponseBody},
+    dev::{BodySize, MessageBody},
-    error::{Error, Result},
     http::{HeaderName, StatusCode},
     service::{ServiceRequest, ServiceResponse},
-    HttpResponse,
+    Error, HttpResponse, Result,
 };
 
 /// Middleware for logging request and response summaries to the terminal.
@@ -290,13 +289,11 @@ where
         let time = *this.time;
         let format = this.format.take();
 
-        Poll::Ready(Ok(res.map_body(move |_, body| {
+        Poll::Ready(Ok(res.map_body(move |_, body| StreamLog {
-            ResponseBody::Body(StreamLog {
             body,
             time,
             format,
             size: 0,
-            })
         })))
     }
 }
@@ -306,7 +303,7 @@ use pin_project::{pin_project, pinned_drop};
 #[pin_project(PinnedDrop)]
 pub struct StreamLog<B> {
     #[pin]
-    body: ResponseBody<B>,
+    body: B,
     format: Option<Format>,
     size: usize,
     time: OffsetDateTime,
@@ -327,7 +324,13 @@ impl<B> PinnedDrop for StreamLog<B> {
     }
 }
 
-impl<B: MessageBody> MessageBody for StreamLog<B> {
+impl<B> MessageBody for StreamLog<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
+    type Error = Error;
+
     fn size(&self) -> BodySize {
         self.body.size()
     }
@@ -335,14 +338,17 @@ impl<B: MessageBody> MessageBody for StreamLog<B> {
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, Error>>> {
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
         let this = self.project();
-        match this.body.poll_next(cx) {
+
-            Poll::Ready(Some(Ok(chunk))) => {
+        // TODO: MSRV 1.51: poll_map_err
+        match ready!(this.body.poll_next(cx)) {
+            Some(Ok(chunk)) => {
                 *this.size += chunk.len();
                 Poll::Ready(Some(Ok(chunk)))
             }
-            val => val,
+            Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
+            None => Poll::Ready(None),
         }
     }
 }

src/responder.rs

@@ -264,7 +264,7 @@ pub(crate) mod tests {
         let resp = srv.call(req).await.unwrap();
         assert_eq!(resp.status(), StatusCode::OK);
         match resp.response().body() {
-            ResponseBody::Body(Body::Bytes(ref b)) => {
+            Body::Bytes(ref b) => {
                 let bytes = b.clone();
                 assert_eq!(bytes, Bytes::from_static(b"some"));
             }
@@ -277,16 +277,28 @@ pub(crate) mod tests {
         fn body(&self) -> &Body;
     }
 
+    impl BodyTest for Body {
+        fn bin_ref(&self) -> &[u8] {
+            match self {
+                Body::Bytes(ref bin) => &bin,
+                _ => unreachable!("bug in test impl"),
+            }
+        }
+        fn body(&self) -> &Body {
+            self
+        }
+    }
+
     impl BodyTest for ResponseBody<Body> {
         fn bin_ref(&self) -> &[u8] {
             match self {
                 ResponseBody::Body(ref b) => match b {
                     Body::Bytes(ref bin) => &bin,
-                    _ => panic!(),
+                    _ => unreachable!("bug in test impl"),
                 },
                 ResponseBody::Other(ref b) => match b {
                     Body::Bytes(ref bin) => &bin,
-                    _ => panic!(),
+                    _ => unreachable!("bug in test impl"),
                 },
             }
         }

src/response/builder.rs

@@ -310,16 +310,19 @@ impl HttpResponseBuilder {
     ///
     /// `HttpResponseBuilder` can not be used after this call.
     #[inline]
-    pub fn body<B: Into<Body>>(&mut self, body: B) -> HttpResponse {
+    pub fn body<B: Into<Body>>(&mut self, body: B) -> HttpResponse<Body> {
-        self.message_body(body.into())
+        match self.message_body(body.into()) {
+            Ok(res) => res,
+            Err(err) => HttpResponse::from_error(err),
+        }
     }
 
     /// Set a body and generate `Response`.
     ///
     /// `HttpResponseBuilder` can not be used after this call.
-    pub fn message_body<B>(&mut self, body: B) -> HttpResponse<B> {
+    pub fn message_body<B>(&mut self, body: B) -> Result<HttpResponse<B>, Error> {
         if let Some(err) = self.err.take() {
-            return HttpResponse::from_error(Error::from(err)).into_body();
+            return Err(err.into());
         }
 
         let res = self
@@ -336,12 +339,12 @@ impl HttpResponseBuilder {
             for cookie in jar.delta() {
                 match HeaderValue::from_str(&cookie.to_string()) {
                     Ok(val) => res.headers_mut().append(header::SET_COOKIE, val),
-                    Err(err) => return HttpResponse::from_error(Error::from(err)).into_body(),
+                    Err(err) => return Err(err.into()),
                 };
             }
         }
 
-        res
+        Ok(res)
     }
 
     /// Set a streaming body and generate `Response`.
@@ -422,7 +425,6 @@ impl Future for HttpResponseBuilder {
     type Output = Result<HttpResponse, Error>;
 
     fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
-        eprintln!("httpresponse future error");
         Poll::Ready(Ok(self.finish()))
     }
 }
@@ -478,42 +480,42 @@ mod tests {
 
     #[actix_rt::test]
     async fn test_json() {
-        let mut resp = HttpResponse::Ok().json(vec!["v1", "v2", "v3"]);
+        let resp = HttpResponse::Ok().json(vec!["v1", "v2", "v3"]);
         let ct = resp.headers().get(CONTENT_TYPE).unwrap();
         assert_eq!(ct, HeaderValue::from_static("application/json"));
         assert_eq!(
-            body::to_bytes(resp.take_body()).await.unwrap().as_ref(),
+            body::to_bytes(resp.into_body()).await.unwrap().as_ref(),
             br#"["v1","v2","v3"]"#
         );
 
-        let mut resp = HttpResponse::Ok().json(&["v1", "v2", "v3"]);
+        let resp = HttpResponse::Ok().json(&["v1", "v2", "v3"]);
         let ct = resp.headers().get(CONTENT_TYPE).unwrap();
         assert_eq!(ct, HeaderValue::from_static("application/json"));
         assert_eq!(
-            body::to_bytes(resp.take_body()).await.unwrap().as_ref(),
+            body::to_bytes(resp.into_body()).await.unwrap().as_ref(),
             br#"["v1","v2","v3"]"#
         );
 
         // content type override
-        let mut resp = HttpResponse::Ok()
+        let resp = HttpResponse::Ok()
             .insert_header((CONTENT_TYPE, "text/json"))
             .json(&vec!["v1", "v2", "v3"]);
         let ct = resp.headers().get(CONTENT_TYPE).unwrap();
         assert_eq!(ct, HeaderValue::from_static("text/json"));
         assert_eq!(
-            body::to_bytes(resp.take_body()).await.unwrap().as_ref(),
+            body::to_bytes(resp.into_body()).await.unwrap().as_ref(),
             br#"["v1","v2","v3"]"#
         );
     }
 
     #[actix_rt::test]
     async fn test_serde_json_in_body() {
-        let mut resp = HttpResponse::Ok().body(
+        let resp = HttpResponse::Ok().body(
             serde_json::to_vec(&serde_json::json!({ "test-key": "test-value" })).unwrap(),
         );
 
         assert_eq!(
-            body::to_bytes(resp.take_body()).await.unwrap().as_ref(),
+            body::to_bytes(resp.into_body()).await.unwrap().as_ref(),
             br#"{"test-key":"test-value"}"#
         );
     }

src/response/response.rs

@@ -8,7 +8,7 @@ use std::{
 };
 
 use actix_http::{
-    body::{Body, MessageBody, ResponseBody},
+    body::{Body, MessageBody},
     http::{header::HeaderMap, StatusCode},
     Extensions, Response, ResponseHead,
 };
@@ -27,7 +27,7 @@ use crate::{error::Error, HttpResponseBuilder};
 /// An HTTP Response
 pub struct HttpResponse<B = Body> {
     res: Response<B>,
-    error: Option<Error>,
+    pub(crate) error: Option<Error>,
 }
 
 impl HttpResponse<Body> {
@@ -56,14 +56,6 @@ impl HttpResponse<Body> {
             error: Some(error),
         }
     }
-
-    /// Convert response to response with body
-    pub fn into_body<B>(self) -> HttpResponse<B> {
-        HttpResponse {
-            res: self.res.into_body(),
-            error: self.error,
-        }
-    }
 }
 
 impl<B> HttpResponse<B> {
@@ -192,7 +184,7 @@ impl<B> HttpResponse<B> {
 
     /// Get body of this response
     #[inline]
-    pub fn body(&self) -> &ResponseBody<B> {
+    pub fn body(&self) -> &B {
         self.res.body()
     }
 
@@ -206,7 +198,7 @@ impl<B> HttpResponse<B> {
     }
 
     /// Split response and body
-    pub fn into_parts(self) -> (HttpResponse<()>, ResponseBody<B>) {
+    pub fn into_parts(self) -> (HttpResponse<()>, B) {
         let (head, body) = self.res.into_parts();
 
         (
@@ -229,7 +221,7 @@ impl<B> HttpResponse<B> {
     /// Set a body and return previous body value
     pub fn map_body<F, B2>(self, f: F) -> HttpResponse<B2>
     where
-        F: FnOnce(&mut ResponseHead, ResponseBody<B>) -> ResponseBody<B2>,
+        F: FnOnce(&mut ResponseHead, B) -> B2,
     {
         HttpResponse {
             res: self.res.map_body(f),
@@ -238,12 +230,16 @@ impl<B> HttpResponse<B> {
     }
 
     /// Extract response body
-    pub fn take_body(&mut self) -> ResponseBody<B> {
+    pub fn into_body(self) -> B {
-        self.res.take_body()
+        self.res.into_body()
     }
 }
 
-impl<B: MessageBody> fmt::Debug for HttpResponse<B> {
+impl<B> fmt::Debug for HttpResponse<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("HttpResponse")
             .field("error", &self.error)
@@ -270,20 +266,25 @@ impl<B> From<HttpResponse<B>> for Response<B> {
 
         // TODO: expose cause somewhere?
         // if let Some(err) = res.error {
-        //     eprintln!("impl<B> From<HttpResponse<B>> for Response<B> let Some(err)");
+        //     return Response::from_error(err);
-        //     return Response::from_error(err).into_body();
         // }
 
         res.res
     }
 }
 
-impl Future for HttpResponse {
+// Future is only implemented for Body payload type because it's the most useful for making simple
+// handlers without async blocks. Making it generic over all MessageBody types requires a future
+// impl on Response which would cause it's body field to be, undesirably, Option<B>.
+//
+// This impl is not particularly efficient due to the Response construction and should probably
+// not be invoked if performance is important. Prefer an async fn/block in such cases.
+impl Future for HttpResponse<Body> {
     type Output = Result<Response<Body>, Error>;
 
     fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
         if let Some(err) = self.error.take() {
-            return Poll::Ready(Ok(Response::from_error(err).into_body()));
+            return Poll::Ready(Err(err));
         }
 
         Poll::Ready(Ok(mem::replace(

src/scope.rs

@@ -578,7 +578,7 @@ mod tests {
     use actix_utils::future::ok;
     use bytes::Bytes;
 
-    use crate::dev::{Body, ResponseBody};
+    use crate::dev::Body;
     use crate::http::{header, HeaderValue, Method, StatusCode};
     use crate::middleware::DefaultHeaders;
     use crate::service::ServiceRequest;
@@ -748,7 +748,7 @@ mod tests {
         assert_eq!(resp.status(), StatusCode::OK);
 
         match resp.response().body() {
-            ResponseBody::Body(Body::Bytes(ref b)) => {
+            Body::Bytes(ref b) => {
                 let bytes = b.clone();
                 assert_eq!(bytes, Bytes::from_static(b"project: project1"));
             }
@@ -849,7 +849,7 @@ mod tests {
         assert_eq!(resp.status(), StatusCode::CREATED);
 
         match resp.response().body() {
-            ResponseBody::Body(Body::Bytes(ref b)) => {
+            Body::Bytes(ref b) => {
                 let bytes = b.clone();
                 assert_eq!(bytes, Bytes::from_static(b"project: project_1"));
             }
@@ -877,7 +877,7 @@ mod tests {
         assert_eq!(resp.status(), StatusCode::CREATED);
 
         match resp.response().body() {
-            ResponseBody::Body(Body::Bytes(ref b)) => {
+            Body::Bytes(ref b) => {
                 let bytes = b.clone();
                 assert_eq!(bytes, Bytes::from_static(b"project: test - 1"));
             }

src/server.rs

@@ -81,6 +81,7 @@ where
     S::Service: 'static,
     // S::Service: 'static,
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     /// Create new HTTP server with application factory
     pub fn new(factory: F) -> Self {
@@ -173,6 +174,16 @@ where
         self
     }
 
+    /// Set max number of threads for each worker's blocking task thread pool.
+    ///
+    /// One thread pool is set up **per worker**; not shared across workers.
+    ///
+    /// By default set to 512 / workers.
+    pub fn worker_max_blocking_threads(mut self, num: usize) -> Self {
+        self.builder = self.builder.worker_max_blocking_threads(num);
+        self
+    }
+
     /// Set server keep-alive setting.
     ///
     /// By default keep alive is set to a 5 seconds.

src/service.rs

@@ -2,7 +2,7 @@ use std::cell::{Ref, RefMut};
 use std::rc::Rc;
 use std::{fmt, net};
 
-use actix_http::body::{Body, MessageBody, ResponseBody};
+use actix_http::body::{Body, MessageBody};
 use actix_http::http::{HeaderMap, Method, StatusCode, Uri, Version};
 use actix_http::{
     Error, Extensions, HttpMessage, Payload, PayloadStream, RequestHead, Response, ResponseHead,
@@ -110,9 +110,9 @@ impl ServiceRequest {
 
     /// Create service response for error
     #[inline]
-    pub fn error_response<B, E: Into<Error>>(self, err: E) -> ServiceResponse<B> {
+    pub fn error_response<E: Into<Error>>(self, err: E) -> ServiceResponse {
         let res = HttpResponse::from_error(err.into());
-        ServiceResponse::new(self.req, res.into_body())
+        ServiceResponse::new(self.req, res)
     }
 
     /// This method returns reference to the request head
@@ -335,22 +335,24 @@ pub struct ServiceResponse<B = Body> {
     response: HttpResponse<B>,
 }
 
+impl ServiceResponse<Body> {
+    /// Create service response from the error
+    pub fn from_err<E: Into<Error>>(err: E, request: HttpRequest) -> Self {
+        let response = HttpResponse::from_error(err.into());
+        ServiceResponse { request, response }
+    }
+}
+
 impl<B> ServiceResponse<B> {
     /// Create service response instance
     pub fn new(request: HttpRequest, response: HttpResponse<B>) -> Self {
         ServiceResponse { request, response }
     }
 
-    /// Create service response from the error
-    pub fn from_err<E: Into<Error>>(err: E, request: HttpRequest) -> Self {
-        let response = HttpResponse::from_error(err.into()).into_body();
-        ServiceResponse { request, response }
-    }
-
     /// Create service response for error
     #[inline]
-    pub fn error_response<E: Into<Error>>(self, err: E) -> Self {
+    pub fn error_response<E: Into<Error>>(self, err: E) -> ServiceResponse {
-        Self::from_err(err, self.request)
+        ServiceResponse::from_err(err, self.request)
     }
 
     /// Create service response
@@ -396,23 +398,18 @@ impl<B> ServiceResponse<B> {
     }
 
     /// Execute closure and in case of error convert it to response.
-    pub fn checked_expr<F, E>(mut self, f: F) -> Self
+    pub fn checked_expr<F, E>(mut self, f: F) -> Result<Self, Error>
     where
         F: FnOnce(&mut Self) -> Result<(), E>,
         E: Into<Error>,
     {
-        match f(&mut self) {
+        f(&mut self).map_err(Into::into)?;
-            Ok(_) => self,
+        Ok(self)
-            Err(err) => {
-                let res = HttpResponse::from_error(err.into());
-                ServiceResponse::new(self.request, res.into_body())
-            }
-        }
     }
 
     /// Extract response body
-    pub fn take_body(&mut self) -> ResponseBody<B> {
+    pub fn into_body(self) -> B {
-        self.response.take_body()
+        self.response.into_body()
     }
 }
 
@@ -420,7 +417,7 @@ impl<B> ServiceResponse<B> {
     /// Set a new body
     pub fn map_body<F, B2>(self, f: F) -> ServiceResponse<B2>
     where
-        F: FnOnce(&mut ResponseHead, ResponseBody<B>) -> ResponseBody<B2>,
+        F: FnOnce(&mut ResponseHead, B) -> B2,
     {
         let response = self.response.map_body(f);
 
@@ -443,7 +440,11 @@ impl<B> From<ServiceResponse<B>> for Response<B> {
     }
 }
 
-impl<B: MessageBody> fmt::Debug for ServiceResponse<B> {
+impl<B> fmt::Debug for ServiceResponse<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let res = writeln!(
             f,

src/test.rs

@@ -4,13 +4,14 @@ use std::{net::SocketAddr, rc::Rc};
 
 pub use actix_http::test::TestBuffer;
 use actix_http::{
+    body,
     http::{header::IntoHeaderPair, Method, StatusCode, Uri, Version},
     test::TestRequest as HttpTestRequest,
     Extensions, Request,
 };
 use actix_router::{Path, ResourceDef, Url};
 use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
-use actix_utils::future::ok;
+use actix_utils::future::{ok, poll_fn};
 use futures_core::Stream;
 use futures_util::StreamExt as _;
 use serde::{de::DeserializeOwned, Serialize};
@@ -151,17 +152,19 @@ pub async fn read_response<S, B>(app: &S, req: Request) -> Bytes
 where
     S: Service<Request, Response = ServiceResponse<B>, Error = Error>,
     B: MessageBody + Unpin,
+    B::Error: Into<Error>,
 {
-    let mut resp = app
+    let resp = app
         .call(req)
         .await
         .unwrap_or_else(|e| panic!("read_response failed at application call: {}", e));
 
-    let mut body = resp.take_body();
+    let body = resp.into_body();
     let mut bytes = BytesMut::new();
 
-    while let Some(item) = body.next().await {
+    actix_rt::pin!(body);
-        bytes.extend_from_slice(&item.unwrap());
+    while let Some(item) = poll_fn(|cx| body.as_mut().poll_next(cx)).await {
+        bytes.extend_from_slice(&item.map_err(Into::into).unwrap());
     }
 
     bytes.freeze()
@@ -193,15 +196,19 @@ where
 ///     assert_eq!(result, Bytes::from_static(b"welcome!"));
 /// }
 /// ```
-pub async fn read_body<B>(mut res: ServiceResponse<B>) -> Bytes
+pub async fn read_body<B>(res: ServiceResponse<B>) -> Bytes
 where
     B: MessageBody + Unpin,
+    B::Error: Into<Error>,
 {
-    let mut body = res.take_body();
+    let body = res.into_body();
     let mut bytes = BytesMut::new();
-    while let Some(item) = body.next().await {
+
-        bytes.extend_from_slice(&item.unwrap());
+    actix_rt::pin!(body);
+    while let Some(item) = poll_fn(|cx| body.as_mut().poll_next(cx)).await {
+        bytes.extend_from_slice(&item.map_err(Into::into).unwrap());
     }
+
     bytes.freeze()
 }
 
@@ -245,6 +252,7 @@ where
 pub async fn read_body_json<T, B>(res: ServiceResponse<B>) -> T
 where
     B: MessageBody + Unpin,
+    B::Error: Into<Error>,
     T: DeserializeOwned,
 {
     let body = read_body(res).await;
@@ -268,6 +276,14 @@ where
     Ok(data.freeze())
 }
 
+pub async fn load_body<B>(body: B) -> Result<Bytes, Error>
+where
+    B: MessageBody + Unpin,
+    B::Error: Into<Error>,
+{
+    body::to_bytes(body).await.map_err(Into::into)
+}
+
 /// Helper function that returns a deserialized response body of a TestRequest
 ///
 /// ```
@@ -306,6 +322,7 @@ pub async fn read_response_json<S, B, T>(app: &S, req: Request) -> T
 where
     S: Service<Request, Response = ServiceResponse<B>, Error = Error>,
     B: MessageBody + Unpin,
+    B::Error: Into<Error>,
     T: DeserializeOwned,
 {
     let body = read_response(app, req).await;

src/types/json.rs

@@ -435,7 +435,7 @@ mod tests {
             header::{self, CONTENT_LENGTH, CONTENT_TYPE},
             StatusCode,
         },
-        test::{load_stream, TestRequest},
+        test::{load_body, TestRequest},
     };
 
     #[derive(Serialize, Deserialize, PartialEq, Debug)]
@@ -492,10 +492,10 @@ mod tests {
             .to_http_parts();
 
         let s = Json::<MyObject>::from_request(&req, &mut pl).await;
-        let mut resp = HttpResponse::from_error(s.err().unwrap());
+        let resp = HttpResponse::from_error(s.err().unwrap());
         assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
 
-        let body = load_stream(resp.take_body()).await.unwrap();
+        let body = load_body(resp.into_body()).await.unwrap();
         let msg: MyObject = serde_json::from_slice(&body).unwrap();
         assert_eq!(msg.name, "invalid request");
     }

tests/test_server.rs

@@ -32,7 +32,7 @@ use rand::{distributions::Alphanumeric, Rng};
 
 use actix_web::dev::BodyEncoding;
 use actix_web::middleware::{Compress, NormalizePath, TrailingSlash};
-use actix_web::{dev, web, App, Error, HttpResponse};
+use actix_web::{web, App, Error, HttpResponse};
 
 const STR: &str = "Hello World Hello World Hello World Hello World Hello World \
                    Hello World Hello World Hello World Hello World Hello World \
@@ -160,9 +160,7 @@ async fn test_body_gzip2() {
     let srv = actix_test::start_with(actix_test::config().h1(), || {
         App::new()
             .wrap(Compress::new(ContentEncoding::Gzip))
-            .service(web::resource("/").route(web::to(|| {
+            .service(web::resource("/").route(web::to(|| HttpResponse::Ok().body(STR))))
-                HttpResponse::Ok().body(STR).into_body::<dev::Body>()
-            })))
     });
 
     let mut response = srv
@@ -903,7 +901,7 @@ async fn test_normalize() {
     let srv = actix_test::start_with(actix_test::config().h1(), || {
         App::new()
             .wrap(NormalizePath::new(TrailingSlash::Trim))
-            .service(web::resource("/one").route(web::to(|| HttpResponse::Ok().finish())))
+            .service(web::resource("/one").route(web::to(|| HttpResponse::Ok())))
     });
 
     let response = srv.get("/one/").send().await.unwrap();