feat(resp): implement full RESP3 support with unified architecture

src/resp/Cargo.toml

@@ -9,4 +9,5 @@ workspace = true
 [dependencies]
 bytes.workspace = true
 thiserror.workspace = true
-nom.workspace = true
+nom.workspace = true
+memchr = "2"

src/resp/src/compat.rs

@@ -0,0 +1,34 @@
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum BooleanMode {
+    Integer,
+    SimpleString,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum DoubleMode {
+    BulkString,
+    IntegerIfWhole,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum MapMode {
+    FlatArray,
+    ArrayOfPairs,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct DownlevelPolicy {
+    pub boolean_mode: BooleanMode,
+    pub double_mode: DoubleMode,
+    pub map_mode: MapMode,
+}
+
+impl Default for DownlevelPolicy {
+    fn default() -> Self {
+        Self {
+            boolean_mode: BooleanMode::Integer,
+            double_mode: DoubleMode::BulkString,
+            map_mode: MapMode::FlatArray,
+        }
+    }
+}

src/resp/src/encode.rs

@@ -370,6 +370,7 @@ impl RespEncode for RespEncoder {
                 }
                 self.append_crlf()
             }
+            _ => self,
         }
     }
 }

src/resp/src/factory.rs

@@ -0,0 +1,46 @@
+// Copyright (c) 2024-present, arana-db Community.  All rights reserved.
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::{
+    compat::DownlevelPolicy,
+    traits::{Decoder, Encoder},
+    types::RespVersion,
+};
+
+pub fn new_decoder(version: RespVersion) -> Box<dyn Decoder> {
+    match version {
+        RespVersion::RESP1 => Box::new(crate::resp1::decoder::Resp1Decoder::default()),
+        RespVersion::RESP2 => Box::new(crate::resp2::decoder::Resp2Decoder::default()),
+        RespVersion::RESP3 => Box::new(crate::resp3::decoder::Resp3Decoder::default()),
+    }
+}
+
+pub fn new_encoder(version: RespVersion) -> Box<dyn Encoder> {
+    match version {
+        RespVersion::RESP1 => Box::new(crate::resp1::encoder::Resp1Encoder::default()),
+        RespVersion::RESP2 => Box::new(crate::resp2::encoder::Resp2Encoder::default()),
+        RespVersion::RESP3 => Box::new(crate::resp3::encoder::Resp3Encoder),
+    }
+}
+
+pub fn new_encoder_with_policy(version: RespVersion, policy: DownlevelPolicy) -> Box<dyn Encoder> {
+    match version {
+        RespVersion::RESP1 => Box::new(crate::resp1::encoder::Resp1Encoder::with_policy(policy)),
+        RespVersion::RESP2 => Box::new(crate::resp2::encoder::Resp2Encoder::with_policy(policy)),
+        RespVersion::RESP3 => Box::new(crate::resp3::encoder::Resp3Encoder),
+    }
+}

src/resp/src/lib.rs

@@ -16,15 +16,30 @@
 // limitations under the License.
 
 pub mod command;
+pub mod compat;
 pub mod encode;
 pub mod error;
 pub mod parse;
 pub mod types;
 
+// Versioned modules
+pub mod resp1;
+pub mod resp2;
+pub mod resp3;
+
+// Unified traits and helpers
+pub mod factory;
+pub mod multi;
+pub mod traits;
+
 pub use command::{Command, CommandType, RespCommand};
+pub use compat::{BooleanMode, DoubleMode, DownlevelPolicy, MapMode};
 pub use encode::{CmdRes, RespEncode};
 pub use error::{RespError, RespResult};
+pub use factory::{new_decoder, new_encoder, new_encoder_with_policy};
+pub use multi::{decode_many, encode_many};
 pub use parse::{Parse, RespParse, RespParseResult};
+pub use traits::{Decoder, Encoder};
 pub use types::{RespData, RespType, RespVersion};
 
 pub const CRLF: &str = "\r\n";

src/resp/src/multi.rs

@@ -0,0 +1,41 @@
+// Copyright (c) 2024-present, arana-db Community.  All rights reserved.
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use bytes::{Bytes, BytesMut};
+
+use crate::{
+    error::RespResult,
+    traits::{Decoder, Encoder},
+    types::RespData,
+};
+
+pub fn decode_many(decoder: &mut dyn Decoder, chunk: Bytes) -> Vec<RespResult<RespData>> {
+    decoder.push(chunk);
+    let mut out = Vec::new();
+    while let Some(frame) = decoder.next() {
+        out.push(frame);
+    }
+    out
+}
+
+pub fn encode_many(encoder: &mut dyn Encoder, values: &[RespData]) -> RespResult<Bytes> {
+    let mut buf = BytesMut::new();
+    for v in values {
+        encoder.encode_into(v, &mut buf)?;
+    }
+    Ok(buf.freeze())
+}

src/resp/src/parse.rs

