ntex_http/

body.rs

//! Traits and structures to aid consuming and writing HTTP payloads.
use std::{error::Error, fmt, marker, mem, pin::Pin, rc::Rc, task::Context, task::Poll};

use futures_core::Stream;
use ntex_bytes::{BytePages, Bytes, BytesMut};

#[derive(Debug, PartialEq, Eq, Copy, Clone)]
/// Size information used when encoding an HTTP message body.
pub enum BodySize {
    /// No body is present and no `Content-Length` header should be emitted.
    ///
    /// An HTTP/1 response that may carry a body is still sent with
    /// `Content-Length: 0`, otherwise its end could not be determined.
    None,
    /// An empty body with a `Content-Length: 0` header.
    Empty,
    /// A body with a known byte length.
    Sized(u64),
    /// A streaming body with an unknown final length.
    Stream,
}

impl BodySize {
    /// Returns `true` if this size represents no payload bytes.
    pub fn is_eof(&self) -> bool {
        matches!(self, BodySize::None | BodySize::Empty | BodySize::Sized(0))
    }
}

/// Interface for types that can be streamed to a peer.
pub trait MessageBody: 'static {
    /// Returns the body size hint.
    fn size(&self) -> BodySize;

    /// Polls the next body chunk.
    fn poll_next_chunk(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>>;
}

impl MessageBody for () {
    #[inline]
    fn size(&self) -> BodySize {
        BodySize::Empty
    }

    #[inline]
    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        Poll::Ready(None)
    }
}

impl<T: MessageBody> MessageBody for Box<T> {
    #[inline]
    fn size(&self) -> BodySize {
        self.as_ref().size()
    }

    #[inline]
    fn poll_next_chunk(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        self.as_mut().poll_next_chunk(cx)
    }
}

#[derive(Debug)]
/// Response body preserving either a concrete body type or a type-erased body.
pub enum ResponseBody<B> {
    /// The original concrete body.
    Body(B),
    /// A type-erased body.
    Other(Body),
}

impl ResponseBody<Body> {
    /// Converts a type-erased response body to another concrete response type.
    pub fn into_body<B>(self) -> ResponseBody<B> {
        match self {
            ResponseBody::Body(b) | ResponseBody::Other(b) => ResponseBody::Other(b),
        }
    }
}

impl From<ResponseBody<Body>> for Body {
    fn from(b: ResponseBody<Body>) -> Self {
        match b {
            ResponseBody::Body(b) | ResponseBody::Other(b) => b,
        }
    }
}

impl<B> From<Body> for ResponseBody<B> {
    fn from(b: Body) -> Self {
        ResponseBody::Other(b)
    }
}

impl<B> ResponseBody<B> {
    #[inline]
    /// Creates a response body from a concrete body value.
    pub fn new(body: B) -> Self {
        ResponseBody::Body(body)
    }

    #[inline]
    #[must_use]
    /// Replaces this response body with [`Body::None`] and returns the original.
    pub fn take_body(&mut self) -> ResponseBody<B> {
        std::mem::replace(self, ResponseBody::Other(Body::None))
    }
}

impl<B: MessageBody> ResponseBody<B> {
    /// Returns the concrete body, if this is [`ResponseBody::Body`].
    pub fn as_ref(&self) -> Option<&B> {
        if let ResponseBody::Body(b) = self {
            Some(b)
        } else {
            None
        }
    }
}

impl<B: MessageBody> MessageBody for ResponseBody<B> {
    #[inline]
    fn size(&self) -> BodySize {
        match self {
            ResponseBody::Body(body) => body.size(),
            ResponseBody::Other(body) => body.size(),
        }
    }

    #[inline]
    fn poll_next_chunk(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        match self {
            ResponseBody::Body(body) => body.poll_next_chunk(cx),
            ResponseBody::Other(body) => body.poll_next_chunk(cx),
        }
    }
}

