Merge pull request #33 from KAIYOHUGO/master

Add Compressor

Cargo.toml

@@ -36,5 +36,11 @@ harness = false
 path = "src/benches/maps_bench.rs"
 
 [workspace]
-members = ["examples/write-a-lot", "examples/leveldb-tool", "examples/word-analyze",  "examples/stresstest", "examples/asyncdb"]
-
+members = [
+    "examples/write-a-lot",
+    "examples/leveldb-tool",
+    "examples/word-analyze",
+    "examples/stresstest",
+    "examples/asyncdb",
+    "examples/mcpe",
+]

examples/leveldb-tool/src/main.rs

@@ -1,6 +1,6 @@
 extern crate rusty_leveldb;
 
-use rusty_leveldb::{LdbIterator, Options, DB};
+use rusty_leveldb::{compressor, CompressorId, LdbIterator, Options, DB};
 
 use std::env::args;
 use std::io::{self, Write};
@@ -59,7 +59,7 @@ fn main() {
     let mut opt = Options::default();
     opt.reuse_logs = false;
     opt.reuse_manifest = false;
-    opt.compression_type = rusty_leveldb::CompressionType::CompressionSnappy;
+    opt.compressor = compressor::SnappyCompressor::ID;
     let mut db = DB::open("tooldb", opt).unwrap();
 
     match args[1].as_str() {

examples/mcpe/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "mcpe"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+miniz_oxide = "0.7.1"
+rusty-leveldb = { path = "../../" }

examples/mcpe/README.md

@@ -0,0 +1,5 @@
+# MCPE
+
+This example show how to customize compression method.
+
+This setup is compatible to [Mojang's leveldb](https://github.com/Mojang/leveldb-mcpe).

examples/mcpe/src/main.rs

@@ -0,0 +1,70 @@
+use miniz_oxide::deflate::{compress_to_vec, compress_to_vec_zlib};
+use miniz_oxide::inflate::{decompress_to_vec, decompress_to_vec_zlib};
+use rusty_leveldb::{Compressor, CompressorList, Options, DB};
+use std::rc::Rc;
+
+struct ZlibCompressor(u8);
+
+impl ZlibCompressor {
+    /// level 0-10
+    pub fn new(level: u8) -> Self {
+        assert!(level <= 10);
+        Self(level)
+    }
+}
+
+impl Compressor for ZlibCompressor {
+    fn encode(&self, block: Vec<u8>) -> rusty_leveldb::Result<Vec<u8>> {
+        Ok(compress_to_vec_zlib(&block, self.0))
+    }
+
+    fn decode(&self, block: Vec<u8>) -> rusty_leveldb::Result<Vec<u8>> {
+        decompress_to_vec_zlib(&block).map_err(|e| rusty_leveldb::Status {
+            code: rusty_leveldb::StatusCode::CompressionError,
+            err: e.to_string(),
+        })
+    }
+}
+
+struct RawZlibCompressor(u8);
+
+impl RawZlibCompressor {
+    /// level 0-10
+    pub fn new(level: u8) -> Self {
+        assert!(level <= 10);
+        Self(level)
+    }
+}
+
+impl Compressor for RawZlibCompressor {
+    fn encode(&self, block: Vec<u8>) -> rusty_leveldb::Result<Vec<u8>> {
+        Ok(compress_to_vec(&block, self.0))
+    }
+
+    fn decode(&self, block: Vec<u8>) -> rusty_leveldb::Result<Vec<u8>> {
+        decompress_to_vec(&block).map_err(|e| rusty_leveldb::Status {
+            code: rusty_leveldb::StatusCode::CompressionError,
+            err: e.to_string(),
+        })
+    }
+}
+
+pub fn mcpe_options(compression_level: u8) -> Options {
+    let mut opt = Options::default();
+    opt.compressor = 0;
+    let mut list = CompressorList::new();
+    list.set_with_id(0, RawZlibCompressor::new(compression_level));
+    list.set_with_id(1, ZlibCompressor::new(compression_level));
+    opt.compressor_list = Rc::new(list);
+    opt
+}
+
+fn main() {
+    let path = "mcpe_db";
+    let compression_level = 10;
+    let opt = mcpe_options(compression_level);
+    let mut db = DB::open(path, opt).unwrap();
+    db.put(b"~local_player", b"NBT data goes here").unwrap();
+    let value = db.get(b"~local_player").unwrap();
+    assert_eq!(&value, b"NBT data goes here")
+}

examples/stresstest/src/main.rs

@@ -1,5 +1,5 @@
 use rand::distributions::{Alphanumeric, DistString};
-use rusty_leveldb::{Options, DB};
+use rusty_leveldb::{compressor, CompressorId, Options, DB};
 
 const KEY_LEN: usize = 4;
 const VAL_LEN: usize = 8;
@@ -12,7 +12,7 @@ fn gen_string(n: usize) -> String {
 
 fn write(db: &mut DB, n: usize) {
     time_test::time_test!("write");
-    for i in 0..n {
+    for _ in 0..n {
         let (k, v) = (gen_string(KEY_LEN), gen_string(VAL_LEN));
 
         db.put(k.as_bytes(), v.as_bytes()).unwrap();
@@ -27,31 +27,31 @@ fn write(db: &mut DB, n: usize) {
 fn read(db: &mut DB, n: usize) -> usize {
     let mut succ = 0;
     time_test::time_test!("read");
-    for i in 0..n {
+    for _ in 0..n {
         let k = gen_string(KEY_LEN);
 
-        if let Some(v) = db.get(k.as_bytes()) {
+        if let Some(_) = db.get(k.as_bytes()) {
             succ += 1;
         }
     }
     succ
 }
 
 fn main() {
-    let N = 100_000;
+    let n = 100_000;
     let m = 10;
     let path = "stresstestdb";
     let mut entries = 0;
 
     for i in 0..m {
         let mut opt = Options::default();
-        opt.compression_type = rusty_leveldb::CompressionType::CompressionSnappy;
+        opt.compressor = compressor::SnappyCompressor::ID;
         let mut db = DB::open(path, opt).unwrap();
-        write(&mut db, N);
-        entries += N;
+        write(&mut db, n);
+        entries += n;
         println!("Wrote {} entries ({}/{})", entries, i + 1, m);
 
-        let s = read(&mut db, N);
-        println!("Read back {} entries (found {}) ({}/{})", N, s, i + 1, m);
+        let s = read(&mut db, n);
+        println!("Read back {} entries (found {}) ({}/{})", n, s, i + 1, m);
     }
 }

examples/word-analyze/src/main.rs

@@ -1,3 +1,4 @@
+use leveldb::CompressorId;
 use rusty_leveldb as leveldb;
 
 use std::fs::OpenOptions;
@@ -35,7 +36,7 @@ fn run(mut db: leveldb::DB) -> io::Result<()> {
 
 fn main() {
     let mut opts = leveldb::Options::default();
-    opts.compression_type = leveldb::CompressionType::CompressionNone;
+    opts.compressor = leveldb::compressor::NoneCompressor::ID;
     let db = leveldb::DB::open("wordsdb", opts).unwrap();
 
     run(db).unwrap();

examples/write-a-lot/src/main.rs

@@ -2,9 +2,7 @@ extern crate rand;
 extern crate rusty_leveldb;
 
 use rand::Rng;
-use rusty_leveldb::CompressionType;
-use rusty_leveldb::Options;
-use rusty_leveldb::DB;
+use rusty_leveldb::{compressor, CompressorId, Options, DB};
 
 use std::error::Error;
 use std::iter::FromIterator;
@@ -31,7 +29,7 @@ fn fill_db(db: &mut DB, entries: usize) -> Result<(), Box<dyn Error>> {
 
 fn main() {
     let mut opt = Options::default();
-    opt.compression_type = CompressionType::CompressionSnappy;
+    opt.compressor = compressor::SnappyCompressor::ID;
     let mut db = DB::open("test1", opt).unwrap();
 
     fill_db(&mut db, 32768).unwrap();

src/compressor.rs

@@ -0,0 +1,77 @@
+/// Custom compression method
+///
+/// ```
+/// # use rusty_leveldb::{Compressor, CompressorId};
+///
+/// #[derive(Debug, Clone, Copy, Default)]
+/// pub struct CustomCompressor;
+///
+/// impl CompressorId for CustomCompressor {
+///     // a unique id to identify what compressor should DB use
+///     const ID: u8 = 42;
+/// }
+///
+/// impl Compressor for CustomCompressor {
+///     fn encode(&self, block: Vec<u8>) -> rusty_leveldb::Result<Vec<u8>> {
+///         // Do something
+///         Ok(block)
+///     }
+///
+///     fn decode(&self, block: Vec<u8>) -> rusty_leveldb::Result<Vec<u8>> {
+///         // Do something
+///         Ok(block)
+///     }
+/// }
+/// ```
+///
+/// See [crate::CompressorList] for usage
+pub trait Compressor {
+    fn encode(&self, block: Vec<u8>) -> crate::Result<Vec<u8>>;
+
+    fn decode(&self, block: Vec<u8>) -> crate::Result<Vec<u8>>;
+}
+
+/// Set default compressor id
+pub trait CompressorId {
+    const ID: u8;
+}
+
+/// A compressor that do **Nothing**
+///
+/// It default id is `0`
+#[derive(Debug, Clone, Copy, Default)]
+pub struct NoneCompressor;
+
+impl CompressorId for NoneCompressor {
+    const ID: u8 = 0;
+}
+
+impl Compressor for NoneCompressor {
+    fn encode(&self, block: Vec<u8>) -> crate::Result<Vec<u8>> {
+        Ok(block)
+    }
+
+    fn decode(&self, block: Vec<u8>) -> crate::Result<Vec<u8>> {
+        Ok(block)
+    }
+}
+
+/// A compressor that compress data with Google's Snappy
+///
+/// It default id is `1`
+#[derive(Debug, Clone, Copy, Default)]
+pub struct SnappyCompressor;
+
+impl CompressorId for SnappyCompressor {
+    const ID: u8 = 1;
+}
+
+impl Compressor for SnappyCompressor {
+    fn encode(&self, block: Vec<u8>) -> crate::Result<Vec<u8>> {
+        Ok(snap::raw::Encoder::new().compress_vec(&block)?)
+    }
+
+    fn decode(&self, block: Vec<u8>) -> crate::Result<Vec<u8>> {
+        Ok(snap::raw::Decoder::new().decompress_vec(&block)?)
+    }
+}

src/db_impl.rs

@@ -890,10 +890,7 @@ impl DB {
         Ok(())
     }
 
-    fn finish_compaction_output(
-        &mut self,
-        cs: &mut CompactionState,
-    ) -> Result<()> {
+    fn finish_compaction_output(&mut self, cs: &mut CompactionState) -> Result<()> {
         assert!(cs.builder.is_some());
         let output_num = cs.current_output().num;
         assert!(output_num > 0);

src/lib.rs

@@ -73,18 +73,21 @@ mod write_batch;
 mod db_impl;
 mod db_iter;
 
+pub mod compressor;
+
 #[cfg(feature = "async")]
 pub use asyncdb::AsyncDB;
 
 pub use cmp::{Cmp, DefaultCmp};
+pub use compressor::{Compressor, CompressorId};
 pub use db_impl::DB;
 pub use db_iter::DBIterator;
 pub use disk_env::PosixDiskEnv;
 pub use env::Env;
 pub use error::{Result, Status, StatusCode};
 pub use filter::{BloomPolicy, FilterPolicy};
 pub use mem_env::MemEnv;
-pub use options::{in_memory, CompressionType, Options};
+pub use options::{in_memory, CompressorList, Options};
 pub use skipmap::SkipMap;
 pub use types::LdbIterator;
 pub use write_batch::WriteBatch;