berkus
/
actix-web
mirror of https://github.com/fafhrd91/actix-web
1
0
Fork 0
Code Issues Releases Wiki Activity

remove responsebody indirection from response

This commit is contained in:
Rob Ede 2021-05-07 10:41:34 +01:00
parent bc322c95a6
commit b3709d1cb1
No known key found for this signature in database
GPG Key ID: 97C636207D3EF933
14 changed files with 291 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
actix-files/src/service.rs
View File

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

85
actix-http/src/h1/dispatcher.rs
View File

@ -17,7 +17,7 @@ use futures_core::ready;
use log::{error, trace}; use log::{error, trace};
use pin_project::pin_project; use pin_project::pin_project;
use crate::body::{Body, BodySize, MessageBody, ResponseBody}; use crate::body::{Body, BodySize, MessageBody};
use crate::config::ServiceConfig; use crate::config::ServiceConfig;
use crate::error::{DispatchError, Error}; use crate::error::{DispatchError, Error};
use crate::error::{ParseError, PayloadError}; use crate::error::{ParseError, PayloadError};
@ -141,7 +141,8 @@ where
None, None,
ExpectCall(#[pin] X::Future), ExpectCall(#[pin] X::Future),
ServiceCall(#[pin] S::Future), ServiceCall(#[pin] S::Future),
SendPayload(#[pin] ResponseBody<B>), SendPayload(#[pin] B),
SendErrorPayload(#[pin] Body),
} }
impl<S, B, X> State<S, B, X> impl<S, B, X> State<S, B, X>
@ -298,7 +299,7 @@ where
fn send_response( fn send_response(
self: Pin<&mut Self>, self: Pin<&mut Self>,
message: Response<()>, message: Response<()>,
body: ResponseBody<B>, body: B,
) -> Result<(), DispatchError> { ) -> Result<(), DispatchError> {
let size = body.size(); let size = body.size();
let mut this = self.project(); let mut this = self.project();
@ -319,6 +320,32 @@ where
Ok(()) Ok(())
} }
fn send_response_any_body(
self: Pin<&mut Self>,
message: Response<()>,
body: Body,
) -> Result<(), DispatchError> {
// TODO: de-dupe impl with send_response
let size = body.size();
let mut this = self.project();
this.codec
.encode(Message::Item((message, size)), &mut this.write_buf)
.map_err(|err| {
if let Some(mut payload) = this.payload.take() {
payload.set_error(PayloadError::Incomplete(None));
}
DispatchError::Io(err)
})?;
this.flags.set(Flags::KEEPALIVE, this.codec.keepalive());
match size {
BodySize::None | BodySize::Empty => this.state.set(State::None),
_ => this.state.set(State::SendErrorPayload(body)),
};
Ok(())
}
fn send_continue(self: Pin<&mut Self>) { fn send_continue(self: Pin<&mut Self>) {
self.project() self.project()
.write_buf .write_buf
@ -353,8 +380,7 @@ where
// send_response would update InnerDispatcher state to SendPayload or // send_response would update InnerDispatcher state to SendPayload or
// None(If response body is empty). // None(If response body is empty).
// continue loop to poll it. // continue loop to poll it.
self.as_mut() self.as_mut().send_response_any_body(res, Body::Empty)?;
.send_response(res, ResponseBody::Other(Body::Empty))?;
} }
// return with upgrade request and poll it exclusively. // return with upgrade request and poll it exclusively.
@ -376,7 +402,7 @@ where
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
let res = Response::from_error(err.into()); let res = Response::from_error(err.into());
let (res, body) = res.replace_body(()); let (res, body) = res.replace_body(());
self.as_mut().send_response(res, body.into_body())?; self.as_mut().send_response_any_body(res, body)?;
} }
// service call pending and could be waiting for more chunk messages. // service call pending and could be waiting for more chunk messages.
@ -392,6 +418,41 @@ where
}, },
StateProj::SendPayload(mut stream) => { 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, // keep populate writer buffer until buffer size limit hit,
// get blocked or finished. // get blocked or finished.
while this.write_buf.len() < super::payload::MAX_BUFFER_SIZE { while this.write_buf.len() < super::payload::MAX_BUFFER_SIZE {
@ -433,12 +494,14 @@ where
let fut = this.flow.service.call(req); let fut = this.flow.service.call(req);
this.state.set(State::ServiceCall(fut)); this.state.set(State::ServiceCall(fut));
} }
// send expect error as response // send expect error as response
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
let res = Response::from_error(err.into()); let res = Response::from_error(err.into());
let (res, body) = res.replace_body(()); let (res, body) = res.replace_body(());
self.as_mut().send_response(res, body.into_body())?; self.as_mut().send_response_any_body(res, body)?;
} }
// expect must be solved before progress can be made. // expect must be solved before progress can be made.
Poll::Pending => return Ok(PollResponse::DoNothing), Poll::Pending => return Ok(PollResponse::DoNothing),
}, },
@ -486,7 +549,7 @@ where
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
let res = Response::from_error(err.into()); let res = Response::from_error(err.into());
let (res, body) = res.replace_body(()); let (res, body) = res.replace_body(());
return self.send_response(res, body.into_body()); return self.send_response_any_body(res, body);
} }
} }
} }
@ -506,7 +569,7 @@ where
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
let res = Response::from_error(err.into()); let res = Response::from_error(err.into());
let (res, body) = res.replace_body(()); let (res, body) = res.replace_body(());
self.send_response(res, body.into_body()) self.send_response_any_body(res, body)
} }
}; };
} }
@ -706,10 +769,10 @@ where
} else { } else {
// timeout on first request (slow request) return 408 // timeout on first request (slow request) return 408
trace!("Slow request timeout"); trace!("Slow request timeout");
let _ = self.as_mut().send_response( let _ = self.as_mut().send_response_any_body(
Response::new(StatusCode::REQUEST_TIMEOUT) Response::new(StatusCode::REQUEST_TIMEOUT)
.drop_body(), .drop_body(),
ResponseBody::Other(Body::Empty), Body::Empty,
); );
this = self.project(); this = self.project();
this.flags.insert(Flags::STARTED | Flags::SHUTDOWN); this.flags.insert(Flags::STARTED | Flags::SHUTDOWN);

17
actix-http/src/h1/utils.rs
View File

@ -4,7 +4,7 @@ use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Framed}; use actix_codec::{AsyncRead, AsyncWrite, Framed};
use crate::body::{BodySize, MessageBody, ResponseBody}; use crate::body::{BodySize, MessageBody};
use crate::error::Error; use crate::error::Error;
use crate::h1::{Codec, Message}; use crate::h1::{Codec, Message};
use crate::response::Response; use crate::response::Response;
@ -14,7 +14,7 @@ use crate::response::Response;
pub struct SendResponse<T, B> { pub struct SendResponse<T, B> {
res: Option<Message<(Response<()>, BodySize)>>, res: Option<Message<(Response<()>, BodySize)>>,
#[pin] #[pin]
body: Option<ResponseBody<B>>, body: Option<B>,
#[pin] #[pin]
framed: Option<Framed<T, Codec>>, framed: Option<Framed<T, Codec>>,
} }
@ -62,7 +62,18 @@ where
.unwrap() .unwrap()
.is_write_buf_full() .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) => { Poll::Ready(item) => {
// body is done when item is None // body is done when item is None
body_done = item.is_none(); body_done = item.is_none();

69
actix-http/src/h2/dispatcher.rs
View File

@ -12,7 +12,7 @@ use h2::{
use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING}; use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING};
use log::{error, trace}; use log::{error, trace};
use crate::body::{BodySize, MessageBody, ResponseBody}; use crate::body::{Body, BodySize, MessageBody};
use crate::config::ServiceConfig; use crate::config::ServiceConfig;
use crate::error::{DispatchError, Error}; use crate::error::{DispatchError, Error};
use crate::message::ResponseHead; use crate::message::ResponseHead;
@ -135,7 +135,8 @@ struct ServiceResponse<F, I, E, B> {
#[pin_project::pin_project(project = ServiceResponseStateProj)] #[pin_project::pin_project(project = ServiceResponseStateProj)]
enum ServiceResponseState<F, B> { enum ServiceResponseState<F, B> {
ServiceCall(#[pin] F, Option<SendResponse<Bytes>>), 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> impl<F, I, E, B> ServiceResponse<F, I, E, B>
@ -280,9 +281,8 @@ where
if size.is_eof() { if size.is_eof() {
Poll::Ready(()) Poll::Ready(())
} else { } else {
this.state.set(ServiceResponseState::SendPayload( this.state.set(ServiceResponseState::SendErrorPayload(
stream, stream, body,
body.into_body(),
)); ));
self.poll(cx) self.poll(cx)
} }
@ -331,8 +331,65 @@ where
*this.buffer = Some(chunk); *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)) => { 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(()); return Poll::Ready(());
} }
}, },

