redis_oxide/
protocol.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    /// Encode a command with arguments
60    pub fn encode_command(command: &str, args: &[RespValue]) -> RedisResult<Bytes> {
61        let mut buf = BytesMut::new();
62
63        // Create array with command + args
64        let total_len = 1 + args.len();
65        buf.put_u8(b'*');
66        buf.put_slice(total_len.to_string().as_bytes());
67        buf.put_slice(CRLF);
68
69        // Encode command
70        buf.put_u8(b'$');
71        buf.put_slice(command.len().to_string().as_bytes());
72        buf.put_slice(CRLF);
73        buf.put_slice(command.as_bytes());
74        buf.put_slice(CRLF);
75
76        // Encode arguments
77        for arg in args {
78            Self::encode(arg, &mut buf)?;
79        }
80
81        Ok(buf.freeze())
82    }
83}
84
85/// Decodes RESP values from bytes
86pub struct RespDecoder;
87
88impl RespDecoder {
89    /// Decode a RESP value from a buffer
90    pub fn decode(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
91        if !buf.has_remaining() {
92            return Ok(None);
93        }
94
95        let type_byte = buf.chunk()[0];
96
97        match type_byte {
98            b'+' => Self::decode_simple_string(buf),
99            b'-' => Self::decode_error(buf),
100            b':' => Self::decode_integer(buf),
101            b'$' => Self::decode_bulk_string(buf),
102            b'*' => Self::decode_array(buf),
103            _ => Err(RedisError::Protocol(format!(
104                "Invalid RESP type byte: {}",
105                type_byte as char
106            ))),
107        }
108    }
109
110    fn decode_simple_string(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
111        buf.advance(1); // Skip '+'
112
113        if let Some(line) = Self::read_line(buf)? {
114            Ok(Some(RespValue::SimpleString(
115                String::from_utf8(line.to_vec())
116                    .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?,
117            )))
118        } else {
119            Ok(None)
120        }
121    }
122
123    fn decode_error(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
124        buf.advance(1); // Skip '-'
125
126        if let Some(line) = Self::read_line(buf)? {
127            Ok(Some(RespValue::Error(
128                String::from_utf8(line.to_vec())
129                    .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?,
130            )))
131        } else {
132            Ok(None)
133        }
134    }
135
136    fn decode_integer(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
137        buf.advance(1); // Skip ':'
138
139        if let Some(line) = Self::read_line(buf)? {
140            let num_str = String::from_utf8(line.to_vec())
141                .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?;
142            let num = num_str
143                .parse::<i64>()
144                .map_err(|e| RedisError::Protocol(format!("Invalid integer: {}", e)))?;
145            Ok(Some(RespValue::Integer(num)))
146        } else {
147            Ok(None)
148        }
149    }
150
151    fn decode_bulk_string(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
152        buf.advance(1); // Skip '$'
153
154        let len_line = match Self::read_line(buf)? {
155            Some(line) => line,
156            None => return Ok(None),
157        };
158
159        let len_str = String::from_utf8(len_line.to_vec())
160            .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?;
161        let len = len_str
162            .parse::<i64>()
163            .map_err(|e| RedisError::Protocol(format!("Invalid bulk string length: {}", e)))?;
164
165        if len == -1 {
166            return Ok(Some(RespValue::Null));
167        }
168
169        let len = len as usize;
170
171        // Check if we have enough data
172        if buf.remaining() < len + 2 {
173            return Ok(None);
174        }
175
176        let data = buf.chunk()[..len].to_vec();
177        buf.advance(len);
178
179        // Skip CRLF
180        if buf.remaining() < 2 {
181            return Ok(None);
182        }
183        buf.advance(2);
184
185        Ok(Some(RespValue::BulkString(Bytes::from(data))))
186    }
187
188    fn decode_array(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<RespValue>> {
189        buf.advance(1); // Skip '*'
190
191        let len_line = match Self::read_line(buf)? {
192            Some(line) => line,
193            None => return Ok(None),
194        };
195
196        let len_str = String::from_utf8(len_line.to_vec())
197            .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?;
198        let len = len_str
199            .parse::<i64>()
200            .map_err(|e| RedisError::Protocol(format!("Invalid array length: {}", e)))?;
201
202        if len == -1 {
203            return Ok(Some(RespValue::Null));
204        }
205
206        let len = len as usize;
207        let mut arr = Vec::with_capacity(len);
208
209        for _ in 0..len {
210            match Self::decode(buf)? {
211                Some(value) => arr.push(value),
212                None => return Ok(None),
213            }
214        }
215
216        Ok(Some(RespValue::Array(arr)))
217    }
218
219    fn read_line(buf: &mut Cursor<&[u8]>) -> RedisResult<Option<Vec<u8>>> {
220        let start = buf.position() as usize;
221        let slice = buf.get_ref();
222
223        // Find CRLF
224        for i in start..slice.len().saturating_sub(1) {
225            if slice[i] == b'\r' && slice[i + 1] == b'\n' {
226                let line = slice[start..i].to_vec();
227                buf.set_position((i + 2) as u64);
228                return Ok(Some(line));
229            }
230        }
231
232        Ok(None)
233    }
234}
235
236#[cfg(test)]
237mod tests {
238    use super::*;
239
240    #[test]
241    fn test_encode_simple_string() {
242        let mut buf = BytesMut::new();
243        let value = RespValue::SimpleString("OK".to_string());
244        RespEncoder::encode(&value, &mut buf).unwrap();
245        assert_eq!(&buf[..], b"+OK\r\n");
246    }
247
248    #[test]
249    fn test_encode_error() {
250        let mut buf = BytesMut::new();
251        let value = RespValue::Error("ERR unknown command".to_string());
252        RespEncoder::encode(&value, &mut buf).unwrap();
253        assert_eq!(&buf[..], b"-ERR unknown command\r\n");
254    }
255
256    #[test]
257    fn test_encode_integer() {
258        let mut buf = BytesMut::new();
259        let value = RespValue::Integer(1000);
260        RespEncoder::encode(&value, &mut buf).unwrap();
261        assert_eq!(&buf[..], b":1000\r\n");
262    }
263
264    #[test]
265    fn test_encode_bulk_string() {
266        let mut buf = BytesMut::new();
267        let value = RespValue::BulkString(Bytes::from("foobar"));
268        RespEncoder::encode(&value, &mut buf).unwrap();
269        assert_eq!(&buf[..], b"$6\r\nfoobar\r\n");
270    }
271
272    #[test]
273    fn test_encode_null() {
274        let mut buf = BytesMut::new();
275        let value = RespValue::Null;
276        RespEncoder::encode(&value, &mut buf).unwrap();
277        assert_eq!(&buf[..], b"$-1\r\n");
278    }
279
280    #[test]
281    fn test_encode_array() {
282        let mut buf = BytesMut::new();
283        let value = RespValue::Array(vec![
284            RespValue::BulkString(Bytes::from("foo")),
285            RespValue::BulkString(Bytes::from("bar")),
286        ]);
287        RespEncoder::encode(&value, &mut buf).unwrap();
288        assert_eq!(&buf[..], b"*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n");
289    }
290
291    #[test]
292    fn test_encode_command() {
293        let bytes =
294            RespEncoder::encode_command("GET", &[RespValue::BulkString(Bytes::from("mykey"))])
295                .unwrap();
296        assert_eq!(&bytes[..], b"*2\r\n$3\r\nGET\r\n$5\r\nmykey\r\n");
297    }
298
299    #[test]
300    fn test_decode_simple_string() {
301        let data = b"+OK\r\n";
302        let mut cursor = Cursor::new(&data[..]);
303        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
304        assert_eq!(value, RespValue::SimpleString("OK".to_string()));
305    }
306
307    #[test]
308    fn test_decode_error() {
309        let data = b"-ERR unknown\r\n";
310        let mut cursor = Cursor::new(&data[..]);
311        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
312        assert_eq!(value, RespValue::Error("ERR unknown".to_string()));
313    }
314
315    #[test]
316    fn test_decode_integer() {
317        let data = b":1000\r\n";
318        let mut cursor = Cursor::new(&data[..]);
319        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
320        assert_eq!(value, RespValue::Integer(1000));
321    }
322
323    #[test]
324    fn test_decode_bulk_string() {
325        let data = b"$6\r\nfoobar\r\n";
326        let mut cursor = Cursor::new(&data[..]);
327        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
328        assert_eq!(value, RespValue::BulkString(Bytes::from("foobar")));
329    }
330
331    #[test]
332    fn test_decode_null() {
333        let data = b"$-1\r\n";
334        let mut cursor = Cursor::new(&data[..]);
335        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
336        assert_eq!(value, RespValue::Null);
337    }
338
339    #[test]
340    fn test_decode_array() {
341        let data = b"*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n";
342        let mut cursor = Cursor::new(&data[..]);
343        let value = RespDecoder::decode(&mut cursor).unwrap().unwrap();
344        assert_eq!(
345            value,
346            RespValue::Array(vec![
347                RespValue::BulkString(Bytes::from("foo")),
348                RespValue::BulkString(Bytes::from("bar")),
349            ])
350        );
351    }
352
353    #[test]
354    fn test_decode_incomplete_data() {
355        let data = b"+OK\r";
356        let mut cursor = Cursor::new(&data[..]);
357        let result = RespDecoder::decode(&mut cursor).unwrap();
358        assert!(result.is_none());
359    }
360
361    #[test]
362    fn test_roundtrip() {
363        let original = RespValue::Array(vec![
364            RespValue::SimpleString("OK".to_string()),
365            RespValue::Integer(42),
366            RespValue::BulkString(Bytes::from("test")),
367            RespValue::Null,
368        ]);
369
370        let mut buf = BytesMut::new();
371        RespEncoder::encode(&original, &mut buf).unwrap();
372
373        let mut cursor = Cursor::new(&buf[..]);
374        let decoded = RespDecoder::decode(&mut cursor).unwrap().unwrap();
375
376        assert_eq!(original, decoded);
377    }
378}
redis_oxide/protocol.rs

redis_oxide/
protocol.rs
