change slabstore struct to use rocks api in rocks.rs file

src/bin/darkfid.rs

@@ -4,7 +4,7 @@ use easy_parallel::Parallel;
 use std::net::SocketAddr;
 
 use drk::service::{ClientProgramOptions, GatewayClient};
-use drk::{slab::Slab, Result};
+use drk::{slab::Slab, Result, rocks::Rocks};
 
 fn setup_addr(address: Option<SocketAddr>, default: SocketAddr) -> SocketAddr {
     match address {
@@ -16,10 +16,12 @@ fn setup_addr(address: Option<SocketAddr>, default: SocketAddr) -> SocketAddr {
 async fn start(executor: Arc<Executor<'_>>, options: ClientProgramOptions) -> Result<()> {
     let connect_addr: SocketAddr = setup_addr(options.connect_addr, "127.0.0.1:3333".parse()?);
     let sub_addr: SocketAddr = setup_addr(options.sub_addr, "127.0.0.1:4444".parse()?);
-    let slabstore_path = options.slabstore_path.as_path();
+    let database_path = options.database_path.as_path();
+
+    let rocks = Rocks::new(database_path)?;
 
     // create gateway client
-    let mut client = GatewayClient::new(connect_addr, slabstore_path)?;
+    let mut client = GatewayClient::new(connect_addr, rocks)?;
 
     // start gateway client
     client.start().await?;
@@ -92,6 +94,7 @@ mod test {
 
         use drk::service::GatewayClient;
         use drk::slab::Slab;
+        use drk::rocks::Rocks;
 
         use log::*;
         use rand::Rng;
@@ -117,10 +120,17 @@ mod test {
                     let mut rng = rand::thread_rng();
                     let rnd: u32 = rng.gen();
 
+
+
+
+                    let path_str = format!("database_{}.db", rnd);
+                    let database_path = Path::new(path_str.as_str());
+                    let rocks = Rocks::new(database_path.clone()).unwrap();
+
                     // create new client and use different slabstore
                     let mut client = GatewayClient::new(
                         "127.0.0.1:3333".parse().unwrap(),
-                        Path::new(&format!("slabstore_{}.db", rnd)),
+                        rocks,
                     )
                     .unwrap();
 
@@ -130,6 +140,7 @@ mod test {
                     // sending slab
                     let _slab = Slab::new("testcoin".to_string(), rnd.to_le_bytes().to_vec());
                     client.put_slab(_slab).await.unwrap();
+                    
                 })
             });
             thread_pools.push(thread);

src/bin/gatewayd.rs

@@ -5,8 +5,7 @@ extern crate clap;
 use async_executor::Executor;
 use easy_parallel::Parallel;
 
-use drk::Result;
-
+use drk::{Result, rocks::Rocks};
 use drk::service::{GatewayService, ProgramOptions};
 
 fn setup_addr(address: Option<SocketAddr>, default: SocketAddr) -> SocketAddr {
@@ -19,9 +18,11 @@ fn setup_addr(address: Option<SocketAddr>, default: SocketAddr) -> SocketAddr {
 async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<()> {
     let accept_addr: SocketAddr = setup_addr(options.accept_addr, "127.0.0.1:3333".parse()?);
     let pub_addr: SocketAddr = setup_addr(options.pub_addr, "127.0.0.1:4444".parse()?);
-    let slabstore_path = options.slabstore_path.as_path();
+    let database_path = options.database_path.as_path();
+
+    let rocks = Rocks::new(database_path)?;
 
-    let gateway = GatewayService::new(accept_addr, pub_addr, slabstore_path)?;
+    let gateway = GatewayService::new(accept_addr, pub_addr, rocks)?;
 
     gateway.start(executor.clone()).await?;
     Ok(())

src/service/gateway.rs

@@ -1,11 +1,10 @@
 use async_std::sync::Arc;
 use std::convert::From;
 use std::net::SocketAddr;
-use std::path::Path;
 
 use super::reqrep::{PeerId, Publisher, RepProtocol, Reply, ReqProtocol, Request, Subscriber};
 use crate::{
-    serial::deserialize, serial::serialize, slab::Slab, slabstore::SlabStore, Error, Result,
+    serial::deserialize, serial::serialize, slab::Slab, slabstore::SlabStore, Error, Result, rocks::Rocks
 };
 
 use async_executor::Executor;
@@ -37,9 +36,9 @@ impl GatewayService {
     pub fn new(
         addr: SocketAddr,
         pub_addr: SocketAddr,
-        slabstore_path: &Path,
+        rocks: Rocks,
     ) -> Result<Arc<GatewayService>> {
-        let slabstore = SlabStore::new(slabstore_path)?;
+        let slabstore = SlabStore::new(rocks)?;
 
         Ok(Arc::new(GatewayService {
             slabstore,
@@ -185,10 +184,10 @@ pub struct GatewayClient {
 }
 
 impl GatewayClient {
-    pub fn new(addr: SocketAddr, path: &Path) -> Result<Self> {
+    pub fn new(addr: SocketAddr, rocks: Rocks) -> Result<Self> {
         let protocol = ReqProtocol::new(addr, String::from("GATEWAY CLIENT"));
 
-        let slabstore = SlabStore::new(path)?;
+        let slabstore = SlabStore::new(rocks)?;
 
         Ok(GatewayClient {
             protocol,

src/slabstore.rs

@@ -1,27 +1,21 @@
-use std::path::Path;
 use std::sync::Arc;
 
+use crate::rocks::{columns, IteratorMode, Rocks};
 use crate::serial::{deserialize, serialize};
 use crate::{slab::Slab, Result};
 
-use rocksdb::{IteratorMode, Options, DB};
-
 pub struct SlabStore {
-    db: DB,
+    rocks: Rocks,
 }
 
 impl SlabStore {
-    pub fn new(path: &Path) -> Result<Arc<Self>> {
-        let mut opt = Options::default();
-        opt.create_if_missing(true);
-
-        let db = DB::open(&opt, path)?;
-
-        Ok(Arc::new(SlabStore { db }))
+    pub fn new(rocks: Rocks) -> Result<Arc<Self>> {
+        Ok(Arc::new(SlabStore { rocks }))
     }
 
     pub fn get(&self, key: Vec<u8>) -> Result<Option<Vec<u8>>> {
-        let value = self.db.get(key)?;
+        let cf = self.rocks.cf_handle::<columns::Slabs>()?;
+        let value = self.rocks.get_cf(cf, key)?;
         Ok(value)
     }
 
@@ -29,9 +23,13 @@ impl SlabStore {
         let slab: Slab = deserialize(&value)?;
         let last_index = self.get_last_index()?;
         let key = last_index + 1;
+
         if slab.get_index() == key {
             let key = serialize(&key);
-            self.db.put(key.clone(), value)?;
+
+            let cf = self.rocks.cf_handle::<columns::Slabs>()?;
+            self.rocks.put_cf(cf, key.clone(), value)?;
+
             Ok(Some(key))
         } else {
             Ok(None)
@@ -39,7 +37,7 @@ impl SlabStore {
     }
 
     pub fn get_value_deserialized(&self, key: Vec<u8>) -> Result<Option<Slab>> {
-        let value = self.db.get(key)?;
+        let value = self.get(key)?;
         match value {
             Some(v) => {
                 let v: Slab = deserialize(&v)?;
@@ -50,7 +48,8 @@ impl SlabStore {
     }
 
     pub fn get_last_index(&self) -> Result<u64> {
-        let last_index = self.db.iterator(IteratorMode::End).next();
+        let cf = self.rocks.cf_handle::<columns::Slabs>()?;
+        let last_index = self.rocks.iterator(cf, IteratorMode::End).next();
         match last_index {
             Some((index, _)) => Ok(deserialize(&index)?),
             None => Ok(0),
@@ -58,15 +57,11 @@ impl SlabStore {
     }
 
     pub fn get_last_index_as_bytes(&self) -> Result<Vec<u8>> {
-        let last_index = self.db.iterator(IteratorMode::End).next();
+        let cf = self.rocks.cf_handle::<columns::Slabs>()?;
+        let last_index = self.rocks.iterator(cf, IteratorMode::End).next();
         match last_index {
             Some((index, _)) => Ok(index.to_vec()),
             None => Ok(serialize::<u64>(&0)),
         }
     }
-
-    pub fn destroy(path: &Path) -> Result<()> {
-        DB::destroy(&Options::default(), path)?;
-        Ok(())
-    }
 }