Merge branch 'master' of github.com:darkrenaissance/darkfi

Cargo.toml

@@ -48,6 +48,7 @@ log = "0.4"
 ctrlc = "3.1.7"
 serde_json = "1.0.61"
 owning_ref = "0.4.1"
+signal-hook = "0.3.8"
 
 smol = "1.2.4"
 futures = "0.3.5"
@@ -85,7 +86,8 @@ bytes = "1.0.1"
 
 # wallet deps
 rocksdb = "0.16.0"
-dirs = "2.0.2"
+dirs = "3.0.2"
+
 [dependencies.rusqlite]
 version = "0.25.1"
 features = ["bundled", "sqlcipher"]
@@ -131,8 +133,8 @@ name = "gatewayd"
 path = "src/bin/gatewayd.rs"
 
 [[bin]]
-name = "demowallet"
-path = "src/bin/demowallet.rs"
+name = "darkfid"
+path = "src/bin/darkfid.rs"
 
 [profile.release]
 debug = 1

src/wallet/schema.sql → res/schema.sql

@@ -6,4 +6,11 @@ CREATE TABLE IF NOT EXISTS keys(
     key_private BLOB NOT NULL
 );
 CREATE INDEX IF NOT EXISTS key_public on keys(key_public);
-
+CREATE TABLE IF NOT EXISTS coins(
+    coin BLOB NOT NULL,
+    witness BLOB NOT NULL,
+    serial BLOB NOT NULL,
+    value INT NOT NULL,
+    coin_blind BLOB NOT NULL,
+    valcom_blind BLOB NOT NULL
+);

darkcli.py → scripts/drk

@@ -1,17 +1,18 @@
-# TODO: refactor into async
+#!/usr/bin/env python
 
 import argparse
 import requests
 import json
 
 def arg_parser(client):
