Add AsyncBufReadExt::read_line

Author: taiki-e

futures-util/src/io/mod.rs

@@ -30,9 +30,8 @@ pub use self::read::Read;
 mod read_exact;
 pub use self::read_exact::ReadExact;
 
-// TODO
-// mod read_line;
-// pub use self::read_line::ReadLine;
+mod read_line;
+pub use self::read_line::ReadLine;
 
 mod read_to_end;
 pub use self::read_to_end::ReadToEnd;
@@ -403,6 +402,65 @@ pub trait AsyncBufReadExt: AsyncBufRead {
     {
         ReadUntil::new(self, byte, buf)
     }
+
+    /// Creates a future which will read all the bytes associated with this I/O
+    /// object into `buf` until a newline (the 0xA byte) or EOF is reached,
+    /// This method is the async equivalent to [`BufRead::read_line`](std::io::BufRead::read_line).
+    ///
+    /// This function will read bytes from the underlying stream until the
+    /// newline delimiter (the 0xA byte) or EOF is found. Once found, all bytes
+    /// up to, and including, the delimiter (if found) will be appended to
+    /// `buf`.
+    ///
+    /// The returned future will resolve to the number of bytes read once the read
+    /// operation is completed.
+    ///
+    /// In the case of an error the buffer and the object will be discarded, with
+    /// the error yielded.
+    ///
+    /// # Errors
+    ///
+    /// This function has the same error semantics as [`read_until`] and will
+    /// also return an error if the read bytes are not valid UTF-8. If an I/O
+    /// error is encountered then `buf` may contain some bytes already read in
+    /// the event that all data read so far was valid UTF-8.
+    ///
+    /// [`read_until`]: AsyncBufReadExt::read_until
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// #![feature(async_await, await_macro)]
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncBufReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let mut cursor = Cursor::new(b"foo\nbar");
+    /// let mut buf = String::new();
+    ///
+    /// // cursor is at 'f'
+    /// let num_bytes = await!(cursor.read_line(&mut buf))?;
+    /// assert_eq!(num_bytes, 4);
+    /// assert_eq!(buf, "foo\n");
+    /// buf.clear();
+    ///
+    /// // cursor is at 'b'
+    /// let num_bytes = await!(cursor.read_line(&mut buf))?;
+    /// assert_eq!(num_bytes, 3);
+    /// assert_eq!(buf, "bar");
+    /// buf.clear();
+    ///
+    /// // cursor is at EOF
+    /// let num_bytes = await!(cursor.read_line(&mut buf))?;
+    /// assert_eq!(num_bytes, 0);
+    /// assert_eq!(buf, "");
+    /// # Ok::<(), Box<std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn read_line<'a>(&'a mut self, buf: &'a mut String) -> ReadLine<'a, Self>
+        where Self: Unpin,
+    {
+        ReadLine::new(self, buf)
+    }
 }
 
 impl<R: AsyncBufRead + ?Sized> AsyncBufReadExt for R {}

futures-util/src/io/read_line.rs

@@ -0,0 +1,51 @@
+use futures_core::future::Future;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncBufRead;
+use std::io;
+use std::pin::Pin;
+use std::str;
+use super::read_until::read_until_internal;
+
+/// Future for the [`read_line`](super::AsyncBufReadExt::read_line) method.
+#[derive(Debug)]
+pub struct ReadLine<'a, R: ?Sized + Unpin> {
+    reader: &'a mut R,
+    buf: &'a mut String,
+}
+
+impl<R: ?Sized + Unpin> Unpin for ReadLine<'_, R> {}
+
+impl<'a, R: AsyncBufRead + ?Sized + Unpin> ReadLine<'a, R> {
+    pub(super) fn new(reader: &'a mut R, buf: &'a mut String) -> Self {
+        Self { reader, buf }
+    }
+}
+
+struct Guard<'a> { buf: &'a mut Vec<u8>, len: usize }
+
+impl Drop for Guard<'_> {
+    fn drop(&mut self) {
+        unsafe { self.buf.set_len(self.len); }
+    }
+}
+
+impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadLine<'_, R> {
+    type Output = io::Result<usize>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Self { reader, buf } = &mut *self;
+        unsafe {
+            let mut g = Guard { len: buf.len(), buf: buf.as_mut_vec() };
+            let ret = ready!(read_until_internal(Pin::new(reader), b'\n', g.buf, cx));
+            if str::from_utf8(&g.buf[g.len..]).is_err() {
+                Poll::Ready(ret.and_then(|_| {
+                    Err(io::Error::new(io::ErrorKind::InvalidData,
+                                "stream did not contain valid UTF-8"))
+                }))
+            } else {
+                g.len = g.buf.len();
+                Poll::Ready(ret)
+            }
+        }
+    }
+}

