#[cfg(feature = "sse")]
pub mod sse;
#[cfg(feature = "sse")]
pub use sse::Sse;
use bytes::Bytes;
use http_body_util::BodyExt;
use http_body_util::Full;
use http_body_util::combinators::UnsyncBoxBody as BoxBody;
use hyper::body::{Body as HyperBody, Frame, SizeHint};
use hyper::header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue};
use hyper::{Response, StatusCode};
#[cfg(feature = "json")]
use serde::Serialize;
use std::pin::Pin;
use std::task::{Context, Poll};
#[derive(Default)]
pub enum Body {
Full(Full<Bytes>),
#[default]
Empty,
Stream(BoxBody<Bytes, crate::http::error::Error>),
}
impl std::fmt::Debug for Body {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Full(full) => f.debug_tuple("Full").field(full).finish(),
Self::Empty => f.write_str("Empty"),
Self::Stream(_) => f.debug_tuple("Stream").field(&"<stream>").finish(),
}
}
}
impl Body {
#[must_use]
pub const fn empty() -> Self {
Self::Empty
}
pub fn full(bytes: Bytes) -> Self {
Self::Full(Full::new(bytes))
}
pub fn stream<B>(body: B) -> Self
where
B: HyperBody<Data = Bytes> + Send + 'static,
B::Error: Into<crate::http::error::Error>,
{
Self::Stream(BoxBody::new(body.map_err(std::convert::Into::into)))
}
pub async fn collect_bytes(self, limit: usize) -> Result<Bytes, crate::http::error::Error> {
match http_body_util::Limited::new(self, limit).collect().await {
Ok(collected) => Ok(collected.to_bytes()),
Err(e) => {
if e.downcast_ref::<http_body_util::LengthLimitError>()
.is_some()
{
Err(crate::http::error::Error::Rejection {
status: StatusCode::PAYLOAD_TOO_LARGE,
message: "Request body exceeds the maximum allowed size".to_string(),
})
} else {
Err(crate::http::error::Error::Rejection {
status: StatusCode::BAD_REQUEST,
message: format!("Failed to read request body: {e}"),
})
}
}
}
}
}
impl HyperBody for Body {
type Data = Bytes;
type Error = crate::http::error::Error;
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
match self.get_mut() {
Self::Full(full) => match Pin::new(full).poll_frame(cx) {
Poll::Ready(Some(Ok(frame))) => Poll::Ready(Some(Ok(frame))),
Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(
crate::http::error::Error::Internal(e.to_string()),
))),
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
},
Self::Empty => Poll::Ready(None),
Self::Stream(stream) => Pin::new(stream).poll_frame(cx),
}
}
fn is_end_stream(&self) -> bool {
match self {
Self::Full(full) => full.is_end_stream(),
Self::Empty => true,
Self::Stream(stream) => stream.is_end_stream(),
}
}
fn size_hint(&self) -> SizeHint {
match self {
Self::Full(full) => full.size_hint(),
Self::Empty => SizeHint::with_exact(0),
Self::Stream(stream) => stream.size_hint(),
}
}
}
pub trait IntoResponse {
fn into_response(self) -> Response<Body>;
}
impl IntoResponse for Response<Body> {
fn into_response(self) -> Self {
self
}
}
impl IntoResponse for Response<Full<Bytes>> {
fn into_response(self) -> Response<Body> {
let (parts, body) = self.into_parts();
Response::from_parts(parts, Body::Full(body))
}
}
impl IntoResponse for StatusCode {
fn into_response(self) -> Response<Body> {
let mut res = Response::new(Body::empty());
*res.status_mut() = self;
res
}
}
impl IntoResponse for String {
fn into_response(self) -> Response<Body> {
let mut res = Response::new(Body::full(Bytes::from(self)));
let _ = res.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("text/plain; charset=utf-8"),
);
res
}
}
impl IntoResponse for &'static str {
fn into_response(self) -> Response<Body> {
let mut res = Response::new(Body::full(Bytes::from_static(self.as_bytes())));
let _ = res.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("text/plain; charset=utf-8"),
);
res
}
}
impl IntoResponse for Vec<u8> {
fn into_response(self) -> Response<Body> {
let mut res = Response::new(Body::full(Bytes::from(self)));
let _ = res.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("application/octet-stream"),
);
res
}
}
impl IntoResponse for &'static [u8] {
fn into_response(self) -> Response<Body> {
let mut res = Response::new(Body::full(Bytes::from_static(self)));
let _ = res.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("application/octet-stream"),
);
res
}
}
#[derive(Debug, Clone)]
pub struct Html<T>(pub T);
impl<T> IntoResponse for Html<T>
where
T: Into<Bytes>,
{
fn into_response(self) -> Response<Body> {
let mut res = Response::new(Body::full(self.0.into()));
let _ = res.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("text/html; charset=utf-8"),
);
res
}
}
#[cfg(feature = "json")]
thread_local! {
static JSON_WRITE_BUF: std::cell::RefCell<bytes::BytesMut> =
std::cell::RefCell::new(bytes::BytesMut::with_capacity(1024));
}
#[cfg(feature = "json")]
struct BytesMutWriter<'a>(&'a mut bytes::BytesMut);
#[cfg(feature = "json")]
impl std::io::Write for BytesMutWriter<'_> {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
#[cfg(feature = "json")]
pub use crate::routing::extract::Json;
#[cfg(feature = "json")]
impl<T> IntoResponse for Json<T>
where
T: Serialize,
{
fn into_response(self) -> Response<Body> {
#[allow(clippy::single_match_else)]
let result = JSON_WRITE_BUF.with(|buf| match buf.try_borrow_mut() {
Ok(mut b) => {
let res = match serde_json::to_writer(BytesMutWriter(&mut b), &self.0) {
Ok(()) => {
let len = b.len();
Ok(b.split_to(len).freeze())
}
Err(err) => {
b.clear();
Err(err)
}
};
if b.capacity() > 65536 {
*b = bytes::BytesMut::with_capacity(1024);
}
res
}
Err(_) => {
let mut b = Vec::with_capacity(1024);
match serde_json::to_writer(&mut b, &self.0) {
Ok(()) => Ok(Bytes::from(b)),
Err(err) => Err(err),
}
}
});
match result {
Ok(bytes) => {
let mut res = Response::new(Body::full(bytes));
let _ = res
.headers_mut()
.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
res
}
Err(err) => {
let mut res = Response::new(Body::full(Bytes::from(format!(
"Failed to serialize JSON: {err}"
))));
*res.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
let _ = res.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("text/plain; charset=utf-8"),
);
res
}
}
}
}
impl<R> IntoResponse for (StatusCode, R)
where
R: IntoResponse,
{
fn into_response(self) -> Response<Body> {
let (status, res) = self;
let mut response = res.into_response();
*response.status_mut() = status;
response
}
}
#[derive(Debug)]
pub struct ResponseParts {
res: Response<Body>,
}
impl ResponseParts {
pub fn headers_mut(&mut self) -> &mut HeaderMap {
self.res.headers_mut()
}
pub fn extensions_mut(&mut self) -> &mut hyper::http::Extensions {
self.res.extensions_mut()
}
}
pub trait IntoResponseParts {
type Error: IntoResponse;
fn into_response_parts(self, res: ResponseParts) -> Result<ResponseParts, Self::Error>;
}
impl IntoResponseParts for HeaderMap {
type Error = std::convert::Infallible;
fn into_response_parts(self, mut res: ResponseParts) -> Result<ResponseParts, Self::Error> {
res.headers_mut().extend(self);
Ok(res)
}
}
impl<T> IntoResponseParts for Option<T>
where
T: IntoResponseParts,
{
type Error = T::Error;
fn into_response_parts(self, res: ResponseParts) -> Result<ResponseParts, Self::Error> {
match self {
Some(parts) => parts.into_response_parts(res),
None => Ok(res),
}
}
}
impl<T> IntoResponseParts for crate::routing::extract::Extension<T>
where
T: Clone + Send + Sync + 'static,
{
type Error = std::convert::Infallible;
fn into_response_parts(self, mut res: ResponseParts) -> Result<ResponseParts, Self::Error> {
res.extensions_mut().insert(self.0);
Ok(res)
}
}
#[derive(Debug, Clone, Copy)]
pub struct AppendHeaders<I>(pub I);
impl<I, K, V> IntoResponseParts for AppendHeaders<I>
where
I: IntoIterator<Item = (K, V)>,
K: TryInto<HeaderName>,
V: TryInto<HeaderValue>,
{
type Error = crate::http::error::Error;
fn into_response_parts(self, mut res: ResponseParts) -> Result<ResponseParts, Self::Error> {
for (key, value) in self.0 {
let key = key
.try_into()
.map_err(|_| crate::http::error::Error::Rejection {
status: StatusCode::INTERNAL_SERVER_ERROR,
message: "AppendHeaders: invalid header name".to_string(),
})?;
let value = value
.try_into()
.map_err(|_| crate::http::error::Error::Rejection {
status: StatusCode::INTERNAL_SERVER_ERROR,
message: "AppendHeaders: invalid header value".to_string(),
})?;
res.headers_mut().append(key, value);
}
Ok(res)
}
}
#[cfg(feature = "cookies")]
impl IntoResponseParts for crate::routing::extract::Cookies {
type Error = std::convert::Infallible;
fn into_response_parts(self, mut res: ResponseParts) -> Result<ResponseParts, Self::Error> {
for cookie in self.jar.delta() {
if let Ok(header_val) = HeaderValue::try_from(cookie.encoded().to_string()) {
res.headers_mut()
.append(hyper::header::SET_COOKIE, header_val);
}
}
Ok(res)
}
}
macro_rules! impl_into_response_for_parts_tuples {
($($T:ident),+) => {
impl<R, $($T),+> IntoResponse for ($($T,)+ R)
where
R: IntoResponse,
$( $T: IntoResponseParts, )+
{
fn into_response(self) -> Response<Body> {
#[allow(non_snake_case)]
let ($($T,)+ res) = self;
let mut parts = ResponseParts { res: res.into_response() };
$(
parts = match $T.into_response_parts(parts) {
Ok(p) => p,
Err(rejection) => return rejection.into_response(),
};
)+
parts.res
}
}
impl<R, $($T),+> IntoResponse for (StatusCode, $($T,)+ R)
where
R: IntoResponse,
$( $T: IntoResponseParts, )+
{
fn into_response(self) -> Response<Body> {
#[allow(non_snake_case)]
let (status, $($T,)+ res) = self;
let mut response = <($($T,)+ R) as IntoResponse>::into_response(($($T,)+ res));
*response.status_mut() = status;
response
}
}
};
}
impl_into_response_for_parts_tuples!(T1);
impl_into_response_for_parts_tuples!(T1, T2);
impl_into_response_for_parts_tuples!(T1, T2, T3);
impl_into_response_for_parts_tuples!(T1, T2, T3, T4);
impl_into_response_for_parts_tuples!(T1, T2, T3, T4, T5);
impl_into_response_for_parts_tuples!(T1, T2, T3, T4, T5, T6);
impl_into_response_for_parts_tuples!(T1, T2, T3, T4, T5, T6, T7);
impl_into_response_for_parts_tuples!(T1, T2, T3, T4, T5, T6, T7, T8);
impl IntoResponse for () {
fn into_response(self) -> Response<Body> {
Response::builder()
.status(StatusCode::OK)
.body(Body::empty())
.unwrap_or_else(|_| Response::new(Body::empty()))
}
}
impl<T, E> IntoResponse for Result<T, E>
where
T: IntoResponse,
E: IntoResponse,
{
fn into_response(self) -> Response<Body> {
match self {
Ok(value) => value.into_response(),
Err(err) => err.into_response(),
}
}
}
impl IntoResponse for std::convert::Infallible {
fn into_response(self) -> Response<Body> {
match self {}
}
}
#[derive(Debug, Clone)]
pub struct Redirect {
status_code: StatusCode,
location: HeaderValue,
}
impl Redirect {
#[must_use]
pub fn to(uri: &str) -> Self {
Self {
status_code: StatusCode::SEE_OTHER,
location: HeaderValue::try_from(uri).unwrap_or_else(|_| HeaderValue::from_static("/")),
}
}
#[must_use]
pub fn temporary(uri: &str) -> Self {
Self {
status_code: StatusCode::TEMPORARY_REDIRECT,
location: HeaderValue::try_from(uri).unwrap_or_else(|_| HeaderValue::from_static("/")),
}
}
#[must_use]
pub fn permanent(uri: &str) -> Self {
Self {
status_code: StatusCode::PERMANENT_REDIRECT,
location: HeaderValue::try_from(uri).unwrap_or_else(|_| HeaderValue::from_static("/")),
}
}
}
impl IntoResponse for Redirect {
fn into_response(self) -> Response<Body> {
let mut resp = Response::new(Body::empty());
*resp.status_mut() = self.status_code;
let _ = resp
.headers_mut()
.insert(hyper::header::LOCATION, self.location);
resp
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::routing::extract::Cookies;
use hyper::HeaderMap;
#[test]
fn test_body_debug_and_size_hint() {
let b1 = Body::empty();
assert!(format!("{b1:?}").contains("Empty"));
assert!(b1.is_end_stream());
assert_eq!(b1.size_hint().exact(), Some(0));
let b2 = Body::full(Bytes::from("test"));
assert!(format!("{b2:?}").contains("Full"));
assert!(!b2.is_end_stream());
assert_eq!(b2.size_hint().exact(), Some(4));
let stream_body = BoxBody::new(
http_body_util::Empty::<Bytes>::new()
.map_err(|e| crate::http::error::Error::Internal(e.to_string())),
);
let b3 = Body::Stream(stream_body);
assert!(format!("{b3:?}").contains("Stream"));
assert!(b3.is_end_stream());
assert_eq!(b3.size_hint().exact(), Some(0));
}
#[tokio::test]
async fn test_body_poll_frame() {
use hyper::body::Body as _;
let mut b1 = Body::full(Bytes::from("a"));
let mut b1_pin = Pin::new(&mut b1);
let cx = &mut Context::from_waker(futures::task::noop_waker_ref());
let f1 = b1_pin.as_mut().poll_frame(cx);
assert!(matches!(f1, Poll::Ready(Some(Ok(_)))));
let f2 = b1_pin.as_mut().poll_frame(cx);
assert!(matches!(f2, Poll::Ready(None)));
let mut b2 = Body::empty();
let f3 = Pin::new(&mut b2).poll_frame(cx);
assert!(matches!(f3, Poll::Ready(None)));
}
#[cfg(feature = "json")]
struct FailSerialize;
#[cfg(feature = "json")]
impl Serialize for FailSerialize {
fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
Err(serde::ser::Error::custom("failed"))
}
}
#[test]
fn test_into_response_implementations() {
let full_resp = Response::new(Full::new(Bytes::from("abc")));
let r1 = full_resp.into_response();
assert_eq!(r1.status(), StatusCode::OK);
let v: Vec<u8> = vec![1, 2, 3];
let r2 = v.into_response();
assert_eq!(
r2.headers().get(CONTENT_TYPE).unwrap(),
"application/octet-stream"
);
let s: &'static [u8] = b"static";
let r3 = s.into_response();
assert_eq!(
r3.headers().get(CONTENT_TYPE).unwrap(),
"application/octet-stream"
);
#[cfg(feature = "json")]
{
let fail_json = Json(FailSerialize);
let r4 = fail_json.into_response();
assert_eq!(r4.status(), StatusCode::INTERNAL_SERVER_ERROR);
}
let mut h = HeaderMap::new();
let _ = h.insert("x-custom", HeaderValue::from_static("val"));
let r5 = (h.clone(), "body").into_response();
assert_eq!(r5.headers().get("x-custom").unwrap(), "val");
let r6 = (StatusCode::CREATED, h.clone(), "body").into_response();
assert_eq!(r6.status(), StatusCode::CREATED);
assert_eq!(r6.headers().get("x-custom").unwrap(), "val");
let cookies = Cookies::new();
let r7 = (StatusCode::ACCEPTED, cookies, "body").into_response();
assert_eq!(r7.status(), StatusCode::ACCEPTED);
let r8 = ().into_response();
assert_eq!(r8.status(), StatusCode::OK);
let res_ok: Result<&str, &str> = Ok("ok");
let r9 = res_ok.into_response();
assert_eq!(r9.status(), StatusCode::OK);
let res_err: Result<&str, &str> = Err("err");
let r10 = res_err.into_response();
assert_eq!(r10.status(), StatusCode::OK);
}
#[test]
fn option_into_response_parts_some_and_none() {
let mut h = HeaderMap::new();
let _ = h.insert("x-opt", HeaderValue::from_static("present"));
let with_some = (Some(h), "body").into_response();
assert_eq!(with_some.headers().get("x-opt").unwrap(), "present");
let with_none = (None::<HeaderMap>, "body").into_response();
assert!(with_none.headers().get("x-opt").is_none());
assert_eq!(with_none.status(), StatusCode::OK);
}
#[test]
fn extension_into_response_parts_inserts_into_response_extensions() {
use crate::routing::extract::Extension;
#[derive(Clone)]
struct Marker(u32);
let resp = (Extension(Marker(42)), "body").into_response();
assert_eq!(resp.extensions().get::<Marker>().unwrap().0, 42);
}
#[test]
fn response_parts_extensions_mut_is_reachable_directly() {
let mut parts = ResponseParts {
res: Response::new(Body::empty()),
};
let _ = parts.extensions_mut().insert(7u32);
assert_eq!(parts.res.extensions().get::<u32>(), Some(&7));
}
#[test]
fn append_headers_appends_without_replacing() {
let mut existing = HeaderMap::new();
let _ = existing.insert("x-multi", HeaderValue::from_static("first"));
let resp = (
existing,
AppendHeaders([("x-multi", "second"), ("x-other", "value")]),
"body",
)
.into_response();
let all: Vec<_> = resp
.headers()
.get_all("x-multi")
.iter()
.map(|v| v.to_str().unwrap())
.collect();
assert_eq!(all, vec!["first", "second"]);
assert_eq!(resp.headers().get("x-other").unwrap(), "value");
}
#[test]
fn append_headers_rejects_an_invalid_header_name() {
let resp = (AppendHeaders([("bad header name", "value")]), "body").into_response();
assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
}
#[test]
fn append_headers_rejects_an_invalid_header_value() {
let resp = (AppendHeaders([("x-ok-name", "bad\nvalue")]), "body").into_response();
assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
}
#[cfg(feature = "json")]
#[test]
fn json_response_shrinks_its_thread_local_buffer_after_a_large_payload() {
let big = "x".repeat(80 * 1024);
let resp = Json(big).into_response();
assert_eq!(resp.status(), StatusCode::OK);
let resp2 = Json("small").into_response();
assert_eq!(resp2.status(), StatusCode::OK);
}
}