Added checksum::Value

value::checksum::Value is a struct that wraps Bytes with an optional
checksum. The idea is have fast comparisions when the two elements are
clearly distinct.

If the checksum are different there is no point in doing any other
comparision.

Cargo.toml

@@ -10,6 +10,7 @@ edition = "2018"
 redis-zero-protocol-parser = {path = "redis-protocol-parser"}
 tokio={version="1", features = ["full", "tracing"] }
 tokio-util={version="^0.6", features = ["full"] }
+crc32fast="^1.2"
 futures = { version = "0.3.0", features = ["thread-pool"]}
 tokio-stream="0.1"
 seahash = "4"

src/cmd/hash.rs

@@ -28,7 +28,7 @@ pub fn hdel(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
             }
             _ => Err(Error::WrongType),
         },
-        || Ok(0_i64.into()),
+        || Ok(0.into()),
     )
 }
 
@@ -37,13 +37,13 @@ pub fn hexists(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
         &args[1],
         |v| match v {
             Value::Hash(h) => Ok(if h.read().get(&args[2]).is_some() {
-                1_i64.into()
+                1.into()
             } else {
-                0_i64.into()
+                0.into()
             }),
             _ => Err(Error::WrongType),
         },
-        || Ok(0_i64.into()),
+        || Ok(0.into()),
     )
 }
 
@@ -141,7 +141,7 @@ pub fn hlen(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
             Value::Hash(h) => Ok((h.read().len() as i64).into()),
             _ => Err(Error::WrongType),
         },
-        || Ok(0_i64.into()),
+        || Ok(0.into()),
     )
 }
 
@@ -277,10 +277,10 @@ pub fn hsetnx(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
                 let mut h = h.write();
 
                 if h.get(&args[2]).is_some() {
-                    Ok(0_i64.into())
+                    Ok(0.into())
                 } else {
                     h.insert(args[2].clone(), args[3].clone());
-                    Ok(1_i64.into())
+                    Ok(1.into())
                 }
             }
             _ => Err(Error::WrongType),
@@ -305,11 +305,11 @@ pub fn hstrlen(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
             Value::Hash(h) => Ok(if let Some(v) = h.read().get(&args[2]) {
                 (v.len() as i64).into()
             } else {
-                0_i64.into()
+                0.into()
             }),
             _ => Err(Error::WrongType),
         },
-        || Ok(0_i64.into()),
+        || Ok(0.into()),
     )
 }
 

src/cmd/mod.rs

@@ -1,6 +1,7 @@
 pub mod client;
 pub mod hash;
 pub mod key;
+pub mod list;
 pub mod string;
 
 #[cfg(test)]

src/cmd/string.rs

@@ -66,7 +66,7 @@ pub fn strlen(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
     match conn.db().get(&args[1]) {
         Value::Blob(x) => Ok((x.len() as i64).into()),
         Value::String(x) => Ok((x.len() as i64).into()),
-        Value::Null => Ok(0_i64.into()),
+        Value::Null => Ok(0.into()),
         _ => Ok(Error::WrongType.into()),
     }
 }

src/db/mod.rs