impl<B: MessageBody + Unpin> Stream for ResponseBody<B> {
    type Item = Result<Bytes, Rc<dyn Error>>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.get_mut() {
            ResponseBody::Body(body) => body.poll_next_chunk(cx),
            ResponseBody::Other(body) => body.poll_next_chunk(cx),
        }
    }
}

/// Type-erased HTTP message body.
pub enum Body {
    /// Empty response. `Content-Length` header is not set.
    ///
    /// See [`BodySize::None`].
    None,
    /// Zero sized response body. `Content-Length` header is set to `0`.
    Empty,
    /// Specific response body.
    Bytes(Bytes),
    /// Generic message body.
    Message(Box<dyn MessageBody>),
}

impl Body {
    /// Creates a body by copying a byte slice.
    pub fn from_slice(s: &[u8]) -> Body {
        Body::Bytes(Bytes::copy_from_slice(s))
    }

    /// Creates a body from a streaming message body.
    pub fn from_message<B: MessageBody>(body: B) -> Body {
        Body::Message(Box::new(body))
    }
}

impl MessageBody for Body {
    #[inline]
    fn size(&self) -> BodySize {
        match self {
            Body::None => BodySize::None,
            Body::Empty => BodySize::Empty,
            Body::Bytes(bin) => BodySize::Sized(bin.len() as u64),
            Body::Message(body) => body.size(),
        }
    }

    fn poll_next_chunk(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        match self {
            Body::None | Body::Empty => Poll::Ready(None),
            Body::Bytes(bin) => {
                let len = bin.len();
                if len == 0 {
                    Poll::Ready(None)
                } else {
                    Poll::Ready(Some(Ok(mem::take(bin))))
                }
            }
            Body::Message(body) => body.poll_next_chunk(cx),
        }
    }
}

impl PartialEq for Body {
    fn eq(&self, other: &Body) -> bool {
        match (self, other) {
            (Body::None, Body::None) | (Body::Empty, Body::Empty) => true,
            (Body::Bytes(b), Body::Bytes(b2)) => b == b2,
            _ => false,
        }
    }
}

impl fmt::Debug for Body {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match *self {
            Body::None => write!(f, "Body::None"),
            Body::Empty => write!(f, "Body::Empty"),
            Body::Bytes(ref b) => write!(f, "Body::Bytes({b:?})"),
            Body::Message(_) => write!(f, "Body::Message(_)"),
        }
    }
}

impl From<&'static str> for Body {
    fn from(s: &'static str) -> Body {
        Body::Bytes(Bytes::from_static(s.as_ref()))
    }
}

impl From<&'static [u8]> for Body {
    fn from(s: &'static [u8]) -> Body {
        Body::Bytes(Bytes::from_static(s))
    }
}

impl From<Vec<u8>> for Body {
    fn from(vec: Vec<u8>) -> Body {
        Body::Bytes(Bytes::from(vec))
    }
}

impl From<String> for Body {
    fn from(s: String) -> Body {
        s.into_bytes().into()
    }
}

impl<'a> From<&'a String> for Body {
    fn from(s: &'a String) -> Body {
        Body::Bytes(Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(&s)))
    }
}

impl From<Bytes> for Body {
    fn from(s: Bytes) -> Body {
        Body::Bytes(s)
    }
}

impl From<BytesMut> for Body {
    fn from(s: BytesMut) -> Body {
        Body::Bytes(s.freeze())
    }
}

impl From<BytePages> for Body {
    fn from(pages: BytePages) -> Body {
        Body::from_message(pages)
    }
}

impl<S> From<SizedStream<S>> for Body
where
    S: Stream<Item = Result<Bytes, Rc<dyn Error>>> + Unpin + 'static,
{
    fn from(s: SizedStream<S>) -> Body {
        Body::from_message(s)
    }
}

impl<S, E> From<BodyStream<S, E>> for Body
where
    S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
    E: Error + 'static,
{
    fn from(s: BodyStream<S, E>) -> Body {
        Body::from_message(s)
    }
}

