remove responsebody indirection from response (#2201)

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};
@@ -141,7 +141,8 @@ where
     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>
@@ -295,11 +296,11 @@ where
         io.poll_flush(cx)
     }
 
-    fn send_response(
+    fn send_response_inner(
         self: Pin<&mut Self>,
         message: Response<()>,
-        body: ResponseBody<B>,
-    ) -> Result<(), DispatchError> {
+        body: &impl MessageBody,
+    ) -> Result<BodySize, DispatchError> {
         let size = body.size();
         let mut this = self.project();
         this.codec
@@ -312,10 +313,35 @@ where
             })?;
 
         this.flags.set(Flags::KEEPALIVE, this.codec.keepalive());
-        match size {
-            BodySize::None | BodySize::Empty => this.state.set(State::None),
-            _ => this.state.set(State::SendPayload(body)),
+
+        Ok(size)
+    }
+
+    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(())
     }
 
@@ -353,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()
-                            .send_response(res, ResponseBody::Other(Body::Empty))?;
+                        self.as_mut().send_error_response(res, Body::Empty)?;
                     }
 
                     // return with upgrade request and poll it exclusively.
@@ -376,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.
@@ -392,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 {
@@ -433,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),
                 },
@@ -486,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);
                         }
                     }
                 }
@@ -506,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)
                         }
                     };
                 }
@@ -626,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)
-                            .drop_body(),
+                        Response::with_body(
+                            StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE,
+                            (),
+                        ),
                     ));
                     this.flags.insert(Flags::READ_DISCONNECT);
                     *this.error = Some(ParseError::TooLarge.into());
@@ -706,10 +770,9 @@ where
                             } else {
                                 // timeout on first request (slow request) return 408
                                 trace!("Slow request timeout");
-                                let _ = self.as_mut().send_response(
-                                    Response::new(StatusCode::REQUEST_TIMEOUT)
-                                        .drop_body(),
-                                    ResponseBody::Other(Body::Empty),
+                                let _ = self.as_mut().send_error_response(
+                                    Response::with_body(StatusCode::REQUEST_TIMEOUT, ()),
+                                    Body::Empty,
                                 );
                                 this = self.project();
                                 this.flags.insert(Flags::STARTED | Flags::SHUTDOWN);

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<()> =
-            Response::new(StatusCode::SWITCHING_PROTOCOLS).into_body::<()>();
+        let mut res = Response::with_body(StatusCode::SWITCHING_PROTOCOLS, ());
         res.headers_mut().insert(DATE, HeaderValue::from_static(""));
         res.headers_mut()
             .insert(CONTENT_LENGTH, HeaderValue::from_static("0"));

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>>,
 }
@@ -62,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();