creat bridge and add refactoring cashier to work with bridge

src/bin/cashierd.rs

@@ -38,7 +38,6 @@ async fn start(executor: Arc<Executor<'_>>, config: Arc<CashierdConfig>) -> Resu
 
     let mut cashier = CashierService::new(
         accept_addr,
-        btc_endpoint,
         wallet.clone(),
         client_wallet.clone(),
         database_path,
@@ -47,7 +46,7 @@ async fn start(executor: Arc<Executor<'_>>, config: Arc<CashierdConfig>) -> Resu
     )
     .await?;
 
-    cashier.start(ex.clone()).await?;
+    cashier.start(ex.clone(), btc_endpoint).await?;
 
     //let rpc_url: std::net::SocketAddr = config.rpc_url.parse()?;
     //let adapter = Arc::new(CashierAdapter::new(wallet.clone())?);

src/rpc/adapters/client_adapter.rs

@@ -1,5 +1,5 @@
 use crate::client::{Client, ClientFailed};
-use crate::serial::serialize;
+use crate::serial::{deserialize, serialize};
 use crate::service::CashierClient;
 use crate::{Error, Result};
 
@@ -110,7 +110,7 @@ impl RpcClientAdapter {
                 .await?;
 
             return Ok(format!(
-                "sending {} dbtc to provided address for withdrawing: {} ",
+                "sending {} drk to provided address for withdrawing: {} ",
                 amount, drk_addr
             ));
         } else {
@@ -123,15 +123,15 @@ impl RpcClientAdapter {
         cashier_client: Arc<Mutex<CashierClient>>,
     ) -> Result<String> {
         let deposit_addr = client.lock().await.state.wallet.get_public_keys()?[0];
-        let btc_public = cashier_client
+        let coin_public = cashier_client
             .lock()
             .await
             .get_address(deposit_addr)
             .await
             .map_err(|err| ClientFailed::from(err))?;
 
-        if let Some(btc_addr) = btc_public {
-            return Ok(btc_addr.to_string());
+        if let Some(coin_addr) = coin_public {
+            return Ok(deserialize(&coin_addr)?);
         } else {
             return Err(Error::from(ClientFailed::UnableToGetDepositAddress));
         }

src/service/bridge.rs

@@ -0,0 +1,101 @@
+use crate::Result;
+
+use async_executor::Executor;
+use async_trait::async_trait;
+
+use async_std::sync::{Arc, Mutex};
+use std::collections::HashMap;
+
+pub struct BridgeRequests {
+    pub asset_id: u64,
+    pub payload: BridgeRequestsPayload,
+}
+
+pub struct BridgeResponse {
+    pub error: u64,
+    pub payload: BridgeResponsePayload,
+}
+
+pub enum BridgeRequestsPayload {
+    SendRequest(Vec<u8>, u64),
+    WatchRequest,
+}
+
+pub enum BridgeResponsePayload {
+    WatchResponse(Vec<u8>, Vec<u8>),
+    SendResponse,
+}
+
+pub struct BridgeSubscribtion {
+    pub sender: async_channel::Sender<BridgeRequests>,
+    pub receiver: async_channel::Receiver<BridgeResponse>,
+}
+
+pub struct Bridge {
+    clients: Mutex<HashMap<u64, Arc<dyn CoinClient + Send + Sync>>>,
+}
+impl Bridge {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self {
+            clients: Mutex::new(HashMap::new()),
+        })
+    }
+
+    pub async fn add_clients(
+        self: Arc<Self>,
+        asset_id: u64,
+        client: Arc<dyn CoinClient + Send + Sync>,
+    ) {
+        self.clients.lock().await.insert(asset_id, client);
+    }
+
+    pub async fn subscribe(self: Arc<Self>, executor: Arc<Executor<'_>>) -> BridgeSubscribtion {
+        let (sender, req) = async_channel::unbounded();
+        let (rep, receiver) = async_channel::unbounded();
+
+        executor
+            .spawn(self.listen_for_new_subscribtion(req.clone(), rep.clone()))
+            .detach();
+
+        BridgeSubscribtion { sender, receiver }
+    }
+
+    pub async fn listen_for_new_subscribtion(
+        self: Arc<Self>,
+        req: async_channel::Receiver<BridgeRequests>,
+        rep: async_channel::Sender<BridgeResponse>,
+    ) -> Result<()> {
+        let req = req.recv().await?;
+        let client = &self.clients.lock().await[&req.asset_id];
+
+        match req.payload {
+            BridgeRequestsPayload::WatchRequest => {
+                let (private, public) = client.watch().await?;
+                let res = BridgeResponse {
+                    error: 0,
+                    payload: BridgeResponsePayload::WatchResponse(private, public),
+                };
+                rep.send(res).await?;
+            }
+            BridgeRequestsPayload::SendRequest(addr, amount) => {
+                client.send(addr, amount).await?;
+                let res = BridgeResponse {
+                    error: 0,
+                    payload: BridgeResponsePayload::SendResponse,
+                };
+                rep.send(res).await?;
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+pub trait CoinClient {
+    // return private and public keys that be watching
+    async fn watch(&self) -> Result<(Vec<u8>, Vec<u8>)>;
+    async fn send(&self, address: Vec<u8>, amount: u64) -> Result<()>;
+}
+
+

src/service/btc.rs

@@ -1,5 +1,6 @@
 use crate::serial::{Decodable, Encodable};
 use crate::Result;
+use super::bridge::CoinClient;
 
 use bitcoin::blockdata::script::Script;
 use bitcoin::network::constants::Network;
@@ -10,6 +11,7 @@ use log::*;
 use rand::distributions::Alphanumeric;
 use rand::{thread_rng, Rng};
 use secp256k1::key::SecretKey;
+use async_trait::async_trait;
 
 use async_std::sync::Arc;
 use std::str::FromStr;
@@ -174,6 +176,47 @@ impl std::fmt::Display for BtcFailed {
     }
 }
 
+
+pub struct BtcClient {
+    client: Arc<ElectrumClient>,
+}
+
+impl BtcClient {
+    pub fn new(client_address: String) -> Result<Self> {
+        let client = ElectrumClient::new(&client_address)
+            .map_err(|err| crate::Error::from(super::BtcFailed::from(err)))?;
+        Ok(Self {
+            client: Arc::new(client),
+        })
+    }
+}
+
+#[async_trait]
+impl CoinClient for BtcClient {
+    async fn watch(&self) -> Result<(Vec<u8>, Vec<u8>)> {
+        //// Generate bitcoin Address
+        let btc_keys = BitcoinKeys::new(self.client.clone())?;
+
+        let btc_pub = btc_keys.clone();
+        let btc_pub = btc_pub.get_pubkey();
+        let btc_priv = btc_keys.clone();
+        let btc_priv = btc_priv.get_privkey();
+
+        // let _ = btc_keys.start_subscribe().await?;
+
+        // start scheduler for checking balance
+        debug!(target: "BRIDGE BITCOIN", "Subscribing for deposit");
+        //let _script = btc_keys.get_script();
+        //
+
+        Ok((btc_priv.to_bytes(), btc_pub.to_bytes()))
+    }
+    async fn send(&self, _address: Vec<u8>, _amount: u64) -> Result<()> {
+        // TODO
+        Ok(())
+    }
+}
+
 impl From<crate::error::Error> for BtcFailed {
     fn from(err: crate::error::Error) -> BtcFailed {
         BtcFailed::BtcError(err.to_string())

src/service/cashier.rs

@@ -1,4 +1,4 @@
-use super::btc::{BitcoinKeys, PubAddress};
+use super::bridge;
 use super::reqrep::{PeerId, RepProtocol, Reply, ReqProtocol, Request};
 use crate::blockchain::Rocks;
 use crate::client::Client;
@@ -10,7 +10,6 @@ use ff::Field;
 use rand::rngs::OsRng;
 
 use async_executor::Executor;
-use electrum_client::Client as ElectrumClient;
 use log::*;
 
 use async_std::sync::{Arc, Mutex};
@@ -31,14 +30,12 @@ enum CashierCommand {
 pub struct CashierService {
     addr: SocketAddr,
     wallet: CashierDbPtr,
-    btc_client: Arc<ElectrumClient>,
     client: Arc<Mutex<Client>>,
 }
 
 impl CashierService {
     pub async fn new(
         addr: SocketAddr,
-        btc_endpoint: String,
         wallet: CashierDbPtr,
         client_wallet: WalletPtr,
         cashier_database_path: PathBuf,
@@ -46,13 +43,6 @@ impl CashierService {
         params_paths: (PathBuf, PathBuf),
     ) -> Result<CashierService> {
         // Pull address from config later
-        let client_address = btc_endpoint;
-
-        // create btc client
-        let btc_client = Arc::new(
-            ElectrumClient::new(&client_address)
-                .map_err(|err| crate::Error::from(super::BtcFailed::from(err)))?,
-        );
 
         let rocks = Rocks::new(&cashier_database_path)?;
 
@@ -63,11 +53,14 @@ impl CashierService {
         Ok(CashierService {
             addr,
             wallet,
-            btc_client,
             client,
         })
     }
-    pub async fn start(&mut self, executor: Arc<Executor<'_>>) -> Result<()> {
+    pub async fn start(
+        &mut self,
+        executor: Arc<Executor<'_>>,
+        client_address: String,
+    ) -> Result<()> {
         debug!(target: "CASHIER DAEMON", "Start Cashier");
         let service_name = String::from("CASHIER DAEMON");
 
@@ -78,19 +71,25 @@ impl CashierService {
         self.wallet.init_db()?;
 
         let wallet = self.wallet.clone();
-        let btc_client = self.btc_client.clone();
+
+        let bridge = bridge::Bridge::new();
+
+        #[cfg(feature = "default")]
+        let btc_client = super::btc::BtcClient::new(client_address)?;
+        #[cfg(feature = "default")]
+        bridge.clone().add_clients(1, Arc::new(btc_client)).await;
 
         let handle_request_task = executor.spawn(Self::handle_request_loop(
             send.clone(),
             recv.clone(),
             wallet.clone(),
-            btc_client.clone(),
+            bridge.clone(),
             executor.clone(),
         ));
 
         self.client.lock().await.start().await?;
 
-        let (notify, recv_queue) = async_channel::unbounded::<(jubjub::SubgroupPoint, u64)>();
+        let (notify, recv_coin) = async_channel::unbounded::<(jubjub::SubgroupPoint, u64)>();
 
         let cashier_client_subscriber_task =
             executor.spawn(Client::connect_to_subscriber_from_cashier(
@@ -101,15 +100,40 @@ impl CashierService {
             ));
 
         let wallet = self.wallet.clone();
+
+        let ex = executor.clone();
         let subscribe_to_withdraw_keys_task = executor.spawn(async move {
             loop {
-                let (pub_key, amount) = recv_queue.recv().await.expect("Receive Own Coin");
+                let bridge = bridge.clone();
+                let bridge_subscribtion  = bridge.subscribe(ex.clone()).await;
+                let (pub_key, amount) = recv_coin.recv().await.expect("Receive Own Coin");
                 debug!(target: "CASHIER DAEMON", "Receive coin with following address and amount: {}, {}", pub_key, amount);
-                let btc_addr = wallet.get_withdraw_coin_public_key_by_dkey_public(&pub_key, &serialize(&1)).expect("Get btc_key by pub_key");
-                if let Some(addr) =  btc_addr {
-                    // TODO send equivalent amount of btc to this address
-                    // then delete this btc_addr from withdraw_keys records
-                    wallet.delete_withdraw_key_record(&addr, &serialize(&1) ).expect("Delete withdraw key record");
+                let coin_addr = wallet.get_withdraw_coin_public_key_by_dkey_public(&pub_key, &serialize(&1))
+                    .expect("Get coin_key by pub_key");
+                if let Some(addr) =  coin_addr {
+                    // send equivalent amount of coin to this address
+                    bridge_subscribtion.sender.send(
+                        bridge::BridgeRequests {
+                            asset_id: 1,
+                            payload: bridge::BridgeRequestsPayload::SendRequest(addr.clone(), amount)
+                        }
+                    ).await.expect("send request to bridge");
+
+                    let res = bridge_subscribtion.receiver.recv().await.expect("bridge resonse");
+
+                    if res.error == 0 {
+                        match res.payload {
+                            bridge::BridgeResponsePayload::SendResponse => {
+                                // then delete this coin addr from withdraw_keys records
+                                wallet.delete_withdraw_key_record(&addr, &serialize(&1) )
+                                    .expect("Delete withdraw key record");
+                            }
+                            _ => {}
+                        }
+
+                    }
+
+
                 }
 
             }
@@ -124,7 +148,7 @@ impl CashierService {
         Ok(())
     }
 
-    async fn _mint_dbtc(&mut self, dkey_pub: jubjub::SubgroupPoint, value: u64) -> Result<()> {
+    async fn _mint_coin(&mut self, dkey_pub: jubjub::SubgroupPoint, value: u64) -> Result<()> {
         self.client
             .lock()
             .await
@@ -137,16 +161,18 @@ impl CashierService {
         send_queue: async_channel::Sender<(PeerId, Reply)>,
         recv_queue: async_channel::Receiver<(PeerId, Request)>,
         wallet: CashierDbPtr,
-        btc_client: Arc<ElectrumClient>,
+        bridge: Arc<bridge::Bridge>,
         executor: Arc<Executor<'_>>,
     ) -> Result<()> {
         loop {
             match recv_queue.recv().await {
                 Ok(msg) => {
+                    let bridge = bridge.clone();
+                    let bridge_subscribtion = bridge.subscribe(executor.clone()).await;
                     let _ = executor
                         .spawn(Self::handle_request(
                             msg,
-                            btc_client.clone(),
+                            bridge_subscribtion,
                             wallet.clone(),
                             send_queue.clone(),
                         ))
@@ -161,7 +187,7 @@ impl CashierService {
     }
     async fn handle_request(
         msg: (PeerId, Request),
-        btc_client: Arc<ElectrumClient>,
+        bridge_subscribtion: bridge::BridgeSubscribtion,
         cashier_wallet: CashierDbPtr,
         send_queue: async_channel::Sender<(PeerId, Reply)>,
     ) -> Result<()> {
@@ -179,41 +205,41 @@ impl CashierService {
                 let _check =
                     cashier_wallet.get_deposit_coin_keys_by_dkey_public(&dpub, &serialize(&1));
 
-                // Generate bitcoin Address
-                let btc_keys = BitcoinKeys::new(btc_client)?;
-
-                let btc_pub = btc_keys.get_pubkey();
-                let btc_priv = btc_keys.get_privkey();
-
-                let _script = btc_keys.get_script();
-
-                // add pairings to db
-                let _result = cashier_wallet.put_exchange_keys(
-                    &dpub,
-                    &btc_priv.to_bytes(),
-                    &btc_pub.to_bytes(),
-                    &serialize(&1),
-                );
-
-                let mut reply = Reply::from(&request, CashierError::NoError as u32, vec![]);
-
-                reply.set_payload(btc_pub.to_bytes());
-
-                // send reply
-                send_queue.send((peer, reply)).await?;
-
-                // start scheduler for checking balance
-                debug!(target: "CASHIER DAEMON", "Subscribing for deposit");
-
-                let _ = btc_keys.start_subscribe().await?;
-
-                //self.mint_dbtc(deserialize(&zkpub).unwrap(), 100);
+                bridge_subscribtion
+                    .sender
+                    .send(bridge::BridgeRequests {
+                        asset_id: 1,
+                        payload: bridge::BridgeRequestsPayload::WatchRequest,
+                    })
+                    .await?;
+
+                let bridge_res = bridge_subscribtion.receiver.recv().await?;
+
+                match bridge_res.payload {
+                    bridge::BridgeResponsePayload::WatchResponse(coin_priv, coin_pub) => {
+                        // add pairings to db
+                        let _result = cashier_wallet.put_exchange_keys(
+                            &dpub,
+                            &coin_priv,
+                            &coin_pub,
+                            &serialize(&1),
+                        );
+
+                        let mut reply = Reply::from(&request, CashierError::NoError as u32, vec![]);
+
+                        reply.set_payload(coin_pub);
+
+                        // send reply
+                        send_queue.send((peer, reply)).await?;
+                    }
+                    _ => {}
+                }
 
                 debug!(target: "CASHIER DAEMON","Waiting for address balance");
             }
             1 => {
                 debug!(target: "CASHIER DAEMON", "Received withdraw request");
-                let btc_address = request.get_payload();
+                let coin_address = request.get_payload();
                 //let btc_address: String = deserialize(&btc_address)?;
                 //let btc_address = bitcoin::util::address::Address::from_str(&btc_address)
                 //   .map_err(|err| crate::Error::from(super::BtcFailed::from(err)))?;
@@ -221,7 +247,7 @@ impl CashierService {
                 let cashier_public: jubjub::SubgroupPoint;
 
                 if let Some(addr) = cashier_wallet
-                    .get_withdraw_keys_by_coin_public_key(&btc_address, &serialize(&1))?
+                    .get_withdraw_keys_by_coin_public_key(&coin_address, &serialize(&1))?
                 {
                     cashier_public = addr.0;
                 } else {
@@ -230,7 +256,7 @@ impl CashierService {
                         zcash_primitives::constants::SPENDING_KEY_GENERATOR * cashier_secret;
 
                     cashier_wallet.put_withdraw_keys(
-                        &btc_address,
+                        &coin_address,
                         &cashier_public,
                         &cashier_secret,
                         &serialize(&1),
@@ -269,13 +295,16 @@ impl CashierClient {
         Ok(())
     }
 
-    pub async fn withdraw(&mut self, btc_address: String) -> Result<Option<jubjub::SubgroupPoint>> {
+    pub async fn withdraw(
+        &mut self,
+        coin_address: String,
+    ) -> Result<Option<jubjub::SubgroupPoint>> {
         let handle_error = Arc::new(handle_error);
         let rep = self
             .protocol
             .request(
                 CashierCommand::Withdraw as u8,
-                serialize(&btc_address),
+                serialize(&coin_address),
                 handle_error,
             )
             .await?;
@@ -287,10 +316,7 @@ impl CashierClient {
         Ok(None)
     }
 
-    pub async fn get_address(
-        &mut self,
-        index: jubjub::SubgroupPoint,
-    ) -> Result<Option<PubAddress>> {
+    pub async fn get_address(&mut self, index: jubjub::SubgroupPoint) -> Result<Option<Vec<u8>>> {
         let handle_error = Arc::new(handle_error);
         let rep = self
             .protocol
@@ -302,8 +328,7 @@ impl CashierClient {
             .await?;
 
         if let Some(key) = rep {
-            let address = BitcoinKeys::address_from_slice(&key)?;
-            return Ok(Some(address));
+            return Ok(Some(key));
         }
         Ok(None)
     }

src/service/mod.rs

@@ -1,11 +1,17 @@
 pub mod cashier;
 pub mod gateway;
 pub mod reqrep;
+pub mod bridge;
 
+#[cfg(feature = "default")]
 pub mod btc;
+#[cfg(feature = "default")]
+pub use btc::{BitcoinKeys, PubAddress, BtcFailed, BtcResult};
+
+#[cfg(feature = "sol")]
+pub mod sol;
 
 pub use gateway::{GatewayClient, GatewayService, GatewaySlabsSubscriber};
 
 pub use cashier::{CashierClient, CashierService};
 
-pub use btc::{BitcoinKeys, PubAddress, BtcFailed, BtcResult};