Update redis-protocol 0.4 to 0.5 (#1556)

Cargo.toml

@@ -43,7 +43,7 @@ rcgen = "0.12.0"
 subprocess = "0.2.7"
 chacha20poly1305 = { version = "0.10.0", features = ["std"] }
 csv = "1.2.0"
-redis-protocol = { version = "4.0.1", features = ["decode-mut"] }
+redis-protocol = { version = "5.0.0", features = ["bytes"] }
 bincode = "1.3.1"
 futures = "0.3"
 hex = "0.4.3"

shotover-proxy/tests/redis_int_tests/basic_driver_tests.rs

@@ -1309,7 +1309,7 @@ async fn read_redis_message(connection: &mut TcpStream) -> RedisFrame {
     let mut buffer = BytesMut::new();
     loop {
         if let Ok(Some((result, len))) =
-            redis_protocol::resp2::decode::decode(&buffer.clone().freeze())
+            redis_protocol::resp2::decode::decode_bytes(&buffer.clone().freeze())
         {
             let _ = buffer.split_to(len);
             return result;

shotover/Cargo.toml

@@ -38,7 +38,6 @@ redis = [
     "dep:redis-protocol",
     "dep:csv",
     "dep:crc16",
-    "dep:bytes-utils"
 ]
 opensearch = [
     "dep:atoi",
@@ -52,7 +51,6 @@ atomic_enum = "0.3.0"
 axum = { version = "0.7", default-features = false, features = ["tokio", "tracing", "http1"] }
 pretty-hex = "0.4.0"
 tokio-stream = "0.1.2"
-bytes-utils = { version = "0.1.1", optional = true }
 derivative = "2.1.1"
 cached = { version = "0.49", features = ["async"], optional = true }
 governor = { version = "0.6", default-features = false, features = ["std", "jitter", "quanta"] }

shotover/src/codec/redis.rs

@@ -8,8 +8,8 @@ use crate::message::{Encodable, Message, MessageId, Messages};
 use anyhow::{anyhow, Result};
 use bytes::BytesMut;
 use metrics::Histogram;
-use redis_protocol::resp2::prelude::decode_mut;
-use redis_protocol::resp2::prelude::encode_bytes;
+use redis_protocol::resp2::decode::decode_bytes_mut;
+use redis_protocol::resp2::encode::extend_encode;
 use tokio_util::codec::{Decoder, Encoder};
 
 #[derive(Clone)]
@@ -100,7 +100,7 @@ impl Decoder for RedisDecoder {
     //       Once thats done we will have pubsub support for both RedisSinkSingle AND RedisSinkCluster. Progress!
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         let received_at = Instant::now();
-        match decode_mut(src)
+        match decode_bytes_mut(src)
             .map_err(|e| CodecReadError::Parser(anyhow!(e).context("Error decoding redis frame")))?
         {
             Some((frame, _size, bytes)) => {
@@ -241,7 +241,7 @@ impl Encoder<Messages> for RedisEncoder {
                 }
                 Encodable::Frame(frame) => {
                     let item = frame.into_redis().unwrap();
-                    encode_bytes(dst, &item)
+                    extend_encode(dst, &item)
                         .map(|_| ())
                         .map_err(|e| anyhow!("Redis encoding error: {} - {:#?}", e, item))
                 }

shotover/src/frame/mod.rs

@@ -12,7 +12,7 @@ use kafka::KafkaFrame;
 #[cfg(feature = "opensearch")]
 pub use opensearch::OpenSearchFrame;
 #[cfg(feature = "redis")]
-pub use redis_protocol::resp2::types::Frame as RedisFrame;
+pub use redis_protocol::resp2::types::BytesFrame as RedisFrame;
 use std::fmt::{Display, Formatter, Result as FmtResult};
 
 #[cfg(feature = "cassandra")]
@@ -131,7 +131,7 @@ impl Frame {
                 CassandraFrame::from_bytes(bytes, codec_state.as_cassandra()).map(Frame::Cassandra)
             }
             #[cfg(feature = "redis")]
-            MessageType::Redis => redis_protocol::resp2::decode::decode(&bytes)
+            MessageType::Redis => redis_protocol::resp2::decode::decode_bytes(&bytes)
                 .map(|x| Frame::Redis(x.unwrap().0))
                 .map_err(|e| anyhow!("{e:?}")),
             #[cfg(feature = "kafka")]

shotover/src/transforms/redis/sink_cluster.rs

@@ -14,7 +14,6 @@ use crate::transforms::{
 use anyhow::{anyhow, bail, ensure, Context, Result};
 use async_trait::async_trait;
 use bytes::Bytes;
-use bytes_utils::string::Str;
 use derivative::Derivative;
 use futures::stream::FuturesOrdered;
 use futures::stream::FuturesUnordered;
@@ -24,7 +23,8 @@ use metrics::counter;
 use rand::rngs::SmallRng;
 use rand::seq::IteratorRandom;
 use rand::SeedableRng;
-use redis_protocol::types::Redirection;
+use redis_protocol::bytes_utils::string::Str;
+use redis_protocol::resp2::types::Resp2Frame;
 use serde::{Deserialize, Serialize};
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::Arc;
@@ -1054,7 +1054,7 @@ impl Transform for RedisSinkCluster {
             let mut response = response?;
             match response.frame() {
                 Some(Frame::Redis(frame)) => {
-                    match frame.to_redirection() {
+                    match Redirection::parse(frame) {
                         Some(Redirection::Moved { slot, server }) => {
                             debug!("Got MOVE {} {}", slot, server);
 
@@ -1088,6 +1088,33 @@ impl Transform for RedisSinkCluster {
     }
 }
 
+enum Redirection {
+    Moved { slot: u16, server: String },
+    Ask { slot: u16, server: String },
+}
+
+impl Redirection {
+    fn parse(frame: &RedisFrame) -> Option<Redirection> {
+        match frame {
+            RedisFrame::Error(err) => {
+                let mut tokens = err.split(' ');
+                match tokens.next()? {
+                    "MOVED" => Some(Redirection::Moved {
+                        slot: tokens.next()?.parse().ok()?,
+                        server: tokens.next()?.to_owned(),
+                    }),
+                    "ASK" => Some(Redirection::Ask {
+                        slot: tokens.next()?.parse().ok()?,
+                        server: tokens.next()?.to_owned(),
+                    }),
+                    _ => None,
+                }
+            }
+            _ => None,
+        }
+    }
+}
+
 #[derive(Clone, PartialEq, Eq, Hash, Derivative)]
 #[derivative(Debug)]
 pub struct UsernamePasswordToken {