add bincode io

Author: Migorithm

duva-client/src/broker/mod.rs

@@ -14,7 +14,7 @@ use duva::prelude::tokio::sync::mpsc::Receiver;
 use duva::prelude::tokio::sync::mpsc::Sender;
 use duva::prelude::uuid::Uuid;
 use duva::prelude::{
-    ConnectionRequest, ConnectionRequests, ConnectionResponse, ConnectionResponses,
+    BinBytes, ConnectionRequest, ConnectionRequests, ConnectionResponse, ConnectionResponses,
 };
 use duva::prelude::{PeerIdentifier, tokio};
 use duva::prelude::{Topology, anyhow};
@@ -97,9 +97,9 @@ impl Broker {
                         context.expected_result_cnt = result_count;
                         queue.push(context);
                     } else {
-                        context.callback(QueryIO::Err(
-                            "Failed to route command. Try again after ttl time".into(),
-                        ));
+                        context.callback(QueryIO::Err(BinBytes::new(
+                            "Failed to route command. Try again after ttl time",
+                        )));
                     };
                 },
             }

duva-client/src/command.rs

@@ -1,5 +1,6 @@
 use duva::domains::caches::cache_manager::IndexedValueCodec;
 use duva::domains::replications::LogEntry;
+use duva::prelude::BinBytes;
 use duva::prelude::anyhow::{self, Context};
 use duva::presentation::clients::request::NonMutatingAction;
 use duva::{
@@ -30,7 +31,7 @@ impl CommandQueue {
         }
 
         let result =
-            context.get_result().unwrap_or_else(|err| QueryIO::Err(err.to_string().into()));
+            context.get_result().unwrap_or_else(|err| QueryIO::Err(BinBytes::new(err.to_string())));
         context.callback(result);
     }
 }
@@ -93,7 +94,7 @@ impl InputContext {
 
                     count += num;
                 }
