| 1 | use std::mem; |
| 2 | use std::pin::Pin; |
| 3 | use std::str; |
| 4 | |
| 5 | use pin_project_lite::pin_project; |
| 6 | |
| 7 | use super::read_until_internal; |
| 8 | use crate::io::{self, BufRead}; |
| 9 | use crate::stream::Stream; |
| 10 | use crate::task::{Context, Poll}; |
| 11 | |
| 12 | pin_project! { |
| 13 | /// A stream of lines in a byte stream. |
| 14 | /// |
| 15 | /// This stream is created by the [`lines`] method on types that implement [`BufRead`]. |
| 16 | /// |
| 17 | /// This type is an async version of [`std::io::Lines`]. |
| 18 | /// |
| 19 | /// [`lines`]: trait.BufRead.html#method.lines |
| 20 | /// [`BufRead`]: trait.BufRead.html |
| 21 | /// [`std::io::Lines`]: https://doc.rust-lang.org/std/io/struct.Lines.html |
| 22 | #[derive (Debug)] |
| 23 | pub struct Lines<R> { |
| 24 | #[pin] |
| 25 | pub(crate) reader: R, |
| 26 | pub(crate) buf: String, |
| 27 | pub(crate) bytes: Vec<u8>, |
| 28 | pub(crate) read: usize, |
| 29 | } |
| 30 | } |
| 31 | |
| 32 | impl<R: BufRead> Stream for Lines<R> { |
| 33 | type Item = io::Result<String>; |
| 34 | |
| 35 | fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { |
| 36 | let this: Projection<'_, R> = self.project(); |
| 37 | let n: usize = futures_core::ready!(read_line_internal( |
| 38 | this.reader, |
| 39 | cx, |
| 40 | this.buf, |
| 41 | this.bytes, |
| 42 | this.read |
| 43 | ))?; |
| 44 | if n == 0 && this.buf.is_empty() { |
| 45 | return Poll::Ready(None); |
| 46 | } |
| 47 | if this.buf.ends_with(' \n' ) { |
| 48 | this.buf.pop(); |
| 49 | if this.buf.ends_with(' \r' ) { |
| 50 | this.buf.pop(); |
| 51 | } |
| 52 | } |
| 53 | Poll::Ready(Some(Ok(mem::replace(dest:this.buf, src:String::new())))) |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | pub fn read_line_internal<R: BufRead + ?Sized>( |
| 58 | reader: Pin<&mut R>, |
| 59 | cx: &mut Context<'_>, |
| 60 | buf: &mut String, |
| 61 | bytes: &mut Vec<u8>, |
| 62 | read: &mut usize, |
| 63 | ) -> Poll<io::Result<usize>> { |
| 64 | let ret: Result = futures_core::ready!(read_until_internal(reader, cx, b' \n' , bytes, read)); |
| 65 | if str::from_utf8(&bytes).is_err() { |
| 66 | Poll::Ready(ret.and_then(|_| { |
| 67 | Err(io::Error::new( |
| 68 | kind:io::ErrorKind::InvalidData, |
| 69 | error:"stream did not contain valid UTF-8" , |
| 70 | )) |
| 71 | })) |
| 72 | } else { |
| 73 | debug_assert!(buf.is_empty()); |
| 74 | debug_assert_eq!(*read, 0); |
| 75 | // Safety: `bytes` is a valid UTF-8 because `str::from_utf8` returned `Ok`. |
| 76 | mem::swap(x:unsafe { buf.as_mut_vec() }, y:bytes); |
| 77 | Poll::Ready(ret) |
| 78 | } |
| 79 | } |
| 80 | |