Skip to content

feat(response): introduce trailers support #2788

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 4 commits into
base: master
Choose a base branch
from
+204 −0
Draft

feat(response): introduce trailers support #2788

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c994c9c
feat(response): introduce trailers support
0x676e67 Aug 8, 2025
a853453
fix build
0x676e67 Aug 8, 2025
a60050a
test: remove `#[cfg(feature = "stream")]`
0x676e67 Aug 8, 2025
21a49c8
Merge branch 'master' into trailers
0x676e67 Aug 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 112 additions & 0 deletions src/async_impl/response.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,19 @@
use std::fmt;
use std::net::SocketAddr;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use bytes::Bytes;
use http_body::{Frame, SizeHint};
use http_body_util::BodyExt;
use hyper::{HeaderMap, StatusCode, Version};
use hyper_util::client::legacy::connect::HttpInfo;
#[cfg(feature = "json")]
use serde::de::DeserializeOwned;
#[cfg(feature = "json")]
use serde_json;
use tokio::sync::oneshot::{self, Receiver};
use tokio::time::Sleep;
use url::Url;

Expand All @@ -31,6 +34,7 @@ pub struct Response {
// Boxed to save space (11 words to 1 word), and it's not accessed
// frequently internally.
url: Box<Url>,
trailers_rx: Receiver<HeaderMap>,
}

impl Response {
Expand All @@ -42,16 +46,20 @@ impl Response {
read_timeout: Option<Duration>,
) -> Response {
let (mut parts, body) = res.into_parts();
let (body, trailers_rx) = extract_trailers_from_body(body);

let decoder = Decoder::detect(
&mut parts.headers,
super::body::response(body, total_timeout, read_timeout),
accepts,
);

let res = hyper::Response::from_parts(parts, decoder);

Response {
res,
url: Box::new(url),
trailers_rx,
}
}

Expand Down Expand Up @@ -424,6 +432,22 @@ impl Response {
}
}

/// Get the response trailers if available.
///
/// Trailers are additional headers sent after the response body in HTTP/1.1 chunked
/// encoding or HTTP/2 responses. They are typically used for metadata that can only
/// be determined after processing the entire response body.
#[inline]
pub async fn trailers(&mut self) -> crate::Result<Option<HeaderMap>> {
match self.trailers_rx.try_recv() {
Copy link

@anuraaga anuraaga Sep 25, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because this channel isn't actually used for synchronization, does it make sense to use a simpler OnceLock instead? Then trailers can be returned as a reference to match headers and doesn't need to be async.

aaf7bb1

Copy link
Contributor Author

@0x676e67 0x676e67 Sep 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Feel free to open a PR, I might not have the time to implement it myself.

Ok(trailers) => Ok(Some(trailers)),
Err(err) => match err {
oneshot::error::TryRecvError::Empty => Ok(None),
oneshot::error::TryRecvError::Closed => Err(crate::error::body(err)),
},
}
}

// private

// The Response's body is an implementation detail.
Expand Down Expand Up @@ -462,6 +486,9 @@ impl<T: Into<Body>> From<http::Response<T>> for Response {

let (mut parts, body) = r.into_parts();
let body: crate::async_impl::body::Body = body.into();

let (body, trailers_rx) = extract_trailers_from_body(body);

let decoder = Decoder::detect(
&mut parts.headers,
ResponseBody::new(body.map_err(Into::into)),
Expand All @@ -476,6 +503,7 @@ impl<T: Into<Body>> From<http::Response<T>> for Response {
Response {
res,
url: Box::new(url),
trailers_rx,
}
}
}
Expand All @@ -490,6 +518,90 @@ impl From<Response> for http::Response<Body> {
}
}

pin_project_lite::pin_project! {
/// A body wrapper that extracts HTTP trailers while preserving the original size hint.
///
/// This wrapper monitors HTTP frames for trailers and sends them through a oneshot
/// channel when found, while maintaining the original body's size hint and other
/// characteristics to ensure `content_length()` continues to work correctly.
///
/// HTTP trailers are additional headers sent after the response body in chunked
/// encoding (HTTP/1.1) or HTTP/2 responses. They are useful for metadata that
/// can only be determined after processing the entire response body, such as
/// checksums or final status information.
pub struct TrailerExtractingBody<B> {
#[pin]
inner: B,
trailers_tx: Option<oneshot::Sender<HeaderMap>>,
}
}

