From 9f69cc92fef05915fc720fb63cf2dd58ee1c3f03 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 21 Jan 2020 10:04:20 +0000 Subject: [PATCH] Moved MixTrafficController to separate module --- nym-client/src/client/mix_traffic.rs | 38 ++++++++++++++++++++ nym-client/src/{clients => client}/mod.rs | 43 +++-------------------- nym-client/src/commands/tcpsocket.rs | 2 +- nym-client/src/commands/websocket.rs | 2 +- nym-client/src/lib.rs | 2 +- nym-client/src/main.rs | 2 +- nym-client/src/persistence/pathfinder.rs | 2 +- nym-client/src/sockets/tcp.rs | 10 +++--- nym-client/src/sockets/ws.rs | 4 +-- 9 files changed, 55 insertions(+), 50 deletions(-) create mode 100644 nym-client/src/client/mix_traffic.rs rename nym-client/src/{clients => client}/mod.rs (89%) diff --git a/nym-client/src/client/mix_traffic.rs b/nym-client/src/client/mix_traffic.rs new file mode 100644 index 0000000000..ef43645311 --- /dev/null +++ b/nym-client/src/client/mix_traffic.rs @@ -0,0 +1,38 @@ +use futures::channel::mpsc; +use futures::StreamExt; +use log::*; +use sphinx::SphinxPacket; +use std::io; +use std::io::prelude::*; +use std::net::SocketAddr; + +pub(crate) struct MixMessage(SocketAddr, SphinxPacket); + +impl MixMessage { + pub(crate) fn new(address: SocketAddr, packet: SphinxPacket) -> Self { + MixMessage(address, packet) + } +} + +pub(crate) struct MixTrafficController; + +impl MixTrafficController { + // this was way more difficult to implement than what this code may suggest... + pub(crate) async fn run(mut rx: mpsc::UnboundedReceiver) { + let mix_client = mix_client::MixClient::new(); + while let Some(mix_message) = rx.next().await { + info!( + "[MIX TRAFFIC CONTROL] - got a mix_message for {:?}", + mix_message.0 + ); + let send_res = mix_client.send(mix_message.1, mix_message.0).await; + match send_res { + Ok(_) => { + print!("."); + io::stdout().flush().ok().expect("Could not flush stdout"); + } + Err(e) => error!("We failed to send the message :( - {:?}", e), + }; + } + } +} diff --git a/nym-client/src/clients/mod.rs b/nym-client/src/client/mod.rs similarity index 89% rename from nym-client/src/clients/mod.rs rename to nym-client/src/client/mod.rs index 06d3c79ff1..bfd9e7198d 100644 --- a/nym-client/src/clients/mod.rs +++ b/nym-client/src/client/mod.rs @@ -1,4 +1,5 @@ use crate::built_info; +use crate::client::mix_traffic::{MixMessage, MixTrafficController}; use crate::sockets::tcp; use crate::sockets::ws; use crate::utils; @@ -11,54 +12,20 @@ use futures::{SinkExt, StreamExt}; use log::*; use sfw_provider_requests::AuthToken; use sphinx::route::{Destination, DestinationAddressBytes}; -use sphinx::SphinxPacket; -use std::io; -use std::io::prelude::*; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use tokio::runtime::Runtime; use topology::NymTopology; +mod mix_traffic; + const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5; // seconds const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 0.5; // seconds; const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds; -// provider-poller sends polls service provider; receives messages -// provider-poller sends (TX) to ReceivedBufferController (RX) -// ReceivedBufferController sends (TX) to ... ??Client?? -// outQueueController sends (TX) to TrafficStreamController (RX) -// TrafficStreamController sends messages to mixnet -// ... ??Client?? sends (TX) to outQueueController (RX) -// Loop cover traffic stream just sends messages to mixnet without any channel communication - -struct MixMessage(SocketAddr, SphinxPacket); - -struct MixTrafficController; - -impl MixTrafficController { - // this was way more difficult to implement than what this code may suggest... - async fn run(mut rx: mpsc::UnboundedReceiver) { - let mix_client = mix_client::MixClient::new(); - while let Some(mix_message) = rx.next().await { - info!( - "[MIX TRAFFIC CONTROL] - got a mix_message for {:?}", - mix_message.0 - ); - let send_res = mix_client.send(mix_message.1, mix_message.0).await; - match send_res { - Ok(_) => { - print!("."); - io::stdout().flush().ok().expect("Could not flush stdout"); - } - Err(e) => error!("We failed to send the message :( - {:?}", e), - }; - } - } -} - pub type BufferResponse = oneshot::Sender>>; struct ReceivedMessagesBuffer { @@ -163,7 +130,7 @@ impl NymClient { tokio::time::delay_for(delay_duration).await; let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology); - tx.send(MixMessage(cover_message.0, cover_message.1)) + tx.send(MixMessage::new(cover_message.0, cover_message.1)) .await .unwrap(); } @@ -197,7 +164,7 @@ impl NymClient { }; mix_tx - .send(MixMessage(traffic_message.0, traffic_message.1)) + .send(MixMessage::new(traffic_message.0, traffic_message.1)) .await .unwrap(); diff --git a/nym-client/src/commands/tcpsocket.rs b/nym-client/src/commands/tcpsocket.rs index 3ed3662c7b..053981d446 100644 --- a/nym-client/src/commands/tcpsocket.rs +++ b/nym-client/src/commands/tcpsocket.rs @@ -1,4 +1,4 @@ -use crate::clients::{NymClient, SocketType}; +use crate::client::{NymClient, SocketType}; use crate::persistence::pemstore; use clap::ArgMatches; diff --git a/nym-client/src/commands/websocket.rs b/nym-client/src/commands/websocket.rs index 99dbc1ecc6..0a66193a6f 100644 --- a/nym-client/src/commands/websocket.rs +++ b/nym-client/src/commands/websocket.rs @@ -1,4 +1,4 @@ -use crate::clients::{NymClient, SocketType}; +use crate::client::{NymClient, SocketType}; use crate::persistence::pemstore; use clap::ArgMatches; diff --git a/nym-client/src/lib.rs b/nym-client/src/lib.rs index 5fa9078897..a7760d6e36 100644 --- a/nym-client/src/lib.rs +++ b/nym-client/src/lib.rs @@ -1,7 +1,7 @@ #![recursion_limit = "256"] pub mod built_info; -pub mod clients; +pub mod client; pub mod commands; pub mod persistence; pub mod sockets; diff --git a/nym-client/src/main.rs b/nym-client/src/main.rs index 8455bc7287..64daf4e515 100644 --- a/nym-client/src/main.rs +++ b/nym-client/src/main.rs @@ -3,7 +3,7 @@ use clap::{App, Arg, ArgMatches, SubCommand}; pub mod built_info; -pub mod clients; +pub mod client; mod commands; mod persistence; mod sockets; diff --git a/nym-client/src/persistence/pathfinder.rs b/nym-client/src/persistence/pathfinder.rs index 08110b9ae8..3e5d82bb78 100644 --- a/nym-client/src/persistence/pathfinder.rs +++ b/nym-client/src/persistence/pathfinder.rs @@ -9,7 +9,7 @@ pub struct Pathfinder { impl Pathfinder { pub fn new(id: String) -> Pathfinder { let os_config_dir = dirs::config_dir().unwrap(); // grabs the OS default config dir - let config_dir = os_config_dir.join("nym").join("clients").join(id); + let config_dir = os_config_dir.join("nym").join("client").join(id); let private_mix_key = config_dir.join("private.pem"); let public_mix_key = config_dir.join("public.pem"); Pathfinder { diff --git a/nym-client/src/sockets/tcp.rs b/nym-client/src/sockets/tcp.rs index 66d5a3dbf6..a6cf018bec 100644 --- a/nym-client/src/sockets/tcp.rs +++ b/nym-client/src/sockets/tcp.rs @@ -1,5 +1,5 @@ -use crate::clients::BufferResponse; -use crate::clients::InputMessage; +use crate::client::BufferResponse; +use crate::client::InputMessage; use directory_client::presence::Topology; use futures::channel::{mpsc, oneshot}; use futures::future::FutureExt; @@ -130,7 +130,7 @@ impl ClientRequest { } async fn handle_get_clients(topology: &Topology) -> ServerResponse { - println!("get clients handle"); + println!("get client handle"); let clients = topology .mix_provider_nodes .iter() @@ -189,8 +189,8 @@ fn encode_fetched_messages(messages: Vec>) -> Vec { } fn encode_list_of_clients(clients: Vec>) -> Vec { - println!("clients: {:?}", clients); - // we can just concat all clients since all of them got to be 32 bytes long + println!("client: {:?}", clients); + // we can just concat all client since all of them got to be 32 bytes long // (if not, then we have bigger problem somewhere up the line) // converts [[1,2,3],[4,5,6],...] into [1,2,3,4,5,6,...] diff --git a/nym-client/src/sockets/ws.rs b/nym-client/src/sockets/ws.rs index ab01fca011..685964c898 100644 --- a/nym-client/src/sockets/ws.rs +++ b/nym-client/src/sockets/ws.rs @@ -1,5 +1,5 @@ -use crate::clients::BufferResponse; -use crate::clients::InputMessage; +use crate::client::BufferResponse; +use crate::client::InputMessage; use directory_client::presence::Topology; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; use futures::channel::{mpsc, oneshot};