Moved MixTrafficController to separate module

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<MixMessage>) {
+        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),
+            };
+        }
+    }
+}

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<MixMessage>) {
-        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<Vec<Vec<u8>>>;
 
 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();
 

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;

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;

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;

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;

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 {

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<u8>>) -> Vec<u8> {
 }
 
 fn encode_list_of_clients(clients: Vec<Vec<u8>>) -> Vec<u8> {
-    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,...]

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};