-    parser = argparse.ArgumentParser(prog='dark',
+    parser = argparse.ArgumentParser(prog='drk',
                                           usage='%(prog)s [commands]',
                                           description="""DarkFi wallet
                                           command-line tool""")
     parser.add_argument("-k", "--key", action='store_true', help="Generate a new keypair")
     parser.add_argument("-i", "--info", action='store_true', help="Request info from daemon")
     parser.add_argument("-s", "--stop", action='store_true', help="Send a stop signal to the daemon")
+    parser.add_argument("-n", "--new", action='store_true', help="Generate a new wallet")
     parser.add_argument("-hi", "--hello", action='store_true', help="Say hello")
     args = parser.parse_args()
 
@@ -22,6 +23,13 @@ def arg_parser(client):
         except Exception:
             raise
 
+    if args.new:
+        try:
+            print("Attemping to generate a new wallet...")
+            client.new_wallet(client.payload)
+        except Exception:
+            raise
+
     if args.info:
         try:
             print("Info was entered")
@@ -46,6 +54,7 @@ def arg_parser(client):
             raise
 
 
+# TODO: refactor into async
 class DarkClient:
     # TODO: generate random ID (4 byte unsigned int) (rand range 0 - max size
     # uint32
@@ -85,6 +94,13 @@ class DarkClient:
         payload['id'] = "0"
         hello = self.__request(payload)
         print(hello)
+    
+    def new_wallet(self, payload):
+        payload['method'] = "new_wallet"
+        payload['jsonrpc'] = "2.0"
+        payload['id'] = "0"
+        wallet = self.__request(payload)
+        print(wallet)
 
     def __request(self, payload):
         response = requests.post(self.url, json=payload).json()

src/bin/darkfid.rs

@@ -0,0 +1,142 @@
+use async_executor::Executor;
+use async_std::sync::Arc;
+use easy_parallel::Parallel;
+use std::net::SocketAddr;
+
+use drk::service::{ClientProgramOptions, GatewayClient};
+use drk::{slab::Slab, Result};
+
+fn setup_addr(address: Option<SocketAddr>, default: SocketAddr) -> SocketAddr {
+    match address {
+        Some(addr) => addr,
+        None => default,
+    }
+}
+
+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();
+
+    // create gateway client
+    let mut client = GatewayClient::new(connect_addr, slabstore_path)?;
+
+    // start gateway client
+    client.start().await?;
+
+    // start subscribe to gateway publisher
+
+    let subscriber = GatewayClient::start_subscriber(sub_addr).await?;
+    let slabstore = client.get_slabstore();
+    let subscribe_task = executor.spawn(GatewayClient::subscribe(subscriber, slabstore));
+
+    // TEST
+    let _slab = Slab::new("testcoin".to_string(), vec![0, 0, 0, 0]);
+    //client.put_slab(_slab).await?;
+
+    subscribe_task.cancel().await;
+    Ok(())
+}
+
+fn main() -> Result<()> {
+    use simplelog::*;
+
+    let ex = Arc::new(Executor::new());
+    let (signal, shutdown) = async_channel::unbounded::<()>();
+
+    let options = ClientProgramOptions::load()?;
+
+    let logger_config = ConfigBuilder::new().set_time_format_str("%T%.6f").build();
+
+    let debug_level = if options.verbose {
+        LevelFilter::Debug
+    } else {
+        LevelFilter::Off
+    };
+
+    CombinedLogger::init(vec![
+        TermLogger::new(debug_level, logger_config, TerminalMode::Mixed).unwrap(),
+        WriteLogger::new(
+            LevelFilter::Debug,
+            Config::default(),
+            std::fs::File::create(options.log_path.as_path()).unwrap(),
+        ),
+    ])
+        .unwrap();
+
+    let ex2 = ex.clone();
+
+    let (_, result) = Parallel::new()
+        // Run four executor threads.
+        .each(0..3, |_| smol::future::block_on(ex.run(shutdown.recv())))
+        // Run the main future on the current thread.
+        .finish(|| {
+            smol::future::block_on(async move {
+                start(ex2, options).await?;
+                drop(signal);
+                Ok::<(), drk::Error>(())
+            })
+        });
+
+    result
+}
+
+// $ cargo test --bin darkfid
+// run 10 clients simultaneously
+#[cfg(test)]
+mod test {
+
+    #[test]
+    fn test_darkfid_client() {
+        use std::path::Path;
+
+        use drk::service::GatewayClient;
+        use drk::slab::Slab;
+
+        use log::*;
+        use rand::Rng;
+        use simplelog::*;
+
+        let logger_config = ConfigBuilder::new().set_time_format_str("%T%.6f").build();
+
+        CombinedLogger::init(vec![
+            TermLogger::new(LevelFilter::Debug, logger_config, TerminalMode::Mixed).unwrap(),
+            WriteLogger::new(
+                LevelFilter::Debug,
+                Config::default(),
+                std::fs::File::create(Path::new("/tmp/dar.log")).unwrap(),
+            ),
+        ])
+            .unwrap();
+
+        let mut thread_pools: Vec<std::thread::JoinHandle<()>> = vec![];
+
+        for _ in 0..10 {
+            let thread = std::thread::spawn(|| {
+                smol::future::block_on(async move {
+                    let mut rng = rand::thread_rng();
+                    let rnd: u32 = rng.gen();
+
+                    // 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)),
+                    )
+                        .unwrap();
+
+                    // start client
+                    client.start().await.unwrap();
+
+                    // 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);
+        }
+        for t in thread_pools {
+            t.join().unwrap();
+        }
+    }
+}

src/bin/demowallet.rs

@@ -1,49 +0,0 @@
-use async_executor::Executor;
-use async_std::sync::{Arc, Mutex};
-use easy_parallel::Parallel;
-
-use drk::service::{fetch_slabs_loop, GatewayClient};
-use drk::Result;
-
-async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
-    let mut client = GatewayClient::new("127.0.0.1:3333".parse()?);
-
-    client.start().await?;
-    println!("connected to a server");
-
-    let slabs = Arc::new(Mutex::new(vec![]));
-
-    let subscriber = client.subscribe("127.0.0.1:4444".parse()?).await?;
-
-    println!("subscription ready");
-
-    let fetch_loop_task = executor.spawn(fetch_slabs_loop(subscriber.clone(), slabs.clone()));
-
-    client.put_slab(vec![0, 0, 0, 0]).await?;
-    client.put_slab(vec![0, 0, 0, 0]).await?;
-    client.put_slab(vec![0, 0, 0, 0]).await?;
-
-    fetch_loop_task.cancel().await;
-
-    Ok(())
-}
-
-fn main() -> Result<()> {
-    let ex = Arc::new(Executor::new());
-    let (signal, shutdown) = async_channel::unbounded::<()>();
-    let ex2 = ex.clone();
-
-    let (_, result) = Parallel::new()
-        // Run four executor threads.
-        .each(0..3, |_| smol::future::block_on(ex.run(shutdown.recv())))
-        // Run the main future on the current thread.
-        .finish(|| {
-            smol::future::block_on(async move {
-                start(ex2).await?;
-                drop(signal);
-                Ok::<(), drk::Error>(())
-            })
-        });
-
-    result
-}

src/bin/gatewayd.rs

@@ -19,8 +19,9 @@ 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 gateway = GatewayService::new(accept_addr, pub_addr);
+    let gateway = GatewayService::new(accept_addr, pub_addr, slabstore_path)?;
 
     gateway.start(executor.clone()).await?;
     Ok(())

src/bin/tx-test.rs

@@ -0,0 +1,326 @@
+use async_std::sync;
+use log::*;
+use bellman::groth16;
+use rocksdb::DB;
+use std::fs::File;
+use rusqlite::{Statement, MappedRows, Connection};
+use bls12_381::Bls12;
+use ff::{Field, PrimeField};
+use rand::rngs::OsRng;
+use std::path::Path;
+use drk::{Result, Error};
+
+use drk::crypto::{
+    coin::Coin,
+    load_params,
+    merkle::{CommitmentTree, IncrementalWitness},
+    merkle_node::{hash_coin, MerkleNode},
+    note::{EncryptedNote, Note},
+    nullifier::Nullifier,
+    save_params, setup_mint_prover, setup_spend_prover,
+};
+use drk::serial::{Decodable, Encodable};
+use drk::state::{state_transition, ProgramState, StateUpdate};
+use drk::tx;
+
+struct MemoryState {
+    // The entire merkle tree state
+    tree: CommitmentTree<MerkleNode>,
+    // List of all previous and the current merkle roots
+    // This is the hashed value of all the children.
+    merkle_roots: Vec<MerkleNode>,
+    // Nullifiers prevent double spending
+    nullifiers: Vec<Nullifier>,
+    // All received coins
+    // NOTE: we need maybe a flag to keep track of which ones are spent
+    // Maybe the spend field links to a tx hash:input index
+    // We should also keep track of the tx hash:output index where this
+    // coin was received
+    own_coins: Vec<(Coin, Note, jubjub::Fr, IncrementalWitness<MerkleNode>)>,
+
+    // Mint verifying key used by ZK
+    mint_pvk: groth16::PreparedVerifyingKey<Bls12>,
+    // Spend verifying key used by ZK
+    spend_pvk: groth16::PreparedVerifyingKey<Bls12>,
+
+    // Public key of the cashier
+    cashier_public: jubjub::SubgroupPoint,
+    // List of all our secret keys
+    secrets: Vec<jubjub::Fr>,
+}
+
+impl ProgramState for MemoryState {
+    fn is_valid_cashier_public_key(&self, public: &jubjub::SubgroupPoint) -> bool {
+        public == &self.cashier_public
+    }
+    // rocksdb
+    fn is_valid_merkle(&self, merkle_root: &MerkleNode) -> bool {
+        self.merkle_roots.iter().any(|m| *m == *merkle_root)
+    }
+    // rocksdb
+    fn nullifier_exists(&self, nullifier: &Nullifier) -> bool {
+        self.nullifiers.iter().any(|n| n.repr == nullifier.repr)
+    }
+
+    // loaded from disk
+    fn mint_pvk(&self) -> &groth16::PreparedVerifyingKey<Bls12> {
+        &self.mint_pvk
+    }
+    // loaded from disk
+    fn spend_pvk(&self) -> &groth16::PreparedVerifyingKey<Bls12> {
+        &self.spend_pvk
+    }
+}
+
+impl MemoryState {
+    fn apply(&mut self, mut update: StateUpdate) {
+        // Extend our list of nullifiers with the ones from the update
+        self.nullifiers.append(&mut update.nullifiers);
+
+        // merkle tree is rocksdb
+        // encrpt note is sql
+
+        // Update merkle tree and witnesses
+        for (coin, enc_note) in update.coins.into_iter().zip(update.enc_notes.into_iter()) {
+            // Add the new coins to the merkle tree
+            let node = MerkleNode::from_coin(&coin);
+            self.tree.append(node).expect("Append to merkle tree");
+
+            // Keep track of all merkle roots that have existed
+            self.merkle_roots.push(self.tree.root());
+
+            // own coins is sql
+            // Also update all the coin witnesses
+            for (_, _, _, witness) in self.own_coins.iter_mut() {
+                witness.append(node).expect("append to witness");
+            }
+
+            // sql
+            if let Some((note, secret)) = self.try_decrypt_note(enc_note) {
+                // We need to keep track of the witness for this coin.
+                // This allows us to prove inclusion of the coin in the merkle tree with ZK.
+                // Just as we update the merkle tree with every new coin, so we do the same with
+                // the witness.
+
+                // Derive the current witness from the current tree.
+                // This is done right after we add our coin to the tree (but before any other
+                // coins are added)
+
+                // Make a new witness for this coin
+                let witness = IncrementalWitness::from_tree(&self.tree);
+                self.own_coins.push((coin, note, secret, witness));
+            }
+        }
+    }
+
+    // sql
+    fn try_decrypt_note(&self, ciphertext: EncryptedNote) -> Option<(Note, jubjub::Fr)> {
+        debug!(target: "adapter", "try_decrypt_note() [START]");
+        let path = dirs::home_dir()
+            .expect("Cannot find home directory.")
+            .as_path()
+            .join(".config/darkfi/wallet.db");
+        debug!(target: "adapter", "try_decrypt_note() [FOUND PATH]");
+        println!("Found path: {:?}", &path);
+        debug!(target: "adapter", "try_decrypt_note() [TRY DB CONNECT]");
+        let connect = Connection::open(&path).expect("Failed to connect to database.");
+        let mut stmt = connect.prepare("SELECT key_private FROM keys").ok()?;
+        let key_iter = stmt.query_map::<String, _, _>([], |row| row.get(0)).ok()?;
+        for key in key_iter {
+            println!("Found key {:?}", key.unwrap());
+        }
+        // Loop through all our secret keys...
+        
+        for secret in &self.secrets {
+            // ... attempt to decrypt the note ...
+            match ciphertext.decrypt(secret) {
+                Ok(note) => {
+                    // ... and return the decrypted note for this coin.
+                    return Some((note, secret.clone()));
+                }
+                Err(_) => {}
+            }
+        }
+        // We weren't able to decrypt the note with any of our keys.
+        None
+    }
+}
+
+fn main() {
+    // Auto create trusted ceremony parameters if they don't exist
+    if !Path::new("mint.params").exists() {
+        let params = setup_mint_prover();
+        save_params("mint.params", &params).expect("Failed to create mint.params.");
+    }
+    if !Path::new("spend.params").exists() {
+        let params = setup_spend_prover();
+        save_params("spend.params", &params).expect("Failed to create save.params");
+    }
+
+    // Load trusted setup parameters
+    let (mint_params, mint_pvk) = load_params("mint.params").expect("params should load");
+    let (spend_params, spend_pvk) = load_params("spend.params").expect("params should load");
+
+    // Where is cashier private key stored? Does node have its own wallet schema
+    // Cashier creates a secret key
+    let cashier_secret = jubjub::Fr::random(&mut OsRng);
+    // This is their public key
+    let cashier_public = zcash_primitives::constants::SPENDING_KEY_GENERATOR * cashier_secret;
+
+    // Wallet 1 creates a secret key
+    let secret = jubjub::Fr::random(&mut OsRng);
+    // This is their public key
+    let public = zcash_primitives::constants::SPENDING_KEY_GENERATOR * secret;
+
+    let mut state = MemoryState {
+        tree: CommitmentTree::empty(),
+        merkle_roots: vec![],
+        nullifiers: vec![],
+        own_coins: vec![],
+        mint_pvk,
+        spend_pvk,
+        cashier_public,
+        secrets: vec![secret.clone()],
+    };
+
+    // Step 1: Cashier deposits to wallet1's address
+
+    // Create the deposit for 110 BTC
+    // Clear inputs are visible to everyone on the network
+    let builder = tx::TransactionBuilder {
+        clear_inputs: vec![tx::TransactionBuilderClearInputInfo {
+            value: 110,
+            signature_secret: cashier_secret,
+        }],
+        inputs: vec![],
+        outputs: vec![tx::TransactionBuilderOutputInfo { value: 110, public }],
+    };
+
+    // We will 'compile' the tx, and then serialize it to this Vec<u8>
+    let mut tx_data = vec![];
+    {
+        // Build the tx
+        let tx = builder.build(&mint_params, &spend_params);
+        // Now serialize it
+        tx.encode(&mut tx_data).expect("encode tx");
+    }
+
+    // Step 1 is completed.
+    // Tx data is posted to the blockchain
+
+    // Step 2: wallet1 receive's payment from the cashier
+
+    // Wallet1 is receiving tx, and for every new coin it finds, it adds to its
+    // merkle tree
+    {
+        // Here we simulate 5 fake random coins, adding them to our tree.
+        let tree = &mut state.tree;
+        for i in 0..5 {
+            // Don't worry about any of the code in this block
+            // We're just filling the tree with fake coins
+            let cmu = MerkleNode::new(bls12_381::Scalar::random(&mut OsRng).to_repr());
+            tree.append(cmu);
+
+            let root = tree.root();
+            state.merkle_roots.push(root.into());
+        }
+    }
+
+    // Now we receive the tx data
+    {
+        let tx = tx::Transaction::decode(&tx_data[..]).unwrap();
+
+        let update = state_transition(&state, tx).expect("step 2 state transition failed");
+        // Our state impl is memory online for this demo
+        // but in the real version, this function will be async
+        // and using the databases.
+        state.apply(update);
+    }
+
+    // Wallet1 has received payment from the cashier.
+    // Step 2 is complete.
+    assert_eq!(state.own_coins.len(), 1);
+    //let (coin, note, secret, witness) = &mut state.own_coins[0];
+
+    let merkle_path = {
+        let tree = &mut state.tree;
+        let (coin, _, _, witness) = &mut state.own_coins[0];
+        // Check this is the 6th coin we added
+        assert_eq!(witness.position(), 5);
+        assert_eq!(tree.root(), witness.root());
+
+        // Add some more random coins in
+        for i in 0..10 {
+            // Don't worry about any of the code in this block
+            // We're just filling the tree with fake coins
+            let cmu = MerkleNode::new(bls12_381::Scalar::random(&mut OsRng).to_repr());
+            tree.append(cmu);
+            witness.append(cmu);
+            assert_eq!(tree.root(), witness.root());
+
+            let root = tree.root();
+            state.merkle_roots.push(root.into());
+        }
+
+        assert_eq!(state.merkle_roots.len(), 16);
+
+        // This is the value we need to spend the coin
+        // We use the witness and the merkle root (both in sync with each other)
+        // to prove our coin exists inside the tree.
+        // The coin is not revealed publicly but is proved to exist inside
+        // a merkle tree. Only the root will be revealed, and then the
+        // verifier checks that merkle root actually existed before.
+        let merkle_path = witness.path().unwrap();
+
+        // Just test the path is good because we just added a bunch of fake coins
+        let node = MerkleNode::from_coin(&coin);
+        let root = tree.root();
+        drop(tree);
+        drop(witness);
+        assert_eq!(merkle_path.root(node), root);
+        let root = root.into();
+        assert!(state.is_valid_merkle(&root));
+
+        merkle_path
+    };
+
+    // Step 3: wallet1 sends payment to wallet2
+
+    // Wallet1 now wishes to send the coin to wallet2
+
+    // The receiving wallet has a secret key
+    let secret2 = jubjub::Fr::random(&mut OsRng);
+    // This is their public key to receive payment
+    let public2 = zcash_primitives::constants::SPENDING_KEY_GENERATOR * secret2;
+
+    // Make a spend tx
+
+    // Construct a new tx spending the coin
+    // We need the decrypted note and our private key
+    let builder = tx::TransactionBuilder {
+        clear_inputs: vec![],
+        inputs: vec![tx::TransactionBuilderInputInfo {
+            merkle_path,
+            secret: secret.clone(),
+            note: state.own_coins[0].1.clone(),
+        }],
+        // We can add more outputs to this list.
+        // The only constraint is that sum(value in) == sum(value out)
+        outputs: vec![tx::TransactionBuilderOutputInfo {
+            value: 110,
+            public: public2,
+        }],
+    };
+    // Build the tx
+    let mut tx_data = vec![];
+    {
+        let tx = builder.build(&mint_params, &spend_params);
+        tx.encode(&mut tx_data).expect("encode tx");
+    }
+    // Verify it's valid
+    {
+        let tx = tx::Transaction::decode(&tx_data[..]).unwrap();
+        let update = state_transition(&state, tx).expect("step 3 state transition failed");
+        state.apply(update);
+    }
+}

src/crypto/mod.rs

@@ -2,8 +2,8 @@ pub mod coin;
 pub mod diffie_hellman;
 pub mod fr_serial;
 pub mod merkle;
-pub mod mint_proof;
 pub mod merkle_node;
+pub mod mint_proof;
 pub mod note;
 pub mod nullifier;
 pub mod schnorr;

src/error.rs

@@ -35,8 +35,11 @@ pub enum Error {
     Utf8Error,
     NoteDecryptionFailed,
     ServicesError(&'static str),
-    ZMQError,
+    ZMQError(String),
     VerifyFailed,
+    TryIntoError,
+    TryFromError,
+    RocksdbError(String),
 }
 
 impl std::error::Error for Error {}
@@ -69,16 +72,25 @@ impl fmt::Display for Error {
             Error::Utf8Error => f.write_str("Malformed UTF8"),
             Error::NoteDecryptionFailed => f.write_str("Unable to decrypt mint note"),
             Error::ServicesError(ref err) => write!(f, "Services error: {}", err),
-            Error::ZMQError => f.write_str("ZMQ error"),
+            Error::ZMQError(ref err) => write!(f, "ZMQError: {}", err),
             Error::VerifyFailed => f.write_str("Verify failed"),
+            Error::TryIntoError => f.write_str("TryInto error"),
+            Error::TryFromError => f.write_str("TryFrom error"),
+            Error::RocksdbError(ref err) => write!(f, "Rocksdb Error: {}", err),
         }
     }
 }
 
 // TODO: Match statement to parse external errors into strings.
 impl From<zeromq::ZmqError> for Error {
-    fn from(_err: zeromq::ZmqError) -> Error {
-        Error::ZMQError
+    fn from(err: zeromq::ZmqError) -> Error {
+        Error::ZMQError(err.to_string())
+    }
+}
+
+impl From<rocksdb::Error> for Error {
+    fn from(err: rocksdb::Error) -> Error {
+        Error::RocksdbError(err.to_string())
     }
 }
 

src/lib.rs

@@ -16,6 +16,8 @@ pub mod net;
 pub mod rpc;
 pub mod serial;
 pub mod service;
+pub mod slab;
+pub mod slabstore;
 pub mod state;
 pub mod system;
 pub mod tx;

src/rpc/adapter.rs

@@ -1,5 +1,11 @@
-// Adapter class goes here
-//use crate::rpc::jsonserver::JsonRpcInterface;
+use crate::serial;
+use crate::Result;
+use ff::Field;
+use log::*;
+use rand::rngs::OsRng;
+use rusqlite::{named_params, Connection};
+use std::fs::File;
+use std::io::prelude::*;
 use std::sync::Arc;
 
 // Dummy adapter for now
@@ -10,9 +16,70 @@ impl RpcAdapter {
         Arc::new(Self {})
     }
 
-    pub async fn get_info() {}
+    pub async fn db_connect() -> Connection {
+        let path = dirs::home_dir()
+            .expect("Cannot find home directory.")
+            .as_path()
+            .join(".config/darkfi/wallet.db");
+        let connector = Connection::open(&path);
+        connector.expect("Failed to connect to database.")
+    }
 
-    pub async fn key_gen() {}
+    pub async fn key_gen() -> Result<()> {
+        debug!(target: "adapter", "key_gen() [START]");
+        let path = dirs::home_dir()
+            .expect("Cannot find home directory.")
+            .as_path()
+            .join(".config/darkfi/wallet.db");
+        debug!(target: "adapter", "key_gen() [FOUND PATH]");
+        println!("Found path: {:?}", &path);
+        debug!(target: "adapter", "key_gen() [TRY DB CONNECT]");
+        let connect = Connection::open(&path).expect("Failed to connect to database.");
+        // TODO: assign new ID on each run
+        debug!(target: "adapter", "key_gen() [Assigning ID...]");
+        let id = 0;
+        debug!(target: "adapter", "key_gen() [Generating private key...]");
+        let secret: jubjub::Fr = jubjub::Fr::random(&mut OsRng);
+        debug!(target: "adapter", "key_gen() [Generating public key...]");
+        let public = zcash_primitives::constants::SPENDING_KEY_GENERATOR * secret;
+        let pubkey = serial::serialize(&public);
+        let privkey = serial::serialize(&secret);
+        connect.execute(
+            "INSERT INTO keys(key_id, key_private, key_public)
+            VALUES (:id, :privkey, :pubkey)",
+            named_params!{":id": id,
+                           ":privkey": privkey,
+                           ":pubkey": pubkey
+                          }
+        )?;
+        Ok(())
+    }
+
+    pub async fn new_wallet() -> Result<()> {
+        debug!(target: "adapter", "new_wallet() [START]");
+        let path = dirs::home_dir()
+            .expect("Cannot find home directory.")
+            .as_path()
+            .join(".config/darkfi/wallet.db");
+        debug!(target: "adapter", "new_wallet() [FOUND PATH]");
+        println!("Found path: {:?}", &path);
+        debug!(target: "adapter", "new_wallet() [TRY DB CONNECT]");
+        let connect = Connection::open(&path).expect("Failed to connect to database.");
+        let contents = include_str!("../../res/schema.sql");
+        Ok(connect.execute_batch(&contents)?)
+    }
+
+    //pub async fn decrypt(conn: &Connection, password: )
+    // TODO: getting an error when i call this function- does not implement send
+    pub async fn save_key(conn: &Connection, pubkey: Vec<u8>, privkey: Vec<u8>) -> Result<()> {
+        // loads the walle
+        let mut db_file = File::open("wallet.sql")?;
+        let mut contents = String::new();
+        db_file.read_to_string(&mut contents)?;
+        Ok(conn.execute_batch(&mut contents)?)
+    }
+
+    pub async fn get_info() {}
 
     pub async fn say_hello() {}
 

src/rpc/jsonserver.rs

@@ -154,10 +154,18 @@ impl RpcInterface {
             RpcAdapter::stop().await;
             Ok(jsonrpc_core::Value::Null)
         });
-        io.add_method("key_gen", move |_| async move {
-            RpcAdapter::key_gen().await;
+        io.add_method("new_wallet", move |_| async move {
+            println!("New wallet method called...");
+            RpcAdapter::new_wallet().await;
             Ok(jsonrpc_core::Value::Null)
         });
+        io.add_method("key_gen", move |_| async move {
+            println!("Key generation method called...");
+            RpcAdapter::key_gen().await.expect("Failed to generate key");
+            Ok(jsonrpc_core::Value::String(
+                "Attempted key generation".into(),
+            ))
+        });
         debug!(target: "rpc", "JsonRpcInterface::handle_input() [END]");
         Ok(io)
     }

src/service/gateway.rs

@@ -1,16 +1,26 @@
-use async_std::sync::{Arc, Mutex};
-use std::convert::TryInto;
+use async_std::sync::Arc;
+use std::convert::From;
 use std::net::SocketAddr;
+use std::path::Path;
 
-use super::reqrep::{Publisher, RepProtocol, Reply, ReqProtocol, Request, Subscriber};
-use crate::{Error, Result};
+use super::reqrep::{PeerId, Publisher, RepProtocol, Reply, ReqProtocol, Request, Subscriber};
+use crate::{
+    serial::deserialize, serial::serialize, slab::Slab, slabstore::SlabStore, Error, Result,
+};
 
 use async_executor::Executor;
-
 use log::*;
 
 pub type Slabs = Vec<Vec<u8>>;
 
+
+#[repr(u8)]
+enum GatewayError {
+    NoError,
+    UpdateIndex,
+    IndexNotExist,
+}
+
 #[repr(u8)]
 enum GatewayCommand {
     PutSlab,
@@ -19,135 +29,258 @@ enum GatewayCommand {
 }
 
 pub struct GatewayService {
-    slabs: Mutex<Slabs>,
+    slabstore: Arc<SlabStore>,
     addr: SocketAddr,
-    publisher: Mutex<Publisher>,
+    pub_addr: SocketAddr,
 }
 
 impl GatewayService {
-    pub fn new(addr: SocketAddr, pub_addr: SocketAddr) -> Arc<GatewayService> {
-        let slabs = Mutex::new(vec![]);
-        let publisher = Mutex::new(Publisher::new(pub_addr));
-        Arc::new(GatewayService {
-            slabs,
+    pub fn new(addr: SocketAddr, pub_addr: SocketAddr, slabstore_path: &Path) -> Result<Arc<GatewayService>> {
+        let slabstore = SlabStore::new(slabstore_path)?;
+
+        Ok(Arc::new(GatewayService {
+            slabstore,
             addr,
-            publisher,
-        })
+            pub_addr,
+        }))
     }
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
-        let mut socket = RepProtocol::new(self.addr.clone());
+        let service_name = String::from("GATEWAY DAEMON");
 
-        let (send, recv) = socket.start().await?;
-        info!("server started: bind to {}", self.addr.to_string());
+        let mut protocol = RepProtocol::new(self.addr.clone(), service_name.clone());
 
-        self.publisher.lock().await.start().await?;
+        let (send, recv) = protocol.start().await?;
 
-        info!("publisher started");
+        let (publish_queue, publish_recv_queue) = async_channel::unbounded::<Vec<u8>>();
+        let publisher_task = executor.spawn(Self::start_publisher(
+                self.pub_addr,
+                service_name,
+                publish_recv_queue.clone(),
+        ));
 
-        let handle_request_task = executor.spawn(self.handle_request(send.clone(), recv.clone()));
+        let handle_request_task = executor.spawn(self.handle_request_loop(
+                send.clone(),
+                recv.clone(),
+                publish_queue.clone(),
+                executor.clone(),
+        ));
 
-        socket.run().await?;
+        protocol.run(executor.clone()).await?;
 
-        handle_request_task.cancel().await;
+        let _ = publisher_task.cancel().await;
+        let _ = handle_request_task.cancel().await;
         Ok(())
     }
 
-    async fn handle_request(
-        self: Arc<Self>,
-        send_queue: async_channel::Sender<Reply>,
-        recv_queue: async_channel::Receiver<Request>,
+    async fn start_publisher(
+        pub_addr: SocketAddr,
+        service_name: String,
+        publish_recv_queue: async_channel::Receiver<Vec<u8>>,
     ) -> Result<()> {
-        let data = vec![];
+        let mut publisher = Publisher::new(pub_addr, service_name);
+        publisher.start(publish_recv_queue).await?;
+        Ok(())
+    }
 
+    async fn handle_request_loop(
+        self: Arc<Self>,
+        send_queue: async_channel::Sender<(PeerId, Reply)>,
+        recv_queue: async_channel::Receiver<(PeerId, Request)>,
+        publish_queue: async_channel::Sender<Vec<u8>>,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
         loop {
             match recv_queue.recv().await {
-                Ok(request) => {
-                    match request.get_command() {
-                        0 => {
-                            // PUTSLAB
-                            let slab = request.get_payload();
-                            self.slabs.lock().await.push(slab.clone());
-
-                            // publish to all subscribes
-                            self.publisher.lock().await.publish(slab).await?;
-
-                            info!("received putslab msg");
-                        }
-                        1 => {
-                            // GETSLAB
-                            info!("received getslab msg");
-                        }
-                        2 => {
-                            // GETLASTINDEX
-                            info!("received getlastindex msg");
-                        }
-                        _ => {
-                            return Err(Error::ServicesError("wrong command"));
-                        }
+                Ok(msg) => {
+                    let slabstore = self.slabstore.clone();
+                    let _ = executor
+                        .spawn(Self::handle_request(
+                                msg,
+                                slabstore,
+                                send_queue.clone(),
+                                publish_queue.clone(),
+                        ))
+                        .detach();
                     }
-                    let rep = Reply::from(&request, 0, data.clone());
-                    send_queue.send(rep.into()).await?;
+                Err(_) => {
+                    break;
                 }
-                Err(_) => {}
             }
         }
+        Ok(())
+    }
+
+    async fn handle_request(
+        msg: (PeerId, Request),
+        slabstore: Arc<SlabStore>,
+        send_queue: async_channel::Sender<(PeerId, Reply)>,
+        publish_queue: async_channel::Sender<Vec<u8>>,
+    ) -> Result<()> {
+        let request = msg.1;
+        let peer = msg.0;
+        match request.get_command() {
+            0 => {
+                // PUTSLAB
+
+                let slab = request.get_payload();
+
+                // add to slabstore
+                let error = slabstore.put(slab.clone())?;
+
+                let mut reply = Reply::from(&request, GatewayError::NoError as u32, vec![]);
+
+
+                if let None = error {
+                    reply.set_error(GatewayError::UpdateIndex as u32);
+                }
+
+                // send reply
+                send_queue.send((peer, reply)).await?;
+
+                // publish to all subscribes
+                publish_queue.send(slab).await?;
+
+                info!("Received putslab msg");
+            }
+            1 => {
+                let index = request.get_payload();
+                let slab = slabstore.get(index)?;
+
+                let mut reply = Reply::from(&request, GatewayError::NoError as u32, vec![]);
+
+                if let Some(payload) = slab {
+                    reply.set_payload(payload);
+                } else {
+                    reply.set_error(GatewayError::IndexNotExist as u32);
+                }
+
+                send_queue.send((peer, reply)).await?;
+
+                // GETSLAB
+                info!("Received getslab msg");
+            }
+            2 => {
+                let index = slabstore.get_last_index_as_bytes()?;
+
+                let reply = Reply::from(&request, GatewayError::NoError as u32, index);
+                send_queue.send((peer, reply)).await?;
+
+                // GETLASTINDEX
+                info!("Received getlastindex msg");
+            }
+            _ => {
+                return Err(Error::ServicesError("received wrong command"));
+            }
+        }
+        Ok(())
     }
 }
 
 pub struct GatewayClient {
     protocol: ReqProtocol,
+    slabstore: Arc<SlabStore>,
 }
 
 impl GatewayClient {
-    pub fn new(addr: SocketAddr) -> GatewayClient {
-        let protocol = ReqProtocol::new(addr);
-        GatewayClient { protocol }
+    pub fn new(addr: SocketAddr, path: &Path) -> Result<Self> {
+        let protocol = ReqProtocol::new(addr, String::from("GATEWAY CLIENT"));
+
+        let slabstore = SlabStore::new(path)?;
+
+        Ok(GatewayClient {
+            protocol,
+            slabstore,
+        })
     }
+
     pub async fn start(&mut self) -> Result<()> {
         self.protocol.start().await?;
+        self.sync().await?;
+
         Ok(())
     }
 
-    pub async fn subscribe(&self, sub_addr: SocketAddr) -> Result<Arc<Mutex<Subscriber>>> {
-        let mut subscriber = Subscriber::new(sub_addr);
-        subscriber.start().await?;
-        Ok(Arc::new(Mutex::new(subscriber)))
-    }
+    pub async fn sync(&mut self) -> Result<u64> {
+        info!("Start Syncing");
+        let local_last_index = self.slabstore.get_last_index()?;
+
+        let last_index = self.get_last_index().await?;
+
+        assert!(last_index >= local_last_index);
+
+        if last_index > 0  {
+            for index in (local_last_index + 1)..(last_index + 1) {
+                if let None = self.get_slab(index).await? {
+                    warn!("Index not exist");
+                    break;
+                }
+            }
+        }
+
+
+
+        info!("End Syncing");
+        Ok(last_index)
 
-    pub async fn get_slab(&mut self, index: u32) -> Result<Vec<u8>> {
-        self.protocol
-            .request(GatewayCommand::GetSlab as u8, index.to_be_bytes().to_vec())
-            .await
     }
 
-    pub async fn put_slab(&mut self, data: Vec<u8>) -> Result<()> {
-        self.protocol
-            .request(GatewayCommand::PutSlab as u8, data.clone())
+    pub async fn get_slab(&mut self, index: u64) -> Result<Option<Vec<u8>>> {
+        let rep = self
+            .protocol
+            .request(GatewayCommand::GetSlab as u8, serialize(&index))
             .await?;
+
+        if let Some(slab) = rep{
+            self.slabstore.put(slab.clone())?;
+            return Ok(Some(slab));
+        }
+        Ok(None)
+    }
+
+    pub async fn put_slab(&mut self, mut slab: Slab) -> Result<()> {
+        loop{
+            let last_index = self.sync().await?;
+            slab.set_index(last_index + 1);
+            let slab = serialize(&slab);
+
+            let rep = self.protocol
+                .request(GatewayCommand::PutSlab as u8, slab.clone())
+                .await?;
+
+            if let Some(_) =  rep{
+                break;
+            }
+        }
         Ok(())
     }
-    pub async fn get_last_index(&mut self) -> Result<u32> {
+
+    pub async fn get_last_index(&mut self) -> Result<u64> {
         let rep = self
             .protocol
             .request(GatewayCommand::GetLastIndex as u8, vec![])
             .await?;
-        let rep: [u8; 4] = rep.try_into().unwrap();
-        Ok(u32::from_be_bytes(rep))
+        if let Some(index) = rep {
+            return Ok(deserialize(&index)?);
+        }
+        Ok(0)
+    }
+
+    pub fn get_slabstore(&self) -> Arc<SlabStore> {
+        self.slabstore.clone()
     }
-}
 
-pub async fn fetch_slabs_loop(
-    subscriber: Arc<Mutex<Subscriber>>,
-    slabs: Arc<Mutex<Slabs>>,
-) -> Result<()> {
-    loop {
-        let slab: Vec<u8>;
-        {
-            let mut subscriber = subscriber.lock().await;
+    pub async fn start_subscriber(sub_addr: SocketAddr) -> Result<Subscriber> {
+        let mut subscriber = Subscriber::new(sub_addr, String::from("GATEWAY CLIENT"));
+        subscriber.start().await?;
+        Ok(subscriber)
+    }
+
+    pub async fn subscribe(mut subscriber: Subscriber, slabstore: Arc<SlabStore>) -> Result<()> {
+        loop {
+            let slab: Vec<u8>;
             slab = subscriber.fetch().await?;
+            slabstore.put(slab)?;
         }
-        info!("received new slab from subscriber");
-        slabs.lock().await.push(slab);
     }
 }

src/service/mod.rs

@@ -2,5 +2,5 @@ pub mod gateway;
 pub mod options;
 pub mod reqrep;
 
-pub use gateway::{fetch_slabs_loop, GatewayClient, GatewayService};
-pub use options::ProgramOptions;
+pub use gateway::{GatewayClient, GatewayService};
+pub use options::{ClientProgramOptions, ProgramOptions};

src/service/options.rs

@@ -5,6 +5,7 @@ pub struct ProgramOptions {
     pub accept_addr: Option<SocketAddr>,
     pub pub_addr: Option<SocketAddr>,
     pub verbose: bool,
+    pub slabstore_path: Box<std::path::PathBuf>,
     pub log_path: Box<std::path::PathBuf>,
 }
 
@@ -17,6 +18,7 @@ impl ProgramOptions {
             (@arg ACCEPT: -a --accept +takes_value "Accept add//ress")
             (@arg PUB_ADDR: -p --pubaddr +takes_value "Publisher addr")
             (@arg VERBOSE: -v --verbose "Increase verbosity")
+            (@arg SLABSTORE_PATH: --slabstore +takes_value "slabstore path")
             (@arg LOG_PATH: --log +takes_value "Logfile path")
         )
         .get_matches();
@@ -35,6 +37,15 @@ impl ProgramOptions {
 
         let verbose = app.is_present("VERBOSE");
 
+        let slabstore_path = Box::new(
+            if let Some(slabstore_path) = app.value_of("SLABSTORE_PATH") {
+                std::path::Path::new(slabstore_path)
+            } else {
+                std::path::Path::new("slabstore.db")
+            }
+            .to_path_buf(),
+        );
+
         let log_path = Box::new(
             if let Some(log_path) = app.value_of("LOG_PATH") {
                 std::path::Path::new(log_path)
@@ -48,6 +59,71 @@ impl ProgramOptions {
             accept_addr,
             pub_addr,
             verbose,
+            slabstore_path,
+            log_path,
+        })
+    }
+}
+
+pub struct ClientProgramOptions {
+    pub connect_addr: Option<SocketAddr>,
+    pub sub_addr: Option<SocketAddr>,
+    pub verbose: bool,
+    pub slabstore_path: Box<std::path::PathBuf>,
+    pub log_path: Box<std::path::PathBuf>,
+}
+
+impl ClientProgramOptions {
+    pub fn load() -> Result<Self> {
+        let app = clap_app!(dfi =>
+            (version: "0.1.0")
+            (author: "Amir Taaki <amir@dyne.org>")
+            (about: "Run Service Client")
+            (@arg CONNECT: -c --connect +takes_value "Connect add//ress")
+            (@arg SUB_ADDR: -s --subaddr +takes_value "Subscriber addr")
+            (@arg VERBOSE: -v --verbose "Increase verbosity")
+            (@arg SLABSTORE_PATH: --slabstore +takes_value "slabstore path")
+            (@arg LOG_PATH: --log +takes_value "Logfile path")
+        )
+        .get_matches();
+
+        let connect_addr = if let Some(connect_addr) = app.value_of("CONNECT") {
+            Some(connect_addr.parse()?)
+        } else {
+            None
+        };
+
+        let sub_addr = if let Some(sub_addr) = app.value_of("SUB_ADDR") {
+            Some(sub_addr.parse()?)
+        } else {
+            None
+        };
+
+        let verbose = app.is_present("VERBOSE");
+
+        let slabstore_path = Box::new(
+            if let Some(slabstore_path) = app.value_of("SLABSTORE_PATH") {
+                std::path::Path::new(slabstore_path)
+            } else {
+                std::path::Path::new("slabstore_client.db")
+            }
+            .to_path_buf(),
+        );
+
+        let log_path = Box::new(
+            if let Some(log_path) = app.value_of("LOG_PATH") {
+                std::path::Path::new(log_path)
+            } else {
+                std::path::Path::new("/tmp/darkfid_service_daemon.log")
+            }
+            .to_path_buf(),
+        );
+
+        Ok(ClientProgramOptions {
+            connect_addr,
+            sub_addr,
+            verbose,
+            slabstore_path,
             log_path,
         })
     }

src/service/reqrep.rs

@@ -1,17 +1,25 @@
+use async_std::sync::Arc;
+use std::convert::TryFrom;
 use std::io;
 use std::net::SocketAddr;
 
 use crate::serial::{deserialize, serialize};
 use crate::{Decodable, Encodable, Result};
 
+use async_executor::Executor;
 use bytes::Bytes;
 use futures::FutureExt;
+use log::*;
 use rand::Rng;
+use signal_hook::{consts::SIGINT, iterator::Signals};
 use zeromq::*;
 
+pub type PeerId  = Vec<u8>;
+
 enum NetEvent {
     Receive(zeromq::ZmqMessage),
-    Send(Reply),
+    Send((PeerId, Reply)),
+    Stop,
 }
 
 pub fn addr_to_string(addr: SocketAddr) -> String {
@@ -20,20 +28,21 @@ pub fn addr_to_string(addr: SocketAddr) -> String {
 
 pub struct RepProtocol {
     addr: SocketAddr,
-    socket: zeromq::RepSocket,
-    recv_queue: async_channel::Receiver<Reply>,
-    send_queue: async_channel::Sender<Request>,
+    socket: zeromq::RouterSocket,
+    recv_queue: async_channel::Receiver<(PeerId, Reply)>,
+    send_queue: async_channel::Sender<(PeerId, Request)>,
     channels: (
-        async_channel::Sender<Reply>,
-        async_channel::Receiver<Request>,
+        async_channel::Sender<(PeerId, Reply)>,
+        async_channel::Receiver<(PeerId, Request)>,
     ),
+    service_name: String,
 }
 
 impl RepProtocol {
-    pub fn new(addr: SocketAddr) -> RepProtocol {
-        let socket = zeromq::RepSocket::new();
-        let (send_queue, recv_channel) = async_channel::unbounded::<Request>();
-        let (send_channel, recv_queue) = async_channel::unbounded::<Reply>();
+    pub fn new(addr: SocketAddr, service_name: String) -> RepProtocol {
+        let socket = zeromq::RouterSocket::new();
+        let (send_queue, recv_channel) = async_channel::unbounded::<(PeerId, Request)>();
+        let (send_channel, recv_queue) = async_channel::unbounded::<(PeerId, Reply)>();
 
         let channels = (send_channel.clone(), recv_channel.clone());
 
@@ -43,101 +52,169 @@ impl RepProtocol {
             recv_queue,
             send_queue,
             channels,
+            service_name,
         }
     }
 
     pub async fn start(
         &mut self,
     ) -> Result<(
-        async_channel::Sender<Reply>,
-        async_channel::Receiver<Request>,
+    async_channel::Sender<(PeerId, Reply)>,
+    async_channel::Receiver<(PeerId, Request)>,
     )> {
         let addr = addr_to_string(self.addr);
         self.socket.bind(addr.as_str()).await?;
+        info!("{} SERVICE: Bound To {}", self.service_name, addr);
         Ok(self.channels.clone())
     }
 
-    pub async fn run(&mut self) -> Result<()> {
+    pub async fn run(&mut self, executor: Arc<Executor<'_>>) -> Result<()> {
+        info!("{} SERVICE: Running", self.service_name);
+
+        let (stop_s, stop_r) = async_channel::unbounded::<()>();
+
+        let mut signals = Signals::new(&[SIGINT])?;
+
+        let stop_task = executor.spawn(async move {
+            for _ in signals.forever() {
+                stop_s.send(()).await?;
+                break;
+            }
+            Ok::<(), crate::Error>(())
+        });
+
         loop {
             let event = futures::select! {
-                request = self.socket.recv().fuse() => NetEvent::Receive(request?),
-                reply = self.recv_queue.recv().fuse() => NetEvent::Send(reply?)
+                msg = self.socket.recv().fuse() => NetEvent::Receive(msg?),
+                msg = self.recv_queue.recv().fuse() => NetEvent::Send(msg?),
+                _ = stop_r.recv().fuse() => NetEvent::Stop
             };
 
             match event {
-                NetEvent::Receive(request) => {
-                    let request: &Bytes = request.get(0).unwrap();
-                    let request: Vec<u8> = request.to_vec();
-                    let req: Request = deserialize(&request)?;
-                    self.send_queue.send(req).await?;
+                NetEvent::Receive(msg) => {
+                    if let Some(peer) = msg.get(0) {
+                        if let Some(request) = msg.get(1) {
+                            let request: Vec<u8> = request.to_vec();
+                            let request: Request = deserialize(&request)?;
+                            self.send_queue.send((peer.to_vec(), request)).await?;
+                        }
+                    }
                 }
-                NetEvent::Send(reply) => {
+                NetEvent::Send((peer, reply)) => {
+                    let peer = Bytes::from(peer);
+                    let mut msg: Vec<Bytes> = vec![peer];
                     let reply: Vec<u8> = serialize(&reply);
                     let reply = Bytes::from(reply);
-                    self.socket.send(reply.into()).await?;
+                    msg.push(reply);
+
+                    let reply = zeromq::ZmqMessage::try_from(msg)
+                        .map_err(|_| crate::Error::TryFromError)?;
+
+                    self.socket.send(reply).await?;
                 }
+                NetEvent::Stop => break,
             }
         }
+        let _ = stop_task.cancel().await;
+        warn!("{} SERVICE: Stopped", self.service_name);
+        Ok(())
     }
 }
 
 pub struct ReqProtocol {
     addr: SocketAddr,
-    socket: zeromq::ReqSocket,
+    socket: zeromq::DealerSocket,
+    service_name: String,
 }
 
 impl ReqProtocol {
-    pub fn new(addr: SocketAddr) -> ReqProtocol {
-        let socket = zeromq::ReqSocket::new();
-        ReqProtocol { addr, socket }
+    pub fn new(addr: SocketAddr, service_name: String) -> ReqProtocol {
+        let socket = zeromq::DealerSocket::new();
+        ReqProtocol {
+            addr,
+            socket,
+            service_name,
+        }
     }
 
     pub async fn start(&mut self) -> Result<()> {
         let addr = addr_to_string(self.addr);
         self.socket.connect(addr.as_str()).await?;
+        info!("{} SERVICE: Connected To {}", self.service_name, self.addr);
         Ok(())
     }
 
-    pub async fn request(&mut self, command: u8, data: Vec<u8>) -> Result<Vec<u8>> {
+    pub async fn request(&mut self, command: u8, data: Vec<u8>) -> Result<Option<Vec<u8>>> {
         let request = Request::new(command, data);
         let req = serialize(&request);
         let req = bytes::Bytes::from(req);
+        let req: zeromq::ZmqMessage = req.into();
 
-        self.socket.send(req.into()).await?;
+        self.socket.send(req).await?;
+        info!(
+            "{} SERVICE: Sent Request {{ command: {} }}",
+            self.service_name, command
+        );
 
         let rep: zeromq::ZmqMessage = self.socket.recv().await?;
-        let rep: &Bytes = rep.get(0).unwrap();
-        let rep: Vec<u8> = rep.to_vec();
+        if let Some(reply) = rep.get(0) {
+            let reply: Vec<u8> = reply.to_vec();
 
-        let reply: Reply = deserialize(&rep)?;
+            let reply: Reply = deserialize(&reply)?;
 
-        if reply.has_error() {
-            return Err(crate::Error::ServicesError("response has an error"));
-        }
+            info!(
+                "{} SERVICE: Received Reply {{ error: {} }}",
+                self.service_name,
+                reply.has_error()
+            );
+
+            // TODO return error status code instead of None
+            if reply.has_error() {
+                warn!("Reply has an error {}", reply.get_error());
+                return Ok(None);
+            }
 
-        assert!(reply.get_id() == request.get_id());
+            assert!(reply.get_id() == request.get_id());
 
-        Ok(reply.get_payload())
+            Ok(Some(reply.get_payload()))
+        } else {
+            Err(crate::Error::ZMQError(
+                    "Couldn't parse ZmqMessage".to_string(),
+            ))
+        }
     }
 }
 
 pub struct Publisher {
     addr: SocketAddr,
     socket: zeromq::PubSocket,
+    service_name: String,
 }
 
 impl Publisher {
-    pub fn new(addr: SocketAddr) -> Publisher {
+    pub fn new(addr: SocketAddr, service_name: String) -> Publisher {
         let socket = zeromq::PubSocket::new();
-        Publisher { addr, socket }
+        Publisher {
+            addr,
+            socket,
+            service_name,
+        }
     }
-    pub async fn start(&mut self) -> Result<()> {
+
+    pub async fn start(&mut self, recv_queue: async_channel::Receiver<Vec<u8>>) -> Result<()> {
         let addr = addr_to_string(self.addr);
         self.socket.bind(addr.as_str()).await?;
-        Ok(())
+        info!(
+            "{} PUBLISHER SERVICE : Bound To {}",
+            self.service_name, addr
+        );
+        loop {
+            let msg = recv_queue.recv().await?;
+            self.publish(msg).await?;
+        }
     }
 
-    pub async fn publish(&mut self, data: Vec<u8>) -> Result<()> {
+    async fn publish(&mut self, data: Vec<u8>) -> Result<()> {
         let data = Bytes::from(data);
         self.socket.send(data.into()).await?;
         Ok(())
@@ -147,12 +224,17 @@ impl Publisher {
 pub struct Subscriber {
     addr: SocketAddr,
     socket: zeromq::SubSocket,
+    service_name: String,
 }
 
 impl Subscriber {
-    pub fn new(addr: SocketAddr) -> Subscriber {
+    pub fn new(addr: SocketAddr, service_name: String) -> Subscriber {
         let socket = zeromq::SubSocket::new();
-        Subscriber { addr, socket }
+        Subscriber {
+            addr,
+            socket,
+            service_name,
+        }
     }
 
     pub async fn start(&mut self) -> Result<()> {
@@ -160,15 +242,24 @@ impl Subscriber {
         self.socket.connect(addr.as_str()).await?;
 
         self.socket.subscribe("").await?;
-
+        info!(
+            "{} SUBSCRIBER SERVICE : Connected To {}",
+            self.service_name, addr
+        );
         Ok(())
     }
 
     pub async fn fetch(&mut self) -> Result<Vec<u8>> {
         let data = self.socket.recv().await?;
-        let data: &Bytes = data.get(0).unwrap();
-        let data = data.to_vec();
-        Ok(data)
+        match data.get(0) {
+            Some(d) => {
+                let data = d.to_vec();
+                Ok(data)
+            }
+            None => Err(crate::Error::ZMQError(
+                    "Couldn't parse ZmqMessage".to_string(),
+            )),
+        }
     }
 }
 
@@ -230,10 +321,22 @@ impl Reply {
         }
     }
 
+    pub fn get_error(&self) -> u32 {
+        self.error
+    }
+
     pub fn get_payload(&self) -> Vec<u8> {
         self.payload.clone()
     }
 
+    pub fn set_payload(&mut self, payload: Vec<u8>) {
+        self.payload = payload;
+    }
+
+    pub fn set_error(&mut self, error: u32) {
+        self.error = error;
+    }
+
     pub fn get_id(&self) -> u32 {
         self.id
     }

src/slab.rs

@@ -0,0 +1,51 @@
+use crate::serial::{Decodable, Encodable};
+use crate::Result;
+
+pub struct Slab {
+    asset_type: String,
+    index: u64,
+    payload: Vec<u8>,
+}
+
+impl Slab {
+    pub fn new(asset_type: String, payload: Vec<u8>) -> Self {
+        let index = 0;
+        Slab {
+            asset_type,
+            index,
+            payload,
+        }
+    }
+
+    pub fn set_index(&mut self, index: u64) {
+        self.index = index;
+    }
+
+    pub fn get_index(&self) -> u64 {
+        self.index
+    }
+
+    pub fn get_payload(&self) -> Vec<u8> {
+        self.payload.clone()
+    }
+}
+
+impl Encodable for Slab {
+    fn encode<S: std::io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.asset_type.encode(&mut s)?;
+        len += self.index.encode(&mut s)?;
+        len += self.payload.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for Slab {
+    fn decode<D: std::io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            asset_type: Decodable::decode(&mut d)?,
+            index: Decodable::decode(&mut d)?,
+            payload: Decodable::decode(&mut d)?,
+        })
+    }
+}

src/slabstore.rs

@@ -0,0 +1,72 @@
+use std::path::Path;
+use std::sync::Arc;
+
+use crate::serial::{deserialize, serialize};
+use crate::{slab::Slab, Result};
+
+use rocksdb::{IteratorMode, Options, DB};
+
+pub struct SlabStore {
+    db: DB,
+}
+
+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 get(&self, key: Vec<u8>) -> Result<Option<Vec<u8>>> {
+        let value = self.db.get(key)?;
+        Ok(value)
+    }
+
+    pub fn put(&self, value: Vec<u8>) -> Result<Option<Vec<u8>>> {
+        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)?;
+            Ok(Some(key))
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub fn get_value_deserialized(&self, key: Vec<u8>) -> Result<Option<Slab>> {
+        let value = self.db.get(key)?;
+        match value {
+            Some(v) => {
+                let v: Slab = deserialize(&v)?;
+                Ok(Some(v))
+            }
+            None => Ok(None),
+        }
+    }
+
+    pub fn get_last_index(&self) -> Result<u64> {
+        let last_index = self.db.iterator(IteratorMode::End).next();
+        match last_index {
+            Some((index, _)) => Ok(deserialize(&index)?),
+            None => Ok(0),
+        }
+    }
+
+    pub fn get_last_index_as_bytes(&self) -> Result<Vec<u8>> {
+        let last_index = self.db.iterator(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(())
+    }
+}

src/tx/builder.rs

@@ -8,7 +8,8 @@ use super::{
     Transaction, TransactionClearInput, TransactionInput, TransactionOutput,
 };
 use crate::crypto::{
-    create_mint_proof, create_spend_proof, merkle::MerklePath, merkle_node::MerkleNode, note::Note, schnorr,
+    create_mint_proof, create_spend_proof, merkle::MerklePath, merkle_node::MerkleNode, note::Note,
+    schnorr,
 };
 use crate::serial::Encodable;
 

src/wallet/mod.rs

@@ -1 +0,0 @@
-// Empty mod.rs file for now