Bridge: change struct name of TokenClient to NetworkClient & minor

changes

src/bin/sol-test → src/bin/sol-test.rs

@@ -42,7 +42,7 @@ async fn run() -> Result<()> {
 
     let bridge_res = bridge_subscribtion.receiver.recv().await?;
 
-    // XXX this will not work 
+    // XXX this will not work
     match bridge_res.payload {
         bridge::BridgeResponsePayload::Watch(_, token_pub) => {
             println!("watch this address {}", token_pub);
@@ -57,19 +57,22 @@ async fn run() -> Result<()> {
 
 fn main() -> Result<()> {
     #[cfg(feature = "sol")]
-    let args = clap_app!(darkfid =>
-        (@arg verbose: -v --verbose "Increase verbosity")
-    )
-    .get_matches();
-
-    let loglevel = if args.is_present("verbose") {
-        log::Level::Debug
-    } else {
-        log::Level::Info
-    };
-
-    simple_logger::init_with_level(loglevel)?;
+    {
+        let args = clap_app!(darkfid =>
+            (@arg verbose: -v --verbose "Increase verbosity")
+        )
+        .get_matches();
+
+        let loglevel = if args.is_present("verbose") {
+            log::Level::Debug
+        } else {
+            log::Level::Info
+        };
+
+        simple_logger::init_with_level(loglevel)?;
+    }
 
+    #[cfg(feature = "sol")]
     smol::block_on(run())?;
     Ok(())
 }

src/service/bridge.rs

@@ -49,7 +49,7 @@ pub struct TokenNotification {
 }
 
 pub struct Bridge {
-    clients: Mutex<HashMap<String, Arc<dyn TokenClient + Send + Sync>>>,
+    clients: Mutex<HashMap<String, Arc<dyn NetworkClient + Send + Sync>>>,
     //notifiers: Mutex<HashMap<Vec<u8>, async_channel::Receiver<TokenNotification>>>,
 }
 
@@ -64,7 +64,7 @@ impl Bridge {
     pub async fn add_clients(
         self: Arc<Self>,
         network: String,
-        client: Arc<dyn TokenClient + Send + Sync>,
+        client: Arc<dyn NetworkClient + Send + Sync>,
     ) -> Result<()> {
         //let notifier = client.get_notifier().await?;
 
@@ -97,6 +97,7 @@ impl Bridge {
         rep: async_channel::Sender<BridgeResponse>,
     ) -> Result<()> {
         let req = req.recv().await?;
+
         let network = req.network;
 
         if !self.clients.lock().await.contains_key(&network) {
@@ -108,7 +109,13 @@ impl Bridge {
             return Ok(());
         }
 
-        let client = &self.clients.lock().await[&network];
+
+        let client: Arc<dyn NetworkClient + Send + Sync>;
+        // avoid deadlock
+        {
+            let c = &self.clients.lock().await[&network];
+            client = c.clone();
+        }
 
         match req.payload {
             BridgeRequestsPayload::Watch(val) => match val {
@@ -146,17 +153,17 @@ impl Bridge {
 }
 
 #[async_trait]
-pub trait TokenClient {
-    async fn subscribe(&self) -> Result<TokenSubscribtion>;
+pub trait NetworkClient {
+    async fn subscribe(self: Arc<Self>) -> Result<TokenSubscribtion>;
 
     // should check if the keypair in not already subscribed
     async fn subscribe_with_keypair(
-        &self,
+        self: Arc<Self>,
         private_key: Vec<u8>,
         public_key: Vec<u8>,
     ) -> Result<String>;
 
-    async fn get_notifier(&self) -> Result<async_channel::Receiver<TokenNotification>>;
+    async fn get_notifier(self: Arc<Self>) -> Result<async_channel::Receiver<TokenNotification>>;
 
-    async fn send(&self, address: Vec<u8>, amount: u64) -> Result<()>;
+    async fn send(self: Arc<Self>, address: Vec<u8>, amount: u64) -> Result<()>;
 }

src/service/btc.rs

@@ -1,4 +1,4 @@
-use super::bridge::{TokenClient, TokenNotification, TokenSubscribtion};
+use super::bridge::{NetworkClient, TokenNotification, TokenSubscribtion};
 use crate::serial::{serialize, Decodable, Encodable};
 use crate::Result;
 
@@ -155,8 +155,8 @@ impl BtcClient {
 }
 
 #[async_trait]
-impl TokenClient for BtcClient {
-    async fn subscribe(&self) -> Result<TokenSubscribtion> {
+impl NetworkClient for BtcClient {
+    async fn subscribe(self: Arc<Self>) -> Result<TokenSubscribtion> {
         //// Generate bitcoin Address
         let btc_keys = BitcoinKeys::new(self.client.clone(), self.network)?;
 
@@ -178,7 +178,7 @@ impl TokenClient for BtcClient {
     }
 
     async fn subscribe_with_keypair(
-        &self,
+        self: Arc<Self>,
         _private_key: Vec<u8>,
         _public_key: Vec<u8>,
     ) -> Result<String> {
@@ -186,12 +186,12 @@ impl TokenClient for BtcClient {
         Ok(String::new())
     }
 
-    async fn get_notifier(&self) -> Result<async_channel::Receiver<TokenNotification>> {
+    async fn get_notifier(self: Arc<Self>) -> Result<async_channel::Receiver<TokenNotification>> {
         // TODO this not implemented yet
         let (_, notifier) = async_channel::unbounded();
         Ok(notifier)
     }
-    async fn send(&self, _address: Vec<u8>, _amount: u64) -> Result<()> {
+    async fn send(self: Arc<Self>, _address: Vec<u8>, _amount: u64) -> Result<()> {
         // TODO this not implemented yet
         Ok(())
     }

src/service/sol.rs

@@ -5,7 +5,7 @@ use crate::rpc::{
 use crate::serial::{deserialize, serialize, Decodable, Encodable};
 use crate::{Error, Result};
 
-use super::bridge::{TokenClient, TokenNotification, TokenSubscribtion};
+use super::bridge::{NetworkClient, TokenNotification, TokenSubscribtion};
 
 use async_native_tls::TlsConnector;
 use async_std::sync::{Arc, Mutex};
@@ -79,7 +79,7 @@ impl SolClient {
         let v: std::collections::HashMap<String, Value> =
             serde_json::from_str(&data).map_err(|err| Error::from(err))?;
 
-        // XXX this for testing 
+        // XXX this for testing
         if v.contains_key(&String::from("result")) {
             json_res = JsonResult::Resp(JsonResponse {
                 jsonrpc: v["jsonrpc"].clone(),
@@ -170,7 +170,7 @@ impl SolClient {
 
                         self.unsubscribe(sub_id, &keypair.pubkey()).await?;
 
-                        self.send_to_main_account(&keypair, new_bal)?;
+                        //self.send_to_main_account(&keypair, new_bal)?;
 
                         debug!(
                             target: "SOL BRIDGE",
@@ -187,7 +187,7 @@ impl SolClient {
         Ok(())
     }
 
-    fn send_to_main_account(&self, keypair: &Keypair, amount: u64) -> SolResult<()> {
+    pub fn send_to_main_account(&self, keypair: &Keypair, amount: u64) -> SolResult<()> {
         debug!(
             target: "SOL BRIDGE",
             "send received token to main account"
@@ -207,7 +207,7 @@ impl SolClient {
         Ok(())
     }
 
-    async fn handle_subscribe_request(&self, keypair: Keypair) -> Result<()> {
+    async fn handle_subscribe_request(self: Arc<Self>, keypair: Keypair) -> Result<()> {
         debug!(
             target: "SOL BRIDGE",
             "Handle subscribe request"
@@ -299,14 +299,15 @@ impl SolClient {
 }
 
 #[async_trait]
-impl TokenClient for SolClient {
-    async fn subscribe(&self) -> Result<TokenSubscribtion> {
+impl NetworkClient for SolClient {
+    async fn subscribe(self: Arc<Self>) -> Result<TokenSubscribtion> {
         let keypair = Keypair::generate(&mut OsRng);
 
         let public_key = keypair.pubkey().to_string();
         let secret_key = serialize(&keypair);
 
-        self.handle_subscribe_request(keypair).await?;
+        let self2 = self.clone();
+        smol::spawn(self2.handle_subscribe_request(keypair)).detach();
 
         Ok(TokenSubscribtion {
             secret_key,
@@ -316,7 +317,7 @@ impl TokenClient for SolClient {
 
     // in solana case private key it's the same as keypair
     async fn subscribe_with_keypair(
-        &self,
+        self: Arc<Self>,
         private_key: Vec<u8>,
         _public_key: Vec<u8>,
     ) -> Result<String> {
@@ -324,16 +325,17 @@ impl TokenClient for SolClient {
 
         let public_key = keypair.pubkey().to_string();
 
-        self.handle_subscribe_request(keypair).await?;
+        let self2 = self.clone();
+        smol::spawn(self2.handle_subscribe_request(keypair)).detach();
 
         Ok(public_key)
     }
 
-    async fn get_notifier(&self) -> Result<async_channel::Receiver<TokenNotification>> {
+    async fn get_notifier(self: Arc<Self>) -> Result<async_channel::Receiver<TokenNotification>> {
         Ok(self.notify_channel.1.clone())
     }
 
-    async fn send(&self, address: Vec<u8>, amount: u64) -> Result<()> {
+    async fn send(self: Arc<Self>, address: Vec<u8>, amount: u64) -> Result<()> {
         let rpc = RpcClient::new(RPC_SERVER.to_string());
         let address: Pubkey = deserialize(&address)?;
         let instruction = system_instruction::transfer(&self.keypair.pubkey(), &address, amount);