feat(io): implement Read::bytes

Author: dignifiedquire

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(())
+        })
+    }
+}

src/io/read/mod.rs

@@ -1,3 +1,4 @@
+mod bytes;
 mod read;
 mod read_exact;
 mod read_to_end;
@@ -303,6 +304,39 @@ cfg_if! {
             /// }) }
             /// ```
             fn by_ref(&mut self) -> &mut Self where Self: Sized { unreachable!() }
+
+
+            /// 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::io::prelude::*;
+            /// use async_std::fs::File;
+            ///
+            /// fn main() -> io::Result<()> { async_std::task:.block_on(async {
+            ///     let mut f = File::open("foo.txt").await?;
+            ///     let mut s = f.bytes();
+            ///
+            ///     while let Some(v) = s.next().await {
+            ///         println!("{}", byte.unwrap());
+            ///     }
+            ///     Ok(())
+            /// }) }
+            /// ```
+            fn bytes(self) -> Bytes<Self> where Self: Sized {
+                unreachable!{}
+            }
         }
 
         impl<T: Read + Unpin + ?Sized> Read for Box<T> {
@@ -417,6 +451,13 @@ pub trait ReadExt: futures_io::AsyncRead {
     {
         self
     }
+
+    fn bytes(self) -> bytes::Bytes<Self>
+    where
+        Self: Sized,
+    {
+        bytes::Bytes { inner: self }
+    }
 }
 
 impl<T: futures_io::AsyncRead + ?Sized> ReadExt for T {}