@@ -35,7 +35,7 @@ use crate::{
     types::{RespData, RespVersion},
 };
 
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, PartialEq)]
 pub enum RespParseResult {
     Complete(RespData),
     Incomplete,

src/resp/src/resp1/decoder.rs

@@ -0,0 +1,52 @@
+use std::collections::VecDeque;
+
+use bytes::Bytes;
+
+use crate::{
+    error::RespResult,
+    parse::{Parse, RespParse, RespParseResult},
+    traits::Decoder,
+    types::{RespData, RespVersion},
+};
+
+#[derive(Default)]
+pub struct Resp1Decoder {
+    inner: RespParse,
+    out: VecDeque<RespResult<RespData>>,
+}
+
+impl Resp1Decoder {
+    pub fn new() -> Self {
+        Self {
+            inner: RespParse::new(RespVersion::RESP1),
+            out: VecDeque::new(),
+        }
+    }
+}
+
+impl Decoder for Resp1Decoder {
+    fn push(&mut self, data: Bytes) {
+        let mut res = self.inner.parse(data);
+        loop {
+            match res {
+                RespParseResult::Complete(d) => self.out.push_back(Ok(d)),
+                RespParseResult::Error(e) => self.out.push_back(Err(e)),
+                RespParseResult::Incomplete => break,
+            }
+            res = self.inner.parse(Bytes::new());
+        }
+    }
+
+    fn next(&mut self) -> Option<RespResult<RespData>> {
+        self.out.pop_front()
+    }
+
+    fn reset(&mut self) {
+        self.inner.reset();
+        self.out.clear();
+    }
+
+    fn version(&self) -> RespVersion {
+        RespVersion::RESP1
+    }
+}

src/resp/src/resp1/encoder.rs

@@ -0,0 +1,123 @@
+use bytes::{Bytes, BytesMut};
+
+use crate::{
+    compat::{BooleanMode, DoubleMode, DownlevelPolicy, MapMode},
+    encode::{RespEncode, RespEncoder},
+    error::RespResult,
+    traits::Encoder,
+    types::{RespData, RespVersion},
+};
+
+#[derive(Default)]
+pub struct Resp1Encoder {
+    inner: RespEncoder,
+    policy: DownlevelPolicy,
+}
+
+impl Resp1Encoder {
+    pub fn new() -> Self {
+        Self {
+            inner: RespEncoder::new(RespVersion::RESP1),
+            policy: DownlevelPolicy::default(),
+        }
+    }
+
+    pub fn with_policy(policy: DownlevelPolicy) -> Self {
+        Self {
+            inner: RespEncoder::new(RespVersion::RESP1),
+            policy,
+        }
+    }
+}
+
+impl Encoder for Resp1Encoder {
+    fn encode_one(&mut self, data: &RespData) -> RespResult<Bytes> {
+        self.inner.clear();
+        self.encode_downleveled(data);
+        Ok(self.inner.get_response())
+    }
+
+    fn encode_into(&mut self, data: &RespData, out: &mut BytesMut) -> RespResult<()> {
+        let bytes = self.encode_one(data)?;
+        out.extend_from_slice(&bytes);
+        Ok(())
+    }
+
+    fn version(&self) -> RespVersion {
+        RespVersion::RESP1
+    }
+}
+
+impl Resp1Encoder {
+    fn encode_downleveled(&mut self, data: &RespData) {
+        match data {
+            // RESP1 uses inline/simple/integer/bulk/array semantics basically consistent
+            RespData::SimpleString(_)
+            | RespData::Error(_)
+            | RespData::Integer(_)
+            | RespData::BulkString(_)
+            | RespData::Array(_)
+            | RespData::Inline(_) => {
+                self.inner.encode_resp_data(data);
+            }
+            // Downlevel mapping strategy consistent with RESP2
+            RespData::Null => {
+                self.inner.set_line_string("$-1");
+            }
+            RespData::Boolean(b) => {
+                match self.policy.boolean_mode {
+                    BooleanMode::Integer => self.inner.append_integer(if *b { 1 } else { 0 }),
+                    BooleanMode::SimpleString => {
+                        if *b {
+                            self.inner.append_simple_string("OK")
+                        } else {
+                            self.inner.append_simple_string("ERR")
+                        }
+                    }
+                };
+            }
+            RespData::Double(v) => {
+                if let DoubleMode::IntegerIfWhole = self.policy.double_mode {
+                    if v.fract() == 0.0 && v.is_finite() {
+                        self.inner.append_integer(*v as i64);
+                        return;
+                    }
+                }
+                self.inner.append_string(&format!("{}", v));
+            }
+            crate::types::RespData::BulkError(msg) => {
+                self.inner
+                    .append_string_raw(&format!("-{}\r\n", String::from_utf8_lossy(msg)));
+            }
+            crate::types::RespData::VerbatimString { data, .. } => {
+                self.inner.append_bulk_string(data);
+            }
+            crate::types::RespData::BigNumber(s) => {
+                self.inner.append_string(s);
+            }
+            crate::types::RespData::Map(entries) => match self.policy.map_mode {
+                MapMode::FlatArray => {
+                    self.inner.append_array_len((entries.len() * 2) as i64);
+                    for (k, v) in entries {
+                        self.encode_downleveled(k);
+                        self.encode_downleveled(v);
+                    }
+                }
+                MapMode::ArrayOfPairs => {
+                    self.inner.append_array_len(entries.len() as i64);
+                    for (k, v) in entries {
+                        self.inner.append_array_len(2);
+                        self.encode_downleveled(k);
+                        self.encode_downleveled(v);
+                    }
+                }
+            },
+            crate::types::RespData::Set(items) | crate::types::RespData::Push(items) => {
+                self.inner.append_array_len(items.len() as i64);
+                for it in items {
+                    self.encode_downleveled(it);
+                }
+            }
+        }
+    }
+}

src/resp/src/resp1/mod.rs

@@ -0,0 +1,2 @@
+pub mod decoder;
+pub mod encoder;