futures-util/src/io/read_until.rs

@@ -20,28 +20,37 @@ impl<'a, R: AsyncBufRead + ?Sized + Unpin> ReadUntil<'a, R> {
     }
 }
 
+pub(super) fn read_until_internal<R: AsyncBufRead + ?Sized + Unpin>(
+    mut reader: Pin<&mut R>,
+    byte: u8,
+    buf: &mut Vec<u8>,
+    cx: &mut Context<'_>,
+) -> Poll<io::Result<usize>> {
+    let mut read = 0;
+    loop {
+        let (done, used) = {
+            let available = try_ready!(reader.as_mut().poll_fill_buf(cx));
+            if let Some(i) = memchr::memchr(byte, available) {
+                buf.extend_from_slice(&available[..=i]);
+                (true, i + 1)
+            } else {
+                buf.extend_from_slice(available);
+                (false, available.len())
+            }
+        };
+        reader.as_mut().consume(used);
+        read += used;
+        if done || used == 0 {
+            return Poll::Ready(Ok(read));
+        }
+    }
+}
+
 impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadUntil<'_, R> {
     type Output = io::Result<usize>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = &mut *self;
-        let mut read = 0;
-        loop {
-            let (done, used) = {
-                let available = try_ready!(Pin::new(&mut this.reader).poll_fill_buf(cx));
-                if let Some(i) = memchr::memchr(this.byte, available) {
-                    this.buf.extend_from_slice(&available[..=i]);
-                    (true, i + 1)
-                } else {
-                    this.buf.extend_from_slice(available);
-                    (false, available.len())
-                }
-            };
-            Pin::new(&mut this.reader).consume(used);
-            read += used;
-            if done || used == 0 {
-                return Poll::Ready(Ok(read));
-            }
-        }
+        let Self { reader, byte, buf } = &mut *self;
+        read_until_internal(Pin::new(reader), *byte, buf, cx)
     }
 }

futures/src/lib.rs

@@ -273,8 +273,8 @@ pub mod io {
     };
     pub use futures_util::io::{
         AsyncReadExt, AsyncWriteExt, AsyncSeekExt, AsyncBufReadExt, AllowStdIo,
-        Close, CopyInto, Flush, Read, ReadExact, ReadHalf, ReadToEnd, ReadUntil,
-        Seek, Window, WriteAll, WriteHalf,
+        Close, CopyInto, Flush, Read, ReadExact, ReadHalf, ReadLine, ReadToEnd,
+        ReadUntil, Seek, Window, WriteAll, WriteHalf,
     };
 }
 

futures/tests/io_read_line.rs

@@ -0,0 +1,22 @@
+use futures::executor::block_on;
+use futures::io::AsyncBufReadExt;
+use std::io::Cursor;
+
+#[test]
+fn read_line() {
+    let mut buf = Cursor::new(&b"12"[..]);
+    let mut v = String::new();
+    assert_eq!(block_on(buf.read_line(&mut v)).unwrap(), 2);
+    assert_eq!(v, "12");
+
+    let mut buf = Cursor::new(&b"12\n\n"[..]);
+    let mut v = String::new();
+    assert_eq!(block_on(buf.read_line(&mut v)).unwrap(), 3);
+    assert_eq!(v, "12\n");
+    v.truncate(0);
+    assert_eq!(block_on(buf.read_line(&mut v)).unwrap(), 1);
+    assert_eq!(v, "\n");
+    v.truncate(0);
+    assert_eq!(block_on(buf.read_line(&mut v)).unwrap(), 0);
+    assert_eq!(v, "");
+}