lib_resp/
decoder.rs

1use nom::Err;
2use std::io::{BufRead, Error, ErrorKind, Result};
3
4use super::{Parser, Value};
5
6/// Streaming decoder implementation for [BufRead](std::io::BufRead) implementors
7pub struct Decoder<T: BufRead> {
8    src: T,
9    multi_buf: Vec<Value>,
10}
11
12impl<T> Decoder<T>
13where
14    T: BufRead,
15{
16    pub fn new(src: T) -> Self {
17        Decoder {
18            src,
19            multi_buf: Vec::new(),
20        }
21    }
22
23    /// Attempts to read a single value from the stream, then parse it.
24    pub fn decode(&mut self) -> Result<Option<Value>> {
25        let (ret, consumed) = {
26            let buf = self.src.fill_buf()?;
27
28            match Parser::parse(buf) {
29                Ok((i, o)) => (Ok(Some(o)), buf.len() - i.len()),
30
31                Err(Err::Incomplete(_)) => (Ok(None), 0),
32
33                Err(_) => (
34                    Err(Error::new(ErrorKind::InvalidData, "Invalid RESP")),
35                    buf.len(),
36                ),
37            }
38        };
39
40        if consumed != 0 {
41            self.src.consume(consumed);
42        }
43
44        ret
45    }
46
47    /// Attempts to read *all* values from the stream, then parse them.
48    ///
49    /// NOTE: Parsed values will be stored until the stream is empty.
50    pub fn decode_all(&mut self) -> Result<Option<Vec<Value>>> {
51        loop {
52            match self.decode()? {
53                None => break,
54                Some(value) => self.multi_buf.push(value),
55            }
56        }
57
58        if self.src.fill_buf()?.len() > 0 {
59            return Ok(None);
60        }
61
62        Ok(Some(::std::mem::replace(&mut self.multi_buf, Vec::new())))
63    }
64}
lib_resp/decoder.rs

lib_resp/
decoder.rs
