Bridge: WIP refactroing the old functions & add new functions and types

src/service/bridge.rs

@@ -18,7 +18,7 @@ pub struct BridgeResponse {
 }
 
 pub enum BridgeRequestsPayload {
-    SendRequest(Vec<u8>, u64),
+    SendRequest(Vec<u8>, u64), // send (address, amount)
     WatchRequest,
 }
 
@@ -32,13 +32,26 @@ pub struct BridgeSubscribtion {
     pub receiver: async_channel::Receiver<BridgeResponse>,
 }
 
+pub struct CoinSubscribtion {
+    pub secret_key: Vec<u8>,
+    pub public_key: Vec<u8>,
+}
+
+pub struct CoinNotification {
+    pub secret_key: Vec<u8>,
+    pub received_balance: u64,
+}
+
 pub struct Bridge {
     clients: Mutex<HashMap<Vec<u8>, Arc<dyn CoinClient + Send + Sync>>>,
+    notifiers: Mutex<HashMap<Vec<u8>, async_channel::Receiver<CoinNotification>>>,
 }
+
 impl Bridge {
     pub fn new() -> Arc<Self> {
         Arc::new(Self {
             clients: Mutex::new(HashMap::new()),
+            notifiers: Mutex::new(HashMap::new()),
         })
     }
 
@@ -46,11 +59,21 @@ impl Bridge {
         self: Arc<Self>,
         asset_id: jubjub::Fr,
         client: Arc<dyn CoinClient + Send + Sync>,
-    ) {
+    ) -> Result<()> {
         let asset_id = serialize(&asset_id);
-        self.clients.lock().await.insert(asset_id, client);
+
+        let notifier = client.get_notifier().await?;
+
+        self.clients.lock().await.insert(asset_id.clone(), client);
+        self.notifiers
+            .lock()
+            .await
+            .insert(asset_id, notifier.clone());
+        Ok(())
     }
 
+    pub async fn listen(self: Arc<Self>) {}
+
     pub async fn subscribe(self: Arc<Self>, executor: Arc<Executor<'_>>) -> BridgeSubscribtion {
         let (sender, req) = async_channel::unbounded();
         let (rep, receiver) = async_channel::unbounded();
@@ -62,7 +85,7 @@ impl Bridge {
         BridgeSubscribtion { sender, receiver }
     }
 
-    pub async fn listen_for_new_subscribtion(
+    async fn listen_for_new_subscribtion(
         self: Arc<Self>,
         req: async_channel::Receiver<BridgeRequests>,
         rep: async_channel::Sender<BridgeResponse>,
@@ -73,10 +96,10 @@ impl Bridge {
 
         match req.payload {
             BridgeRequestsPayload::WatchRequest => {
-                let (private, public) = client.watch().await?;
+                let sub = client.subscribe().await?;
                 let res = BridgeResponse {
                     error: 0,
-                    payload: BridgeResponsePayload::WatchResponse(private, public),
+                    payload: BridgeResponsePayload::WatchResponse(sub.secret_key, sub.public_key),
                 };
                 rep.send(res).await?;
             }
@@ -96,7 +119,7 @@ impl Bridge {
 
 #[async_trait]
 pub trait CoinClient {
-    // return private and public keys that be watching
-    async fn watch(&self) -> Result<(Vec<u8>, Vec<u8>)>;
+    async fn subscribe(&self) -> Result<CoinSubscribtion>;
+    async fn get_notifier(&self) -> Result<async_channel::Receiver<CoinNotification>>;
     async fn send(&self, address: Vec<u8>, amount: u64) -> Result<()>;
 }

src/service/btc.rs

@@ -1,13 +1,13 @@
-use super::bridge::CoinClient;
-use crate::serial::{Decodable, Encodable, serialize};
+use super::bridge::{CoinClient, CoinNotification, CoinSubscribtion};
+use crate::serial::{serialize, Decodable, Encodable};
 use crate::Result;
 
 use async_trait::async_trait;
 use bitcoin::blockdata::script::Script;
+use bitcoin::hash_types::Txid;
 use bitcoin::network::constants::Network;
 use bitcoin::util::address::Address;
 use bitcoin::util::ecdsa::{PrivateKey, PublicKey};
-use bitcoin::hash_types::Txid;
 use electrum_client::{Client as ElectrumClient, ElectrumApi};
 use log::*;
 use rand::distributions::Alphanumeric;
@@ -33,10 +33,7 @@ pub struct BitcoinKeys {
 }
 
 impl BitcoinKeys {
-    pub fn new(
-        btc_client: Arc<ElectrumClient>,
-        network: Network,
-    ) -> Result<Arc<BitcoinKeys>> {
+    pub fn new(btc_client: Arc<ElectrumClient>, network: Network) -> Result<Arc<BitcoinKeys>> {
         let context = secp256k1::Secp256k1::new();
 
         // Probably not good enough for release
@@ -138,17 +135,15 @@ impl BitcoinKeys {
     pub fn get_script(&self) -> &Script {
         &self.script
     }
-
 }
 
-
 pub struct BtcClient {
     client: Arc<ElectrumClient>,
     network: Network,
 }
 
 impl BtcClient {
-    pub fn new( btc_endpoint: (Network, String) ) -> Result<Self> {
+    pub fn new(btc_endpoint: (Network, String)) -> Result<Self> {
         let (network, client_address) = btc_endpoint;
         let client = ElectrumClient::new(&client_address)
             .map_err(|err| crate::Error::from(super::BtcFailed::from(err)))?;
@@ -161,7 +156,7 @@ impl BtcClient {
 
 #[async_trait]
 impl CoinClient for BtcClient {
-    async fn watch(&self) -> Result<(Vec<u8>, Vec<u8>)> {
+    async fn subscribe(&self) -> Result<CoinSubscribtion> {
         //// Generate bitcoin Address
         let btc_keys = BitcoinKeys::new(self.client.clone(), self.network)?;
 
@@ -176,9 +171,17 @@ impl CoinClient for BtcClient {
         let (_txid, _balance) = btc_keys.start_subscribe().await?;
         //let _script = btc_keys.get_script();
 
-        Ok((serialize(&btc_priv.to_bytes()), serialize(&btc_pub.to_bytes())))
+        Ok(CoinSubscribtion {
+            secret_key: serialize(&btc_priv.to_bytes()),
+            public_key: serialize(&btc_pub.to_bytes()),
+        })
     }
 
+    async fn get_notifier(&self) -> Result<async_channel::Receiver<CoinNotification>> {
+        // TODO this not implemented yet
+        let (_, notifier) = async_channel::unbounded();
+        Ok(notifier)
+    }
     async fn send(&self, _address: Vec<u8>, _amount: u64) -> Result<()> {
         // TODO
 
@@ -237,7 +240,6 @@ impl Decodable for bitcoin::PrivateKey {
     }
 }
 
-
 #[derive(Debug)]
 pub enum BtcFailed {
     NotEnoughValue(u64),
@@ -259,7 +261,9 @@ impl std::fmt::Display for BtcFailed {
                 write!(f, "Unable to create Electrum Client: {}", err)
             }
             BtcFailed::ElectrumError(ref err) => write!(f, "could not parse BTC address: {}", err),
-            BtcFailed::DecodeAndEncodeError(ref err) => write!(f, "Decode and decode keys error: {}", err),
+            BtcFailed::DecodeAndEncodeError(ref err) => {
+                write!(f, "Decode and decode keys error: {}", err)
+            }
             BtcFailed::BtcError(i) => {
                 write!(f, "BtcFailed: {}", i)
             }
@@ -267,7 +271,6 @@ impl std::fmt::Display for BtcFailed {
     }
 }
 
-
 impl From<crate::error::Error> for BtcFailed {
     fn from(err: crate::error::Error) -> BtcFailed {
         BtcFailed::BtcError(err.to_string())

src/service/sol.rs

@@ -2,7 +2,7 @@ use crate::rpc::{jsonrpc, jsonrpc::JsonResult};
 use crate::serial::{deserialize, serialize, Decodable, Encodable};
 use crate::{Error, Result};
 
-use super::bridge::CoinClient;
+use super::bridge::{ CoinSubscribtion, CoinNotification, CoinClient};
 
 use async_trait::async_trait;
 
@@ -40,17 +40,15 @@ struct SubscribeParams {
 pub struct SolClient {
     keypair: Keypair,
 
-    // subscription hashmap using pubkey as an index
-    subscriptions: Arc<Mutex<HashMap<String, (Vec<u8>, u64)>>>,
+    // subscriptions hashmap using pubkey as an index and a value of (keypair, amount)
+    subscriptions: Arc<Mutex<HashMap<Pubkey, (Keypair, u64)>>>,
 
-    // notify when get new update
     notify_channel: (
-        async_channel::Sender<(Vec<u8>, u64)>,
-        async_channel::Receiver<(Vec<u8>, u64)>,
+        async_channel::Sender<CoinNotification>,
+        async_channel::Receiver<CoinNotification>,
     ),
 
-    // send subscription msg to websocket
-    watch_channel: (
+    subscribe_channel: (
         async_channel::Sender<jsonrpc::JsonRequest>,
         async_channel::Receiver<jsonrpc::JsonRequest>,
     ),
@@ -61,22 +59,16 @@ impl SolClient {
         let keypair: Keypair = deserialize(&keypair)?;
 
         let notify_channel = async_channel::unbounded();
-        let watch_channel = async_channel::unbounded();
+        let subscribe_channel = async_channel::unbounded();
 
         Ok(Arc::new(Self {
             keypair,
             subscriptions: Arc::new(Mutex::new(HashMap::new())),
             notify_channel,
-            watch_channel,
+            subscribe_channel,
         }))
     }
 
-    pub async fn subscribe_to_notify_channel(
-        self: Arc<Self>,
-    ) -> Result<async_channel::Receiver<(Vec<u8>, u64)>> {
-        Ok(self.notify_channel.1.clone())
-    }
-
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> SolResult<()> {
         // WebSocket handshake/connect
         let (ws_stream, _) = connect_async(WSS_SERVER).await?;
@@ -86,8 +78,10 @@ impl SolClient {
         let self2 = self.clone();
         let _: async_executor::Task<Result<()>> = executor.spawn(async move {
             loop {
-                let sub_msg = self2.watch_channel.1.recv().await?;
+                // recv a request for websocket
+                let sub_msg = self2.subscribe_channel.1.recv().await?;
 
+                // write the request to websocket
                 write
                     .send(Message::Text(serde_json::to_string(&sub_msg)?))
                     .await
@@ -96,8 +90,8 @@ impl SolClient {
         });
 
         read.for_each(|message| async {
-            let self2 = self.clone();
-            self2
+            // read ws msg
+            self.clone()
                 .read_ws_msg(message)
                 .await
                 .expect("read from websocket");
@@ -116,22 +110,26 @@ impl SolClient {
 
         match v {
             JsonResult::Resp(r) => {
-                if let Some(sub_id) = r.result.as_i64() {
-                    debug!(
-                        target: "SOL BRIDGE",
-                        "Successfully get response : {:?}",
-                        sub_id
-                    );
-                }
+                // receive a response with subscription id
+                let sub_id = r.result.as_i64().ok_or(Error::ParseIntError)?;
+                debug!(
+                    target: "SOL BRIDGE",
+                    "Successfully get response : {:?}",
+                    sub_id
+                );
             }
 
             JsonResult::Err(e) => {
+                // receive an error
                 debug!(
                         target: "SOL BRIDGE",
                         "Error on subscription: {:?}", e.error.message.to_string());
             }
 
             JsonResult::Notif(n) => {
+                // receive notification once an account get updated
+
+                // get values from the notification
                 let new_bal = n.params["result"]["value"]["lamports"]
                     .as_u64()
                     .ok_or(Error::ParseIntError)?;
@@ -139,45 +137,46 @@ impl SolClient {
                 let owner_pubkey = n.params["result"]["value"]["owner"]
                     .as_str()
                     .ok_or(Error::ParseFailed("Error Parse serde_json Value to &str"))?;
-
-                let (keypair, old_balance) = self.subscriptions.lock().await[owner_pubkey].clone();
+                
+                let owner_pubkey: Pubkey = Pubkey::from_str(&owner_pubkey)?;
 
                 let sub_id = n.params["subscription"]
                     .as_u64()
                     .ok_or(Error::ParseIntError)?;
 
-                if new_bal > old_balance {
-                    let received_balance = new_bal - old_balance;
+                // get the keypair and old_balance from the subscriptions list
+                let (keypair, old_balance) = &self.subscriptions.lock().await[&owner_pubkey];
 
-                    let keypair: Keypair = deserialize(&keypair)?;
+                if new_bal > old_balance.to_owned() {
+                    let received_balance = new_bal - old_balance;
 
-                    self.send_to_main_account(keypair)?;
+                    self.send_to_main_account(&keypair)?;
 
                     self.notify_channel
                         .0
-                        .send((
-                            serialize(&Pubkey::from_str(owner_pubkey)?),
+                        .send(CoinNotification {
+                            secret_key: serialize(keypair),
                             received_balance,
-                        ))
+                        })
                         .await
                         .map_err(|err| Error::from(err))?;
 
-                    SolClient::unsubscribe(self.watch_channel.0.clone(), sub_id).await?;
+                    self.unsubscribe(sub_id, &owner_pubkey).await?;
 
                     debug!(
                         target: "SOL BRIDGE",
                         "Received {} lamports, to the pubkey: {} ",
                         received_balance, owner_pubkey.to_string(),
                     );
-                } else if new_bal < old_balance {
-                    SolClient::unsubscribe(self.watch_channel.0.clone(), sub_id).await?;
+                } else if new_bal < old_balance.to_owned() {
+                    self.unsubscribe(sub_id, &owner_pubkey).await?;
                 }
             }
         }
         Ok(())
     }
 
-    fn send_to_main_account(&self, keypair: Keypair) -> SolResult<()> {
+    fn send_to_main_account(&self, keypair: &Keypair) -> SolResult<()> {
         let rpc = RpcClient::new(RPC_SERVER.to_string());
 
         let amount = rpc.get_balance(&keypair.pubkey())?;
@@ -189,25 +188,23 @@ impl SolClient {
         let bhq = BlockhashQuery::default();
         match bhq.get_blockhash_and_fee_calculator(&rpc, rpc.commitment()) {
             Err(_) => panic!("Couldn't connect to RPC"),
-            Ok(v) => tx.sign(&[&keypair], v.0),
+            Ok(v) => tx.sign(&[keypair], v.0),
         }
         let _signature = rpc.send_and_confirm_transaction(&tx)?;
         Ok(())
     }
 
-    async fn unsubscribe(
-        watch_channel_sender: async_channel::Sender<jsonrpc::JsonRequest>,
-        sub_id: u64,
-    ) -> Result<()> {
+    async fn unsubscribe(&self, sub_id: u64, pubkey: &Pubkey) -> Result<()> {
         let sub_msg = jsonrpc::request(json!("accountUnsubscribe"), json!([json!(sub_id)]));
-        watch_channel_sender.send(sub_msg).await?;
+        self.subscribe_channel.0.send(sub_msg).await?;
+        self.subscriptions.lock().await.remove(pubkey);
         Ok(())
     }
 }
 
 #[async_trait]
 impl CoinClient for SolClient {
-    async fn watch(&self) -> Result<(Vec<u8>, Vec<u8>)> {
+    async fn subscribe(&self) -> Result<CoinSubscribtion> {
         let keypair = Keypair::generate(&mut OsRng);
 
         // Parameters for subscription to events related to `pubkey`.
@@ -227,16 +224,24 @@ impl CoinClient for SolClient {
             .get_balance(&keypair.pubkey())
             .map_err(|err| SolFailed::from(err))?;
 
+        let public_key = serialize(&keypair.pubkey());
+        // NOTE we send keypair for sol as secret_key
+        let secret_key = serialize(&keypair);
+
+        // add to subscriptions list
         self.subscriptions
             .lock()
             .await
-            .insert(keypair.pubkey().to_string(), (serialize(&keypair), balance));
+            .insert(keypair.pubkey(), (keypair, balance));
+
+        //  send
+        self.subscribe_channel.0.send(sub_msg).await?;
 
-        self.watch_channel.0.send(sub_msg).await?;
+        Ok(CoinSubscribtion { secret_key, public_key})
+    }
 
-        let pubkey = serialize(&keypair.pubkey());
-        let keypair = serialize(&keypair);
-        Ok((pubkey, keypair))
+    async fn get_notifier(&self) -> Result<async_channel::Receiver<CoinNotification>>{
+        Ok(self.notify_channel.1.clone())
     }
 
     async fn send(&self, address: Vec<u8>, amount: u64) -> Result<()> {