@@ -90,12 +90,12 @@ impl Db {
         entries
             .get_mut(key)
             .filter(|x| x.is_valid())
-            .map_or(0_i64.into(), |x| {
+            .map_or(0.into(), |x| {
                 if x.has_ttl() {
                     x.persist();
-                    1_i64.into()
+                    1.into()
                 } else {
-                    0_i64.into()
+                    0.into()
                 }
             })
     }
@@ -107,15 +107,15 @@ impl Db {
         entries
             .get_mut(key)
             .filter(|x| x.is_valid())
-            .map_or(0_i64.into(), |x| {
+            .map_or(0.into(), |x| {
                 self.expirations.lock().unwrap().add(key, expires_at);
                 x.set_ttl(expires_at);
-                1_i64.into()
+                1.into()
             })
     }
 
     pub fn del(&self, keys: &[Bytes]) -> Value {
-        let mut deleted = 0_i64;
+        let mut deleted = 0;
         let mut expirations = self.expirations.lock().unwrap();
         keys.iter()
             .map(|key| {
@@ -133,7 +133,7 @@ impl Db {
     }
 
     pub fn exists(&self, keys: &[Bytes]) -> Value {
-        let mut matches = 0_i64;
+        let mut matches = 0;
         keys.iter()
             .map(|key| {
                 let entries = self.entries[self.get_slot(key)].read().unwrap();

src/dispatcher.rs

@@ -22,6 +22,18 @@ fn do_command(_conn: &Connection, _args: &[Bytes]) -> Result<Value, Error> {
 }
 
 dispatcher! {
+    list {
+        lrange {
+            cmd::list::lrange,
+            [""],
+            4,
+        },
+        rpush {
+            cmd::list::rpush,
+            [""],
+            -3,
+        },
+    },
     hash {
         hdel {
             cmd::hash::hdel,

src/value/checksum.rs

@@ -0,0 +1,106 @@
+use crate::value;
+use bytes::Bytes;
+use crc32fast::Hasher;
+
+#[derive(Debug, Clone)]
+pub struct Value {
+    bytes: Bytes,
+    checksum: Option<u32>,
+}
+
+impl Value {
+    pub fn new(bytes: Bytes) -> Self {
+        let checksum = Self::calculate_checksum(&bytes);
+        Self { bytes, checksum }
+    }
+
+    pub fn clone_value(&self) -> value::Value {
+        value::Value::Blob(self.bytes.clone())
+    }
+
+    pub fn has_checksum(&self) -> bool {
+        self.checksum.is_some()
+    }
+
+    fn calculate_checksum(bytes: &Bytes) -> Option<u32> {
+        if bytes.len() < 1024 {
+            None
+        } else {
+            let mut hasher = Hasher::new();
+            hasher.update(bytes);
+            Some(hasher.finalize())
+        }
+    }
+}
+
+impl PartialEq for Value {
+    fn eq(&self, other: &Value) -> bool {
+        if self.checksum == other.checksum && self.bytes.len() == other.bytes.len() {
+            // The data have the same checksum now perform a more extensive
+            // comparision
+            return self.bytes.eq(&other.bytes);
+        }
+
+        false
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::bytes;
+
+    #[test]
+    fn does_not_have_checksum() {
+        let data = Value::new(bytes!(b"one"));
+        assert!(!data.has_checksum())
+    }
+
+    #[test]
+    fn has_checksum() {
+        let data = Value::new(bytes!(
+            b"
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+        "
+        ));
+        assert!(data.has_checksum())
+    }
+
+    #[test]
+    fn compare() {
+        let data1 = Value::new(bytes!(
+            b"
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+            one one one one one one one one one one one one one one one one one one
+        "
+        ));
+        assert!(data1.clone() == data1.clone());
+
+        let data2 = Value::new(bytes!(b"one"));
+        assert!(data2 == data2.clone());
+        assert!(data1 != data2);
+    }
+}

src/value/locked.rs

@@ -0,0 +1,55 @@
+use std::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
+
+#[derive(Debug)]
+pub struct Value<T: Clone + PartialEq>(pub RwLock<T>);
+
+impl<T: Clone + PartialEq> Clone for Value<T> {
+    fn clone(&self) -> Self {
+        Self(RwLock::new(self.0.read().unwrap().clone()))
+    }
+}
+
+impl<T: PartialEq + Clone> PartialEq for Value<T> {
+    fn eq(&self, other: &Value<T>) -> bool {
+        self.0.read().unwrap().eq(&other.0.read().unwrap())
+    }
+}
+
+impl<T: PartialEq + Clone> Value<T> {
+    pub fn new(obj: T) -> Self {
+        Self(RwLock::new(obj))
+    }
+
+    pub fn write(&self) -> RwLockWriteGuard<'_, T> {
+        self.0.write().unwrap()
+    }
+
+    pub fn read(&self) -> RwLockReadGuard<'_, T> {
+        self.0.read().unwrap()
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn locked_eq1() {
+        let a = Value::new(1);
+        let b = Value::new(1);
+        assert!(a == b);
+    }
+
+    #[test]
+    fn locked_eq2() {
+        let a = Value::new(1);
+        let b = Value::new(2);
+        assert!(a != b);
+    }
+
+    #[test]
+    fn locked_clone() {
+        let a = Value::new((1, 2, 3));
+        assert!(a == a.clone());
+    }
+}

src/value.rs → src/value/mod.rs

@@ -1,45 +1,25 @@
+pub mod checksum;
+pub mod locked;
+
 use crate::{error::Error, value_try_from, value_vec_try_from};
 use bytes::{Bytes, BytesMut};
 use redis_zero_protocol_parser::Value as ParsedValue;
 use std::{
-    collections::HashMap,
+    collections::{HashMap, LinkedList},
     convert::{TryFrom, TryInto},
     str::FromStr,
-    sync::{RwLock, RwLockReadGuard, RwLockWriteGuard},
 };
 
-#[derive(Debug)]
-pub struct LockedValue<T: Clone + PartialEq>(pub RwLock<T>);
-
-impl<T: Clone + PartialEq> Clone for LockedValue<T> {
-    fn clone(&self) -> Self {
-        Self(RwLock::new(self.0.read().unwrap().clone()))
-    }
-}
-
-impl<T: PartialEq + Clone> PartialEq for LockedValue<T> {
-    fn eq(&self, other: &LockedValue<T>) -> bool {
-        self.0.read().unwrap().eq(&other.0.read().unwrap())
-    }
-}
-
-impl<T: PartialEq + Clone> LockedValue<T> {
-    pub fn new(obj: T) -> Self {
-        Self(RwLock::new(obj))
-    }
-
-    pub fn write(&self) -> RwLockWriteGuard<'_, T> {
-        self.0.write().unwrap()
-    }
-
-    pub fn read(&self) -> RwLockReadGuard<'_, T> {
-        self.0.read().unwrap()
-    }
-}
+#[derive(Debug, PartialEq, Clone)]
+/// Stores binary data as Bytes but has built-in verification to make
+/// comparisions faster. The verification (CRC32) is a quicker process to
+/// compare two objects, which will return MAYBE or NOT.
+pub struct BytesWithChecksum((Bytes, u32));
 
 #[derive(Debug, PartialEq, Clone)]
 pub enum Value {
-    Hash(LockedValue<HashMap<Bytes, Bytes>>),
+    Hash(locked::Value<HashMap<Bytes, Bytes>>),
+    List(locked::Value<LinkedList<checksum::Value>>),
     Array(Vec<Value>),
     Blob(Bytes),
     String(String),
@@ -138,6 +118,7 @@ impl<'a> From<&ParsedValue<'a>> for Value {
 }
 
 value_try_from!(f64, Value::Float);
+value_try_from!(i32, Value::Integer);
 value_try_from!(i64, Value::Integer);
 value_try_from!(i128, Value::BigInteger);
 
@@ -149,7 +130,13 @@ impl From<&str> for Value {
 
 impl From<HashMap<Bytes, Bytes>> for Value {
     fn from(value: HashMap<Bytes, Bytes>) -> Value {
-        Value::Hash(LockedValue::new(value))
+        Value::Hash(locked::Value::new(value))
+    }
+}
+
+impl From<LinkedList<checksum::Value>> for Value {
+    fn from(value: LinkedList<checksum::Value>) -> Value {
+        Value::List(locked::Value::new(value))
     }
 }