Merge pull request #58 from nymtech/bugfix/presence_client_crash

Bugfix/presence client crash

mixnode/Cargo.toml

@@ -12,6 +12,7 @@ base64 = "0.11.0"
 clap = "2.33.0"
 curve25519-dalek = "1.2.3"
 futures = "0.3.1"
+log = "0.4.8"
 tokio = { version = "0.2", features = ["full"] }
 
 ## internal

mixnode/src/node/metrics.rs

@@ -4,6 +4,7 @@ use directory_client::DirectoryClient;
 use futures::channel::mpsc;
 use futures::lock::Mutex;
 use futures::StreamExt;
+use log::{debug, error};
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
@@ -80,14 +81,15 @@ impl MetricsReporter {
             tokio::time::delay_for(delay_duration).await;
             let (received, sent) =
                 MetricsReporter::acquire_and_reset_metrics(metrics.clone()).await;
-            directory_client
-                .metrics_post
-                .post(&MixMetric {
-                    pub_key: pub_key_str.clone(),
-                    received,
-                    sent,
-                })
-                .unwrap();
+
+            match directory_client.metrics_post.post(&MixMetric {
+                pub_key: pub_key_str.clone(),
+                received,
+                sent,
+            }) {
+                Err(err) => error!("failed to send metrics - {:?}", err),
+                Ok(_) => debug!("sent metrics information"),
+            }
         }
     }
 }

mixnode/src/node/mod.rs

@@ -1,4 +1,5 @@
 use crate::mix_peer::MixPeer;
+use crate::node;
 use crate::node::metrics::MetricsReporter;
 use curve25519_dalek::montgomery::MontgomeryPoint;
 use curve25519_dalek::scalar::Scalar;
@@ -214,7 +215,7 @@ impl MixNode {
         }
     }
 
-    pub fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
+    pub fn start(&self, config: node::Config) -> Result<(), Box<dyn std::error::Error>> {
         // Create the runtime, probably later move it to MixNode itself?
         let mut rt = Runtime::new()?;
 
@@ -227,6 +228,11 @@ impl MixNode {
         let pub_key_str =
             base64::encode_config(&self.public_key.to_bytes().to_vec(), base64::URL_SAFE);
 
+        rt.spawn({
+            let presence_notifier = presence::Notifier::new(&config);
+            presence_notifier.run()
+        });
+
         let metrics = MetricsReporter::new().add_arc_mutex();
         rt.spawn(MetricsReporter::run_received_metrics_control(
             metrics.clone(),

mixnode/src/node/presence.rs

@@ -2,7 +2,7 @@ use crate::node;
 use directory_client::presence::MixNodePresence;
 use directory_client::requests::presence_mixnodes_post::PresenceMixNodesPoster;
 use directory_client::DirectoryClient;
-use std::thread;
+use log::{debug, error};
 use std::time::Duration;
 
 pub struct Notifier {
@@ -30,16 +30,18 @@ impl Notifier {
     }
 
     pub fn notify(&self) {
-        self.net_client
-            .presence_mix_nodes_post
-            .post(&self.presence)
-            .unwrap();
+        match self.net_client.presence_mix_nodes_post.post(&self.presence) {
+            Err(err) => error!("failed to send presence - {:?}", err),
+            Ok(_) => debug!("sent presence information"),
+        }
     }
 
-    pub fn run(&self) {
+    pub async fn run(self) {
+        let delay_duration = Duration::from_secs(5);
+
         loop {
             self.notify();
-            thread::sleep(Duration::from_secs(5));
+            tokio::time::delay_for(delay_duration).await;
         }
     }
 }

mixnode/src/node/runner.rs

@@ -1,11 +1,8 @@
 use crate::banner;
 use crate::node;
-use crate::node::presence;
 use crate::node::MixNode;
 use clap::ArgMatches;
-
 use std::net::ToSocketAddrs;
-use std::thread;
 
 fn print_binding_warning(address: &str) {
     println!("\n##### WARNING #####");
@@ -29,11 +26,7 @@ pub fn start(matches: &ArgMatches) {
     );
 
     let mix = MixNode::new(&config);
-    thread::spawn(move || {
-        let notifier = presence::Notifier::new(&config);
-        notifier.run();
-    });
-    mix.start().unwrap();
+    mix.start(config).unwrap();
 }
 
 fn new_config(matches: &ArgMatches) -> node::Config {

sfw-provider/Cargo.toml

@@ -13,6 +13,7 @@ clap = "2.33.0"
 curve25519-dalek = "1.2.3"
 hex = "0.4.0"
 futures = "0.3.1"
+log = "0.4.8"
 rand = "0.7.2"
 tokio = { version = "0.2.4", features = ["full"] }
 sha2 = "0.8.0"

sfw-provider/src/provider/presence.rs

@@ -4,6 +4,7 @@ use directory_client::presence::MixProviderPresence;
 use directory_client::requests::presence_providers_post::PresenceMixProviderPoster;
 use directory_client::DirectoryClient;
 use futures::lock::Mutex as FMutex;
+use log::{debug, error};
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
@@ -52,10 +53,10 @@ impl Notifier {
     }
 
     pub fn notify(&self, presence: MixProviderPresence) {
-        self.net_client
-            .presence_providers_post
-            .post(&presence)
-            .unwrap();
+        match self.net_client.presence_providers_post.post(&presence) {
+            Err(err) => error!("failed to send presence - {:?}", err),
+            Ok(_) => debug!("sent presence information"),
+        }
     }
 
     pub async fn run(self) {