How are response trailers managed? #3620
-
|
Hello, I am developing a small gRPC proxy using Whevener the proxy responds by forwarding the response, it works as expected. However, as soon as the response is buffered, I get this error on the client side:
Here is the part of the code: pub async fn forward(request: Request<Body>) -> Result<Response<Body>, hyper::Error> {
let forwarded_request = ...;
let client = Client::builder().http2_only(true).build_http();
let response = client.request(forwarded_request).await.unwrap();
// Getting a "server closed the stream without sending trailers" error from client with this
// let (parts, body) = resp.into_parts();
// let body_bytes = hyper::body::to_bytes(body).await.unwrap();
// let response = Response::from_parts(parts, Body::from(body_bytes));
Ok(response)
}I am not familiar with what trailers are. The initial request has a "te": "trailers" header, so I am guessing my response is missing something when it is buffered by the proxy. Are "trailers" something I can inspect in an Thanks in advance! |
Beta Was this translation helpful? Give feedback.
Replies: 1 comment 2 replies
-
|
The trailers are basically "headers" that come at the end of the body. They sorta exist in HTTP/1.1, but are a better supported thing in HTTP/2 and 3, so gRPC uses them to indicate the "status" of a message. hyper (0.14.x) exposes them in the body as However, gRPC makes heavy use of streaming bodies, so you might find buffering till the end in a proxy causes problems. Some gRPC services assume they can send a message sporadically or on an interval, and expect a streamed reply back, such that buffering would disrupt that. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks a lot for your explanations @seanmonstar ! I tried a few things now trying to reinject trailers such that I can still return an pub async fn forward(request: Request<Body>) -> Result<Response<Body>, hyper::Error> {
let forwarded_request: Request<Body> = ...;
let client = Client::builder().http2_only(true).build_http();
let response = client.request(forwarded_request).await.unwrap();
let (parts, body) = response.into_parts();
let collected: Collected<Bytes> = body.collect().await.unwrap();
let response = Response::from_parts(parts, Body::from(collected)); // <- this doesn't work, how to get a Response<Body> from parts and Collected<Bytes> ?
Ok(response)
}I am sorry for requesting your help again :/ |
Beta Was this translation helpful? Give feedback.
-
|
I ended up migrating to hyper 1.2 and made a custom body for buffered content, inspired from Here is the code in case someone is interested: use bytes::{Buf, BytesMut};
use futures::Future;
use hyper::body::{Body, Frame};
use hyper::HeaderMap;
use std::convert::Infallible;
use std::hash::Hash;
use std::pin::Pin;
use std::task::{Context, Poll};
use pin_project_lite::pin_project;
pin_project! {
pub struct BufferBody<T>
where
T: Body,
T: ?Sized,
{
pub(crate) collected: Option<BufferedBody>,
#[pin]
pub(crate) body: T,
}
}
impl<T: Body + ?Sized> Future for BufferBody<T> {
type Output = Result<BufferedBody, T::Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> std::task::Poll<Self::Output> {
let mut me = self.project();
loop {
let frame = futures_core::ready!(me.body.as_mut().poll_frame(cx));
let frame = if let Some(frame) = frame {
frame?
} else {
return Poll::Ready(Ok(me.collected.take().expect("polled after complete")));
};
me.collected.as_mut().unwrap().push_frame(frame);
}
}
}
#[derive(Debug, Default, Clone)]
pub struct BufferedBody {
bufs: BytesMut,
trailers: Option<HeaderMap>,
}
impl BufferedBody {
pub fn trailers(&self) -> Option<&HeaderMap> {
self.trailers.as_ref()
}
pub(crate) fn push_frame<B>(&mut self, frame: Frame<B>)
where
B: Buf,
{
let frame = match frame.into_data() {
Ok(mut data) => {
while data.has_remaining() {
self.bufs.extend(data.chunk());
data.advance(data.remaining());
}
return;
}
Err(frame) => frame,
};
if let Ok(trailers) = frame.into_trailers() {
if let Some(current) = &mut self.trailers {
current.extend(trailers);
} else {
self.trailers = Some(trailers);
}
};
}
pub fn collect_buffered<T>(body: T) -> BufferBody<T>
where
T: Body,
T: Sized,
{
BufferBody {
body: body,
collected: Some(BufferedBody::default()),
}
}
}
impl Body for BufferedBody {
type Data = BytesMut;
type Error = Infallible;
fn poll_frame(
mut self: Pin<&mut Self>, _: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
let frame = if self.bufs.len() > 0
/* Shall we skip this frame if body is empty? */
{
let frame = Frame::data(self.bufs.to_owned());
self.bufs.clear();
frame
} else if let Some(trailers) = self.trailers.take() {
Frame::trailers(trailers)
} else {
return Poll::Ready(None);
};
Poll::Ready(Some(Ok(frame)))
}
} |
Beta Was this translation helpful? Give feedback.
The trailers are basically "headers" that come at the end of the body. They sorta exist in HTTP/1.1, but are a better supported thing in HTTP/2 and 3, so gRPC uses them to indicate the "status" of a message. hyper (0.14.x) exposes them in the body as
poll_trailers(). If you make use ofhyper::body::HttpBody, the trait, there's a replacement forto_bytes(), which isHttpBody::collect(). It will also have collected the trailers (to_bytes()just tosses them).However, gRPC makes heavy use of streaming bodies, so you might find buffering till the end in a proxy causes problems. Some gRPC services assume they can send a message sporadically or on an interval, and expect a streamed reply back, …