Merge pull request #18 from Miaxos/feat-parse-proper-map-resp3

Feat: handle resp3 map

Author: Miaxos

Cargo.lock

@@ -23,6 +23,7 @@ default = []
 
 [dependencies]
 anyhow.workspace = true
+async-recursion = "1"
 atoi_simd = "0.15"
 coarsetime = "0.1"
 crc = "3"
@@ -34,6 +35,7 @@ derive_builder.workspace = true
 dotenv.workspace = true
 futures = "0.3"
 futures-locks = "0.7"
+indexmap = "2"
 local-sync = "0.1"
 monoio = { workspace = true, features = ["bytes", "sync", "iouring"] }
 rustc-hash = "1.1.0"

app/roster/Cargo.toml

@@ -2,13 +2,13 @@ use std::fmt::Debug;
 use std::io::{self, Cursor};
 
 use bytes::BytesMut;
-use monoio::buf::IoBuf;
 use monoio::io::{
-    AsyncReadRent, AsyncWriteRent, BufReader, BufWriter, OwnedReadHalf,
-    OwnedWriteHalf, Splitable,
+    AsyncReadRent, BufReader, BufWriter, OwnedReadHalf, OwnedWriteHalf,
+    Splitable,
 };
 use monoio::net::TcpStream;
 
+use super::frame::write::write_frame;
 use super::frame::Frame;
 
 /// Send and receive `Frame` values from a remote peer.