6
actix-http/src/message.rs
View File

@ -389,12 +389,6 @@ impl BoxedResponseHead {
pub fn new(status: StatusCode) -> Self { pub fn new(status: StatusCode) -> Self {
RESPONSE_POOL.with(|p| p.get_message(status)) 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 { impl std::ops::Deref for BoxedResponseHead {

84
actix-http/src/response.rs
View File

@ -2,17 +2,13 @@
use std::{ use std::{
cell::{Ref, RefMut}, cell::{Ref, RefMut},
fmt, fmt, str,
future::Future,
pin::Pin,
str,
task::{Context, Poll},
}; };
use bytes::{Bytes, BytesMut}; use bytes::{Bytes, BytesMut};
use crate::{ use crate::{
body::{Body, MessageBody, ResponseBody}, body::{Body, MessageBody},
error::Error, error::Error,
extensions::Extensions, extensions::Extensions,
http::{HeaderMap, StatusCode}, http::{HeaderMap, StatusCode},
@ -23,7 +19,7 @@ use crate::{
/// An HTTP response. /// An HTTP response.
pub struct Response<B> { pub struct Response<B> {
pub(crate) head: BoxedResponseHead, pub(crate) head: BoxedResponseHead,
pub(crate) body: ResponseBody<B>, pub(crate) body: B,
pub(crate) error: Option<Error>, pub(crate) error: Option<Error>,
} }
@ -33,7 +29,7 @@ impl Response<Body> {
pub fn new(status: StatusCode) -> Response<Body> { pub fn new(status: StatusCode) -> Response<Body> {
Response { Response {
head: BoxedResponseHead::new(status), head: BoxedResponseHead::new(status),
body: ResponseBody::Body(Body::Empty), body: Body::Empty,
error: None, error: None,
} }
} }
@ -84,18 +80,18 @@ impl Response<Body> {
resp resp
} }
/// Convert response to response with body // /// Convert response to response with body
pub fn into_body<B>(self) -> Response<B> { // pub fn into_body<B>(self) -> Response<B> {
let b = match self.body { // let b = match self.body {
ResponseBody::Body(b) => b, // ResponseBody::Body(b) => b,
ResponseBody::Other(b) => b, // ResponseBody::Other(b) => b,
}; // };
Response { // Response {
head: self.head, // head: self.head,
error: self.error, // error: self.error,
body: ResponseBody::Other(b), // body: ResponseBody::Other(b),
} // }
} // }
} }
impl<B> Response<B> { impl<B> Response<B> {
@ -104,7 +100,7 @@ impl<B> Response<B> {
pub fn with_body(status: StatusCode, body: B) -> Response<B> { pub fn with_body(status: StatusCode, body: B) -> Response<B> {
Response { Response {
head: BoxedResponseHead::new(status), head: BoxedResponseHead::new(status),
body: ResponseBody::Body(body), body,
error: None, error: None,
} }
} }
@ -176,7 +172,7 @@ impl<B> Response<B> {
/// Get body of this response /// Get body of this response
#[inline] #[inline]
pub fn body(&self) -> &ResponseBody<B> { pub fn body(&self) -> &B {
&self.body &self.body
} }
@ -184,17 +180,17 @@ impl<B> Response<B> {
pub fn set_body<B2>(self, body: B2) -> Response<B2> { pub fn set_body<B2>(self, body: B2) -> Response<B2> {
Response { Response {
head: self.head, head: self.head,
body: ResponseBody::Body(body), body,
error: None, error: None,
} }
} }
/// Split response and body /// Split response and body
pub fn into_parts(self) -> (Response<()>, ResponseBody<B>) { pub fn into_parts(self) -> (Response<()>, B) {
( (
Response { Response {
head: self.head, head: self.head,
body: ResponseBody::Body(()), body: (),
error: self.error, error: self.error,
}, },
self.body, self.body,
@ -205,17 +201,17 @@ impl<B> Response<B> {
pub fn drop_body(self) -> Response<()> { pub fn drop_body(self) -> Response<()> {
Response { Response {
head: self.head, head: self.head,
body: ResponseBody::Body(()), body: (),
error: None, error: None,
} }
} }
/// Set a body and return previous body value /// Set a body and return 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 { Response {
head: self.head, head: self.head,
body: ResponseBody::Body(body), body,
error: self.error, error: self.error,
}, },
self.body, self.body,
@ -225,7 +221,7 @@ impl<B> Response<B> {
/// Set a body and return previous body value /// Set a body and return previous body value
pub fn map_body<F, B2>(mut self, f: F) -> Response<B2> pub fn map_body<F, B2>(mut self, f: F) -> Response<B2>
where where
F: FnOnce(&mut ResponseHead, ResponseBody<B>) -> ResponseBody<B2>, F: FnOnce(&mut ResponseHead, B) -> B2,
{ {
let body = f(&mut self.head, self.body); let body = f(&mut self.head, self.body);
@ -236,9 +232,14 @@ impl<B> Response<B> {
} }
} }
// /// Extract response body
// pub fn take_body(&mut self) -> ResponseBody<B> {
// self.body.take_body()
// }
/// Extract response body /// Extract response body
pub fn take_body(&mut self) -> ResponseBody<B> { pub fn into_body(self) -> B {
self.body.take_body() self.body
} }
} }
@ -264,17 +265,18 @@ where
} }
} }
impl<B: Unpin> Future for Response<B> { // TODO: document why this is needed
type Output = Result<Response<B>, Error>; // impl<B: Unpin> Future for Response<B> {
// type Output = Result<Response<B>, Infallible>;
fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> { // fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
Poll::Ready(Ok(Response { // Poll::Ready(Ok(Response {
head: self.head.take(), // head: self.head.take(),
body: self.body.take_body(), // body: self.body.take_body(),
error: self.error.take(), // error: self.error.take(),
})) // }))
} // }
} // }
/// Helper converters /// Helper converters
impl<I: Into<Response<Body>>, E: Into<Error>> From<Result<I, E>> for Response<Body> { impl<I: Into<Response<Body>>, E: Into<Error>> From<Result<I, E>> for Response<Body> {

11
actix-http/src/response_builder.rs
View File

@ -13,7 +13,7 @@ use bytes::Bytes;
use futures_core::Stream; use futures_core::Stream;
use crate::{ use crate::{
body::{Body, BodyStream, ResponseBody}, body::{Body, BodyStream},
error::{Error, HttpError}, error::{Error, HttpError},
header::{self, IntoHeaderPair, IntoHeaderValue}, header::{self, IntoHeaderPair, IntoHeaderValue},
message::{BoxedResponseHead, ConnectionType, ResponseHead}, message::{BoxedResponseHead, ConnectionType, ResponseHead},
@ -242,15 +242,16 @@ impl ResponseBuilder {
/// ///
/// This `ResponseBuilder` will be left in a useless state. /// 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) -> Response<B> {
if let Some(e) = self.err.take() { // TODO: put error handling back somehow
return Response::from(Error::from(e)).into_body(); // if let Some(e) = self.err.take() {
} // return Response::from(Error::from(e)).into_body();
// }
let response = self.head.take().expect("cannot reuse response builder"); let response = self.head.take().expect("cannot reuse response builder");
Response { Response {
head: response, head: response,
body: ResponseBody::Body(body), body,
error: None, error: None,
} }
} }

7
src/middleware/compat.rs
View File

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

14
src/middleware/compress.rs
View File

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

17
src/middleware/logger.rs
View File

@ -21,7 +21,7 @@ use regex::{Regex, RegexSet};
use time::OffsetDateTime; use time::OffsetDateTime;
use crate::{ use crate::{
dev::{BodySize, MessageBody, ResponseBody}, dev::{BodySize, MessageBody},
http::{HeaderName, StatusCode}, http::{HeaderName, StatusCode},
service::{ServiceRequest, ServiceResponse}, service::{ServiceRequest, ServiceResponse},
Error, HttpResponse, Result, Error, HttpResponse, Result,
@ -289,13 +289,11 @@ where
let time = *this.time; let time = *this.time;
let format = this.format.take(); 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, body,
time, time,
format, format,
size: 0, size: 0,
})
}))) })))
} }
} }
@ -305,7 +303,7 @@ use pin_project::{pin_project, pinned_drop};
#[pin_project(PinnedDrop)] #[pin_project(PinnedDrop)]
pub struct StreamLog<B> { pub struct StreamLog<B> {
#[pin] #[pin]
body: ResponseBody<B>, body: B,
format: Option<Format>, format: Option<Format>,
size: usize, size: usize,
time: OffsetDateTime, time: OffsetDateTime,
@ -342,12 +340,15 @@ where
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Self::Error>>> { ) -> Poll<Option<Result<Bytes, Self::Error>>> {
let this = self.project(); 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(); *this.size += chunk.len();
Poll::Ready(Some(Ok(chunk))) Poll::Ready(Some(Ok(chunk)))
} }
val => val, Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
None => Poll::Ready(None),
} }
} }
} }

