diff options
| author | Alex Auvolat <alex@adnab.me> | 2024-02-13 11:24:56 +0100 |
|---|---|---|
| committer | Alex Auvolat <alex@adnab.me> | 2024-02-13 11:36:28 +0100 |
| commit | cf2af186fcc0c8f581a966454b6cd4720d3821f0 (patch) | |
| tree | 37a978ba9ffb780fc828cff7b8ec93662d50884f /src/api/helpers.rs | |
| parent | db48dd3d6c1f9e86a62e9b8edfce2c1620bcd5f3 (diff) | |
| parent | 823078b4cdaf93e09de0847c5eaa75beb7b26b7f (diff) | |
| download | garage-cf2af186fcc0c8f581a966454b6cd4720d3821f0.tar.gz garage-cf2af186fcc0c8f581a966454b6cd4720d3821f0.zip | |
Merge branch 'main' into next-0.10
Diffstat (limited to 'src/api/helpers.rs')
| -rw-r--r-- | src/api/helpers.rs | 68 |
1 files changed, 62 insertions, 6 deletions
diff --git a/src/api/helpers.rs b/src/api/helpers.rs index 1d55ebd5..5f488912 100644 --- a/src/api/helpers.rs +++ b/src/api/helpers.rs @@ -1,7 +1,17 @@ -use hyper::{Body, Request, Response}; +use std::convert::Infallible; + +use futures::{Stream, StreamExt, TryStreamExt}; + +use http_body_util::{BodyExt, Full as FullBody}; +use hyper::{ + body::{Body, Bytes}, + Request, Response, +}; use idna::domain_to_unicode; use serde::{Deserialize, Serialize}; +use garage_util::error::Error as GarageError; + use crate::common_error::{CommonError as Error, *}; /// What kind of authorization is required to perform a given action @@ -138,18 +148,64 @@ pub fn key_after_prefix(pfx: &str) -> Option<String> { None } -pub async fn parse_json_body<T: for<'de> Deserialize<'de>>(req: Request<Body>) -> Result<T, Error> { - let body = hyper::body::to_bytes(req.into_body()).await?; +// =============== body helpers ================= + +pub type EmptyBody = http_body_util::Empty<bytes::Bytes>; +pub type ErrorBody = FullBody<bytes::Bytes>; +pub type BoxBody<E> = http_body_util::combinators::BoxBody<bytes::Bytes, E>; + +pub fn string_body<E>(s: String) -> BoxBody<E> { + bytes_body(bytes::Bytes::from(s.into_bytes())) +} +pub fn bytes_body<E>(b: bytes::Bytes) -> BoxBody<E> { + BoxBody::new(FullBody::new(b).map_err(|_: Infallible| unreachable!())) +} +pub fn empty_body<E>() -> BoxBody<E> { + BoxBody::new(http_body_util::Empty::new().map_err(|_: Infallible| unreachable!())) +} +pub fn error_body(s: String) -> ErrorBody { + ErrorBody::from(bytes::Bytes::from(s.into_bytes())) +} + +pub async fn parse_json_body<T, B, E>(req: Request<B>) -> Result<T, E> +where + T: for<'de> Deserialize<'de>, + B: Body, + E: From<<B as Body>::Error> + From<Error>, +{ + let body = req.into_body().collect().await?.to_bytes(); let resp: T = serde_json::from_slice(&body).ok_or_bad_request("Invalid JSON")?; Ok(resp) } -pub fn json_ok_response<T: Serialize>(res: &T) -> Result<Response<Body>, Error> { - let resp_json = serde_json::to_string_pretty(res).map_err(garage_util::error::Error::from)?; +pub fn json_ok_response<E, T: Serialize>(res: &T) -> Result<Response<BoxBody<E>>, E> +where + E: From<Error>, +{ + let resp_json = serde_json::to_string_pretty(res) + .map_err(GarageError::from) + .map_err(Error::from)?; Ok(Response::builder() .status(hyper::StatusCode::OK) .header(http::header::CONTENT_TYPE, "application/json") - .body(Body::from(resp_json))?) + .body(string_body(resp_json)) + .unwrap()) +} + +pub fn body_stream<B, E>(body: B) -> impl Stream<Item = Result<Bytes, E>> +where + B: Body<Data = Bytes>, + <B as Body>::Error: Into<E>, + E: From<Error>, +{ + let stream = http_body_util::BodyStream::new(body); + let stream = TryStreamExt::map_err(stream, Into::into); + stream.map(|x| { + x.and_then(|f| { + f.into_data() + .map_err(|_| E::from(Error::bad_request("non-data frame"))) + }) + }) } pub fn is_default<T: Default + PartialEq>(v: &T) -> bool { |