actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-08-21 21:25:36 +02:00
Code Issues Releases Wiki Activity

make AnyBody generic on Body type (#2448)

Browse Source
This commit is contained in:
Rob Ede
2021-11-16 21:41:35 +00:00
committed by GitHub
parent 13cf5a9e44
commit d8cbb879dd
18 changed files with 283 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
actix-http/src/body/body.rs
View File

@@ -8,6 +8,7 @@ use std::{
use bytes::{Bytes, BytesMut};
use futures_core::Stream;
use pin_project::pin_project;
use crate::error::Error;
@@ -16,15 +17,17 @@ use super::{BodySize, BodyStream, MessageBody, MessageBodyMapErr, SizedStream};
pub type Body = AnyBody;
/// Represents various types of HTTP message body.
pub enum AnyBody {
#[pin_project(project = AnyBodyProj)]
#[derive(Clone)]
pub enum AnyBody<B = BoxBody> {
/// Empty response. `Content-Length` header is not set.
None,
/// Specific response body.
/// Complete, in-memory response body.
Bytes(Bytes),
/// Generic message body.
Stream(BoxAnyBody),
/// Generic / Other message body.
Body(#[pin] B),
}
impl AnyBody {
@@ -33,29 +36,60 @@ impl AnyBody {
Self::Bytes(Bytes::new())
}
/// Create body from slice (copy)
pub fn from_slice(s: &[u8]) -> Self {
Self::Bytes(Bytes::copy_from_slice(s))
}
/// Create body from generic message body.
pub fn from_message<B>(body: B) -> Self
/// Create boxed body from generic message body.
pub fn new_boxed<B>(body: B) -> Self
where
B: MessageBody + 'static,
B::Error: Into<Box<dyn StdError + 'static>>,
{
Self::Stream(BoxAnyBody::from_body(body))
Self::Body(BoxBody::from_body(body))
}
/// Constructs new `AnyBody` instance from a slice of bytes by copying it.
///
/// If your bytes container is owned, it may be cheaper to use a `From` impl.
pub fn copy_from_slice(s: &[u8]) -> Self {
Self::Bytes(Bytes::copy_from_slice(s))
}
#[doc(hidden)]
#[deprecated(since = "4.0.0", note = "Renamed to `copy_from_slice`.")]
pub fn from_slice(s: &[u8]) -> Self {
Self::Bytes(Bytes::copy_from_slice(s))
}
}
impl MessageBody for AnyBody {
impl<B> AnyBody<B>
where
B: MessageBody + 'static,
B::Error: Into<Box<dyn StdError + 'static>>,
{
/// Create body from generic message body.
pub fn new(body: B) -> Self {
Self::Body(body)
}
pub fn into_boxed(self) -> AnyBody {
match self {
Self::None => AnyBody::None,
Self::Bytes(bytes) => AnyBody::Bytes(bytes),
Self::Body(body) => AnyBody::new_boxed(body),
}
}
}
impl<B> MessageBody for AnyBody<B>
where
B: MessageBody,
B::Error: Into<Box<dyn StdError>> + 'static,
{
type Error = Error;
fn size(&self) -> BodySize {
match self {
AnyBody::None => BodySize::None,
AnyBody::Bytes(ref bin) => BodySize::Sized(bin.len() as u64),
AnyBody::Stream(ref body) => body.size(),
AnyBody::Body(ref body) => body.size(),
}
}
@@ -63,9 +97,9 @@ impl MessageBody for AnyBody {
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
match self.get_mut() {
AnyBody::None => Poll::Ready(None),
AnyBody::Bytes(ref mut bin) => {
match self.project() {
AnyBodyProj::None => Poll::Ready(None),
AnyBodyProj::Bytes(bin) => {
let len = bin.len();
if len == 0 {
Poll::Ready(None)
@@ -74,8 +108,7 @@ impl MessageBody for AnyBody {
}
}
AnyBody::Stream(body) => body
.as_pin_mut()
AnyBodyProj::Body(body) => body
.poll_next(cx)
.map_err(|err| Error::new_body().with_cause(err)),
}
@@ -90,30 +123,30 @@ impl PartialEq for AnyBody {
AnyBody::Bytes(ref b2) => b == b2,
_ => false,
},
AnyBody::Stream(_) => false,
AnyBody::Body(_) => false,
}
}
}
impl fmt::Debug for AnyBody {
impl<S: fmt::Debug> fmt::Debug for AnyBody<S> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
AnyBody::None => write!(f, "AnyBody::None"),
AnyBody::Bytes(ref b) => write!(f, "AnyBody::Bytes({:?})", b),
AnyBody::Stream(_) => write!(f, "AnyBody::Message(_)"),
AnyBody::Bytes(ref bytes) => write!(f, "AnyBody::Bytes({:?})", bytes),
AnyBody::Body(ref stream) => write!(f, "AnyBody::Message({:?})", stream),
}
}
}
impl From<&'static str> for AnyBody {
fn from(s: &'static str) -> Body {
AnyBody::Bytes(Bytes::from_static(s.as_ref()))
fn from(string: &'static str) -> Body {
AnyBody::Bytes(Bytes::from_static(string.as_ref()))
}
}
impl From<&'static [u8]> for AnyBody {
fn from(s: &'static [u8]) -> Body {
AnyBody::Bytes(Bytes::from_static(s))
fn from(bytes: &'static [u8]) -> Body {
AnyBody::Bytes(Bytes::from_static(bytes))
}
}
@@ -124,20 +157,20 @@ impl From<Vec<u8>> for AnyBody {
}
impl From<String> for AnyBody {
fn from(s: String) -> Body {
s.into_bytes().into()
fn from(string: String) -> Body {
string.into_bytes().into()
}
}
impl From<&'_ String> for AnyBody {
fn from(s: &String) -> Body {
AnyBody::Bytes(Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(&s)))
fn from(string: &String) -> Body {
AnyBody::Bytes(Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(&string)))
}
}
impl From<Cow<'_, str>> for AnyBody {
fn from(s: Cow<'_, str>) -> Body {
match s {
fn from(string: Cow<'_, str>) -> Body {
match string {
Cow::Owned(s) => AnyBody::from(s),
Cow::Borrowed(s) => {
AnyBody::Bytes(Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(s)))
@@ -147,14 +180,14 @@ impl From<Cow<'_, str>> for AnyBody {
}
impl From<Bytes> for AnyBody {
fn from(s: Bytes) -> Body {
AnyBody::Bytes(s)
fn from(bytes: Bytes) -> Body {
AnyBody::Bytes(bytes)
}
}
impl From<BytesMut> for AnyBody {
fn from(s: BytesMut) -> Body {
AnyBody::Bytes(s.freeze())
fn from(bytes: BytesMut) -> Body {
AnyBody::Bytes(bytes.freeze())
}
}
@@ -163,8 +196,8 @@ where
S: Stream<Item = Result<Bytes, E>> + 'static,
E: Into<Box<dyn StdError>> + 'static,
{
fn from(s: SizedStream<S>) -> Body {
AnyBody::from_message(s)
fn from(stream: SizedStream<S>) -> Body {
AnyBody::new_boxed(stream)
}
}
@@ -173,15 +206,15 @@ where
S: Stream<Item = Result<Bytes, E>> + 'static,
E: Into<Box<dyn StdError>> + 'static,
{
fn from(s: BodyStream<S>) -> Body {
AnyBody::from_message(s)
fn from(stream: BodyStream<S>) -> Body {
AnyBody::new_boxed(stream)
}
}
/// A boxed message body with boxed errors.
pub struct BoxAnyBody(Pin<Box<dyn MessageBody<Error = Box<dyn StdError + 'static>>>>);
pub struct BoxBody(Pin<Box<dyn MessageBody<Error = Box<dyn StdError>>>>);
impl BoxAnyBody {
impl BoxBody {
/// Boxes a `MessageBody` and any errors it generates.
pub fn from_body<B>(body: B) -> Self
where
@@ -195,18 +228,18 @@ impl BoxAnyBody {
/// 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>>)> {
) -> Pin<&mut (dyn MessageBody<Error = Box<dyn StdError>>)> {
self.0.as_mut()
}
}
impl fmt::Debug for BoxAnyBody {
impl fmt::Debug for BoxBody {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("BoxAnyBody(dyn MessageBody)")
}
}
impl MessageBody for BoxAnyBody {
impl MessageBody for BoxBody {
type Error = Error;
fn size(&self) -> BodySize {
@@ -223,3 +256,52 @@ impl MessageBody for BoxAnyBody {
.map_err(|err| Error::new_body().with_cause(err))
}
}
#[cfg(test)]
mod tests {
use std::marker::PhantomPinned;
use static_assertions::{assert_impl_all, assert_not_impl_all};
use super::*;
use crate::body::to_bytes;
struct PinType(PhantomPinned);
impl MessageBody for PinType {
type Error = crate::Error;
fn size(&self) -> BodySize {
unimplemented!()
}
fn poll_next(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
unimplemented!()
}
}
assert_impl_all!(AnyBody<()>: MessageBody, fmt::Debug, Send, Sync, Unpin);
assert_impl_all!(AnyBody<AnyBody<()>>: MessageBody, fmt::Debug, Send, Sync, Unpin);
assert_impl_all!(AnyBody<Bytes>: MessageBody, fmt::Debug, Send, Sync, Unpin);
assert_impl_all!(AnyBody: MessageBody, fmt::Debug, Unpin);
assert_impl_all!(BoxBody: MessageBody, fmt::Debug, Unpin);
assert_impl_all!(AnyBody<PinType>: MessageBody);
assert_not_impl_all!(AnyBody: Send, Sync, Unpin);
assert_not_impl_all!(BoxBody: Send, Sync, Unpin);
assert_not_impl_all!(AnyBody<PinType>: Send, Sync, Unpin);
#[actix_rt::test]
async fn nested_boxed_body() {
let body = AnyBody::copy_from_slice(&[1, 2, 3]);
let boxed_body = BoxBody::from_body(BoxBody::from_body(body));
assert_eq!(
to_bytes(boxed_body).await.unwrap(),
Bytes::from(vec![1, 2, 3]),
);
}
}

36
actix-http/src/body/body_stream.rs
View File

@@ -75,10 +75,22 @@ mod tests {
use derive_more::{Display, Error};
use futures_core::ready;
use futures_util::{stream, FutureExt as _};
use static_assertions::{assert_impl_all, assert_not_impl_all};
use super::*;
use crate::body::to_bytes;
assert_impl_all!(BodyStream<stream::Empty<Result<Bytes, crate::Error>>>: MessageBody);
assert_impl_all!(BodyStream<stream::Empty<Result<Bytes, &'static str>>>: MessageBody);
assert_impl_all!(BodyStream<stream::Repeat<Result<Bytes, &'static str>>>: MessageBody);
assert_impl_all!(BodyStream<stream::Empty<Result<Bytes, Infallible>>>: MessageBody);
assert_impl_all!(BodyStream<stream::Repeat<Result<Bytes, Infallible>>>: MessageBody);
assert_not_impl_all!(BodyStream<stream::Empty<Bytes>>: MessageBody);
assert_not_impl_all!(BodyStream<stream::Repeat<Bytes>>: MessageBody);
// crate::Error is not Clone
assert_not_impl_all!(BodyStream<stream::Repeat<Result<Bytes, crate::Error>>>: MessageBody);
#[actix_rt::test]
async fn skips_empty_chunks() {
let body = BodyStream::new(stream::iter(
@@ -124,6 +136,30 @@ mod tests {
assert!(matches!(to_bytes(body).await, Err(StreamErr)));
}
#[actix_rt::test]
async fn stream_string_error() {
// `&'static str` does not impl `Error`
// but it does impl `Into<Box<dyn Error>>`
let body = BodyStream::new(stream::once(async { Err("stringy error") }));
assert!(matches!(to_bytes(body).await, Err("stringy error")));
}
#[actix_rt::test]
async fn stream_boxed_error() {
// `Box<dyn Error>` does not impl `Error`
// but it does impl `Into<Box<dyn Error>>`
let body = BodyStream::new(stream::once(async {
Err(Box::<dyn StdError>::from("stringy error"))
}));
assert_eq!(
to_bytes(body).await.unwrap_err().to_string(),
"stringy error"
);
}
#[actix_rt::test]
async fn stream_delayed_error() {
let body =

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

@@ -11,15 +11,13 @@ use futures_core::ready;
mod body;
mod body_stream;
mod message_body;
mod response_body;
mod size;
mod sized_stream;
pub use self::body::{AnyBody, Body, BoxAnyBody};
pub use self::body::{AnyBody, Body, BoxBody};
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;
@@ -108,10 +106,10 @@ mod tests {
assert_eq!(Body::from(b"test".as_ref()).size(), BodySize::Sized(4));
assert_eq!(Body::from(b"test".as_ref()).get_ref(), b"test");
assert_eq!(
Body::from_slice(b"test".as_ref()).size(),
Body::copy_from_slice(b"test".as_ref()).size(),
BodySize::Sized(4)
);
assert_eq!(Body::from_slice(b"test".as_ref()).get_ref(), b"test");
assert_eq!(Body::copy_from_slice(b"test".as_ref()).get_ref(), b"test");
let sb = Bytes::from(&b"test"[..]);
pin!(sb);

84
actix-http/src/body/response_body.rs
View File

@@ -1,84 +0,0 @@
use std::{
mem,
pin::Pin,
task::{Context, Poll},
};
use bytes::Bytes;
use futures_core::Stream;
use pin_project::pin_project;
use crate::error::Error;
use super::{Body, BodySize, MessageBody};
#[pin_project(project = ResponseBodyProj)]
pub enum ResponseBody<B> {
Body(#[pin] B),
Other(Body),
}
impl ResponseBody<Body> {
pub fn into_body<B>(self) -> ResponseBody<B> {
match self {
ResponseBody::Body(b) => ResponseBody::Other(b),
ResponseBody::Other(b) => ResponseBody::Other(b),
}
}
}
impl<B> ResponseBody<B> {
pub fn take_body(&mut self) -> ResponseBody<B> {
mem::replace(self, ResponseBody::Other(Body::None))
}
}
impl<B: MessageBody> ResponseBody<B> {
pub fn as_ref(&self) -> Option<&B> {
if let ResponseBody::Body(ref b) = self {
Some(b)
} else {
None
}
}
}
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(),
ResponseBody::Other(ref body) => body.size(),
}
}
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
Stream::poll_next(self, cx)
}
}
impl<B> Stream for ResponseBody<B>
where
B: MessageBody,
B::Error: Into<Error>,
{
type Item = Result<Bytes, Error>;
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
match self.project() {
ResponseBodyProj::Body(body) => body.poll_next(cx).map_err(Into::into),
ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx),
}
}
}

45
actix-http/src/body/sized_stream.rs
View File

@@ -72,10 +72,22 @@ mod tests {
use actix_rt::pin;
use actix_utils::future::poll_fn;
use futures_util::stream;
use static_assertions::{assert_impl_all, assert_not_impl_all};
use super::*;
use crate::body::to_bytes;
assert_impl_all!(SizedStream<stream::Empty<Result<Bytes, crate::Error>>>: MessageBody);
assert_impl_all!(SizedStream<stream::Empty<Result<Bytes, &'static str>>>: MessageBody);
assert_impl_all!(SizedStream<stream::Repeat<Result<Bytes, &'static str>>>: MessageBody);
assert_impl_all!(SizedStream<stream::Empty<Result<Bytes, Infallible>>>: MessageBody);
assert_impl_all!(SizedStream<stream::Repeat<Result<Bytes, Infallible>>>: MessageBody);
assert_not_impl_all!(SizedStream<stream::Empty<Bytes>>: MessageBody);
assert_not_impl_all!(SizedStream<stream::Repeat<Bytes>>: MessageBody);
// crate::Error is not Clone
assert_not_impl_all!(SizedStream<stream::Repeat<Result<Bytes, crate::Error>>>: MessageBody);
#[actix_rt::test]
async fn skips_empty_chunks() {
let body = SizedStream::new(
@@ -119,4 +131,37 @@ mod tests {
assert_eq!(to_bytes(body).await.ok(), Some(Bytes::from("12")));
}
#[actix_rt::test]
async fn stream_string_error() {
// `&'static str` does not impl `Error`
// but it does impl `Into<Box<dyn Error>>`
let body = SizedStream::new(0, stream::once(async { Err("stringy error") }));
assert_eq!(to_bytes(body).await, Ok(Bytes::new()));
let body = SizedStream::new(1, stream::once(async { Err("stringy error") }));
assert!(matches!(to_bytes(body).await, Err("stringy error")));
}
#[actix_rt::test]
async fn stream_boxed_error() {
// `Box<dyn Error>` does not impl `Error`
// but it does impl `Into<Box<dyn Error>>`
let body = SizedStream::new(
0,
stream::once(async { Err(Box::<dyn StdError>::from("stringy error")) }),
);
assert_eq!(to_bytes(body).await.unwrap(), Bytes::new());
let body = SizedStream::new(
1,
stream::once(async { Err(Box::<dyn StdError>::from("stringy error")) }),
);
assert_eq!(
to_bytes(body).await.unwrap_err().to_string(),
"stringy error"
);
}
}