redis_event/
resp.rs

1/*!
2Redis Serialization Protocol相关的解析代码
3*/
4
5use std::io::{Read, Result};
6
7use byteorder::ReadBytesExt;
8
9use crate::to_string;
10
11/// Redis Serialization Protocol解析
12pub trait RespDecode: Read {
13    /// 读取并解析Redis响应
14    fn decode_resp(&mut self) -> Result<Resp> {
15        match self.decode_type()? {
16            Type::String => Ok(Resp::String(self.decode_string()?)),
17            Type::Int => self.decode_int(),
18            Type::Error => Ok(Resp::Error(self.decode_string()?)),
19            Type::BulkString => self.decode_bulk_string(),
20            Type::Array => self.decode_array(),
21        }
22    }
23    /// 读取解析Redis响应的类型
24    fn decode_type(&mut self) -> Result<Type> {
25        loop {
26            let b = self.read_u8()?;
27            if b == LF {
28                continue;
29            } else {
30                match b {
31                    PLUS => return Ok(Type::String),
32                    MINUS => return Ok(Type::Error),
33                    COLON => return Ok(Type::Int),
34                    DOLLAR => return Ok(Type::BulkString),
35                    STAR => return Ok(Type::Array),
36                    _ => panic!("Unexpected Data Type: {}", b),
37                }
38            }
39        }
40    }
41    /// 解析Simple String响应
42    fn decode_string(&mut self) -> Result<String> {
43        let mut buf = vec![];
44        loop {
45            let byte = self.read_u8()?;
46            if byte != CR {
47                buf.push(byte);
48            } else {
49                break;
50            }
51        }
52        if self.read_u8()? == LF {
53            Ok(to_string(buf))
54        } else {
55            panic!("Expect LF after CR");
56        }
57    }
58
59    /// 解析Integer响应
60    fn decode_int(&mut self) -> Result<Resp> {
61        let s = self.decode_string()?;
62        let i = s.parse::<i64>().unwrap();
63        return Ok(Resp::Int(i));
64    }
65
66    /// 解析Bulk String响应
67    fn decode_bulk_string(&mut self) -> Result<Resp> {
68        let r = self.decode_int()?;
69        if let Resp::Int(i) = r {
70            if i > 0 {
71                let mut buf = vec![0; i as usize];
72                self.read_exact(&mut buf)?;
73                let mut end = vec![0; 2];
74                self.read_exact(&mut end)?;
75                if !end.eq(&[CR, LF]) {
76                    panic!("Expected CRLF");
77                } else {
78                    return Ok(Resp::BulkBytes(buf));
79                }
80            } else {
81                self.read_exact(&mut [0; 2])?;
82                return Ok(Resp::BulkBytes(vec![0; 0]));
83            }
84        } else {
85            panic!("Expected Int Response");
86        }
87    }
88
89    /// 解析Array响应
90    fn decode_array(&mut self) -> Result<Resp> {
91        let r = self.decode_int()?;
92        if let Resp::Int(i) = r {
93            let mut arr = Vec::with_capacity(i as usize);
94            for _ in 0..i {
95                let resp = self.decode_resp()?;
96                arr.push(resp);
97            }
98            return Ok(Resp::Array(arr));
99        } else {
100            panic!("Expected Int Response");
101        }
102    }
103}
104
105impl<R: Read + ?Sized> RespDecode for R {}
106
107pub enum Type {
108    String,
109    Error,
110    Int,
111    BulkString,
112    Array,
113}
114
115#[derive(Debug)]
116pub enum Resp {
117    String(String),
118    Int(i64),
119    Error(String),
120    BulkBytes(Vec<u8>),
121    Array(Vec<Resp>),
122}
123
124// 回车换行，在redis响应中一般表示终结符，或用作分隔符以分隔数据
125pub(crate) const CR: u8 = b'\r';
126pub(crate) const LF: u8 = b'\n';
127// 代表array响应
128pub(crate) const STAR: u8 = b'*';
129// 代表bulk string响应
130pub(crate) const DOLLAR: u8 = b'$';
131// 代表simple string响应
132pub(crate) const PLUS: u8 = b'+';
133// 代表error响应
134pub(crate) const MINUS: u8 = b'-';
135// 代表integer响应
136pub(crate) const COLON: u8 = b':';
137
138#[cfg(test)]
139mod test {
140    use crate::resp::{Resp, RespDecode};
141    use std::io::Cursor;
142
143    #[test]
144    fn test_decode_array() {
145        let b = b"*2\r\n$6\r\nSELECT\r\n$1\r\n0\r\n";
146        let mut cursor = Cursor::new(b);
147        let r = cursor.decode_resp();
148        match r {
149            Ok(resp) => match resp {
150                Resp::Array(arr) => {
151                    let mut data = Vec::new();
152                    for x in arr {
153                        match x {
154                            Resp::BulkBytes(bytes) => data.push(bytes),
155                            _ => panic!("wrong type"),
156                        }
157                    }
158                    assert!(b"SELECT".eq(data.get(0).unwrap().as_slice()));
159                    assert!(b"0".eq(data.get(1).unwrap().as_slice()));
160                }
161                _ => panic!("wrong type"),
162            },
163            Err(err) => panic!(err),
164        }
165    }
166}
redis_event/resp.rs

redis_event/
resp.rs
