From e681e297ef49b9bdf16d41bf931bdf23186a839d Mon Sep 17 00:00:00 2001
From: dignifiedquire <dignifiedquire@users.noreply.github.com>
Date: Fri, 27 Sep 2019 14:58:14 +0200
Subject: [PATCH] feat(io): implement Read::bytes

---
 src/io/read/bytes.rs | 60 ++++++++++++++++++++++++++++++++++++++++++++
 src/io/read/mod.rs   | 36 ++++++++++++++++++++++++++
 2 files changed, 96 insertions(+)
 create mode 100644 src/io/read/bytes.rs

diff --git a/src/io/read/bytes.rs b/src/io/read/bytes.rs
new file mode 100644
index 00000000..a45ee165
--- /dev/null
+++ b/src/io/read/bytes.rs
@@ -0,0 +1,60 @@
+use std::pin::Pin;
+
+use crate::io::{self, Read};
+use crate::stream::stream::Stream;
+use crate::task::{Context, Poll};
+
+/// An iterator over `u8` values of a reader.
+///
+/// This struct is generally created by calling [`bytes`] on a reader.
+/// Please see the documentation of [`bytes`] for more details.
+///
+/// [`bytes`]: trait.Read.html#method.bytes
+#[derive(Debug)]
+pub struct Bytes<T> {
+    pub(crate) inner: T,
+}
+
+impl<T: Read + Unpin> Stream for Bytes<T> {
+    type Item = io::Result<u8>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut byte = 0;
+
+        let rd = Pin::new(&mut self.inner);
+
+        match futures_core::ready!(rd.poll_read(cx, std::slice::from_mut(&mut byte))) {
+            Ok(0) => Poll::Ready(None),
+            Ok(..) => Poll::Ready(Some(Ok(byte))),
+            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => Poll::Pending,
+            Err(e) => Poll::Ready(Some(Err(e))),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::io;
+    use crate::prelude::*;
+    use crate::task;
+
+    #[test]
+    fn test_bytes_basics() -> std::io::Result<()> {
+        task::block_on(async move {
+            let raw: Vec<u8> = vec![0, 1, 2, 3, 4, 5, 6, 7, 8];
+            let source: io::Cursor<Vec<u8>> = io::Cursor::new(raw.clone());
+
+            let mut s = source.bytes();
+
+            // TODO(@dignifiedquire): Use collect, once it is stable.
+            let mut result = Vec::new();
+            while let Some(byte) = s.next().await {
+                result.push(byte?);
+            }
+
+            assert_eq!(result, raw);
+
+            Ok(())
+        })
+    }
+}
diff --git a/src/io/read/mod.rs b/src/io/read/mod.rs
index 59e9bfba..c6b5bad0 100644
--- a/src/io/read/mod.rs
+++ b/src/io/read/mod.rs
@@ -1,3 +1,4 @@
+mod bytes;
 mod read;
 mod read_exact;
 mod read_to_end;
@@ -341,6 +342,41 @@ extension_trait! {
             ```
         "#]
         fn by_ref(&mut self) -> &mut Self where Self: Sized { self }
+
+
+        #[doc=r#"
+            Transforms this `Read` instance to a `Stream` over its bytes.
+           
+            The returned type implements `Stream` where the `Item` is
+            `Result<u8, io::Error>`.
+            The yielded item is `Ok` if a byte was successfully read and `Err`
+            otherwise. EOF is mapped to returning `None` from this iterator.
+           
+            # Examples
+           
+            [`File`][file]s implement `Read`:
+           
+            [file]: ../fs/struct.File.html
+           
+            ```no_run
+            use async_std::io;
+            use async_std::prelude::*;
+            use async_std::fs::File;
+           
+            fn main() -> io::Result<()> { async_std::task::block_on(async {
+                let f = File::open("foo.txt").await?;
+                let mut s = f.bytes();
+           
+                while let Some(byte) = s.next().await {
+                    println!("{}", byte.unwrap());
+                }
+                Ok(())
+            }) }
+            ```
+        "#]
+        fn bytes(self) -> bytes::Bytes<Self> where Self: Sized {
+            bytes::Bytes { inner: self }
+        }
     }
 
     impl<T: Read + Unpin + ?Sized> Read for Box<T> {