broker_tokio/io/util/
lines.rs

1use crate::io::util::read_line::read_line_internal;
2use crate::io::AsyncBufRead;
3
4use pin_project_lite::pin_project;
5use std::io;
6use std::mem;
7use std::pin::Pin;
8use std::task::{Context, Poll};
9
10pin_project! {
11    /// Stream for the [`lines`](crate::io::AsyncBufReadExt::lines) method.
12    #[derive(Debug)]
13    #[must_use = "streams do nothing unless polled"]
14    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
15    pub struct Lines<R> {
16        #[pin]
17        reader: R,
18        buf: String,
19        bytes: Vec<u8>,
20        read: usize,
21    }
22}
23
24pub(crate) fn lines<R>(reader: R) -> Lines<R>
25where
26    R: AsyncBufRead,
27{
28    Lines {
29        reader,
30        buf: String::new(),
31        bytes: Vec::new(),
32        read: 0,
33    }
34}
35
36impl<R> Lines<R>
37where
38    R: AsyncBufRead + Unpin,
39{
40    /// Returns the next line in the stream.
41    ///
42    /// # Examples
43    ///
44    /// ```
45    /// # use tokio::io::AsyncBufRead;
46    /// use tokio::io::AsyncBufReadExt;
47    ///
48    /// # async fn dox(my_buf_read: impl AsyncBufRead + Unpin) -> std::io::Result<()> {
49    /// let mut lines = my_buf_read.lines();
50    ///
51    /// while let Some(line) = lines.next_line().await? {
52    ///     println!("length = {}", line.len())
53    /// }
54    /// # Ok(())
55    /// # }
56    /// ```
57    pub async fn next_line(&mut self) -> io::Result<Option<String>> {
58        use crate::future::poll_fn;
59
60        poll_fn(|cx| Pin::new(&mut *self).poll_next_line(cx)).await
61    }
62}
63
64impl<R> Lines<R>
65where
66    R: AsyncBufRead,
67{
68    #[doc(hidden)]
69    pub fn poll_next_line(
70        self: Pin<&mut Self>,
71        cx: &mut Context<'_>,
72    ) -> Poll<io::Result<Option<String>>> {
73        let me = self.project();
74
75        let n = ready!(read_line_internal(me.reader, cx, me.buf, me.bytes, me.read))?;
76
77        if n == 0 && me.buf.is_empty() {
78            return Poll::Ready(Ok(None));
79        }
80
81        if me.buf.ends_with('\n') {
82            me.buf.pop();
83
84            if me.buf.ends_with('\r') {
85                me.buf.pop();
86            }
87        }
88
89        Poll::Ready(Ok(Some(mem::replace(me.buf, String::new()))))
90    }
91}
92
93#[cfg(feature = "stream")]
94impl<R: AsyncBufRead> crate::stream::Stream for Lines<R> {
95    type Item = io::Result<String>;
96
97    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
98        Poll::Ready(match ready!(self.poll_next_line(cx)) {
99            Ok(Some(line)) => Some(Ok(line)),
100            Ok(None) => None,
101            Err(err) => Some(Err(err)),
102        })
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109
110    #[test]
111    fn assert_unpin() {
112        crate::is_unpin::<Lines<()>>();
113    }
114}
broker_tokio/io/util/lines.rs

broker_tokio/io/util/
lines.rs