impl<S> From<BoxedBodyStream<S>> for Body
where
    S: Stream<Item = Result<Bytes, Rc<dyn Error>>> + Unpin + 'static,
{
    fn from(s: BoxedBodyStream<S>) -> Body {
        Body::from_message(s)
    }
}

impl MessageBody for Bytes {
    fn size(&self) -> BodySize {
        BodySize::Sized(self.len() as u64)
    }

    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        if self.is_empty() {
            Poll::Ready(None)
        } else {
            Poll::Ready(Some(Ok(mem::take(self))))
        }
    }
}

impl MessageBody for BytesMut {
    fn size(&self) -> BodySize {
        BodySize::Sized(self.len() as u64)
    }

    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        if self.is_empty() {
            Poll::Ready(None)
        } else {
            Poll::Ready(Some(Ok(mem::take(self).freeze())))
        }
    }
}

impl MessageBody for BytePages {
    fn size(&self) -> BodySize {
        BodySize::Sized(self.len() as u64)
    }

    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        if let Some(page) = self.take() {
            Poll::Ready(Some(Ok(page.freeze())))
        } else {
            Poll::Ready(None)
        }
    }
}

impl MessageBody for &'static str {
    fn size(&self) -> BodySize {
        BodySize::Sized(self.len() as u64)
    }

    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        if self.is_empty() {
            Poll::Ready(None)
        } else {
            Poll::Ready(Some(Ok(Bytes::from_static(mem::take(self).as_ref()))))
        }
    }
}

impl MessageBody for &'static [u8] {
    fn size(&self) -> BodySize {
        BodySize::Sized(self.len() as u64)
    }

    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        if self.is_empty() {
            Poll::Ready(None)
        } else {
            Poll::Ready(Some(Ok(Bytes::from_static(mem::take(self)))))
        }
    }
}

impl MessageBody for Vec<u8> {
    fn size(&self) -> BodySize {
        BodySize::Sized(self.len() as u64)
    }

    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        if self.is_empty() {
            Poll::Ready(None)
        } else {
            Poll::Ready(Some(Ok(Bytes::from(mem::take(self)))))
        }
    }
}

impl MessageBody for String {
    fn size(&self) -> BodySize {
        BodySize::Sized(self.len() as u64)
    }

    fn poll_next_chunk(
        &mut self,
        _: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        if self.is_empty() {
            Poll::Ready(None)
        } else {
            Poll::Ready(Some(Ok(Bytes::from(mem::take(self).into_bytes()))))
        }
    }
}

/// Type represent streaming body.
///
/// Response does not contain `content-length` header and appropriate transfer encoding is used.
pub struct BodyStream<S, E> {
    stream: S,
    _t: marker::PhantomData<E>,
}

impl<S, E> BodyStream<S, E>
where
    S: Stream<Item = Result<Bytes, E>> + Unpin,
    E: Error,
{
    pub fn new(stream: S) -> Self {
        BodyStream {
            stream,
            _t: marker::PhantomData,
        }
    }
}

impl<S, E> fmt::Debug for BodyStream<S, E> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("BodyStream")
            .field("stream", &std::any::type_name::<S>())
            .field("error", &std::any::type_name::<E>())
            .finish()
    }
}

impl<S, E> MessageBody for BodyStream<S, E>
where
    S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
    E: Error + 'static,
{
    fn size(&self) -> BodySize {
        BodySize::Stream
    }

    /// Attempts to pull out the next value of the underlying [`Stream`].
    ///
    /// Empty values are skipped to prevent [`BodyStream`]'s transmission being
    /// ended on a zero-length chunk, but rather proceed until the underlying
    /// [`Stream`] ends.
    fn poll_next_chunk(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        loop {
            return Poll::Ready(match Pin::new(&mut self.stream).poll_next(cx) {
                Poll::Ready(Some(Ok(ref bytes))) if bytes.is_empty() => continue,
                Poll::Ready(opt) => opt.map(|res| {
                    res.map_err(|e| {
                        let e: Rc<dyn Error> = Rc::new(e);
                        e
                    })
                }),
                Poll::Pending => return Poll::Pending,
            });
        }
    }
}

