[dep-upgrade-202402] refactor use of BodyStream

1 files changed, 22 insertions, 1 deletions

diff --git a/src/api/helpers.rs b/src/api/helpers.rs
index ba7b1599..5f488912 100644
--- a/src/api/helpers.rs
+++ b/src/api/helpers.rs
@@ -1,7 +1,12 @@
 use std::convert::Infallible;
 
+use futures::{Stream, StreamExt, TryStreamExt};
+
 use http_body_util::{BodyExt, Full as FullBody};
-use hyper::{body::Body, Request, Response};
+use hyper::{
+	body::{Body, Bytes},
+	Request, Response,
+};
 use idna::domain_to_unicode;
 use serde::{Deserialize, Serialize};
 
@@ -187,6 +192,22 @@ where
 		.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 {
 	*v == T::default()
 }