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 | |