/// Type represent streaming body.
/// Response does not contain `content-length` header and appropriate transfer encoding is used.
pub struct BoxedBodyStream<S> {
    stream: S,
}

impl<S> BoxedBodyStream<S>
where
    S: Stream<Item = Result<Bytes, Rc<dyn Error>>> + Unpin,
{
    pub fn new(stream: S) -> Self {
        BoxedBodyStream { stream }
    }
}

impl<S> fmt::Debug for BoxedBodyStream<S> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("BoxedBodyStream")
            .field("stream", &std::any::type_name::<S>())
            .finish()
    }
}

impl<S> MessageBody for BoxedBodyStream<S>
where
    S: Stream<Item = Result<Bytes, Rc<dyn Error>>> + Unpin + 'static,
{
    fn size(&self) -> BodySize {
        BodySize::Stream
    }

    /// Attempts to pull out the next value of the underlying [`Stream`].
    ///
    /// Empty values are skipped to prevent [`BodyStream`]'s transmission being
    /// ended on a zero-length chunk, but rather proceed until the underlying
    /// [`Stream`] ends.
    fn poll_next_chunk(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        loop {
            return Poll::Ready(match Pin::new(&mut self.stream).poll_next(cx) {
                Poll::Ready(Some(Ok(ref bytes))) if bytes.is_empty() => continue,
                Poll::Ready(opt) => opt,
                Poll::Pending => return Poll::Pending,
            });
        }
    }
}

/// Type represent streaming body. This body implementation should be used
/// if total size of stream is known. Data get sent as is without using transfer encoding.
pub struct SizedStream<S> {
    size: u64,
    stream: S,
}

impl<S> SizedStream<S>
where
    S: Stream<Item = Result<Bytes, Rc<dyn Error>>> + Unpin,
{
    pub fn new(size: u64, stream: S) -> Self {
        SizedStream { size, stream }
    }
}

impl<S> fmt::Debug for SizedStream<S> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SizedStream")
            .field("size", &self.size)
            .field("stream", &std::any::type_name::<S>())
            .finish()
    }
}