-                Ok(QueryIO::SimpleString(count.to_string().into()))
+                Ok(QueryIO::SimpleString(BinBytes::new(count.to_string())))
             },
             ClientAction::Mutating(LogEntry::Delete { keys: _ }) => {
                 let mut count = 0;
@@ -106,7 +107,7 @@ impl InputContext {
 
                     count += decoded_value;
                 }
-                Ok(QueryIO::SimpleString(count.to_string().into()))
+                Ok(QueryIO::SimpleString(BinBytes::new(count.to_string())))
             },
             _ => {
                 if res.len() != 1 {

duva-client/src/controller.rs

@@ -46,14 +46,14 @@ fn render_return(kind: ClientAction, query_io: QueryIO) -> Response {
             | ClusterInfo,
         ) => match query_io {
             QueryIO::Null => Response::Null,
-            QueryIO::SimpleString(value) => Response::String(value),
-            QueryIO::BulkString(value) => Response::String(value),
-            QueryIO::Err(value) => Response::Error(value),
+            QueryIO::SimpleString(value) => Response::String(value.into()),
+            QueryIO::BulkString(value) => Response::String(value.into()),
+            QueryIO::Err(value) => Response::Error(value.into()),
             _err => Response::FormatError,
         },
         Mutating(LogEntry::Delete { .. }) | NonMutating(Exists { .. } | LLen { .. }) => {
             if let QueryIO::Err(value) = query_io {
-                return Response::Error(value);
+                return Response::Error(value.into());
             }
 
             let QueryIO::SimpleString(value) = query_io else {
@@ -79,7 +79,7 @@ fn render_return(kind: ClientAction, query_io: QueryIO) -> Response {
                 let s: Option<i64> = IndexedValueCodec::decode_value(s);
                 Response::Integer(s.unwrap().to_string().into())
             },
-            QueryIO::Err(value) => Response::Error(value),
+            QueryIO::Err(value) => Response::Error(value.into()),
 
             _ => Response::FormatError,
         },
@@ -92,13 +92,13 @@ fn render_return(kind: ClientAction, query_io: QueryIO) -> Response {
         Mutating(LogEntry::Set { .. } | LogEntry::LTrim { .. } | LogEntry::LSet { .. }) => {
             match query_io {
                 QueryIO::SimpleString(_) => Response::String("OK".into()),
-                QueryIO::Err(value) => Response::Error(value),
+                QueryIO::Err(value) => Response::Error(value.into()),
                 _ => Response::FormatError,
             }
         },
         NonMutating(ClusterMeet { .. } | ClusterReshard) => match query_io {
             QueryIO::Null => Response::String("OK".into()),
-            QueryIO::Err(value) => Response::Error(value),
+            QueryIO::Err(value) => Response::Error(value.into()),
             _ => Response::FormatError,
         },
         Mutating(LogEntry::Append { .. }) => match query_io {
@@ -107,7 +107,7 @@ fn render_return(kind: ClientAction, query_io: QueryIO) -> Response {
                 let s: Option<i64> = IndexedValueCodec::decode_value(s);
                 Response::String(s.unwrap().to_string().into())
             },
-            QueryIO::Err(value) => Response::Error(value),
+            QueryIO::Err(value) => Response::Error(value.into()),
             _ => Response::FormatError,
         },
         Mutating(LogEntry::LPop { .. } | LogEntry::RPop { .. })
@@ -137,11 +137,11 @@ fn render_return(kind: ClientAction, query_io: QueryIO) -> Response {
                     let QueryIO::BulkString(value) = item else {
                         return Response::FormatError;
                     };
-                    nodes.push(Response::String(value));
+                    nodes.push(Response::String(value.into()));
                 }
                 Response::Array(nodes)
             },
-            QueryIO::Err(value) => Response::Error(value),
+            QueryIO::Err(value) => Response::Error(value.into()),
             _ => Response::FormatError,
         },
 

duva/Cargo.toml

@@ -38,7 +38,3 @@ lzf = { version = "1.0.0" }
 [dev-dependencies]
 tempfile = "3.19.1"
 criterion = "0.7.0"
-
-[[bench]]
-name = "benchmark"
-harness = false

duva/benches/benchmark.rs

@@ -148,6 +148,8 @@ impl From<ErrorKind> for IoError {
 
 #[cfg(test)]
 pub mod test_tokio_stream_impl {
+    use crate::types::BinBytes;
+
     use super::*;
     #[derive(Debug, PartialEq, bincode::Encode, bincode::Decode)]
     struct TestMessage {
@@ -228,10 +230,12 @@ pub mod test_tokio_stream_impl {
         assert_eq!(parsed_values.len(), 2);
         assert_eq!(
             parsed_values[0],
-            QueryIO::SimpleString("FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0".into())
+            QueryIO::SimpleString(BinBytes::new(
+                "FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0"
+            ))
         );
 
-        assert_eq!(parsed_values[1], QueryIO::SimpleString("PEERS 127.0.0.1:6378".into()));
+        assert_eq!(parsed_values[1], QueryIO::SimpleString(BinBytes::new("PEERS 127.0.0.1:6378")));
     }
 
     #[tokio::test]

duva/benches/benchmark_query.rs

@@ -1,15 +1,9 @@
-use bincode::{
-    BorrowDecode, Decode,
-    de::Decoder,
-    enc::Encoder,
-    error::{DecodeError, EncodeError},
-};
 use bytes::Bytes;
 use chrono::{DateTime, Utc};
 
 use crate::{
     domains::caches::cache_objects::{THasExpiry, types::quicklist::QuickList},
-    from_to, make_smart_pointer,
+    types::BinBytes,
 };
 
 #[derive(Debug, PartialEq, Eq, Clone, Default, bincode::Encode, bincode::Decode)]
@@ -49,35 +43,10 @@ impl CacheValue {
 pub enum TypedValue {
     #[default]
     Null,
-    String(BytesWrapper),
+    String(BinBytes),
     List(QuickList),
 }
 
-#[derive(Debug, PartialEq, Eq, Clone, Default)]
-pub(crate) struct BytesWrapper(Bytes); // To minimize bincode implementation boilerplate as Bytes does not implement bincode traits
-make_smart_pointer!(BytesWrapper, Bytes);
-from_to!(Bytes, BytesWrapper);
-
-impl bincode::Encode for BytesWrapper {
-    fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
-        self.0.as_ref().encode(encoder)
-    }
-}
-impl<Ctx> Decode<Ctx> for BytesWrapper {
-    fn decode<D: Decoder>(decoder: &mut D) -> Result<Self, DecodeError> {
-        let vec: Vec<u8> = Decode::decode(decoder)?;
-        Ok(BytesWrapper(Bytes::from(vec)))
-    }
-}
-impl<'de, Ctx> BorrowDecode<'de, Ctx> for BytesWrapper {
-    fn borrow_decode<D: bincode::de::BorrowDecoder<'de>>(
-        decoder: &mut D,
-    ) -> Result<Self, DecodeError> {
-        let slice: &'de [u8] = BorrowDecode::borrow_decode(decoder)?;
-        Ok(BytesWrapper(Bytes::copy_from_slice(slice)))
-    }
-}
-
 pub const WRONG_TYPE_ERR_MSG: &str =
     "WRONGTYPE Operation against a key holding the wrong kind of value";
 

duva/src/adapters/io/tokio_stream.rs

@@ -1,7 +1,7 @@
 use crate::domains::replications::*;
+use crate::types::BinBytes;
 use crate::{domains::QueryIO, err, from_to, make_smart_pointer};
 use anyhow::Context;
-use bytes::Bytes;
 
 pub(crate) struct HandShakeRequest {
     pub(crate) command: HandShakeRequestEnum,
@@ -58,14 +58,14 @@ impl HandShakeRequest {
         }
     }
 
-    pub(crate) fn extract_capa(&self) -> anyhow::Result<Vec<(Bytes, Bytes)>> {
+    pub(crate) fn extract_capa(&self) -> anyhow::Result<Vec<(BinBytes, BinBytes)>> {
         self.match_query(HandShakeRequestEnum::ReplConf)?;
         if self.args.is_empty() || !self.args.len().is_multiple_of(2) {
             return Err(anyhow::anyhow!("Invalid number of arguments"));
         }
 
         // Process pairs directly using chunks_exact
-        let capabilities: Vec<(Bytes, Bytes)> = self
+        let capabilities: Vec<(BinBytes, BinBytes)> = self
             .args
             .chunks_exact(2)
             .filter_map(|chunk| match (&chunk[0], &chunk[1]) {
@@ -79,7 +79,7 @@ impl HandShakeRequest {
             .collect();
 
         // Validate last capability is psync2
-        if capabilities.last().context("No capabilities given")?.1 != "psync2" {
+        if *capabilities.last().context("No capabilities given")?.1 != "psync2" {
             return Err(anyhow::anyhow!("psync2 must be given as the last capability"));
         }
         Ok(capabilities)

duva/src/domains/caches/cache_objects/value.rs

@@ -12,7 +12,7 @@ use crate::domains::peers::identifier::PeerIdentifier;
 
 use crate::domains::peers::peer::Peer;
 use crate::domains::peers::service::PeerListener;
-use bytes::Bytes;
+use crate::types::BinBytes;
 
 // The following is used only when the node is in leader mode
 #[derive(Debug)]
@@ -46,7 +46,7 @@ impl InboundStream {
         let cmd = self.extract_cmd().await?;
         cmd.match_query(HandShakeRequestEnum::Ping)?;
 
-        self.w.write(QueryIO::SimpleString("PONG".into())).await?;
+        self.w.write(QueryIO::SimpleString(BinBytes::new("PONG"))).await?;
         Ok(())
     }
 
@@ -55,15 +55,15 @@ impl InboundStream {
 
         let port = cmd.extract_listening_port()?;
 
-        self.w.write(QueryIO::SimpleString("OK".into())).await?;
+        self.w.write(QueryIO::SimpleString(BinBytes::new("OK"))).await?;
 
         Ok(port)
     }
 
-    async fn recv_replconf_capa(&mut self) -> anyhow::Result<Vec<(Bytes, Bytes)>> {
+    async fn recv_replconf_capa(&mut self) -> anyhow::Result<Vec<(BinBytes, BinBytes)>> {
         let cmd = self.extract_cmd().await?;
         let capa_val_vec = cmd.extract_capa()?;
-        self.w.write(QueryIO::SimpleString("OK".into())).await?;
+        self.w.write(QueryIO::SimpleString(BinBytes::new("OK"))).await?;
         Ok(capa_val_vec)
     }
     async fn recv_psync(&mut self) -> anyhow::Result<(ReplicationId, u64, ReplicationRole, u64)> {
@@ -81,10 +81,9 @@ impl InboundStream {
         );
 
         self.w
-            .write(QueryIO::SimpleString(
-                format!("FULLRESYNC {id} {self_replid} {self_repl_offset} {self_role} {term}")
-                    .into(),
-            ))
+            .write(QueryIO::SimpleString(BinBytes::new(format!(
+                "FULLRESYNC {id} {self_replid} {self_repl_offset} {self_role} {term}"
+            ))))
             .await?;
         self.recv_ok().await?;
         Ok((inbound_repl_id, offset, role, term))