1 files changed, 65 insertions, 0 deletions
diff --git a/tokio-util/src/io/read_buf.rs b/tokio-util/src/io/read_buf.rs
new file mode 100644
index 00000000..d617fa6f
--- /dev/null
+++ b/tokio-util/src/io/read_buf.rs
@@ -0,0 +1,65 @@
+use bytes::BufMut;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::AsyncRead;
+
+/// Read data from an `AsyncRead` into an implementer of the [`Buf`] trait.
+///
+/// [`Buf`]: bytes::Buf
+///
+/// # Example
+///
+/// ```
+/// use bytes::{Bytes, BytesMut};
+/// use tokio::stream;
+/// use tokio::io::Result;
+/// use tokio_util::io::{StreamReader, read_buf};
+/// # #[tokio::main]
+/// # async fn main() -> std::io::Result<()> {
+///
+/// // Create a reader from an iterator. This particular reader will always be
+/// // ready.
+/// let mut read = StreamReader::new(stream::iter(vec![Result::Ok(Bytes::from_static(&[0, 1, 2, 3]))]));
+///
+/// let mut buf = BytesMut::new();
+/// let mut reads = 0;
+///
+/// loop {
+///     reads += 1;
+///     let n = read_buf(&mut read, &mut buf).await?;
+///
+///     if n == 0 {
+///         break;
+///     }
+/// }
+///
+/// // one or more reads might be necessary.
+/// assert!(reads >= 1);
+/// assert_eq!(&buf[..], &[0, 1, 2, 3]);
+/// # Ok(())
+/// # }
+/// ```
+pub async fn read_buf<R, B>(read: &mut R, buf: &mut B) -> io::Result<usize>
+where
+    R: AsyncRead + Unpin,
+    B: BufMut,
+{
+    return ReadBufFn(read, buf).await;
+
+    struct ReadBufFn<'a, R, B>(&'a mut R, &'a mut B);
+
+    impl<'a, R, B> Future for ReadBufFn<'a, R, B>
+    where
+        R: AsyncRead + Unpin,
+        B: BufMut,
+    {
+        type Output = io::Result<usize>;
+
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            let this = &mut *self;
+            super::poll_read_buf(Pin::new(this.0), cx, this.1)
+        }
+    }
+}