io: add AsyncBufReadExt::split (#1642)

add a `split` method to `AsyncBufReadExt`, analogous to `std::io::BufRead::split`.

tokio-io/src/io/async_buf_read_ext.rs

@@ -1,6 +1,7 @@
 use crate::io::lines::{lines, Lines};
 use crate::io::read_line::{read_line, ReadLine};
 use crate::io::read_until::{read_until, ReadUntil};
+use crate::io::split::{split, Split};
 use crate::AsyncBufRead;
 
 /// An extension trait which adds utility methods to `AsyncBufRead` types.
@@ -55,6 +56,30 @@ pub trait AsyncBufReadExt: AsyncBufRead {
         read_line(self, buf)
     }
 
+    /// Returns a stream of the contents of this reader split on the byte
+    /// `byte`.
+    ///
+    /// This method is the async equivalent to
+    /// [`BufRead::split`](std::io::BufRead::split).
+    ///
+    /// The stream returned from this function will yield instances of
+    /// [`io::Result`]`<`[`Vec<u8>`]`>`. Each vector returned will *not* have
+    /// the delimiter byte at the end.
+    ///
+    /// [`io::Result`]: std::io::Result
+    /// [`Vec<u8>`]: std::vec::Vec
+    ///
+    /// # Errors
+    ///
+    /// Each item of the stream has the same error semantics as
+    /// [`AsyncBufReadExt::read_until`](AsyncBufReadExt::read_until).
+    fn split(self, byte: u8) -> Split<Self>
+    where
+        Self: Sized,
+    {
+        split(self, byte)
+    }
+
     /// Returns a stream over the lines of this reader.
     /// This method is the async equivalent to [`BufRead::lines`](std::io::BufRead::lines).
     ///

tokio-io/src/io/mod.rs

@@ -18,6 +18,7 @@ mod read_until;
 mod repeat;
 mod shutdown;
 mod sink;
+mod split;
 mod take;
 mod write;
 mod write_all;

tokio-io/src/io/split.rs

@@ -0,0 +1,67 @@
+use super::read_until::read_until_internal;
+use crate::AsyncBufRead;
+
+use futures_core::{ready, Stream};
+use pin_project::{pin_project, project};
+use std::io;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Stream for the [`split`](crate::io::AsyncBufReadExt::split) method.
+#[pin_project]
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Split<R> {
+    #[pin]
+    reader: R,
+    buf: Vec<u8>,
+    delim: u8,
+    read: usize,
+}
+
+pub(crate) fn split<R>(reader: R, delim: u8) -> Split<R>
+where
+    R: AsyncBufRead,
+{
+    Split {
+        reader,
+        buf: Vec::new(),
+        delim,
+        read: 0,
+    }
+}
+
+impl<R: AsyncBufRead> Stream for Split<R> {
+    type Item = io::Result<Vec<u8>>;
+
+    #[project]
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        #[project]
+        let Split {
+            reader,
+            buf,
+            delim,
+            read,
+        } = self.project();
+
+        let n = ready!(read_until_internal(reader, cx, *delim, buf, read))?;
+        if n == 0 && buf.is_empty() {
+            return Poll::Ready(None);
+        }
+        if buf.last() == Some(&delim) {
+            buf.pop();
+        }
+        Poll::Ready(Some(Ok(mem::replace(buf, Vec::new()))))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn assert_unpin() {
+        crate::is_unpin::<Split<()>>();
+    }
+}