Add BufRead::split (#312)

* add BufRead::split

Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com>

* fix docs

Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com>

* Update src/io/buf_read/mod.rs

Co-Authored-By: Stjepan Glavina <stjepang@gmail.com>

src/io/buf_read/mod.rs

@@ -1,8 +1,11 @@
 mod lines;
 mod read_line;
 mod read_until;
+mod split;
 
 pub use lines::Lines;
+pub use split::Split;
+
 use read_line::ReadLineFuture;
 use read_until::ReadUntilFuture;
 
@@ -226,6 +229,57 @@ extension_trait! {
                 read: 0,
             }
         }
+
+        #[doc = r#"
+            Returns a stream over the contents of this reader split on the byte `byte`.
+
+            The stream returned from this function will return instances of
+            [`io::Result`]`<`[`Vec<u8>`]`>`. Each vector returned will *not* have
+            the delimiter byte at the end.
+
+            This function will yield errors whenever [`read_until`] would have
+            also yielded an error.
+
+            [`io::Result`]: type.Result.html
+            [`Vec<u8>`]: ../vec/struct.Vec.html
+            [`read_until`]: #method.read_until
+
+            # Examples
+
+            [`std::io::Cursor`][`Cursor`] is a type that implements `BufRead`. In
+            this example, we use [`Cursor`] to iterate over all hyphen delimited
+            segments in a byte slice
+
+            [`Cursor`]: struct.Cursor.html
+
+            ```
+            # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
+            #
+            use async_std::prelude::*;
+            use async_std::io;
+
+            let cursor = io::Cursor::new(b"lorem-ipsum-dolor");
+
+            let mut split_iter = cursor.split(b'-').map(|l| l.unwrap());
+            assert_eq!(split_iter.next().await, Some(b"lorem".to_vec()));
+            assert_eq!(split_iter.next().await, Some(b"ipsum".to_vec()));
+            assert_eq!(split_iter.next().await, Some(b"dolor".to_vec()));
+            assert_eq!(split_iter.next().await, None);
+            #
+            # Ok(()) }) }
+            ```
+        "#]
+        fn split(self, byte: u8) -> Split<Self>
+        where
+            Self: Sized,
+        {
+            Split {
+                reader: self,
+                buf: Vec::new(),
+                delim: byte,
+                read: 0,
+            }
+        }
     }
 
     impl<T: BufRead + Unpin + ?Sized> BufRead for Box<T> {

src/io/buf_read/split.rs

@@ -0,0 +1,46 @@
+use std::mem;
+use std::pin::Pin;
+
+use super::read_until_internal;
+use crate::io::{self, BufRead};
+use crate::stream::Stream;
+use crate::task::{Context, Poll};
+
+/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
+///
+/// This stream is created by the [`split`] method on types that implement [`BufRead`].
+///
+/// This type is an async version of [`std::io::Split`].
+///
+/// [`split`]: trait.BufRead.html#method.lines
+/// [`BufRead`]: trait.BufRead.html
+/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
+#[derive(Debug)]
+pub struct Split<R> {
+    pub(crate) reader: R,
+    pub(crate) buf: Vec<u8>,
+    pub(crate) read: usize,
+    pub(crate) delim: u8,
+}
+
+impl<R: BufRead> Stream for Split<R> {
+    type Item = io::Result<Vec<u8>>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let Self {
+            reader,
+            buf,
+            read,
+            delim,
+        } = unsafe { self.get_unchecked_mut() };
+        let reader = unsafe { Pin::new_unchecked(reader) };
+        let n = futures_core::ready!(read_until_internal(reader, cx, *delim, buf, read))?;
+        if n == 0 && buf.is_empty() {
+            return Poll::Ready(None);
+        }
+        if buf[buf.len() - 1] == *delim {
+            buf.pop();
+        }
+        Poll::Ready(Some(Ok(mem::replace(buf, vec![]))))
+    }
+}