diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index 0be475c82e..ae69b29795 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -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>>, - ) { - 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> { 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 => { diff --git a/nym-client/src/client/provider_poller.rs b/nym-client/src/client/provider_poller.rs new file mode 100644 index 0000000000..f55b11cde0 --- /dev/null +++ b/nym-client/src/client/provider_poller.rs @@ -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>>, +} + +impl ProviderPoller { + pub(crate) fn new( + poller_tx: mpsc::UnboundedSender>>, + provider_client_listener_address: SocketAddr, + client_address: DestinationAddressBytes, + auth_token: Option, + ) -> 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; + } + } +}