@@ -179,83 +179,7 @@ impl WriteConnection {
     /// *buffered* write stream. The data will be written to the buffer.
     /// Once the buffer is full, it is flushed to the underlying socket.
     pub async fn write_frame(&mut self, frame: &Frame) -> io::Result<()> {
-        // Arrays are encoded by encoding each entry. All other frame types are
-        // considered literals. For now, mini-redis is not able to encode
-        // recursive frame structures. See below for more details.
-        match frame {
-            Frame::Array(val) => {
-                // Encode the length of the array.
-                self.stream_w.write(&[b'*']).await.0?;
-                self.write_decimal(val.len() as u64).await?;
-
-                // Iterate and encode each entry in the array.
-                for entry in &**val {
-                    self.write_value(entry).await?;
-                }
-            }
-            // The frame type is a literal. Encode the value directly.
-            _ => self.write_value(frame).await?,
-        }
-
-        // Ensure the encoded frame is written to the socket. The calls above
-        // are to the buffered stream and writes. Calling `flush` writes the
-        // remaining contents of the buffer to the socket.
-        self.stream_w.flush().await
-    }
-
-    /// Write a frame literal to the stream
-    async fn write_value(&mut self, frame: &Frame) -> io::Result<()> {
-        match frame {
-            Frame::Simple(val) => {
-                self.stream_w.write(&[b'+']).await.0?;
-                self.stream_w.write(val.as_bytes().slice(..)).await.0?;
-                self.stream_w.write(&[b'\r', b'\n']).await.0?;
-            }
-            Frame::Error(val) => {
-                self.stream_w.write(&[b'-']).await.0?;
-                self.stream_w.write(val.as_bytes().slice(..)).await.0?;
-                self.stream_w.write(&[b'\r', b'\n']).await.0?;
-            }
-            Frame::Integer(val) => {
-                self.stream_w.write(&[b':']).await.0?;
-                self.write_decimal(*val).await?;
-            }
-            Frame::Null => {
-                self.stream_w.write(b"$-1\r\n").await.0?;
-            }
-            Frame::Bulk(val) => {
-                let len = val.len();
-
-                self.stream_w.write([b'$'].as_slice()).await.0?;
-                self.write_decimal(len as u64).await?;
-                self.stream_w.write(val.slice(..)).await.0?;
-                self.stream_w.write(&[b'\r', b'\n']).await.0?;
-            }
-            // Encoding an `Array` from within a value cannot be done using a
-            // recursive strategy. In general, async fns do not support
-            // recursion. Mini-redis has not needed to encode nested arrays yet,
-            // so for now it is skipped.
-            Frame::Array(_val) => unreachable!(),
-        }
-
-        Ok(())
-    }
-
-    /// Write a decimal frame to the stream_w
-    async fn write_decimal(&mut self, val: u64) -> io::Result<()> {
-        use std::io::Write;
-
-        // Convert the value to a string
-        let buf = vec![0u8; 20];
-        let mut buf = Cursor::new(buf);
-        write!(&mut buf, "{}", val)?;
-
-        let pos = buf.position() as usize;
-
-        self.stream_w.write(buf.into_inner().slice(..pos)).await.0?;
-        self.stream_w.write(b"\r\n").await.0?;
-
-        Ok(())
+        write_frame(&mut self.stream_w, frame).await
     }
 
     pub fn into_inner(self) -> OwnedWriteHalf<TcpStream> {

app/roster/src/application/server/connection.rs

@@ -8,16 +8,49 @@ use std::string::FromUtf8Error;
 
 use bytes::{Buf, Bytes, BytesMut};
 use bytestring::ByteString;
+use indexmap::IndexMap;
+
+pub(crate) mod write;
 
 /// A frame in the Redis protocol.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub enum Frame {
     Simple(ByteString),
     Error(ByteString),
     Integer(u64),
     Bulk(Bytes),
     Null,
     Array(Vec<Frame>),
+    Map(IndexMap<Frame, Frame>),
+}
+
+// TODO(@miaxos): hacky hash derivation for now, to test a little.
+impl core::hash::Hash for Frame {
+    fn hash<H: core::hash::Hasher>(&self, ra_expand_state: &mut H) {
+        core::mem::discriminant(self).hash(ra_expand_state);
+        match self {
+            Frame::Simple(f0) => {
+                f0.hash(ra_expand_state);
+            }
+            Frame::Error(f0) => {
+                f0.hash(ra_expand_state);
+            }
+            Frame::Integer(f0) => {
+                f0.hash(ra_expand_state);
+            }
+            Frame::Bulk(f0) => {
+                f0.hash(ra_expand_state);
+            }
+            Frame::Null => {}
+            Frame::Array(f0) => {
+                f0.hash(ra_expand_state);
+            }
+            Frame::Map(_) => {
+                // TODO: Should test the behavior of redis in this case.
+                unimplemented!("")
+            }
+        }
+    }
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -68,6 +101,16 @@ impl Frame {
 
                 Ok(())
             }
+            b'%' => {
+                let len = get_decimal_mut(src)?;
+
+                // Key and value frames
+                for _ in 0..(len * 2) {
+                    Frame::check(src)?;
+                }
+
+                Ok(())
+            }
             actual => Err(format!(
                 "protocol error; invalid frame type byte `{}`",
                 actual
@@ -137,6 +180,18 @@ impl Frame {
 
                 Ok(Frame::Array(out))
             }
+            b'%' => {
+                let len = get_decimal(src)?.try_into()?;
+                let mut out = IndexMap::with_capacity(len);
+
+                for _ in (0..(len * 2)).step_by(2) {
+                    let key = Frame::parse(src)?;
+                    let value = Frame::parse(src)?;
+                    out.insert(key, value);
+                }
+
+                Ok(Frame::Map(out))
+            }
             _ => unimplemented!(),
         }
     }
@@ -321,4 +376,16 @@ mod tests {
             assert!(Frame::check(&mut cur).is_ok());
         }
     }
+
+    #[test]
+    fn test_map_frame() {
+        let test_case: Vec<&[u8]> =
+            vec![b"%2\r\n+first\r\n:1\r\n+second\r\n:2\r\n"];
+
+        for t in test_case {
+            let b = BytesMut::from(t);
+            let mut cur = Cursor::new(&b);
+            assert!(Frame::check(&mut cur).is_ok());
+        }
+    }
 }