redis_oxide/protocol/
resp2.rs

1//! RESP2 protocol implementation
2//!
3//! This module implements the Redis Serialization Protocol (RESP2) for
4//! encoding and decoding Redis commands and responses.
5
6use crate::core::{
7    error::{RedisError, RedisResult},
8    value::RespValue,
9};
10use bytes::{Buf, BufMut, Bytes, BytesMut};
11use std::io::Cursor;
12
13const CRLF: &[u8] = b"\r\n";
14
15/// Encodes a RESP value into bytes
16pub struct RespEncoder;
17
18impl RespEncoder {
19    /// Encode a RESP value into a buffer
20    pub fn encode(value: &RespValue, buf: &mut BytesMut) -> RedisResult<()> {
21        match value {
22            RespValue::SimpleString(s) => {
23                buf.put_u8(b'+');
24                buf.put_slice(s.as_bytes());
25                buf.put_slice(CRLF);
26            }
27            RespValue::Error(e) => {
28                buf.put_u8(b'-');
29                buf.put_slice(e.as_bytes());
30                buf.put_slice(CRLF);
31            }
32            RespValue::Integer(i) => {
33                buf.put_u8(b':');
34                buf.put_slice(i.to_string().as_bytes());
35                buf.put_slice(CRLF);
36            }
37            RespValue::BulkString(data) => {
38                buf.put_u8(b'$');
39                buf.put_slice(data.len().to_string().as_bytes());
40                buf.put_slice(CRLF);
41                buf.put_slice(data);
42                buf.put_slice(CRLF);
43            }
44            RespValue::Null => {
45                buf.put_slice(b"$-1\r\n");
46            }
47            RespValue::Array(arr) => {
48                buf.put_u8(b'*');
49                buf.put_slice(arr.len().to_string().as_bytes());
50                buf.put_slice(CRLF);
51                for item in arr {
52                    Self::encode(item, buf)?;
53                }
54            }
55        }
56        Ok(())
57    }
58
59    fn encode_bulk_argument(data: &[u8], buf: &mut BytesMut) {
60        buf.put_u8(b'$');
61        buf.put_slice(data.len().to_string().as_bytes());
62        buf.put_slice(CRLF);
63        buf.put_slice(data);
64        buf.put_slice(CRLF);
65    }
66
67    fn encode_command_arg(arg: &RespValue, buf: &mut BytesMut) -> RedisResult<()> {
68        match arg {
69            RespValue::SimpleString(s) | RespValue::Error(s) => {
70                Self::encode_bulk_argument(s.as_bytes(), buf);
71            }
72            RespValue::Integer(i) => {
73                Self::encode_bulk_argument(i.to_string().as_bytes(), buf);
74            }
75            RespValue::BulkString(data) => {
76                Self::encode_bulk_argument(data, buf);
77            }
78            RespValue::Null => {
79                buf.put_slice(b"$-1\r\n");
80            }
81            RespValue::Array(_) => {
82                Self::encode(arg, buf)?;
83            }
84        }
85        Ok(())
86    }
87
88    /// Encode a command with arguments
89    pub fn encode_command(command: &str, args: &[RespValue]) -> RedisResult<Bytes> {
90        let mut buf = BytesMut::new();
91
92        // Create array with command + args
93        let total_len = 1 + args.len();
94        buf.put_u8(b'*');
95        buf.put_slice(total_len.to_string().as_bytes());
96        buf.put_slice(CRLF);
97
98        // Encode command
99        buf.put_u8(b'$');
100        buf.put_slice(command.len().to_string().as_bytes());
101        buf.put_slice(CRLF);
102        buf.put_slice(command.as_bytes());
103        buf.put_slice(CRLF);
104
105        // Encode arguments
106        for arg in args {
107            Self::encode_command_arg(arg, &mut buf)?;
108        }
109
110        Ok(buf.freeze())
111    }
112}
113
114/// Decodes RESP values from bytes
115pub struct RespDecoder;
116
117impl RespDecoder {
118    /// Decode a RESP value from a buffer
119    pub fn decode(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
120        if !buf.has_remaining() {
121            return Ok(None);
122        }
123
124        let type_byte = buf.chunk()[0];
125
126        match type_byte {
127            b'+' => Self::decode_simple_string(buf),
128            b'-' => Self::decode_error(buf),
129            b':' => Self::decode_integer(buf),
130            b'$' => Self::decode_bulk_string(buf),
131            b'*' => Self::decode_array(buf),
132            _ => Err(RedisError::Protocol(format!(
133                "Invalid RESP type byte: {}",
134                type_byte as char
135            ))),
136        }
137    }
138
139    fn decode_simple_string(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
140        buf.advance(1); // Skip '+'
141
142        if let Some(line) = Self::read_line(buf)? {
143            Ok(Some(RespValue::SimpleString(
144                String::from_utf8(line.to_vec())
145                    .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?,
146            )))
147        } else {
148            Ok(None)
149        }
150    }
151
152    fn decode_error(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
153        buf.advance(1); // Skip '-'
154
155        if let Some(line) = Self::read_line(buf)? {
156            Ok(Some(RespValue::Error(
157                String::from_utf8(line.to_vec())
158                    .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?,
159            )))
160        } else {
161            Ok(None)
162        }
163    }
164
165    fn decode_integer(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
166        buf.advance(1); // Skip ':'
167
168        if let Some(line) = Self::read_line(buf)? {
169            let num_str = String::from_utf8(line.to_vec())
170                .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?;
171            let num = num_str
172                .parse::<i64>()
173                .map_err(|e| RedisError::Protocol(format!("Invalid integer: {}", e)))?;
174            Ok(Some(RespValue::Integer(num)))
175        } else {
176            Ok(None)
177        }
178    }
179
180    fn decode_bulk_string(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
181        buf.advance(1); // Skip '$'
182
183        let len_line = match Self::read_line(buf)? {
184            Some(line) => line,
185            None => return Ok(None),
186        };
187
188        let len_str = String::from_utf8(len_line.to_vec())
189            .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?;
190        let len = len_str
191            .parse::<i64>()
192            .map_err(|e| RedisError::Protocol(format!("Invalid bulk string length: {}", e)))?;
193
194        if len == -1 {
195            return Ok(Some(RespValue::Null));
196        }
197
198        let len = len as usize;
199
200        // Check if we have enough data
201        if buf.remaining() < len + 2 {
202            return Ok(None);
203        }
204
205        let data = buf.chunk()[..len].to_vec();
206        buf.advance(len);
207
208        // Skip CRLF
209        if buf.remaining() < 2 {
210            return Ok(None);
211        }
212        buf.advance(2);
213
214        Ok(Some(RespValue::BulkString(Bytes::from(data))))
215    }
216
217    fn decode_array(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
218        buf.advance(1); // Skip '*'
219
220        let len_line = match Self::read_line(buf)? {
221            Some(line) => line,
222            None => return Ok(None),
223        };
224
225        let len_str = String::from_utf8(len_line.to_vec())
226            .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?;
227        let len = len_str
228            .parse::<i64>()
229            .map_err(|e| RedisError::Protocol(format!("Invalid array length: {}", e)))?;
230
231        if len == -1 {
232            return Ok(Some(RespValue::Null));
233        }
234
235        let len = len as usize;
236        let mut arr = Vec::with_capacity(len);
237
238        for _ in 0..len {
239            match Self::decode(buf)? {
240                Some(value) => arr.push(value),
241                None => return Ok(None),
242            }
243        }
244
245        Ok(Some(RespValue::Array(arr)))
246    }
247
248    fn read_line(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<Vec<u8>>> {
249        let start = buf.position() as usize;
250        let slice = buf.get_ref();
251
252        // Find CRLF
253        for i in start..slice.len().saturating_sub(1) {
254            if slice[i] == b'\r' && slice[i + 1] == b'\n' {
255                let line = slice[start..i].to_vec();
256                buf.set_position((i + 2) as u64);
257                return Ok(Some(line));
258            }
259        }
260
261        Ok(None)
262    }
263}
264
265#[cfg(test)]
266mod tests {
267    use super::*;
268
269    #[test]
270    fn test_encode_simple_string() {
271        let mut buf = BytesMut::new();
272        let value = RespValue::SimpleString("OK".to_string());
273        RespEncoder::encode(&value, &mut buf).unwrap();
274        assert_eq!(&buf[..], b"+OK\r\n");
275    }
276
277    #[test]
278    fn test_encode_error() {
279        let mut buf = BytesMut::new();
280        let value = RespValue::Error("ERR unknown command".to_string());
281        RespEncoder::encode(&value, &mut buf).unwrap();
282        assert_eq!(&buf[..], b"-ERR unknown command\r\n");
283    }
284
285    #[test]
286    fn test_encode_integer() {
287        let mut buf = BytesMut::new();
288        let value = RespValue::Integer(1000);
289        RespEncoder::encode(&value, &mut buf).unwrap();
290        assert_eq!(&buf[..], b":1000\r\n");
291    }
292
293    #[test]
294    fn test_encode_bulk_string() {
295        let mut buf = BytesMut::new();
296        let value = RespValue::BulkString(Bytes::from("foobar"));
297        RespEncoder::encode(&value, &mut buf).unwrap();
298        assert_eq!(&buf[..], b"$6\r\nfoobar\r\n");
299    }
300
301    #[test]
302    fn test_encode_null() {
303        let mut buf = BytesMut::new();
304        let value = RespValue::Null;
305        RespEncoder::encode(&value, &mut buf).unwrap();
306        assert_eq!(&buf[..], b"$-1\r\n");
307    }
308
309    #[test]
310    fn test_encode_array() {
311        let mut buf = BytesMut::new();
312        let value = RespValue::Array(vec![
313            RespValue::BulkString(Bytes::from("foo")),
314            RespValue::BulkString(Bytes::from("bar")),
315        ]);
316        RespEncoder::encode(&value, &mut buf).unwrap();
317        assert_eq!(&buf[..], b"*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n");
318    }
319
320    #[test]
321    fn test_encode_command() {
322        let bytes =
323            RespEncoder::encode_command("GET", &[RespValue::BulkString(Bytes::from("mykey"))])
324                .unwrap();
325        assert_eq!(&bytes[..], b"*2\r\n$3\r\nGET\r\n$5\r\nmykey\r\n");
326    }
327
328    #[test]
329    fn test_encode_command_arguments_are_bulk_strings() {
330        let bytes = RespEncoder::encode_command(
331            "LRANGE",
332            &[
333                RespValue::from("items"),
334                RespValue::from(0),
335                RespValue::from(-1),
336            ],
337        )
338        .unwrap();
339
340        assert_eq!(
341            &bytes[..],
342            b"*4\r\n$6\r\nLRANGE\r\n$5\r\nitems\r\n$1\r\n0\r\n$2\r\n-1\r\n"
343        );
344    }
345
346    #[test]
347    fn test_decode_simple_string() {
348        let data = b"+OK\r\n";
349        let mut cursor = Cursor::new(&data[..]);
350        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
351        assert_eq!(value, RespValue::SimpleString("OK".to_string()));
352    }
353
354    #[test]
355    fn test_decode_error() {
356        let data = b"-ERR unknown\r\n";
357        let mut cursor = Cursor::new(&data[..]);
358        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
359        assert_eq!(value, RespValue::Error("ERR unknown".to_string()));
360    }
361
362    #[test]
363    fn test_decode_integer() {
364        let data = b":1000\r\n";
365        let mut cursor = Cursor::new(&data[..]);
366        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
367        assert_eq!(value, RespValue::Integer(1000));
368    }
369
370    #[test]
371    fn test_decode_bulk_string() {
372        let data = b"$6\r\nfoobar\r\n";
373        let mut cursor = Cursor::new(&data[..]);
374        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
375        assert_eq!(value, RespValue::BulkString(Bytes::from("foobar")));
376    }
377
378    #[test]
379    fn test_decode_null() {
380        let data = b"$-1\r\n";
381        let mut cursor = Cursor::new(&data[..]);
382        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
383        assert_eq!(value, RespValue::Null);
384    }
385
386    #[test]
387    fn test_decode_array() {
388        let data = b"*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n";
389        let mut cursor = Cursor::new(&data[..]);
390        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
391        assert_eq!(
392            value,
393            RespValue::Array(vec![
394                RespValue::BulkString(Bytes::from("foo")),
395                RespValue::BulkString(Bytes::from("bar")),
396            ])
397        );
398    }
399
400    #[test]
401    fn test_decode_incomplete_data() {
402        let data = b"+OK\r";
403        let mut cursor = Cursor::new(&data[..]);
404        let result = RespDecoder::decode(&mut cursor).unwrap();
405        assert!(result.is_none());
406    }
407
408    #[test]
409    fn test_roundtrip() {
410        let original = RespValue::Array(vec![
411            RespValue::SimpleString("OK".to_string()),
412            RespValue::Integer(42),
413            RespValue::BulkString(Bytes::from("test")),
414            RespValue::Null,
415        ]);
416
417        let mut buf = BytesMut::new();
418        RespEncoder::encode(&original, &mut buf).unwrap();
419
420        let mut cursor = Cursor::new(&buf[..]);
421        let decoded = RespDecoder::decode(&mut cursor).unwrap().unwrap();
422
423        assert_eq!(original, decoded);
424    }
425}
redis_oxide/protocol/resp2.rs

redis_oxide/protocol/
resp2.rs
