tokio_io/io/
read_until.rs

1use std::io::{self, BufRead};
2use std::mem;
3
4use futures::{Future, Poll};
5
6use AsyncRead;
7
8/// A future which can be used to easily read the contents of a stream into a
9/// vector until the delimiter is reached.
10///
11/// Created by the [`read_until`] function.
12///
13/// [`read_until`]: fn.read_until.html
14#[derive(Debug)]
15pub struct ReadUntil<A> {
16    state: State<A>,
17}
18
19#[derive(Debug)]
20enum State<A> {
21    Reading { a: A, byte: u8, buf: Vec<u8> },
22    Empty,
23}
24
25/// Creates a future which will read all the bytes associated with the I/O
26/// object `A` into the buffer provided until the delimiter `byte` is reached.
27/// This method is the async equivalent to [`BufRead::read_until`].
28///
29/// In case of an error the buffer and the object will be discarded, with
30/// the error yielded. In the case of success the object will be destroyed and
31/// the buffer will be returned, with all bytes up to, and including, the delimiter
32/// (if found).
33///
34/// [`BufRead::read_until`]: https://doc.rust-lang.org/std/io/trait.BufRead.html#method.read_until
35pub fn read_until<A>(a: A, byte: u8, buf: Vec<u8>) -> ReadUntil<A>
36where
37    A: AsyncRead + BufRead,
38{
39    ReadUntil {
40        state: State::Reading {
41            a: a,
42            byte: byte,
43            buf: buf,
44        },
45    }
46}
47
48impl<A> Future for ReadUntil<A>
49where
50    A: AsyncRead + BufRead,
51{
52    type Item = (A, Vec<u8>);
53    type Error = io::Error;
54
55    fn poll(&mut self) -> Poll<(A, Vec<u8>), io::Error> {
56        match self.state {
57            State::Reading {
58                ref mut a,
59                byte,
60                ref mut buf,
61            } => {
62                // If we get `Ok(n)`, then we know the stream hit EOF or the delimiter.
63                // and just return it, as we are finished.
64                // If we hit "would block" then all the read data so far
65                // is in our buffer, and otherwise we propagate errors.
66                try_nb!(a.read_until(byte, buf));
67            }
68            State::Empty => panic!("poll ReadUntil after it's done"),
69        }
70
71        match mem::replace(&mut self.state, State::Empty) {
72            State::Reading { a, byte: _, buf } => Ok((a, buf).into()),
73            State::Empty => unreachable!(),
74        }
75    }
76}
tokio_io/io/read_until.rs

tokio_io/io/
read_until.rs
