encoder: move encoder to a independent mod

components/br-stream/Cargo.toml

@@ -18,7 +18,7 @@ tokio = { version = "1.5", features = ["rt-multi-thread", "macros", "time", "syn
 prometheus = { version = "0.13", default-features = false, features = ["nightly"] }
 async-trait = { version = "0.1" }
 thiserror = "1"
-bytes = "0.4"
+bytes = "0.4.12"
 etcd-client = { version = "0.7", features = ["pub-response-field"] }
 kvproto = { git = "https://github.com/pingcap/kvproto.git", branch = "br-stream"}
 tokio-stream = "0.1"
@@ -43,4 +43,7 @@ engine_traits = { path = "../engine_traits", default-features = false }
 raft = { version = "0.6.0-alpha", default-features = false, features = ["protobuf-codec"] }
 raftstore = { path = "../raftstore", default-features = false }
 tikv = { path = "../../", default-features = false }
-online_config = { path = "../online_config" }
+online_config = { path = "../online_config" }
+
+[dev-dependencies]
+rand = "0.8.0"

components/br-stream/src/codec.rs

@@ -0,0 +1,55 @@
+// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
+use bytes::{Buf, Bytes};
+use std::io::prelude::*;
+use std::io::Cursor;
+
+pub struct Encoder;
+
+impl Encoder {
+    // TODO move this function to a indepentent module.
+    pub fn encode_event<'e>(key: &'e [u8], value: &'e [u8]) -> [impl AsRef<[u8]> + 'e; 4] {
+        let key_len = (key.len() as u32).to_le_bytes();
+        let val_len = (value.len() as u32).to_le_bytes();
+        [
+            Either::Left(key_len),
+            Either::Right(key),
+            Either::Left(val_len),
+            Either::Right(value),
+        ]
+    }
+
+    #[allow(dead_code)]
+    pub fn decode_event(e: &[u8]) -> (Vec<u8>, Vec<u8>) {
+        let mut buf = Cursor::new(Bytes::from(e));
+        let len = buf.get_u32_le() as usize;
+        let mut key = vec![0; len];
+        buf.read_exact(key.as_mut_slice()).unwrap();
+        let len = buf.get_u32_le() as usize;
+        let mut val = vec![0; len];
+        buf.read_exact(buffer.as_mut_slice()).unwrap();
+        (key, val)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rand::Rng;
+
+    #[test]
+    fn test_encode_decode() {
+        let mut rng = rand::thread_rng();
+        for _i in 0..10 {
+            let key: Vec<u8> = (0..100).map(|_| rng.gen_range(0..255)).collect();
+            let val: Vec<u8> = (0..100).map(|_| rng.gen_range(0..255)).collect();
+            let e = Encoder::encode_event(&key, &val);
+            let mut event = vec![];
+            for s in e {
+                event.push(s.into());
+            }
+            let (decoded_key, decoded_val) = Encoder::decode_event(&event);
+            assert_eq!(key, decoded_key);
+            assert_eq!(val, decoded_val);
+        }
+    }
+}

components/br-stream/src/endpoint.rs

@@ -138,17 +138,6 @@ where
             }
         }
     }
-    // TODO move this function to a indepentent module.
-    pub fn encode_event<'e>(key: &'e [u8], value: &'e [u8]) -> [impl AsRef<[u8]> + 'e; 4] {
-        let key_len = (key.len() as u32).to_le_bytes();
-        let val_len = (value.len() as u32).to_le_bytes();
-        [
-            Either::Left(key_len),
-            Either::Right(key),
-            Either::Left(val_len),
-            Either::Right(value),
-        ]
-    }
 
     fn backup_batch(&self, batch: CmdBatch) {
         let mut sw = StopWatch::new();

components/br-stream/src/lib.rs

@@ -1,6 +1,6 @@
 // Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
 #![feature(inherent_ascii_escape)]
-
+mod codec;
 pub mod config;
 mod endpoint;
 pub mod errors;