Redis from scratch: Frame

Let’s write a Redis client from scratch!

The foundation of client-server interaction is the REdis Serialization Protocol (RESP), which defines the data types and how they encode into sequences of bytes. According to the official docs, there are five main types of data frames:

type	encoding	notes
simple string	+<data><CRLF>	binary unsafe
error	-<data><CRLF>	binary unsafe
integer	:<num><CRLF>	binary unsafe, number must be in range of signed 64-bit integer
bulk	$<nbytes><CRLF><bytes><CRLF>	binary safe
null	$-1<CRLF>	A special case of a bulk frame
array	*<nelems><CRLF>	can be nested

These types can be trivially represented in an enum:

pub enum Frame {
    Simple(String),
    Error(String),
    Integer(i64),
    Bulk(Bytes),
    Null,
    Array(Vec<Frame>),
}

In addition, serialization is trivial, maybe except for serializing Frame::Array, which requires some non-trivial recursive calls.

Deserialization on the other hand, is not trivial. The core function for parsing Frame from some unknown bytes will be the parse function:

impl Frame {
    pub fn parse(&mut bytes: Bytes) -> Option<Frame> {
        todo!()
    }
}

First we need to consume the first byte to determine what kind of frame we are expecting. For that the method Bytes::get_u8 is used:

use bytes::{Buf, Bytes};

impl Frame {
    pub fn parse(&mut bytes: Bytes) -> Option<Frame> {
        match bytes.get_u8() {
            b'+' => { todo!() }
            b'-' => { todo!() }
            b':' => { todo!() }
            b'$' => { todo!() }
            b'*' => { todo!() }
            _ => (),
        }
        return None;
    }
}

Another procedure common to parsing binary-unsafe types and to extracting the size of binary-safe types is “read until CRLF”:

const CRLF: &str = "\r\n";

impl Frame {
    fn _parse_binary_safe_string(bytes: &mut Bytes) -> Option<String> {
        let mut msg = vec![];

        while bytes.has_remaining() && !bytes.starts_with(CRLF.as_bytes()) {
            msg.push(bytes.get_u8());
        }

        if bytes.has_remaining() {
            // CRLF should be consumed, as well
            bytes.advance(CRLF.as_bytes().len());
            if let Ok(msg) = String::from_utf8(msg) {
                return Some(msg);
            }
        }
        return None;
    }
}

Here is the final implementation:

fn parse(bytes: &mut Bytes) -> Option<Frame> {
    if !bytes.has_remaining() {
        return None;
    }
    match bytes.get_u8() {
        b'+' => {
            if let Some(msg) = Self::_parse_binary_safe_string(bytes) {
                return Some(Frame::Simple(msg));
            }
        }
        b'-' => {
            if let Some(msg) = Self::_parse_binary_safe_string(bytes) {
                return Some(Frame::Error(msg));
            }
        }
        b':' => {
            if let Some(num) = Self::_parse_binary_safe_string(bytes) {
                if let Ok(num) = num.parse::<i64>() {
                    return Some(Frame::Integer(num));
                }
            }
        }
        b'$' => {
            // Check against Null frame
            if bytes.starts_with(b"-1\r\n") {
                return Some(Frame::Null);
            }

            // Read until the first CRLF to parse the number of bytes
            if let Some(nbytes) = Self::_parse_binary_safe_string(bytes) {
                if let Ok(nbytes) = nbytes.parse::<usize>() {
                    // bytes[0..nbytes] should be the content
                    // bytes[nbytes..nbytes+2] should be another CRLF
                    if bytes.remaining() >= nbytes + 2
                        && bytes.slice(nbytes..nbytes + 2).starts_with(CRLF.as_bytes())
                    {
                        let frame = Frame::Bulk(Bytes::from(bytes.slice(0..nbytes)));
                        bytes.advance(nbytes + 2);
                        return Some(frame);
                    }
                }
            }
        }
        b'*' => {
            // Parsing an array: first obtain the number of elements, then
            // fill a Vector with that number of elements
            if let Some(nelems) = Self::_parse_binary_safe_string(bytes) {
                if let Ok(nelems) = nelems.parse::<usize>() {
                    let mut elems = vec![];

                    for _ in 0..nelems {
                        if let Some(frame) = Self::parse(bytes) {
                            elems.push(frame);
                        } else {
                            return None;
                        }
                    }

                    return Some(Frame::Array(elems));
                }
            }
        }
        _ => (),
    }

    return None;
}

Here are the test cases:

#[test]
fn test_simple_string_deserialization() {
    assert_eq!(
        Frame::parse(&mut Bytes::from("+SET\r\n")),
        Some(Frame::Simple("SET".into())),
    );

    assert_eq!(
        Frame::parse(&mut Bytes::from("+SET\r\n+++++")),
        Some(Frame::Simple("SET".into())),
    );

    assert_eq!(Frame::parse(&mut Bytes::from("+SET\r")), None,);

    assert_eq!(
        Frame::parse(&mut Bytes::from("+\r\n")),
        Some(Frame::Simple("".into())),
    );
}

#[test]
fn test_error_deserialization() {
    assert_eq!(
        Frame::parse(&mut Bytes::from("-Key not found\r\n")),
        Some(Frame::Error("Key not found".into())),
    );

    assert_eq!(
        Frame::parse(&mut Bytes::from("-Key not found\r\n-----")),
        Some(Frame::Error("Key not found".into())),
    );

    assert_eq!(Frame::parse(&mut Bytes::from("-Key not found\r")), None,);

    assert_eq!(
        Frame::parse(&mut Bytes::from("-\r\n")),
        Some(Frame::Error("".into())),
    );
}

#[test]
fn test_integer_deserialization() {
    assert_eq!(
        Frame::parse(&mut Bytes::from(":0\r\n")),
        Some(Frame::Integer(0)),
    );

    assert_eq!(
        Frame::parse(&mut Bytes::from(":1\r\n")),
        Some(Frame::Integer(1)),
    );

    assert_eq!(
        Frame::parse(&mut Bytes::from(":-1\r\n")),
        Some(Frame::Integer(-1)),
    );

    assert_eq!(
        Frame::parse(&mut Bytes::from(":9223372036854775807\r\n")),
        Some(Frame::Integer(9223372036854775807)),
    );

    assert_eq!(
        Frame::parse(&mut Bytes::from(":-9223372036854775808\r\n")),
        Some(Frame::Integer(-9223372036854775808)),
    );

    assert_eq!(
        Frame::parse(&mut Bytes::from(":9223372036854775808\r\n")),
        None,
    );
}

#[test]
fn test_bulk_string_deserialization() {
    // Empty bulk string
    assert_eq!(
        Frame::parse(&mut Bytes::from("$0\r\n\r\n")),
        Some(Frame::Bulk(Bytes::new())),
    );

    // Non-empty bulk string
    assert_eq!(
        Frame::parse(&mut Bytes::from("$36\r\n那么古尔丹，代价是什么呢\r\n")),
        Some(Frame::Bulk(Bytes::from("那么古尔丹，代价是什么呢")))
    );

    // Binary unsafe string
    assert_eq!(
        Frame::parse(&mut Bytes::from(b"$2\r\n\r\n\r\n".to_vec())),
        Some(Frame::Bulk(Bytes::from(b"\r\n".to_vec())))
    );

    // Incomplete
    assert_eq!(Frame::parse(&mut Bytes::from("$10\r\n0123456789")), None);

    // Inconsistent number
    assert_eq!(Frame::parse(&mut Bytes::from("$10\r\n0123456\r\n")), None,);

    // Noise at the end
    assert_eq!(
        Frame::parse(&mut Bytes::from("$10\r\n0123456789\r\nxxxxxx")),
        Some(Frame::Bulk(Bytes::from("0123456789")))
    );
}

#[test]
fn test_null_frame_deserialization() {
    assert_eq!(Frame::parse(&mut Bytes::from("$-1\r\n")), Some(Frame::Null));
}

#[test]
fn test_array_deserialization() {
    assert_eq!(
        Frame::parse(&mut Bytes::from("*0\r\n")),
        Some(Frame::Array(vec![]))
    );

    let some_cmd = Frame::Array(vec![
        Frame::Simple("SET".into()),
        Frame::Bulk(Bytes::from("foo")),
        Frame::Bulk(Bytes::from("bar")),
    ]);
    assert_eq!(Frame::parse(&mut some_cmd.serialize()), Some(some_cmd),);

    let some_cmd = Frame::Array(vec![
        Frame::Simple("DEL".into()),
        Frame::Array(vec![
            Frame::Bulk(Bytes::from("key1")),
            Frame::Bulk(Bytes::from("key2")),
            Frame::Bulk(Bytes::from("key3")),
            Frame::Bulk(Bytes::from("key4")),
        ]),
    ]);
    assert_eq!(Frame::parse(&mut some_cmd.serialize()), Some(some_cmd),);

    assert_eq!(Frame::parse(&mut Bytes::from("*3\r\n:0\r\n:1\r\n")), None);

    assert_eq!(
        Frame::parse(&mut Bytes::from("*2\r\n:0\r\n:1\r\n+++++++")),
        Some(Frame::Array(vec![Frame::Integer(0), Frame::Integer(1)])),
    );
}