/rust/registry/src/index.crates.io-1949cf8c6b5b557f/http-body-1.0.1/src/lib.rs

Source
#![deny(
    missing_debug_implementations,
    missing_docs,
    unreachable_pub,
    clippy::missing_safety_doc,
    clippy::undocumented_unsafe_blocks
)]
#![cfg_attr(test, deny(warnings))]

//! Asynchronous HTTP request or response body.
//!
//! See [`Body`] for more details.
//!
//! [`Body`]: trait.Body.html

mod frame;
mod size_hint;

pub use self::frame::Frame;
pub use self::size_hint::SizeHint;

use bytes::{Buf, Bytes};
use std::convert::Infallible;
use std::ops;
use std::pin::Pin;
use std::task::{Context, Poll};

/// Trait representing a streaming body of a Request or Response.
///
/// Individual frames are streamed via the `poll_frame` function, which asynchronously yields
/// instances of [`Frame<Data>`].
///
/// Frames can contain a data buffer of type `Self::Data`. Frames can also contain an optional
/// set of trailers used to finalize the request/response exchange. This is mostly used when using
/// the HTTP/2.0 protocol.
///
/// The `size_hint` function provides insight into the total number of bytes that will be streamed.
pub trait Body {
    /// Values yielded by the `Body`.
    type Data: Buf;

    /// The error type this `Body` might generate.
    type Error;

    #[allow(clippy::type_complexity)]
    /// Attempt to pull out the next data buffer of this stream.
    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>>;

    /// Returns `true` when the end of stream has been reached.
    ///
    /// An end of stream means that `poll_frame` will return `None`.
    ///
    /// A return value of `false` **does not** guarantee that a value will be
    /// returned from `poll_frame`.
    fn is_end_stream(&self) -> bool {
        false
    }

    /// Returns the bounds on the remaining length of the stream.
    ///
    /// When the **exact** remaining length of the stream is known, the upper bound will be set and
    /// will equal the lower bound.
    fn size_hint(&self) -> SizeHint {
        SizeHint::default()
    }
}

impl<T: Body + Unpin + ?Sized> Body for &mut T {
    type Data = T::Data;
    type Error = T::Error;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        Pin::new(&mut **self).poll_frame(cx)
    }

    fn is_end_stream(&self) -> bool {
        Pin::new(&**self).is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        Pin::new(&**self).size_hint()
    }
}

impl<P> Body for Pin<P>
where
    P: Unpin + ops::DerefMut,
    P::Target: Body,
{
    type Data = <<P as ops::Deref>::Target as Body>::Data;
    type Error = <<P as ops::Deref>::Target as Body>::Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        Pin::get_mut(self).as_mut().poll_frame(cx)
    }

    fn is_end_stream(&self) -> bool {
        self.as_ref().is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        self.as_ref().size_hint()
    }
}

impl<T: Body + Unpin + ?Sized> Body for Box<T> {
    type Data = T::Data;
    type Error = T::Error;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        Pin::new(&mut **self).poll_frame(cx)
    }

    fn is_end_stream(&self) -> bool {
        self.as_ref().is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        self.as_ref().size_hint()
    }
}

impl<B: Body> Body for http::Request<B> {
    type Data = B::Data;
    type Error = B::Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        // SAFETY:
        // A pin projection.
        unsafe {
            self.map_unchecked_mut(http::Request::body_mut)
                .poll_frame(cx)
        }
    }

    fn is_end_stream(&self) -> bool {
        self.body().is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        self.body().size_hint()
    }
}

impl<B: Body> Body for http::Response<B> {
    type Data = B::Data;
    type Error = B::Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        // SAFETY:
        // A pin projection.
        unsafe {
            self.map_unchecked_mut(http::Response::body_mut)
                .poll_frame(cx)
        }
    }

    fn is_end_stream(&self) -> bool {
        self.body().is_end_stream()
    }

    fn size_hint(&self) -> SizeHint {
        self.body().size_hint()
    }
}

impl Body for String {
    type Data = Bytes;
    type Error = Infallible;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        if !self.is_empty() {
            let s = std::mem::take(&mut *self);
            Poll::Ready(Some(Ok(Frame::data(s.into_bytes().into()))))
        } else {
            Poll::Ready(None)
        }
    }

    fn is_end_stream(&self) -> bool {
        self.is_empty()
    }

    fn size_hint(&self) -> SizeHint {
        SizeHint::with_exact(self.len() as u64)
    }
}

#[cfg(test)]
fn _assert_bounds() {
    fn can_be_trait_object(_: &dyn Body<Data = std::io::Cursor<Vec<u8>>, Error = std::io::Error>) {}
}

Coverage Report

Created: 2026-02-14 06:16

