GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved provider-polling related functionalities to new module

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2020-01-21 13:08:26 +00:00
parent 7889ab4778
commit 06b96bd2c1
2 changed files with 103 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-client/src/client/mod.rs
+14 -41
View File
@@ -18,8 +18,10 @@ use tokio::runtime::Runtime;
use topology::NymTopology;
mod mix_traffic;
mod provider_poller;
pub mod received_buffer;
// TODO: all of those constants should probably be moved to config file
const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5;
// seconds
const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 0.5;
@@ -123,26 +125,6 @@ impl NymClient {
}
}
async fn start_provider_polling(
provider_client: provider_client::ProviderClient,
mut poller_tx: mpsc::UnboundedSender<Vec<Vec<u8>>>,
) {
let loop_message = &utils::sphinx::LOOP_COVER_MESSAGE_PAYLOAD.to_vec();
let dummy_message = &sfw_provider_requests::DUMMY_MESSAGE_CONTENT.to_vec();
loop {
let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY);
tokio::time::delay_for(delay_duration).await;
info!("[FETCH MSG] - Polling provider...");
let messages = provider_client.retrieve_messages().await.unwrap();
let good_messages = messages
.into_iter()
.filter(|message| message != loop_message && message != dummy_message)
.collect();
// if any of those fails, whole application should blow...
poller_tx.send(good_messages).await.unwrap();
}
}
pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
let score_threshold = 0.0;
println!("Starting nym client");
@@ -192,6 +174,12 @@ impl NymClient {
panic!("No valid path exists in the topology");
}
// channels for intercomponent communication
let (mix_tx, mix_rx) = mpsc::unbounded();
let (poller_input_tx, poller_input_rx) = mpsc::unbounded();
let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) =
mpsc::unbounded();
// this is temporary and assumes there exists only a single provider.
let provider_client_listener_address: SocketAddr = versioned_healthy_topology
.get_mix_provider_nodes()
@@ -199,29 +187,17 @@ impl NymClient {
.expect("Could not get a provider from the supplied network topology, are you using the right directory server?")
.client_listener;
let mut provider_client = provider_client::ProviderClient::new(
let mut provider_poller = provider_poller::ProviderPoller::new(
poller_input_tx,
provider_client_listener_address,
self.address,
self.auth_token,
);
// registration
rt.block_on(async {
match self.auth_token {
None => {
let auth_token = provider_client.register().await.unwrap();
provider_client.update_token(auth_token);
info!("Obtained new token! - {:?}", auth_token);
}
Some(token) => println!("Already got the token! - {:?}", token),
}
});
// channels for intercomponent communication
let (mix_tx, mix_rx) = mpsc::unbounded();
let (poller_input_tx, poller_input_rx) = mpsc::unbounded();
let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) =
mpsc::unbounded();
if let Err(err) = rt.block_on(provider_poller.perform_initial_registration()) {
panic!("Failed to perform initial registration: {:?}", err);
};
let received_messages_buffer_controllers_future = rt.spawn(
ReceivedMessagesBuffer::new()
@@ -242,10 +218,7 @@ impl NymClient {
versioned_healthy_topology.clone(),
));
let provider_polling_future = rt.spawn(NymClient::start_provider_polling(
provider_client,
poller_input_tx,
));
let provider_polling_future = rt.spawn(provider_poller.start_provider_polling());
match self.socket_type {
SocketType::WebSocket => {
nym-client/src/client/provider_poller.rs
+89
View File
@@ -0,0 +1,89 @@
use crate::client::FETCH_MESSAGES_DELAY;
use crate::utils;
use futures::channel::mpsc;
use log::{debug, error, info, trace, warn};
use provider_client::ProviderClientError;
use sfw_provider_requests::AuthToken;
use sphinx::route::DestinationAddressBytes;
use std::net::SocketAddr;
use std::time::Duration;
pub(crate) struct ProviderPoller {
provider_client: provider_client::ProviderClient,
poller_tx: mpsc::UnboundedSender<Vec<Vec<u8>>>,
}
impl ProviderPoller {
pub(crate) fn new(
poller_tx: mpsc::UnboundedSender<Vec<Vec<u8>>>,
provider_client_listener_address: SocketAddr,
client_address: DestinationAddressBytes,
auth_token: Option<AuthToken>,
) -> Self {
ProviderPoller {
provider_client: provider_client::ProviderClient::new(
provider_client_listener_address,
client_address,
auth_token,
),
poller_tx,
}
}
// This method is only temporary until registration is moved to `client init`
pub(crate) async fn perform_initial_registration(&mut self) -> Result<(), ProviderClientError> {
debug!("performing initial provider registration");
if !self.provider_client.is_registered() {
let auth_token = match self.provider_client.register().await {
// in this particular case we can ignore this error
Err(ProviderClientError::ClientAlreadyRegisteredError) => return Ok(()),
Err(err) => return Err(err),
Ok(token) => token,
};
self.provider_client.update_token(auth_token)
} else {
warn!("did not perform registration - we were already registered")
}
Ok(())
}
pub(crate) async fn start_provider_polling(mut self) {
info!("Starting provider poller");
let loop_message = &utils::sphinx::LOOP_COVER_MESSAGE_PAYLOAD.to_vec();
let dummy_message = &sfw_provider_requests::DUMMY_MESSAGE_CONTENT.to_vec();
let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY);
loop {
debug!("Polling provider...");
let messages = match self.provider_client.retrieve_messages().await {
Ok(messages) => messages,
Err(err) => {
let extended_delay_duration =
Duration::from_secs_f64(FETCH_MESSAGES_DELAY * 10.0);
error!("Failed to query the provider for messages... Going to wait {:?} before retrying", extended_delay_duration);
tokio::time::delay_for(extended_delay_duration).await;
continue;
}
};
let good_messages = messages
.into_iter()
.filter(|message| message != loop_message && message != dummy_message)
.collect();
trace!("Obtained the following messages: {:?}", good_messages);
// if this one fails, there's no retrying because it means that either:
// - we run out of memory
// - the receiver channel is closed
// in either case there's no recovery and we can only panic
self.poller_tx.unbounded_send(good_messages).unwrap();
tokio::time::delay_for(delay_duration).await;
}
}
}