aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorQuentin Dufour <quentin@deuxfleurs.fr>2020-05-22 12:25:44 +0200
committerQuentin Dufour <quentin@deuxfleurs.fr>2020-05-22 12:25:44 +0200
commitefbbc17f82b3a2af80ee4e655528fff09f7ca401 (patch)
treedcc78401d28a36dcc74ddfb3be9402c0699f1571 /src
parentfe1d91fe06e314faa54997638f11ae6afb57b0b5 (diff)
downloaddiplonat-efbbc17f82b3a2af80ee4e655528fff09f7ca401.tar.gz
diplonat-efbbc17f82b3a2af80ee4e655528fff09f7ca401.zip
WIP consul actor
Diffstat (limited to 'src')
-rw-r--r--src/consul_actor.rs44
-rw-r--r--src/main.rs1
-rw-r--r--src/messages.rs4
3 files changed, 45 insertions, 4 deletions
diff --git a/src/consul_actor.rs b/src/consul_actor.rs
index 0f09851..c87874d 100644
--- a/src/consul_actor.rs
+++ b/src/consul_actor.rs
@@ -1,15 +1,51 @@
use crate::consul::Consul;
use tokio::sync::watch;
+use tokio::time::delay_for;
+use crate::messages;
+use anyhow::Result;
+use std::cmp;
+use std::time::Duration;
+use log::*;
pub struct ConsulActor {
- consul: Consul
- out:
+ pub rx_open_ports: watch::Receiver<messages::OpenPorts>,
+
+ consul: Consul,
+ node: String,
+ retries: u32,
+ tx_open_ports: watch::Sender<messages::OpenPorts>
+}
+
+fn retry_to_time(retries: u32, max_time: Duration) -> Duration {
+ // 1.2^x seems to be a good value to exponentially increase time at a good pace
+ // eg. 1.2^32 = 341 seconds ~= 5 minutes - ie. after 32 retries we wait 5 minutes
+ return Duration::from_secs(cmp::max(max_time.as_secs(), 1.2f64.powf(retries as f64) as u64))
}
impl ConsulActor {
- fn new(url: &str) -> Self {
+ fn new(url: &str, node: &str) -> Self {
+ let (tx, rx) = watch::channel(messages::OpenPorts{ports: Vec::new() });
return Self {
- consul: Consul::new(url);
+ consul: Consul::new(url),
+ rx_open_ports: rx,
+ tx_open_ports: tx,
+ node: node.to_string(),
+ retries: 0,
};
}
+
+ async fn listen(&mut self) -> Result<()> {
+ loop {
+ let catalog = match self.consul.watch_node(&self.node).await {
+ Ok(c) => c,
+ Err(e) => {
+ self.retries = cmp::min(u32::MAX - 1, self.retries) + 1;
+ let will_retry_in = retry_to_time(self.retries, Duration::from_secs(600));
+ error!("Failed to query consul. Will retry in {}s. {}", will_retry_in.as_secs(), e);
+ delay_for(will_retry_in).await;
+ continue;
+ }
+ };
+ }
+ }
}
diff --git a/src/main.rs b/src/main.rs
index 844bff9..4fee661 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,3 +1,4 @@
+mod messages;
mod consul;
mod consul_actor;
diff --git a/src/messages.rs b/src/messages.rs
index e69de29..719d2c0 100644
--- a/src/messages.rs
+++ b/src/messages.rs
@@ -0,0 +1,4 @@
+#[derive(Debug, Clone)]
+pub struct OpenPorts {
+ pub ports: Vec<u16>
+}