Line	Count	Source
1		#![deny(
2		missing_debug_implementations,
3		missing_docs,
4		unreachable_pub,
5		clippy::missing_safety_doc,
6		clippy::undocumented_unsafe_blocks
7		)]
8		#![cfg_attr(test, deny(warnings))]
9
10		//! Asynchronous HTTP request or response body.
11		//!
12		//! See [`Body`] for more details.
13		//!
14		//! [`Body`]: trait.Body.html
15
16		mod frame;
17		mod size_hint;
18
19		pub use self::frame::Frame;
20		pub use self::size_hint::SizeHint;
21
22		use bytes::{Buf, Bytes};
23		use std::convert::Infallible;
24		use std::ops;
25		use std::pin::Pin;
26		use std::task::{Context, Poll};
27
28		/// Trait representing a streaming body of a Request or Response.
29		///
30		/// Individual frames are streamed via the `poll_frame` function, which asynchronously yields
31		/// instances of [`Frame<Data>`].
32		///
33		/// Frames can contain a data buffer of type `Self::Data`. Frames can also contain an optional
34		/// set of trailers used to finalize the request/response exchange. This is mostly used when using
35		/// the HTTP/2.0 protocol.
36		///
37		/// The `size_hint` function provides insight into the total number of bytes that will be streamed.
38		pub trait Body {
39		/// Values yielded by the `Body`.
40		type Data: Buf;
41
42		/// The error type this `Body` might generate.
43		type Error;
44
45		#[allow(clippy::type_complexity)]
46		/// Attempt to pull out the next data buffer of this stream.
47		fn poll_frame(
48		self: Pin<&mut Self>,
49		cx: &mut Context<'_>,
50		) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>>;
51
52		/// Returns `true` when the end of stream has been reached.
53		///
54		/// An end of stream means that `poll_frame` will return `None`.
55		///
56		/// A return value of `false` does not guarantee that a value will be
57		/// returned from `poll_frame`.
58	0	fn is_end_stream(&self) -> bool {
59	0	false
60	0	}
61
62		/// Returns the bounds on the remaining length of the stream.
63		///
64		/// When the exact remaining length of the stream is known, the upper bound will be set and
65		/// will equal the lower bound.
66	0	fn size_hint(&self) -> SizeHint {
67	0	SizeHint::default()
68	0	} Unexecuted instantiation: <http_body_util::combinators::map_frame::MapFrame<hyper::body::incoming::Incoming, <tonic::codec::decode::Streaming<ztunnel::xds::types::istio::ca::IstioCertificateResponse>>::new<hyper::body::incoming::Incoming, tonic_prost::codec::ProstDecoder<ztunnel::xds::types::istio::ca::IstioCertificateResponse>>::{closure#0}> as http_body::Body>::size_hint Unexecuted instantiation: <http_body_util::combinators::map_frame::MapFrame<hyper::body::incoming::Incoming, <tonic::codec::decode::Streaming<ztunnel::xds::types::service::discovery::v3::DeltaDiscoveryResponse>>::new<hyper::body::incoming::Incoming, tonic_prost::codec::ProstDecoder<ztunnel::xds::types::service::discovery::v3::DeltaDiscoveryResponse>>::{closure#0}> as http_body::Body>::size_hint Unexecuted instantiation: <tonic::codec::encode::EncodeBody<tonic_prost::codec::ProstEncoder<ztunnel::xds::types::istio::ca::IstioCertificateRequest>, tokio_stream::stream_ext::map::Map<tokio_stream::once::Once<ztunnel::xds::types::istio::ca::IstioCertificateRequest>, core::result::Result<ztunnel::xds::types::istio::ca::IstioCertificateRequest, tonic::status::Status>::Ok>> as http_body::Body>::size_hint Unexecuted instantiation: <tonic::codec::encode::EncodeBody<tonic_prost::codec::ProstEncoder<ztunnel::xds::types::service::discovery::v3::DeltaDiscoveryRequest>, tokio_stream::stream_ext::map::Map<async_stream::async_stream::AsyncStream<ztunnel::xds::types::service::discovery::v3::DeltaDiscoveryRequest, <ztunnel::xds::client::AdsClient>::run_internal::{closure#0}::{closure#3}>, core::result::Result<ztunnel::xds::types::service::discovery::v3::DeltaDiscoveryRequest, tonic::status::Status>::Ok>> as http_body::Body>::size_hint Unexecuted instantiation: <_ as http_body::Body>::size_hint
69		}
70
71		impl<T: Body + Unpin + ?Sized> Body for &mut T {
72		type Data = T::Data;
73		type Error = T::Error;
74
75	0	fn poll_frame(
76	0	mut self: Pin<&mut Self>,
77	0	cx: &mut Context<'_>,
78	0	) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
79	0	Pin::new(&mut **self).poll_frame(cx)
80	0	}
81
82	0	fn is_end_stream(&self) -> bool {
83	0	Pin::new(&**self).is_end_stream()
84	0	}
85
86	0	fn size_hint(&self) -> SizeHint {
87	0	Pin::new(&**self).size_hint()
88	0	}
89		}
90
91		impl<P> Body for Pin<P>
92		where
93		P: Unpin + ops::DerefMut,
94		P::Target: Body,
95		{
96		type Data = <<P as ops::Deref>::Target as Body>::Data;
97		type Error = <<P as ops::Deref>::Target as Body>::Error;
98
99	0	fn poll_frame(
100	0	self: Pin<&mut Self>,
101	0	cx: &mut Context<'_>,
102	0	) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
103	0	Pin::get_mut(self).as_mut().poll_frame(cx)
104	0	}
105
106	0	fn is_end_stream(&self) -> bool {
107	0	self.as_ref().is_end_stream()
108	0	} Unexecuted instantiation: <core::pin::Pin<&mut http_body_util::full::Full<bytes::bytes::Bytes>> as http_body::Body>::is_end_stream Unexecuted instantiation: <core::pin::Pin<&mut tonic::body::Body> as http_body::Body>::is_end_stream Unexecuted instantiation: <core::pin::Pin<alloc::boxed::Box<dyn http_body::Body<Error = tonic::status::Status, Data = bytes::bytes::Bytes> + core::marker::Send>> as http_body::Body>::is_end_stream Unexecuted instantiation: <core::pin::Pin<_> as http_body::Body>::is_end_stream
109
110	0	fn size_hint(&self) -> SizeHint {
111	0	self.as_ref().size_hint()
112	0	} Unexecuted instantiation: <core::pin::Pin<alloc::boxed::Box<dyn http_body::Body<Error = tonic::status::Status, Data = bytes::bytes::Bytes> + core::marker::Send>> as http_body::Body>::size_hint Unexecuted instantiation: <core::pin::Pin<_> as http_body::Body>::size_hint
113		}
114
115		impl<T: Body + Unpin + ?Sized> Body for Box<T> {
116		type Data = T::Data;
117		type Error = T::Error;
118
119	0	fn poll_frame(
120	0	mut self: Pin<&mut Self>,
121	0	cx: &mut Context<'_>,
122	0	) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
123	0	Pin::new(&mut **self).poll_frame(cx)
124	0	}
125
126	0	fn is_end_stream(&self) -> bool {
127	0	self.as_ref().is_end_stream()
128	0	}
129
130	0	fn size_hint(&self) -> SizeHint {
131	0	self.as_ref().size_hint()
132	0	}
133		}
134
135		impl<B: Body> Body for http::Request<B> {
136		type Data = B::Data;
137		type Error = B::Error;
138
139	0	fn poll_frame(
140	0	self: Pin<&mut Self>,
141	0	cx: &mut Context<'_>,
142	0	) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
143		// SAFETY:
144		// A pin projection.
145		unsafe {
146	0	self.map_unchecked_mut(http::Request::body_mut)
147	0	.poll_frame(cx)
148		}
149	0	}
150
151	0	fn is_end_stream(&self) -> bool {
152	0	self.body().is_end_stream()
153	0	}
154
155	0	fn size_hint(&self) -> SizeHint {
156	0	self.body().size_hint()
157	0	}
158		}
159
160		impl<B: Body> Body for http::Response<B> {
161		type Data = B::Data;
162		type Error = B::Error;
163
164	0	fn poll_frame(
165	0	self: Pin<&mut Self>,
166	0	cx: &mut Context<'_>,
167	0	) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
168		// SAFETY:
169		// A pin projection.
170		unsafe {
171	0	self.map_unchecked_mut(http::Response::body_mut)
172	0	.poll_frame(cx)
173		}
174	0	}
175
176	0	fn is_end_stream(&self) -> bool {
177	0	self.body().is_end_stream()
178	0	}
179
180	0	fn size_hint(&self) -> SizeHint {
181	0	self.body().size_hint()
182	0	}
183		}
184
185		impl Body for String {
186		type Data = Bytes;
187		type Error = Infallible;
188
189	0	fn poll_frame(
190	0	mut self: Pin<&mut Self>,
191	0	_cx: &mut Context<'_>,
192	0	) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
193	0	if !self.is_empty() {
194	0	let s = std::mem::take(&mut *self);
195	0	Poll::Ready(Some(Ok(Frame::data(s.into_bytes().into()))))
196		} else {
197	0	Poll::Ready(None)
198		}
199	0	}
200
201	0	fn is_end_stream(&self) -> bool {
202	0	self.is_empty()
203	0	}
204
205	0	fn size_hint(&self) -> SizeHint {
206	0	SizeHint::with_exact(self.len() as u64)
207	0	}
208		}
209
210		#[cfg(test)]
211		fn _assert_bounds() {
212		fn can_be_trait_object(_: &dyn Body<Data = std::io::Cursor<Vec<u8>>, Error = std::io::Error>) {}
213		}