impl<S> MessageBody for SizedStream<S>
where
    S: Stream<Item = Result<Bytes, Rc<dyn Error>>> + Unpin + 'static,
{
    fn size(&self) -> BodySize {
        BodySize::Sized(self.size)
    }

    /// Attempts to pull out the next value of the underlying [`Stream`].
    ///
    /// Empty values are skipped to prevent [`SizedStream`]'s transmission being
    /// ended on a zero-length chunk, but rather proceed until the underlying
    /// [`Stream`] ends.
    fn poll_next_chunk(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Bytes, Rc<dyn Error>>>> {
        loop {
            return Poll::Ready(match Pin::new(&mut self.stream).poll_next(cx) {
                Poll::Ready(Some(Ok(ref bytes))) if bytes.is_empty() => continue,
                Poll::Ready(val) => val,
                Poll::Pending => return Poll::Pending,
            });
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{future::poll_fn, future::ready, io};

    use futures_util::stream;
    use ntex::util::BufMut;

    use super::*;

    impl Body {
        pub(crate) fn get_ref(&self) -> &[u8] {
            if let Body::Bytes(bin) = self { bin } else { panic!() }
        }
    }

    #[ntex::test]
    async fn test_size() {
        assert_eq!(32, std::mem::size_of::<Body>());
        assert_eq!(32, std::mem::size_of::<ResponseBody<Bytes>>());
        assert_eq!(40, std::mem::size_of::<ResponseBody<Body>>());
    }

    #[ntex::test]
    async fn test_static_str() {
        assert_eq!(Body::from("").size(), BodySize::Sized(0));
        assert_eq!(Body::from("test").size(), BodySize::Sized(4));
        assert_eq!(Body::from("test").get_ref(), b"test");

        assert_eq!("test".size(), BodySize::Sized(4));
        assert_eq!(
            poll_fn(|cx| "test".poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("test"))
        );
        assert_eq!(
            poll_fn(|cx| "test".poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("test"))
        );
        assert!(poll_fn(|cx| "".poll_next_chunk(cx)).await.is_none());
    }

    #[ntex::test]
    async fn test_pages() {
        let mut pages = BytePages::default();
        pages.put_slice(b"1111");
        pages.append(Bytes::from("2222"));
        pages.append(Bytes::from("3333"));

        let mut body = Body::from(pages);
        assert_eq!(body.size(), BodySize::Sized(12));
        assert_eq!(
            poll_fn(|cx| body.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("111122223333"))
        );
        assert!(poll_fn(|cx| body.poll_next_chunk(cx)).await.is_none());
    }

    #[ntex::test]
    async fn test_static_bytes() {
        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(),
            BodySize::Sized(4)
        );
        assert_eq!(Body::from_slice(b"test".as_ref()).get_ref(), b"test");

        assert_eq!((&b"test"[..]).size(), BodySize::Sized(4));
        assert_eq!(
            poll_fn(|cx| (&b"test"[..]).poll_next_chunk(cx))
                .await
                .unwrap()
                .ok(),
            Some(Bytes::from("test"))
        );
        assert_eq!((&b"test"[..]).size(), BodySize::Sized(4));
        assert!(poll_fn(|cx| (&b""[..]).poll_next_chunk(cx)).await.is_none());
    }

    #[ntex::test]
    async fn test_vec() {
        assert_eq!(Body::from(Vec::from("test")).size(), BodySize::Sized(4));
        assert_eq!(Body::from(Vec::from("test")).get_ref(), b"test");

        assert_eq!(Vec::from("test").size(), BodySize::Sized(4));
        assert_eq!(
            poll_fn(|cx| Vec::from("test").poll_next_chunk(cx))
                .await
                .unwrap()
                .ok(),
            Some(Bytes::from("test"))
        );
        assert_eq!(
            poll_fn(|cx| Vec::from("test").poll_next_chunk(cx))
                .await
                .unwrap()
                .ok(),
            Some(Bytes::from("test"))
        );
        assert!(
            poll_fn(|cx| Vec::<u8>::new().poll_next_chunk(cx))
                .await
                .is_none()
        );
    }

    #[ntex::test]
    async fn test_bytes() {
        let mut b = Bytes::from("test");
        assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4));
        assert_eq!(Body::from(b.clone()).get_ref(), b"test");

        assert_eq!(b.size(), BodySize::Sized(4));
        assert_eq!(
            poll_fn(|cx| b.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("test"))
        );
        assert!(poll_fn(|cx| b.poll_next_chunk(cx)).await.is_none(),);
    }

    #[ntex::test]
    async fn test_bytes_mut() {
        let mut b = Body::from(BytesMut::from("test"));
        assert_eq!(b.size(), BodySize::Sized(4));
        assert_eq!(b.get_ref(), b"test");
        assert_eq!(
            poll_fn(|cx| b.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("test"))
        );
        assert!(poll_fn(|cx| b.poll_next_chunk(cx)).await.is_none(),);

        let mut b = BytesMut::from("test");
        assert_eq!(b.size(), BodySize::Sized(4));
        assert_eq!(
            poll_fn(|cx| b.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("test"))
        );
        assert_eq!(b.size(), BodySize::Sized(0));
        assert!(poll_fn(|cx| b.poll_next_chunk(cx)).await.is_none(),);
    }

    #[ntex::test]
    async fn test_string() {
        let mut b = "test".to_owned();
        assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4));
        assert_eq!(Body::from(b.clone()).get_ref(), b"test");
        assert_eq!(Body::from(&b).size(), BodySize::Sized(4));
        assert_eq!(Body::from(&b).get_ref(), b"test");

        assert_eq!(b.size(), BodySize::Sized(4));
        assert_eq!(
            poll_fn(|cx| b.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("test"))
        );
        assert!(poll_fn(|cx| b.poll_next_chunk(cx)).await.is_none(),);
    }

    #[ntex::test]
    async fn test_unit() {
        assert_eq!(().size(), BodySize::Empty);
        assert!(poll_fn(|cx| ().poll_next_chunk(cx)).await.is_none());
    }

    #[ntex::test]
    async fn test_box() {
        let mut val = Box::new(());
        assert_eq!(val.size(), BodySize::Empty);
        assert!(poll_fn(|cx| val.poll_next_chunk(cx)).await.is_none());
    }

    #[ntex::test]
    #[allow(clippy::eq_op)]
    async fn test_body_eq() {
        assert_eq!(Body::None, Body::None);
        assert_ne!(Body::None, Body::Empty);
        assert_eq!(Body::Empty, Body::Empty);
        assert_ne!(Body::Empty, Body::None);
        assert_eq!(
            Body::Bytes(Bytes::from_static(b"1")),
            Body::Bytes(Bytes::from_static(b"1"))
        );
        assert_ne!(Body::Bytes(Bytes::from_static(b"1")), Body::None);
    }

    #[ntex::test]
    async fn test_body_debug() {
        assert!(format!("{:?}", Body::None).contains("Body::None"));
        assert!(format!("{:?}", Body::Empty).contains("Body::Empty"));
        assert!(format!("{:?}", Body::Bytes(Bytes::from_static(b"1"))).contains('1'));
    }

    #[ntex::test]
    async fn body_stream() {
        let st = BodyStream::new(stream::once(ready(Ok::<_, io::Error>(Bytes::from("1")))));
        assert!(format!("{st:?}").contains("BodyStream"));
        let body: Body = st.into();
        assert!(format!("{body:?}").contains("Body::Message(_)"));
        assert_ne!(body, Body::None);

        let res = ResponseBody::new(body);
        assert!(res.as_ref().is_some());
    }

    #[ntex::test]
    async fn boxed_body_stream() {
        let st = BoxedBodyStream::new(stream::once(ready(Ok::<_, Rc<dyn Error>>(Bytes::from(
            "1",
        )))));
        assert!(format!("{st:?}").contains("BoxedBodyStream"));
        let body: Body = st.into();
        assert!(format!("{body:?}").contains("Body::Message(_)"));
        assert_ne!(body, Body::None);

        let res = ResponseBody::new(body);
        assert!(res.as_ref().is_some());
    }

    #[ntex::test]
    async fn body_skips_empty_chunks() {
        let mut body = BodyStream::new(stream::iter(
            ["1", "", "2"]
                .iter()
                .map(|&v| Ok(Bytes::from(v)) as Result<Bytes, io::Error>),
        ));
        assert_eq!(
            poll_fn(|cx| body.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("1")),
        );
        assert_eq!(
            poll_fn(|cx| body.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("2")),
        );
    }

    #[ntex::test]
    async fn sized_skips_empty_chunks() {
        let mut body = SizedStream::new(
            2,
            stream::iter(["1", "", "2"].iter().map(|&v| Ok(Bytes::from(v)))),
        );
        assert!(format!("{body:?}").contains("SizedStream"));
        assert_eq!(
            poll_fn(|cx| body.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("1")),
        );
        assert_eq!(
            poll_fn(|cx| body.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("2")),
        );
    }

    #[ntex::test]
    async fn boxed_body_skips_empty_chunks() {
        let mut body = BoxedBodyStream::new(stream::iter(
            ["1", "", "2"]
                .iter()
                .map(|&v| Ok(Bytes::from(v)) as Result<Bytes, Rc<dyn Error>>),
        ));
        assert_eq!(
            poll_fn(|cx| body.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("1")),
        );
        assert_eq!(
            poll_fn(|cx| body.poll_next_chunk(cx)).await.unwrap().ok(),
            Some(Bytes::from("2")),
        );
        assert!(poll_fn(|cx| body.poll_next_chunk(cx)).await.is_none());
    }
}