diff --git a/src/clients/mod.rs b/src/clients/mod.rs index f57d48ceb8..cb956ba2b6 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -12,8 +12,10 @@ pub mod validator; // TODO: put that in config once it exists -const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; // assume seconds -const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; // assume seconds; +const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; +// assume seconds +const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; +// assume seconds; const FETCH_MESSAGES_DELAY: f64 = 10.0; // assume seconds; // provider-poller sends polls service provider; receives messages @@ -38,32 +40,38 @@ impl MixTrafficController { pub struct NymClient { // to be replaced by something else I guess - address: DestinationAddressBytes + address: DestinationAddressBytes, + pub input_tx: mpsc::UnboundedSender>, + // to be used by "send" function or socket, etc + input_rx: mpsc::UnboundedReceiver>, } type TripleFutureResult = (Result<(), Box>, Result<(), Box>, Result<(), Box>); impl NymClient { pub fn new(address: DestinationAddressBytes) -> Self { + let (input_tx, input_rx) = mpsc::unbounded::>(); + NymClient { - address + address, + input_tx, + input_rx, } } - async fn start_loop_cover_traffic_stream(&self, mut tx: mpsc::Sender>) -> Result<(), Box> { + async fn start_loop_cover_traffic_stream(mut tx: mpsc::Sender>) -> Result<(), Box> { loop { let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); println!("waiting for {:?}", delay_duration); tokio::time::delay_for(delay_duration).await; println!("waited {:?} - time to send cover message!", delay_duration); - let dummy_message = vec![1,2,3]; - tx.send(dummy_message).await; + let dummy_message = vec![1, 2, 3]; + tx.send(dummy_message).await?; } - } - async fn control_out_queue(&self) -> Result<(), Box> { + async fn control_out_queue(mut mix_tx: mpsc::Sender>, mut input_rx: mpsc::UnboundedReceiver>) -> Result<(), Box> { loop { println!("here I will be sending real traffic (or loop cover if nothing is available)"); let delay_duration = Duration::from_secs_f64(10.0); @@ -72,7 +80,7 @@ impl NymClient { } - async fn start_provider_polling(&self) -> Result<(), Box> { + async fn start_provider_polling() -> Result<(), Box> { loop { println!("here I will be polling provider for messages"); let delay_duration = Duration::from_secs_f64(10.0); @@ -81,20 +89,21 @@ impl NymClient { } - async fn start_traffic(&self) -> TripleFutureResult { + pub fn start(self) -> Result<(), Box> { + println!("starting nym client"); + let mix_chan_buf_size = 64; let (mix_tx, mix_rx) = mpsc::channel(mix_chan_buf_size); - tokio::spawn(MixTrafficController::run(mix_rx)); - futures::future::join3(self.start_loop_cover_traffic_stream(mix_tx), self.control_out_queue(), self.start_provider_polling()).await - } - - pub fn start(&self) -> Result<(), Box> { - println!("starting nym client"); let mut rt = Runtime::new()?; + rt.spawn(MixTrafficController::run(mix_rx)); + rt.block_on(async { - let future_results = self.start_traffic().await; + let future_results = futures::future::join3( + NymClient::start_loop_cover_traffic_stream(mix_tx.clone()), + NymClient::control_out_queue(mix_tx, self.input_rx), + NymClient::start_provider_polling()).await; assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); });