actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2024-11-23 16:21:06 +01:00
Code Issues Releases Wiki Activity

body ergo v4 using any body

Browse Source
This commit is contained in:
Rob Ede 2021-12-08 22:45:54 +00:00
parent fb5b4734a4
commit f12f62ba73
No known key found for this signature in database
GPG Key ID: 97C636207D3EF933
25 changed files with 427 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.toml
View File

@ -118,6 +118,7 @@ futures-util = { version = "0.3.7", default-features = false, features = ["std"]
rand = "0.8"
rcgen = "0.8"
rustls-pemfile = "0.2"
static_assertions = "1"
tls-openssl = { package = "openssl", version = "0.10.9" }
tls-rustls = { package = "rustls", version = "0.20.0" }
zstd = "0.9"

6
actix-http/src/body/message_body.rs
View File

@ -19,7 +19,7 @@ use super::BodySize;
pub trait MessageBody {
// TODO: consider this bound to only fmt::Display since the error type is not really used
// and there is an impl for Into<Box<StdError>> on String
type Error: Into<Box<dyn StdError>>;
type Error: Into<Box<dyn StdError>> + 'static;
/// Body size hint.
fn size(&self) -> BodySize;
@ -272,7 +272,7 @@ impl<B, F, E> MessageBody for MessageBodyMapErr<B, F>
where
B: MessageBody,
F: FnOnce(B::Error) -> E,
E: Into<Box<dyn StdError>>,
E: Into<Box<dyn StdError>> + 'static,
{
type Error = E;
@ -306,6 +306,8 @@ mod tests {
use super::*;
// static_assertions::assert_obj_safe!(MessageBody<()>);
macro_rules! assert_poll_next {
($pin:expr, $exp:expr) => {
assert_eq!(

2
actix-http/src/body/mod.rs
View File

@ -1,5 +1,6 @@
//! Traits and structures to aid consuming and writing HTTP payloads.
// mod any;
mod body_stream;
mod boxed;
mod either;
@ -9,6 +10,7 @@ mod size;
mod sized_stream;
mod utils;
// pub use self::any::AnyBody;
pub use self::body_stream::BodyStream;
pub use self::boxed::BoxBody;
pub use self::either::EitherBody;

15
actix-http/src/encoding/encoder.rs
View File

@ -53,6 +53,15 @@ impl<B: MessageBody> Encoder<B> {
}
}
pub fn not_acceptable(body: Bytes) -> Self {
Encoder {
body: EncoderBody::Bytes { body },
encoder: None,
fut: None,
eof: false,
}
}
pub fn response(encoding: ContentEncoding, head: &mut ResponseHead, body: B) -> Self {
let can_encode = !(head.headers().contains_key(&CONTENT_ENCODING)
|| head.status == StatusCode::SWITCHING_PROTOCOLS
@ -99,6 +108,7 @@ pin_project! {
#[project = EncoderBodyProj]
enum EncoderBody<B> {
None,
Bytes { body: Bytes },
Stream { #[pin] body: B },
}
}
@ -112,6 +122,7 @@ where
fn size(&self) -> BodySize {
match self {
EncoderBody::None => BodySize::None,
EncoderBody::Bytes { body } => body.size(),
EncoderBody::Stream { body } => body.size(),
}
}
@ -122,7 +133,9 @@ where
) -> Poll<Option<Result<Bytes, Self::Error>>> {
match self.project() {
EncoderBodyProj::None => Poll::Ready(None),
EncoderBodyProj::Bytes { body } => {
Pin::new(body).poll_next(cx).map_err(|err| match err {})
}
EncoderBodyProj::Stream { body } => body
.poll_next(cx)
.map_err(|err| EncoderError::Body(err.into())),

2
actix-http/src/encoding/mod.rs
View File

@ -8,7 +8,7 @@ mod decoder;
mod encoder;
pub use self::decoder::Decoder;
pub use self::encoder::Encoder;
pub use self::encoder::{Encoder, EncoderError};
/// Special-purpose writer for streaming (de-)compression.
///

2
actix-http/src/error.rs
View File

@ -9,6 +9,8 @@ use crate::{body::BoxBody, ws, Response};
pub use http::Error as HttpError;
pub use crate::encoding::EncoderError;
pub struct Error {
inner: Box<ErrorInner>,
}

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

@ -2,7 +2,7 @@
use std::{
cell::{Ref, RefMut},
fmt, str,
fmt, mem, str,
};
use bytes::{Bytes, BytesMut};
@ -203,6 +203,12 @@ impl<B> Response<B> {
}
}
impl<B: Default> Response<B> {
pub fn take_body(&mut self) -> B {
mem::take(&mut self.body)
}
}
impl<B> fmt::Debug for Response<B>
where
B: MessageBody,

72
actix-test/src/lib.rs
View File

@ -163,9 +163,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -179,9 +181,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -195,9 +199,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -214,9 +220,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -230,9 +238,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -246,9 +256,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -265,9 +277,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -281,9 +295,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)
@ -297,9 +313,11 @@ where
local_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> =
err.into().error_response().into();
res.map_into_boxed_body()
});
HttpService::build()
.client_timeout(timeout)

146
src/any_body.rs Normal file
View File

@ -0,0 +1,146 @@
use std::{
error::Error as StdError,
mem,
pin::Pin,
task::{Context, Poll},
};
use actix_http::body::{BodySize, BoxBody, MessageBody};
use bytes::Bytes;
use pin_project_lite::pin_project;
use crate::Error;
pin_project! {
#[derive(Debug)]
#[project = AnyBodyProj]
pub enum AnyBody<B = BoxBody> {
None,
Full { body: Bytes },
Stream { #[pin] body: B },
Boxed { body: BoxBody },
}
}
impl<B: MessageBody + 'static> AnyBody<B> {
pub fn into_body<B1>(self) -> AnyBody<B1> {
match self {
AnyBody::None => AnyBody::None,
AnyBody::Full { body } => AnyBody::Full { body },
AnyBody::Stream { body } => AnyBody::Boxed {
body: BoxBody::new(body),
},
AnyBody::Boxed { body } => AnyBody::Boxed { body },
}
}
}
impl<B> Default for AnyBody<B> {
fn default() -> Self {
Self::Full { body: Bytes::new() }
}
}
impl<B> MessageBody for AnyBody<B>
where
B: MessageBody,
B::Error: 'static,
{
type Error = Box<dyn StdError>;
fn size(&self) -> BodySize {
match self {
Self::None => BodySize::None,
Self::Full { body } => body.size(),
Self::Stream { body } => body.size(),
Self::Boxed { body } => body.size(),
}
}
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
match self.project() {
AnyBodyProj::None => Poll::Ready(None),
AnyBodyProj::Full { body } => {
let bytes = mem::take(body);
Poll::Ready(Some(Ok(bytes)))
}
AnyBodyProj::Stream { body } => body.poll_next(cx).map_err(|err| err.into()),
AnyBodyProj::Boxed { body } => body.as_pin_mut().poll_next(cx),
}
}
}
pin_project! {
#[project = EitherAnyBodyProj]
#[derive(Debug)]
pub enum EitherAnyBody<L, R = BoxBody> {
/// A body of type `L`.
Left { #[pin] body: AnyBody<L> },
/// A body of type `R`.
Right { #[pin] body: AnyBody<R> },
}
}
// impl<L> EitherAnyBody<L, BoxBody> {
// /// Creates new `EitherBody` using left variant and boxed right variant.
// pub fn new(body: L) -> Self {
// Self::Left {
// body: AnyBody::Stream { body },
// }
// }
// }
// impl<L, R> EitherAnyBody<L, R> {
// /// Creates new `EitherBody` using left variant.
// pub fn left(body: L) -> Self {
// Self::Left {
// body: AnyBody::Stream { body },
// }
// }
// /// Creates new `EitherBody` using right variant.
// pub fn right(body: R) -> Self {
// Self::Right {
// body: AnyBody::Stream { body },
// }
// }
// }
impl<L, R> MessageBody for EitherAnyBody<L, R>
where
L: MessageBody + 'static,
R: MessageBody + 'static,
{
type Error = Error;
fn size(&self) -> BodySize {
match self {
Self::Left { body } => body.size(),
Self::Right { body } => body.size(),
}
}
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
match self.project() {
EitherAnyBodyProj::Left { body } => body.poll_next(cx).map_err(Error::from),
EitherAnyBodyProj::Right { body } => body.poll_next(cx).map_err(Error::from),
}
}
}
#[cfg(test)]
mod tests {
use static_assertions::assert_eq_size;
use super::*;
assert_eq_size!(AnyBody<()>, [u8; 40]);
assert_eq_size!(AnyBody<u64>, [u8; 40]); // how is this the same size as ()
}

38
src/dev.rs
View File

@ -102,41 +102,3 @@ impl<B> BodyEncoding for crate::HttpResponse<B> {
self
}
}
// TODO: remove this if it doesn't appear to be needed
#[allow(dead_code)]
#[derive(Debug)]
pub(crate) enum AnyBody {
None,
Full { body: crate::web::Bytes },
Boxed { body: actix_http::body::BoxBody },
}
impl crate::body::MessageBody for AnyBody {
type Error = crate::BoxError;
/// Body size hint.
fn size(&self) -> crate::body::BodySize {
match self {
AnyBody::None => crate::body::BodySize::None,
AnyBody::Full { body } => body.size(),
AnyBody::Boxed { body } => body.size(),
}
}
/// Attempt to pull out the next chunk of body bytes.
fn poll_next(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Result<crate::web::Bytes, Self::Error>>> {
match self.get_mut() {
AnyBody::None => std::task::Poll::Ready(None),
AnyBody::Full { body } => {
let bytes = std::mem::take(body);
std::task::Poll::Ready(Some(Ok(bytes)))
}
AnyBody::Boxed { body } => body.as_pin_mut().poll_next(cx),
}
}
}

13
src/error/error.rs
View File

@ -2,7 +2,7 @@ use std::{error::Error as StdError, fmt};
use actix_http::{body::BoxBody, Response};
use crate::{HttpResponse, ResponseError};
use crate::{any_body::AnyBody, HttpResponse, ResponseError};
/// General purpose actix web error.
///
@ -69,8 +69,15 @@ impl<T: ResponseError + 'static> From<T> for Error {
}
}
impl From<Error> for Response<BoxBody> {
fn from(err: Error) -> Response<BoxBody> {
impl From<Error> for Response<AnyBody<BoxBody>> {
fn from(err: Error) -> Self {
err.error_response().into()
}
}
impl From<Error> for actix_http::Response<BoxBody> {
fn from(err: Error) -> Self {
let res: actix_http::Response<_> = err.error_response().into();
res.map_into_boxed_body()
}
}

6
src/error/internal.rs
View File

@ -7,7 +7,7 @@ use actix_http::{
};
use bytes::{BufMut as _, BytesMut};
use crate::{Error, HttpRequest, HttpResponse, Responder, ResponseError};
use crate::{any_body::AnyBody, Error, HttpRequest, HttpResponse, Responder, ResponseError};
/// Wraps errors to alter the generated response status code.
///
@ -91,7 +91,9 @@ where
let mime = mime::TEXT_PLAIN_UTF_8.try_into_value().unwrap();
res.headers_mut().insert(header::CONTENT_TYPE, mime);
res.set_body(BoxBody::new(buf.into_inner()))
res.set_body(AnyBody::Full {
body: buf.into_inner().freeze(),
})
}
InternalErrorType::Response(ref resp) => {

17
src/error/response_error.rs
View File

@ -14,6 +14,7 @@ use actix_http::{
use bytes::BytesMut;
use crate::{
any_body::AnyBody,
error::{downcast_dyn, downcast_get_type_id},
helpers, HttpResponse,
};
@ -33,7 +34,7 @@ pub trait ResponseError: fmt::Debug + fmt::Display {
///
/// By default, the generated response uses a 500 Internal Server Error status code, a
/// `Content-Type` of `text/plain`, and the body is set to `Self`'s `Display` impl.
fn error_response(&self) -> HttpResponse<BoxBody> {
fn error_response(&self) -> HttpResponse {
let mut res = HttpResponse::new(self.status_code());
let mut buf = BytesMut::new();
@ -42,7 +43,7 @@ pub trait ResponseError: fmt::Debug + fmt::Display {
let mime = mime::TEXT_PLAIN_UTF_8.try_into_value().unwrap();
res.headers_mut().insert(header::CONTENT_TYPE, mime);
res.set_body(BoxBody::new(buf))
res.set_body(AnyBody::Full { body: buf.freeze() })
}
downcast_get_type_id!();
@ -50,7 +51,7 @@ pub trait ResponseError: fmt::Debug + fmt::Display {
downcast_dyn!(ResponseError);
impl ResponseError for Box<dyn StdError + 'static> {}
impl ResponseError for Box<dyn StdError> {}
#[cfg(feature = "openssl")]
impl ResponseError for actix_tls::accept::openssl::reexports::Error {}
@ -128,7 +129,15 @@ impl ResponseError for actix_http::error::ContentTypeError {
impl ResponseError for actix_http::ws::HandshakeError {
fn error_response(&self) -> HttpResponse<BoxBody> {
Response::from(self).map_into_boxed_body().into()
Response::from(self)
.map_body(|_, body| AnyBody::Boxed { body })
.into()
}
}
impl ResponseError for actix_http::error::EncoderError {
fn error_response(&self) -> HttpResponse {
todo!("")
}
}

1
src/lib.rs
View File

@ -70,6 +70,7 @@
#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
#![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
mod any_body;
mod app;
mod app_service;
mod config;

26
src/middleware/compress.rs
View File

@ -10,7 +10,7 @@ use std::{
};
use actix_http::{
body::{EitherBody, MessageBody},
body::MessageBody,
encoding::Encoder,
header::{ContentEncoding, ACCEPT_ENCODING},
StatusCode,
@ -22,6 +22,7 @@ use once_cell::sync::Lazy;
use pin_project_lite::pin_project;
use crate::{
any_body::AnyBody,
dev::BodyEncoding,
service::{ServiceRequest, ServiceResponse},
Error, HttpResponse,
@ -61,7 +62,7 @@ where
B: MessageBody,
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Response = ServiceResponse<EitherBody<Encoder<B>>>;
type Response = ServiceResponse<Encoder<AnyBody<B>>>;
type Error = Error;
type Transform = CompressMiddleware<S>;
type InitError = ();
@ -111,7 +112,7 @@ where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
B: MessageBody,
{
type Response = ServiceResponse<EitherBody<Encoder<B>>>;
type Response = ServiceResponse<Encoder<AnyBody<B>>>;
type Error = Error;
type Future = Either<CompressResponse<S, B>, Ready<Result<Self::Response, Self::Error>>>;
@ -146,12 +147,15 @@ where
let res = HttpResponse::with_body(
StatusCode::NOT_ACCEPTABLE,
SUPPORTED_ALGORITHM_NAMES.clone(),
);
)
.map_body(|_, body| match body {
AnyBody::Full { body } => AnyBody::Stream {
body: Encoder::not_acceptable(body),
},
_ => unreachable!("probably"),
});
Either::right(ok(req
.into_response(res)
.map_into_boxed_body()
.map_into_right_body()))
Either::right(ok(req.into_response(res)))
}
}
}
@ -174,7 +178,7 @@ where
B: MessageBody,
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Output = Result<ServiceResponse<EitherBody<Encoder<B>>>, Error>;
type Output = Result<ServiceResponse<Encoder<AnyBody<B>>>, Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
@ -187,8 +191,8 @@ where
*this.encoding
};
Poll::Ready(Ok(resp.map_body(move |head, body| {
EitherBody::left(Encoder::response(enc, head, body))
Poll::Ready(Ok(resp.map_body(move |head, body| AnyBody::Stream {
body: Encoder::response(enc, head, body),
})))
}

21
src/middleware/logger.rs
View File

@ -22,6 +22,7 @@ use regex::{Regex, RegexSet};
use time::{format_description::well_known::Rfc3339, OffsetDateTime};
use crate::{
any_body::AnyBody,
body::{BodySize, MessageBody},
http::header::HeaderName,
service::{ServiceRequest, ServiceResponse},
@ -175,7 +176,7 @@ impl Default for Logger {
}
}
impl<S, B> Transform<S, ServiceRequest> for Logger
impl<S, B: 'static> Transform<S, ServiceRequest> for Logger
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
B: MessageBody,
@ -210,7 +211,7 @@ pub struct LoggerMiddleware<S> {
service: S,
}
impl<S, B> Service<ServiceRequest> for LoggerMiddleware<S>
impl<S, B: 'static> Service<ServiceRequest> for LoggerMiddleware<S>
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
B: MessageBody,
@ -262,7 +263,7 @@ pin_project! {
}
}
impl<S, B> Future for LoggerResponse<S, B>
impl<S, B: 'static> Future for LoggerResponse<S, B>
where
B: MessageBody,
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
@ -290,11 +291,13 @@ where
let time = *this.time;
let format = this.format.take();
Poll::Ready(Ok(res.map_body(move |_, body| StreamLog {
body,
time,
format,
size: 0,
Poll::Ready(Ok(res.map_body(move |_, body| AnyBody::Stream {
body: StreamLog {
body: body.into_body(),
time,
format,
size: 0,
},
})))
}
}
@ -302,7 +305,7 @@ where
pin_project! {
pub struct StreamLog<B> {
#[pin]
body: B,
body: AnyBody<B>,
format: Option<Format>,
size: usize,
time: OffsetDateTime,

45
src/responder.rs
View File

@ -1,7 +1,7 @@
use std::borrow::Cow;
use actix_http::{
body::{BoxBody, EitherBody, MessageBody},
body::{BoxBody, MessageBody},
error::HttpError,
header::HeaderMap,
header::IntoHeaderPair,
@ -9,7 +9,9 @@ use actix_http::{
};
use bytes::{Bytes, BytesMut};
use crate::{BoxError, Error, HttpRequest, HttpResponse, HttpResponseBuilder};
use crate::{
any_body::AnyBody, BoxError, Error, HttpRequest, HttpResponse, HttpResponseBuilder,
};
/// Trait implemented by types that can be converted to an HTTP response.
///
@ -72,7 +74,7 @@ impl Responder for HttpResponse {
}
}
impl Responder for actix_http::Response<BoxBody> {
impl Responder for actix_http::Response<AnyBody<BoxBody>> {
type Body = BoxBody;
#[inline]
@ -95,7 +97,10 @@ impl Responder for actix_http::ResponseBuilder {
#[inline]
fn respond_to(mut self, req: &HttpRequest) -> HttpResponse<Self::Body> {
self.finish().map_into_boxed_body().respond_to(req)
self.finish()
.map_into_boxed_body()
.map_body(|_, body| AnyBody::Boxed { body })
.respond_to(req)
}
}
@ -104,12 +109,12 @@ where
T: Responder,
<T::Body as MessageBody>::Error: Into<BoxError>,
{
type Body = EitherBody<T::Body>;
type Body = T::Body;
fn respond_to(self, req: &HttpRequest) -> HttpResponse<Self::Body> {
match self {
Some(val) => val.respond_to(req).map_into_left_body(),
None => HttpResponse::new(StatusCode::NOT_FOUND).map_into_right_body(),
Some(val) => val.respond_to(req),
None => HttpResponse::new(StatusCode::NOT_FOUND).map_into_body(),
}
}
}
@ -120,12 +125,12 @@ where
<T::Body as MessageBody>::Error: Into<BoxError>,
E: Into<Error>,
{
type Body = EitherBody<T::Body>;
type Body = T::Body;
fn respond_to(self, req: &HttpRequest) -> HttpResponse<Self::Body> {
match self {
Ok(val) => val.respond_to(req).map_into_left_body(),
Err(err) => HttpResponse::from_error(err.into()).map_into_right_body(),
Ok(val) => val.respond_to(req),
Err(err) => HttpResponse::from_error(err.into()).map_into_body(),
}
}
}
@ -146,7 +151,12 @@ macro_rules! impl_responder_by_forward_into_base_response {
type Body = $body;
fn respond_to(self, _: &HttpRequest) -> HttpResponse<Self::Body> {
let res: actix_http::Response<_> = self.into();
let res = actix_http::Response::with_body(
StatusCode::default(),
AnyBody::Full {
body: Bytes::from(self),
},
);
res.into()
}
}
@ -171,7 +181,12 @@ macro_rules! impl_into_string_responder {
fn respond_to(self, _: &HttpRequest) -> HttpResponse<Self::Body> {
let string: String = self.into();
let res: actix_http::Response<_> = string.into();
let res = actix_http::Response::with_body(
StatusCode::default(),
AnyBody::Full {
body: Bytes::from(string),
},
);
res.into()
}
}
@ -250,12 +265,12 @@ where
T: Responder,
<T::Body as MessageBody>::Error: Into<BoxError>,
{
type Body = EitherBody<T::Body>;
type Body = T::Body;
fn respond_to(self, req: &HttpRequest) -> HttpResponse<Self::Body> {
let headers = match self.headers {
Ok(headers) => headers,
Err(err) => return HttpResponse::from_error(err).map_into_right_body(),
Err(err) => return HttpResponse::from_error(err).map_into_body(),
};
let mut res = self.responder.respond_to(req);
@ -269,7 +284,7 @@ where
res.headers_mut().insert(k, v);
}
res.map_into_left_body()
res
}
}

5
src/response/builder.rs
View File

@ -22,6 +22,7 @@ use actix_http::header::HeaderValue;
use cookie::{Cookie, CookieJar};
use crate::{
any_body::AnyBody,
error::{Error, JsonPayloadError},
BoxError, HttpResponse,
};
@ -333,7 +334,7 @@ impl HttpResponseBuilder {
.set_body(body);
#[allow(unused_mut)] // mut is only unused when cookies are disabled
let mut res = HttpResponse::from(res);
let mut res = HttpResponse::from(res.map_body(|_, body| AnyBody::Stream { body }));
#[cfg(feature = "cookies")]
if let Some(ref jar) = self.cookies {
@ -416,7 +417,7 @@ impl From<HttpResponseBuilder> for HttpResponse {
}
}
impl From<HttpResponseBuilder> for Response<BoxBody> {
impl From<HttpResponseBuilder> for Response<AnyBody<BoxBody>> {
fn from(mut builder: HttpResponseBuilder) -> Self {
builder.finish().into()
}

70
src/response/response.rs
View File

@ -8,7 +8,7 @@ use std::{
};
use actix_http::{
body::{BoxBody, EitherBody, MessageBody},
body::{BoxBody, MessageBody, None as NoneBody},
header::HeaderMap,
Extensions, Response, ResponseHead, StatusCode,
};
@ -22,11 +22,11 @@ use {
cookie::Cookie,
};
use crate::{error::Error, HttpResponseBuilder};
use crate::{any_body::AnyBody, error::Error, HttpResponseBuilder};
/// An outgoing response.
pub struct HttpResponse<B = BoxBody> {
res: Response<B>,
res: Response<AnyBody<B>>,
pub(crate) error: Option<Error>,
}
@ -35,7 +35,7 @@ impl HttpResponse<BoxBody> {
#[inline]
pub fn new(status: StatusCode) -> Self {
Self {
res: Response::new(status),
res: Response::with_body(status, AnyBody::default()),
error: None,
}
}
@ -54,6 +54,13 @@ impl HttpResponse<BoxBody> {
response.error = Some(error);
response
}
pub fn map_into_body<B>(self) -> HttpResponse<B>
where
B: MessageBody + 'static,
{
self.map_body(|_, body| body.into_body())
}
}
impl<B> HttpResponse<B> {
@ -61,7 +68,7 @@ impl<B> HttpResponse<B> {
#[inline]
pub fn with_body(status: StatusCode, body: B) -> Self {
Self {
res: Response::with_body(status, body),
res: Response::with_body(status, AnyBody::Stream { body }),
error: None,
}
}
@ -182,12 +189,12 @@ impl<B> HttpResponse<B> {
/// Get body of this response
#[inline]
pub fn body(&self) -> &B {
pub fn body(&self) -> &AnyBody<B> {
self.res.body()
}
/// Set a body
pub fn set_body<B2>(self, body: B2) -> HttpResponse<B2> {
pub fn set_body<B2>(self, body: AnyBody<B2>) -> HttpResponse<B2> {
HttpResponse {
res: self.res.set_body(body),
error: None,
@ -196,12 +203,12 @@ impl<B> HttpResponse<B> {
}
/// Split response and body
pub fn into_parts(self) -> (HttpResponse<()>, B) {
pub fn into_parts(self) -> (HttpResponse<()>, AnyBody<B>) {
let (head, body) = self.res.into_parts();
(
HttpResponse {
res: head,
res: head.map_body(|_, _b| AnyBody::default()),
error: None,
},
body,
@ -211,7 +218,7 @@ impl<B> HttpResponse<B> {
/// Drop request's body
pub fn drop_body(self) -> HttpResponse<()> {
HttpResponse {
res: self.res.drop_body(),
res: self.res.drop_body().map_body(|_, _b| AnyBody::default()),
error: None,
}
}
@ -219,7 +226,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, B) -> B2,
F: FnOnce(&mut ResponseHead, AnyBody<B>) -> AnyBody<B2>,
{
HttpResponse {
res: self.res.map_body(f),
@ -229,27 +236,28 @@ impl<B> HttpResponse<B> {
// TODO: docs for the body map methods below
#[inline]
pub fn map_into_left_body<R>(self) -> HttpResponse<EitherBody<B, R>> {
self.map_body(|_, body| EitherBody::left(body))
}
#[inline]
pub fn map_into_right_body<L>(self) -> HttpResponse<EitherBody<L, B>> {
self.map_body(|_, body| EitherBody::right(body))
}
#[inline]
pub fn map_into_boxed_body(self) -> HttpResponse<BoxBody>
where
B: MessageBody + 'static,
{
// TODO: avoid double boxing with down-casting, if it improves perf
self.map_body(|_, body| BoxBody::new(body))
self.map_body(|_, body| AnyBody::Boxed {
body: match body {
AnyBody::None => BoxBody::new(NoneBody::new()),
AnyBody::Full { body } => BoxBody::new(body),
AnyBody::Stream { body } => BoxBody::new(body),
AnyBody::Boxed { body } => body,
},
})
}
/// Extract response body
pub fn into_body(self) -> B {
pub fn take_body(&mut self) -> AnyBody<B> {
self.res.take_body()
}
/// Extract response body
pub fn into_body(self) -> AnyBody<B> {
self.res.into_body()
}
}
@ -266,8 +274,8 @@ where
}
}
impl<B> From<Response<B>> for HttpResponse<B> {
fn from(res: Response<B>) -> Self {
impl<B> From<Response<AnyBody<B>>> for HttpResponse<B> {
fn from(res: Response<AnyBody<B>>) -> Self {
HttpResponse { res, error: None }
}
}
@ -278,7 +286,7 @@ impl From<Error> for HttpResponse {
}
}
impl<B> From<HttpResponse<B>> for Response<B> {
impl<B> From<HttpResponse<B>> for Response<AnyBody<B>> {
fn from(res: HttpResponse<B>) -> Self {
// this impl will always be called as part of dispatcher
@ -291,14 +299,14 @@ impl<B> From<HttpResponse<B>> for Response<B> {
}
}
// Future is only implemented for BoxBody payload type because it's the most useful for making
// Future is only implemented for default 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<BoxBody> {
type Output = Result<Response<BoxBody>, Error>;
impl Future for HttpResponse {
type Output = Result<Response<AnyBody>, Error>;
fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
if let Some(err) = self.error.take() {
@ -307,7 +315,7 @@ impl Future for HttpResponse<BoxBody> {
Poll::Ready(Ok(mem::replace(
&mut self.res,
Response::new(StatusCode::default()),
Response::with_body(StatusCode::default(), AnyBody::None),
)))
}
}

36
src/server.rs
View File

@ -17,7 +17,7 @@ use actix_tls::accept::openssl::reexports::{AlpnError, SslAcceptor, SslAcceptorB
#[cfg(feature = "rustls")]
use actix_tls::accept::rustls::reexports::ServerConfig as RustlsServerConfig;
use crate::{config::AppConfig, Error};
use crate::{any_body::AnyBody, config::AppConfig, Error};
struct Socket {
scheme: &'static str,
@ -55,7 +55,7 @@ where
S: ServiceFactory<Request, Config = AppConfig>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
S::Response: Into<Response<AnyBody<B>>>,
B: MessageBody,
{
pub(super) factory: F,
@ -75,7 +75,7 @@ where
S: ServiceFactory<Request, Config = AppConfig> + 'static,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
S::Response: Into<Response<AnyBody<B>>> + 'static,
<S::Service as Service<Request>>::Future: 'static,
S::Service: 'static,
@ -300,9 +300,10 @@ where
})
};
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> = err.into().error_response().into();
res.map_into_boxed_body()
});
svc.finish(map_config(fac, move |_| {
AppConfig::new(false, host.clone(), addr)
@ -360,9 +361,10 @@ where
svc
};
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> = err.into().error_response().into();
res.map_into_boxed_body()
});
svc.finish(map_config(fac, move |_| {
AppConfig::new(true, host.clone(), addr)
@ -544,9 +546,10 @@ where
.on_connect_ext(move |io: &_, ext: _| (&*handler)(io as &dyn Any, ext));
}
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> = err.into().error_response().into();
res.map_into_boxed_body()
});
svc.finish(map_config(fac, move |_| config.clone()))
})
@ -585,9 +588,10 @@ where
socket_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
let fac = factory().into_factory().map_err(|err| {
let res: actix_http::Response<_> = err.into().error_response().into();
res.map_into_boxed_body()
});
fn_service(|io: UnixStream| async { Ok((io, Protocol::Http1, None)) }).and_then(
HttpService::build()
@ -610,7 +614,7 @@ where
S: ServiceFactory<Request, Config = AppConfig>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
S::Response: Into<Response<AnyBody<B>>>,
S::Service: 'static,
B: MessageBody,
{

43
src/service.rs
View File

@ -5,7 +5,7 @@ use std::{
};
use actix_http::{
body::{BoxBody, EitherBody, MessageBody},
body::{self, BoxBody, MessageBody},
header::HeaderMap,
Extensions, HttpMessage, Method, Payload, PayloadStream, RequestHead, Response,
ResponseHead, StatusCode, Uri, Version,
@ -19,6 +19,7 @@ use actix_service::{
use cookie::{Cookie, ParseError as CookieParseError};
use crate::{
any_body::AnyBody,
config::{AppConfig, AppService},
dev::ensure_leading_slash,
guard::Guard,
@ -112,7 +113,7 @@ impl ServiceRequest {
/// Create service response
#[inline]
pub fn into_response<B, R: Into<Response<B>>>(self, res: R) -> ServiceResponse<B> {
pub fn into_response<B, R: Into<Response<AnyBody<B>>>>(self, res: R) -> ServiceResponse<B> {
let res = HttpResponse::from(res.into());
ServiceResponse::new(self.req, res)
}
@ -410,9 +411,14 @@ impl<B> ServiceResponse<B> {
self.response.headers_mut()
}
#[inline]
pub fn take_body(&mut self) -> AnyBody<B> {
self.response.take_body()
}
/// Extract response body
#[inline]
pub fn into_body(self) -> B {
pub fn into_body(self) -> AnyBody<B> {
self.response.into_body()
}
@ -420,7 +426,7 @@ impl<B> ServiceResponse<B> {
#[inline]
pub fn map_body<F, B2>(self, f: F) -> ServiceResponse<B2>
where
F: FnOnce(&mut ResponseHead, B) -> B2,
F: FnOnce(&mut ResponseHead, AnyBody<B>) -> AnyBody<B2>,
{
let response = self.response.map_body(f);
@ -430,22 +436,29 @@ impl<B> ServiceResponse<B> {
}
}
#[inline]
pub fn map_into_left_body<R>(self) -> ServiceResponse<EitherBody<B, R>> {
self.map_body(|_, body| EitherBody::left(body))
}
// #[inline]
// pub fn map_into_left_body<R>(self) -> ServiceResponse<EitherBody<B, R>> {
// self.map_body(|_, body| EitherBody::left(body))
// }
#[inline]
pub fn map_into_right_body<L>(self) -> ServiceResponse<EitherBody<L, B>> {
self.map_body(|_, body| EitherBody::right(body))
}
// #[inline]
// pub fn map_into_right_body<L>(self) -> ServiceResponse<EitherBody<L, B>> {
// self.map_body(|_, body| EitherBody::right(body))
// }
#[inline]
pub fn map_into_boxed_body(self) -> ServiceResponse<BoxBody>
where
B: MessageBody + 'static,
{
self.map_body(|_, body| BoxBody::new(body))
self.map_body(|_, body| AnyBody::Stream {
body: match body {
AnyBody::None => BoxBody::new(body::None::new()),
AnyBody::Full { body } => BoxBody::new(body),
AnyBody::Stream { body } => BoxBody::new(body),
AnyBody::Boxed { body } => body,
},
})
}
}
@ -455,8 +468,8 @@ impl<B> From<ServiceResponse<B>> for HttpResponse<B> {
}
}
impl<B> From<ServiceResponse<B>> for Response<B> {
fn from(res: ServiceResponse<B>) -> Response<B> {
impl<B> From<ServiceResponse<B>> for Response<AnyBody<B>> {
fn from(res: ServiceResponse<B>) -> Response<AnyBody<B>> {
res.response.into()
}
}

4
src/test.rs
View File

@ -163,7 +163,7 @@ where
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.extend_from_slice(&item.unwrap());
}
bytes.freeze()
@ -205,7 +205,7 @@ where
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.extend_from_slice(&item.unwrap());
}
bytes.freeze()

34
src/types/either.rs
View File

@ -12,9 +12,9 @@ use futures_core::ready;
use pin_project_lite::pin_project;
use crate::{
body, dev,
dev,
web::{Form, Json},
Error, FromRequest, HttpRequest, HttpResponse, Responder,
Error, FromRequest, HttpRequest,
};
/// Combines two extractor or responder types into a single type.
@ -140,21 +140,23 @@ impl<L, R> Either<L, R> {
}
}
/// See [here](#responder) for example of usage as a handler return type.
impl<L, R> Responder for Either<L, R>
where
L: Responder,
R: Responder,
{
type Body = body::EitherBody<L::Body, R::Body>;
// /// See [here](#responder) for example of usage as a handler return type.
// impl<L, R> Responder for Either<L, R>
// where
// L: Responder,
// R: Responder,
// {
// type Body = EitherAnyBody<L::Body, R::Body>;
fn respond_to(self, req: &HttpRequest) -> HttpResponse<Self::Body> {
match self {
Either::Left(a) => a.respond_to(req).map_into_left_body(),
Either::Right(b) => b.respond_to(req).map_into_right_body(),
}
}
}
// fn respond_to(self, req: &HttpRequest) -> HttpResponse<Self::Body> {
// match self {
// Either::Left(a) => a.respond_to(req).map_body(|_, body| EitherAnyBodyProj::left(body)),
// Either::Right(b) => b
// .respond_to(req)
// .map_body(|_, body| EitherAnyBodyProj::right(body)),
// }
// }
// }
/// A composite error resulting from failure to extract an `Either<L, R>`.
///

13
src/types/form.rs
View File

@ -20,9 +20,8 @@ use serde::{de::DeserializeOwned, Serialize};
#[cfg(feature = "__compress")]
use crate::dev::Decompress;
use crate::{
body::EitherBody, error::UrlencodedError, extract::FromRequest,
http::header::CONTENT_LENGTH, web, Error, HttpMessage, HttpRequest, HttpResponse,
Responder,
error::UrlencodedError, extract::FromRequest, http::header::CONTENT_LENGTH, web, Error,
HttpMessage, HttpRequest, HttpResponse, Responder,
};
/// URL encoded payload extractor and responder.
@ -181,7 +180,7 @@ impl<T: fmt::Display> fmt::Display for Form<T> {
/// See [here](#responder) for example of usage as a handler return type.
impl<T: Serialize> Responder for Form<T> {
type Body = EitherBody<String>;
type Body = String;
fn respond_to(self, _: &HttpRequest) -> HttpResponse<Self::Body> {
match serde_urlencoded::to_string(&self.0) {
@ -189,12 +188,12 @@ impl<T: Serialize> Responder for Form<T> {
.content_type(mime::APPLICATION_WWW_FORM_URLENCODED)
.message_body(body)
{
Ok(res) => res.map_into_left_body(),
Err(err) => HttpResponse::from_error(err).map_into_right_body(),
Ok(res) => res,
Err(err) => HttpResponse::from_error(err).map_into_body(),
},
Err(err) => {
HttpResponse::from_error(UrlencodedError::Serialize(err)).map_into_right_body()
HttpResponse::from_error(UrlencodedError::Serialize(err)).map_into_body()
}
}
}

9
src/types/json.rs
View File

@ -19,7 +19,6 @@ use actix_http::Payload;
#[cfg(feature = "__compress")]
use crate::dev::Decompress;
use crate::{
body::EitherBody,
error::{Error, JsonPayloadError},
extract::FromRequest,
http::header::CONTENT_LENGTH,
@ -117,7 +116,7 @@ impl<T: Serialize> Serialize for Json<T> {
///
/// If serialization failed
impl<T: Serialize> Responder for Json<T> {
type Body = EitherBody<String>;
type Body = String;
fn respond_to(self, _: &HttpRequest) -> HttpResponse<Self::Body> {
match serde_json::to_string(&self.0) {
@ -125,12 +124,12 @@ impl<T: Serialize> Responder for Json<T> {
.content_type(mime::APPLICATION_JSON)
.message_body(body)
{
Ok(res) => res.map_into_left_body(),
Err(err) => HttpResponse::from_error(err).map_into_right_body(),
Ok(res) => res,
Err(err) => HttpResponse::from_error(err).map_into_body(),
},
Err(err) => {
HttpResponse::from_error(JsonPayloadError::Serialize(err)).map_into_right_body()
HttpResponse::from_error(JsonPayloadError::Serialize(err)).map_into_body()
}
}
}