feat(util): add data() method

http-body-util/src/combinators/data.rs

@@ -0,0 +1,29 @@
+use http_body::Body;
+
+use core::future::Future;
+use core::pin::Pin;
+use core::task;
+
+#[must_use = "futures don't do anything unless polled"]
+#[derive(Debug)]
+/// Future that resolves to the next data chunk from `Body`
+pub struct Data<'a, T: ?Sized>(pub(crate) &'a mut T);
+
+impl<'a, T: Body + Unpin + ?Sized> Future for Data<'a, T> {
+    type Output = Option<Result<T::Data, T::Error>>;
+
+    fn poll(mut self: Pin<&mut Self>, ctx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+        match Pin::new(&mut self.0).poll_frame(ctx) {
+            task::Poll::Ready(Some(Ok(frame))) => {
+                if let Ok(data) = frame.into_data() {
+                    task::Poll::Ready(Some(Ok(data)))
+                } else {
+                    task::Poll::Pending
+                }
+            },
+            task::Poll::Ready(Some(Err(e))) => task::Poll::Ready(Some(Err(e))),
+            task::Poll::Ready(None) => task::Poll::Ready(None),
+            task::Poll::Pending => task::Poll::Pending,
+        }
+    }
+}

http-body-util/src/combinators/mod.rs

@@ -6,6 +6,7 @@ mod frame;
 mod map_err;
 mod map_frame;
 mod with_trailers;
+mod data;
 
 pub use self::{
     box_body::{BoxBody, UnsyncBoxBody},
@@ -14,4 +15,5 @@ pub use self::{
     map_err::MapErr,
     map_frame::MapFrame,
     with_trailers::WithTrailers,
+    data::Data,
 };

http-body-util/src/lib.rs

@@ -38,6 +38,15 @@ pub trait BodyExt: http_body::Body {
         combinators::Frame(self)
     }
 
+    /// Returns future that resolves to next data chunk, if any.
+    fn data(&mut self) -> combinators::Data<'_, Self>
+    where
+        Self: Unpin + Sized,
+    {
+        combinators::Data(self)
+    }
+
+
     /// Maps this body's frame to a different kind.
     fn map_frame<F, B>(self, f: F) -> MapFrame<Self, F>
     where