impl<B> TrailerExtractingBody<B> {
fn new(body: B, trailers_tx: oneshot::Sender<HeaderMap>) -> Self
where
B: http_body::Body,
{
Self {
inner: body,
trailers_tx: Some(trailers_tx),
}
}
}

impl<B> http_body::Body for TrailerExtractingBody<B>
where
B: http_body::Body<Data = Bytes>,
{
type Data = Bytes;
type Error = B::Error;

fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
let this = self.project();

match std::task::ready!(this.inner.poll_frame(cx)) {
Some(Ok(mut frame)) => {
if let Some(trailers) = frame.trailers_mut() {
if let Some(tx) = this.trailers_tx.take() {
let _ = tx.send(std::mem::take(trailers));
}
}
Poll::Ready(Some(Ok(frame)))
}
Some(Err(e)) => Poll::Ready(Some(Err(e))),
None => Poll::Ready(None),
}
}

#[inline]
fn size_hint(&self) -> SizeHint {
self.inner.size_hint()
}

#[inline]
fn is_end_stream(&self) -> bool {
self.inner.is_end_stream()
}
}

fn extract_trailers_from_body<B>(
body: B,
) -> (
http_body_util::combinators::BoxBody<Bytes, B::Error>,
oneshot::Receiver<HeaderMap>,
)
where
B: http_body::Body<Data = Bytes> + Send + Sync + 'static,
{
let (trailers_tx, trailers_rx) = oneshot::channel();
let wrapper = TrailerExtractingBody::new(body, trailers_tx);
let boxed_body = http_body_util::BodyExt::boxed(wrapper);

(boxed_body, trailers_rx)
}

#[cfg(test)]
mod tests {
use super::Response;
Expand Down
92 changes: 92 additions & 0 deletions tests/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -600,3 +600,95 @@ async fn error_has_url() {
let err = reqwest::get(u).await.unwrap_err();
assert_eq!(err.url().map(AsRef::as_ref), Some(u), "{err:?}");
}

#[tokio::test]
async fn response_trailers() {
use tokio::io::AsyncWriteExt;

let server = server::low_level_with_response(|_raw_request, client_socket| {
Box::new(async move {
// Send HTTP response with chunked encoding and trailers
client_socket
.write_all(b"HTTP/1.1 200 OK\r\n")
.await
.expect("write status line");

client_socket
.write_all(b"Transfer-Encoding: chunked\r\n")
.await
.expect("write transfer-encoding header");

client_socket
.write_all(b"Trailer: X-Custom-Trailer, X-Checksum\r\n")
.await
.expect("write trailer header");

client_socket
.write_all(b"\r\n")
.await
.expect("write header end");

// Send chunked body
client_socket
.write_all(b"5\r\nHello\r\n")
.await
.expect("write chunk 1");

client_socket
.write_all(b"6\r\nWorld!\r\n")
.await
.expect("write chunk 2");

// Send end chunk
client_socket
.write_all(b"0\r\n")
.await
.expect("write end chunk");

// Send trailers
client_socket
.write_all(b"X-Custom-Trailer: custom-value\r\n")
.await
.expect("write custom trailer");

client_socket
.write_all(b"X-Checksum: abc123\r\n")
.await
.expect("write checksum trailer");

// End of trailers
client_socket
.write_all(b"\r\n")
.await
.expect("write trailers end");
})
});

let client = Client::new();

let mut res = client
.get(&format!("http://{}/trailers", server.addr()))
.send()
.await
.expect("Failed to get response");

assert_eq!(res.status(), reqwest::StatusCode::OK);

// Read the body using chunk() to preserve response ownership
let mut body_content = Vec::new();

while let Some(chunk) = res.chunk().await.expect("Failed to read chunk") {
body_content.extend_from_slice(&chunk);
}

let body = String::from_utf8(body_content).expect("Invalid UTF-8");
assert_eq!(body, "HelloWorld!");

// Now we can check trailers since the response body has been fully consumed
if let Some(trailers) = res.trailers().await.expect("Failed to get trailers") {
assert_eq!(trailers.get("X-Custom-Trailer").unwrap(), "custom-value");
assert_eq!(trailers.get("X-Checksum").unwrap(), "abc123");
} else {
panic!("Expected trailers but got None");
}
}
Loading