9
src/response/builder.rs
View File

@ -318,9 +318,9 @@ impl HttpResponseBuilder {
/// ///
/// `HttpResponseBuilder` can not be used after this call. /// `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) -> HttpResponse<B> {
if let Some(err) = self.err.take() { // if let Some(err) = self.err.take() {
return HttpResponse::from_error(Error::from(err)).into_body(); // return HttpResponse::from_error(Error::from(err)).into_body();
} // }
let res = self let res = self
.res .res
@ -336,7 +336,8 @@ impl HttpResponseBuilder {
for cookie in jar.delta() { for cookie in jar.delta() {
match HeaderValue::from_str(&cookie.to_string()) { match HeaderValue::from_str(&cookie.to_string()) {
Ok(val) => res.headers_mut().append(header::SET_COOKIE, val), Ok(val) => res.headers_mut().append(header::SET_COOKIE, val),
Err(err) => return HttpResponse::from_error(Error::from(err)).into_body(), Err(err) => res.error = Some(err.into()),
// Err(err) => return HttpResponse::from_error(Error::from(err)).into_body(),
}; };
} }
} }

39
src/response/response.rs
View File

@ -8,7 +8,7 @@ use std::{
}; };
use actix_http::{ use actix_http::{
body::{Body, MessageBody, ResponseBody}, body::{Body, MessageBody},
http::{header::HeaderMap, StatusCode}, http::{header::HeaderMap, StatusCode},
Extensions, Response, ResponseHead, Extensions, Response, ResponseHead,
}; };
@ -27,7 +27,7 @@ use crate::{error::Error, HttpResponseBuilder};
/// An HTTP Response /// An HTTP Response
pub struct HttpResponse<B = Body> { pub struct HttpResponse<B = Body> {
res: Response<B>, res: Response<B>,
error: Option<Error>, pub(crate) error: Option<Error>,
} }
impl HttpResponse<Body> { impl HttpResponse<Body> {
@ -57,13 +57,13 @@ impl HttpResponse<Body> {
} }
} }
/// Convert response to response with body // /// Convert response to response with body
pub fn into_body<B>(self) -> HttpResponse<B> { // pub fn into_body<B>(self) -> HttpResponse<B> {
HttpResponse { // HttpResponse {
res: self.res.into_body(), // res: self.res.into_body(),
error: self.error, // error: self.error,
} // }
} // }
} }
impl<B> HttpResponse<B> { impl<B> HttpResponse<B> {
@ -192,7 +192,7 @@ impl<B> HttpResponse<B> {
/// Get body of this response /// Get body of this response
#[inline] #[inline]
pub fn body(&self) -> &ResponseBody<B> { pub fn body(&self) -> &B {
self.res.body() self.res.body()
} }
@ -206,7 +206,7 @@ impl<B> HttpResponse<B> {
} }
/// Split response and body /// 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(); let (head, body) = self.res.into_parts();
( (
@ -229,7 +229,7 @@ impl<B> HttpResponse<B> {
/// Set a body and return previous body value /// Set a body and return previous body value
pub fn map_body<F, B2>(self, f: F) -> HttpResponse<B2> pub fn map_body<F, B2>(self, f: F) -> HttpResponse<B2>
where where
F: FnOnce(&mut ResponseHead, ResponseBody<B>) -> ResponseBody<B2>, F: FnOnce(&mut ResponseHead, B) -> B2,
{ {
HttpResponse { HttpResponse {
res: self.res.map_body(f), res: self.res.map_body(f),
@ -237,9 +237,14 @@ impl<B> HttpResponse<B> {
} }
} }
// /// Extract response body
// pub fn take_body(&mut self) -> ResponseBody<B> {
// self.res.take_body()
// }
/// Extract response body /// 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()
} }
} }
@ -286,9 +291,9 @@ impl Future for HttpResponse {
type Output = Result<Response<Body>, Error>; type Output = Result<Response<Body>, Error>;
fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
if let Some(err) = self.error.take() { // if let Some(err) = self.error.take() {
return Poll::Ready(Ok(Response::from_error(err).into_body())); // return Poll::Ready(Ok(Response::from_error(err).into_body()));
} // }
Poll::Ready(Ok(mem::replace( Poll::Ready(Ok(mem::replace(
&mut self.res, &mut self.res,

44
src/service.rs
View File

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

22
src/test.rs
View File

@ -10,7 +10,7 @@ use actix_http::{
}; };
use actix_router::{Path, ResourceDef, Url}; use actix_router::{Path, ResourceDef, Url};
use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory}; 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_core::Stream;
use futures_util::StreamExt as _; use futures_util::StreamExt as _;
use serde::{de::DeserializeOwned, Serialize}; use serde::{de::DeserializeOwned, Serialize};
@ -153,16 +153,17 @@ where
B: MessageBody + Unpin, B: MessageBody + Unpin,
B::Error: Into<Error>, B::Error: Into<Error>,
{ {
let mut resp = app let resp = app
.call(req) .call(req)
.await .await
.unwrap_or_else(|e| panic!("read_response failed at application call: {}", e)); .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(); 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() bytes.freeze()
@ -194,16 +195,19 @@ where
/// assert_eq!(result, Bytes::from_static(b"welcome!")); /// 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 where
B: MessageBody + Unpin, B: MessageBody + Unpin,
B::Error: Into<Error>, B::Error: Into<Error>,
{ {
let mut body = res.take_body(); let body = res.into_body();
let mut bytes = BytesMut::new(); 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() bytes.freeze()
} }