From 7045db7d70dc926cd86f1316755d2db1e01a1d62 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 22 Jan 2020 10:50:22 +0000 Subject: [PATCH] ibid. --- common/clients/mix-client/src/packet.rs | 9 +++++++-- nym-client/Cargo.toml | 2 -- nym-client/src/client/cover_traffic_stream.rs | 10 ++++++---- nym-client/src/client/provider_poller.rs | 5 ++--- nym-client/src/client/real_traffic_stream.rs | 16 +++++++++------- 5 files changed, 24 insertions(+), 18 deletions(-) diff --git a/common/clients/mix-client/src/packet.rs b/common/clients/mix-client/src/packet.rs index 356d2bbd2e..cc1a027cb6 100644 --- a/common/clients/mix-client/src/packet.rs +++ b/common/clients/mix-client/src/packet.rs @@ -14,7 +14,12 @@ pub fn loop_cover_message( ) -> (SocketAddr, SphinxPacket) { let destination = Destination::new(our_address, surb_id); - encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), topology, LOOP_COVER_MESSAGE_AVERAGE_DELAY) + encapsulate_message( + destination, + LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), + topology, + LOOP_COVER_MESSAGE_AVERAGE_DELAY, + ) } pub fn encapsulate_message( @@ -27,7 +32,7 @@ pub fn encapsulate_message( let provider = providers.pop().unwrap().into(); let route = topology.route_to(provider).unwrap(); - + let delays = sphinx::header::delays::generate(route.len(), average_delay); // build the packet diff --git a/nym-client/Cargo.toml b/nym-client/Cargo.toml index 86e3073c6e..d80d6fcdde 100644 --- a/nym-client/Cargo.toml +++ b/nym-client/Cargo.toml @@ -21,8 +21,6 @@ futures = "0.3.1" hex = "0.4.0" log = "0.4.8" pem = "0.7.0" -rand = "0.7.2" -rand_distr = "0.2.2" reqwest = "0.9.22" serde = { version = "1.0.104", features = ["derive"] } serde_json = "1.0.44" diff --git a/nym-client/src/client/cover_traffic_stream.rs b/nym-client/src/client/cover_traffic_stream.rs index 3405cc2cf9..82f904eec7 100644 --- a/nym-client/src/client/cover_traffic_stream.rs +++ b/nym-client/src/client/cover_traffic_stream.rs @@ -1,6 +1,5 @@ use crate::client::mix_traffic::MixMessage; use crate::client::LOOP_COVER_AVERAGE_DELAY; -use crate::utils; use futures::channel::mpsc; use log::{info, trace}; use sphinx::route::Destination; @@ -17,11 +16,14 @@ pub(crate) async fn start_loop_cover_traffic_stream( info!("Starting loop cover traffic stream"); loop { trace!("next cover message!"); - let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY); + let delay = mix_client::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); tokio::time::delay_for(delay_duration).await; - let cover_message = - utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology); + let cover_message = mix_client::packet::loop_cover_message( + our_info.address, + our_info.identifier, + &topology, + ); // if this one fails, there's no retrying because it means that either: // - we run out of memory diff --git a/nym-client/src/client/provider_poller.rs b/nym-client/src/client/provider_poller.rs index 20d88a049d..aa3924f991 100644 --- a/nym-client/src/client/provider_poller.rs +++ b/nym-client/src/client/provider_poller.rs @@ -1,5 +1,4 @@ use crate::client::FETCH_MESSAGES_DELAY; -use crate::utils; use futures::channel::mpsc; use log::{debug, error, info, trace, warn}; use provider_client::ProviderClientError; @@ -53,7 +52,7 @@ impl ProviderPoller { 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 loop_message = &mix_client::packet::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); @@ -64,7 +63,7 @@ impl ProviderPoller { let messages = match self.provider_client.retrieve_messages().await { Err(err) => { - error!("Failed to query the provider for messages... Going to wait {:?} before retrying", extended_delay_duration); + error!("Failed to query the provider for messages: {:?}, ... Going to wait {:?} before retrying", err, extended_delay_duration); tokio::time::delay_for(extended_delay_duration).await; continue; } diff --git a/nym-client/src/client/real_traffic_stream.rs b/nym-client/src/client/real_traffic_stream.rs index 8750df7727..f89257747b 100644 --- a/nym-client/src/client/real_traffic_stream.rs +++ b/nym-client/src/client/real_traffic_stream.rs @@ -1,18 +1,19 @@ use crate::client::mix_traffic::MixMessage; use crate::client::{InputMessage, MESSAGE_SENDING_AVERAGE_DELAY}; -use crate::utils; use directory_client::presence::Topology; use futures::channel::mpsc; use futures::task::{Context, Poll}; -use futures::{select, Future, Stream, StreamExt}; -use log::{debug, error, info, trace, warn}; +use futures::{Future, Stream, StreamExt}; +use log::{debug, info, trace}; use sphinx::route::Destination; use sphinx::SphinxPacket; use std::net::SocketAddr; use std::pin::Pin; use std::time::Duration; use tokio::time; -use topology::NymTopology; + +// have a rather low value for test sake +const AVERAGE_PACKET_DELAY: f64 = 0.1; pub(crate) struct OutQueueControl { delay: time::Delay, @@ -40,7 +41,7 @@ impl Stream for OutQueueControl { let now = self.delay.deadline(); let next_poisson_delay = - Duration::from_secs_f64(utils::poisson::sample(MESSAGE_SENDING_AVERAGE_DELAY)); + Duration::from_secs_f64(mix_client::poisson::sample(MESSAGE_SENDING_AVERAGE_DELAY)); // The next interval value is `next_poisson_delay` after the one that just // yielded. @@ -56,17 +57,18 @@ impl Stream for OutQueueControl { // if there's an actual message - return it Poll::Ready(Some(real_message)) => { trace!("real message"); - Poll::Ready(Some(utils::sphinx::encapsulate_message( + Poll::Ready(Some(mix_client::packet::encapsulate_message( real_message.0, real_message.1, &self.topology, + AVERAGE_PACKET_DELAY, ))) } // otherwise construct a dummy one _ => { trace!("loop cover message"); - Poll::Ready(Some(utils::sphinx::loop_cover_message( + Poll::Ready(Some(mix_client::packet::loop_cover_message( self.our_info.address, self.our_info.identifier, &self.topology,