diff options
Diffstat (limited to 'src/peering/basalt.rs')
| -rw-r--r-- | src/peering/basalt.rs | 19 |
1 files changed, 11 insertions, 8 deletions
diff --git a/src/peering/basalt.rs b/src/peering/basalt.rs index efbf6e6..cdb0605 100644 --- a/src/peering/basalt.rs +++ b/src/peering/basalt.rs @@ -11,6 +11,8 @@ use serde::{Deserialize, Serialize}; use sodiumoxide::crypto::hash; +use tokio::sync::watch; + use crate::endpoint::*; use crate::netapp::*; use crate::proto::*; @@ -303,18 +305,19 @@ impl Basalt { .collect::<Vec<_>>() } - pub async fn run(self: Arc<Self>) { + pub async fn run(self: Arc<Self>, must_exit: watch::Receiver<bool>) { for peer in self.bootstrap_peers.iter() { tokio::spawn(self.clone().try_connect(*peer)); } - let pushpull_loop = self.clone().run_pushpull_loop(); - let reset_loop = self.run_reset_loop(); - tokio::join!(pushpull_loop, reset_loop); + tokio::join!( + self.clone().run_pushpull_loop(must_exit.clone()), + self.clone().run_reset_loop(must_exit.clone()), + ); } - async fn run_pushpull_loop(self: Arc<Self>) { - loop { + async fn run_pushpull_loop(self: Arc<Self>, must_exit: watch::Receiver<bool>) { + while !*must_exit.borrow() { tokio::time::sleep(self.param.exchange_interval).await; let peers = self.view.read().unwrap().sample(2); @@ -360,8 +363,8 @@ impl Basalt { } } - async fn run_reset_loop(self: Arc<Self>) { - loop { + async fn run_reset_loop(self: Arc<Self>, must_exit: watch::Receiver<bool>) { + while !*must_exit.borrow() { tokio::time::sleep(self.param.reset_interval).await; { |