Merge pull request #91 from nymtech/feature/client-refactor

Feature/client refactor

Cargo.lock

@@ -1256,9 +1256,13 @@ dependencies = [
 name = "mix-client"
 version = "0.1.0"
 dependencies = [
+ "addressing",
  "log",
+ "rand 0.7.2",
+ "rand_distr",
  "sphinx",
  "tokio 0.2.9",
+ "topology",
 ]
 
 [[package]]
@@ -1356,9 +1360,8 @@ dependencies = [
  "log",
  "mix-client",
  "pem",
+ "pemstore",
  "provider-client",
- "rand 0.7.2",
- "rand_distr",
  "reqwest",
  "serde",
  "serde_json",
@@ -1502,6 +1505,14 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "pemstore"
+version = "0.1.0"
+dependencies = [
+ "crypto",
+ "pem",
+]
+
 [[package]]
 name = "percent-encoding"
 version = "1.0.1"

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "common/addressing",
     "common/crypto",
     "common/healthcheck",
+    "common/pemstore",
     "common/topology",
     "mixnode",
     "nym-client",

common/clients/mix-client/Cargo.toml

@@ -8,7 +8,14 @@ edition = "2018"
 
 [dependencies]
 log = "0.4.8"
+rand = "0.7.2"
+rand_distr = "0.2.2"
 tokio = { version = "0.2", features = ["full"] }
 
+## internal
+addressing = {path = "../../addressing"}
+topology = {path = "../../topology"}
+
 ## will be moved to proper dependencies once released
 sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" }
+

common/clients/mix-client/src/lib.rs

@@ -3,6 +3,9 @@ use sphinx::SphinxPacket;
 use std::net::SocketAddr;
 use tokio::prelude::*;
 
+pub mod packet;
+pub mod poisson;
+
 pub struct MixClient {}
 
 impl MixClient {
@@ -17,8 +20,7 @@ impl MixClient {
         mix_addr: SocketAddr,
     ) -> Result<(), Box<dyn std::error::Error>> {
         let bytes = packet.to_bytes();
-
-        info!("socket addr: {:?}", mix_addr);
+        debug!("Sending to the following address: {:?}", mix_addr);
 
         let mut stream = tokio::net::TcpStream::connect(mix_addr).await?;
         stream.write_all(&bytes[..]).await?;

common/clients/mix-client/src/packet.rs

@@ -5,6 +5,7 @@ use std::net::SocketAddr;
 use topology::NymTopology;
 
 pub const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!";
+pub const LOOP_COVER_MESSAGE_AVERAGE_DELAY: f64 = 2.0;
 
 pub fn loop_cover_message<T: NymTopology>(
     our_address: DestinationAddressBytes,
@@ -13,21 +14,25 @@ pub fn loop_cover_message<T: NymTopology>(
 ) -> (SocketAddr, SphinxPacket) {
     let destination = Destination::new(our_address, surb_id);
 
-    encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), topology)
+    encapsulate_message(
+        destination,
+        LOOP_COVER_MESSAGE_PAYLOAD.to_vec(),
+        topology,
+        LOOP_COVER_MESSAGE_AVERAGE_DELAY,
+    )
 }
 
 pub fn encapsulate_message<T: NymTopology>(
     recipient: Destination,
     message: Vec<u8>,
     topology: &T,
+    average_delay: f64,
 ) -> (SocketAddr, SphinxPacket) {
     let mut providers = topology.get_mix_provider_nodes();
     let provider = providers.pop().unwrap().into();
 
     let route = topology.route_to(provider).unwrap();
 
-    // Set average packet delay to an arbitrary but at least not super-slow value for testing.
-    let average_delay = 0.1;
     let delays = sphinx::header::delays::generate(route.len(), average_delay);
 
     // build the packet

common/clients/provider-client/src/lib.rs

@@ -65,7 +65,7 @@ impl ProviderClient {
 
     pub async fn send_request(&self, bytes: Vec<u8>) -> Result<Vec<u8>, ProviderClientError> {
         let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?;
-        info!("keep alive: {:?}", socket.keepalive());
+
         socket.set_keepalive(Some(Duration::from_secs(2))).unwrap();
         socket.write_all(&bytes[..]).await?;
         if let Err(_e) = socket.shutdown(Shutdown::Write) {
@@ -92,7 +92,6 @@ impl ProviderClient {
         let bytes = pull_request.to_bytes();
 
         let response = self.send_request(bytes).await?;
-        info!("Received the following response: {:?}", response);
 
         let parsed_response = PullResponse::from_bytes(&response)?;
         Ok(parsed_response.messages)
@@ -111,4 +110,8 @@ impl ProviderClient {
 
         Ok(parsed_response.auth_token)
     }
+
+    pub fn is_registered(&self) -> bool {
+        self.auth_token.is_some()
+    }
 }

common/pemstore/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "pemstore"
+version = "0.1.0"
+authors = ["Jedrzej Stuczynski <andrew@nymtech.net>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+pem = "0.7.0"
+
+## internal
+crypto = {path = "../crypto"}

common/pemstore/src/pathfinder.rs

@@ -0,0 +1,15 @@
+use std::path::PathBuf;
+
+pub trait PathFinder {
+    fn config_dir(&self) -> PathBuf;
+    fn private_identity_key(&self) -> PathBuf;
+    fn public_identity_key(&self) -> PathBuf;
+
+    // Optional:
+    fn private_encryption_key(&self) -> Option<PathBuf> {
+        None
+    }
+    fn public_encryption_key(&self) -> Option<PathBuf> {
+        None
+    }
+}

common/pemstore/src/pemstore.rs

@@ -1,18 +1,9 @@
-use crate::persistence::pathfinder::Pathfinder;
+use crate::pathfinder::PathFinder;
 use pem::{encode, parse, Pem};
 use std::fs::File;
 use std::io::prelude::*;
 use std::path::PathBuf;
 
-pub fn read_mix_identity_keypair_from_disk(
-    id: String,
-) -> crypto::identity::DummyMixIdentityKeyPair {
-    let pathfinder = Pathfinder::new(id);
-    let pem_store = PemStore::new(pathfinder);
-    let keypair = pem_store.read_identity();
-    keypair
-}
-
 #[allow(dead_code)]
 pub fn read_mix_encryption_keypair_from_disk(_id: String) -> crypto::encryption::x25519::KeyPair {
     unimplemented!()
@@ -25,11 +16,11 @@ pub struct PemStore {
 }
 
 impl PemStore {
-    pub fn new(pathfinder: Pathfinder) -> PemStore {
+    pub fn new<P: PathFinder>(pathfinder: P) -> PemStore {
         PemStore {
-            config_dir: pathfinder.config_dir,
-            private_mix_key: pathfinder.private_mix_key,
-            public_mix_key: pathfinder.public_mix_key,
+            config_dir: pathfinder.config_dir(),
+            private_mix_key: pathfinder.private_identity_key(),
+            public_mix_key: pathfinder.public_identity_key(),
         }
     }
 

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"
@@ -35,6 +33,7 @@ crypto = {path = "../common/crypto"}
 directory-client = { path = "../common/clients/directory-client" }
 healthcheck = { path = "../common/healthcheck" }
 mix-client = { path = "../common/clients/mix-client" }
+pemstore = {path = "../common/pemstore"}
 provider-client = { path = "../common/clients/provider-client" }
 sfw-provider-requests = { path = "../sfw-provider/sfw-provider-requests" }
 topology = {path = "../common/topology" }

nym-client/src/client/cover_traffic_stream.rs

@@ -0,0 +1,35 @@
+use crate::client::mix_traffic::MixMessage;
+use crate::client::LOOP_COVER_AVERAGE_DELAY;
+use futures::channel::mpsc;
+use log::{info, trace};
+use sphinx::route::Destination;
+use std::time::Duration;
+use topology::NymTopology;
+
+pub(crate) async fn start_loop_cover_traffic_stream<T>(
+    tx: mpsc::UnboundedSender<MixMessage>,
+    our_info: Destination,
+    topology: T,
+) where
+    T: NymTopology,
+{
+    info!("Starting loop cover traffic stream");
+    loop {
+        trace!("next cover message!");
+        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 = 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
+        // - the receiver channel is closed
+        // in either case there's no recovery and we can only panic
+        tx.unbounded_send(MixMessage::new(cover_message.0, cover_message.1))
+            .unwrap();
+    }
+}

nym-client/src/client/mix_traffic.rs

@@ -0,0 +1,37 @@
+use futures::channel::mpsc;
+use futures::StreamExt;
+use log::{debug, error, info, trace};
+use sphinx::SphinxPacket;
+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 {
+    pub(crate) async fn run(mut rx: mpsc::UnboundedReceiver<MixMessage>) {
+        info!("Mix Traffic Controller started!");
+        let mix_client = mix_client::MixClient::new();
+        while let Some(mix_message) = rx.next().await {
+            debug!("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(_) => {
+                    trace!("sent a mix message");
+                }
+                // TODO: should there be some kind of threshold of failed messages
+                // that if reached, the application blows?
+                Err(e) => error!(
+                    "We failed to send the message to {} :( - {:?}",
+                    mix_message.0, e
+                ),
+            };
+        }
+    }
+}

nym-client/src/client/mod.rs

@@ -0,0 +1,265 @@
+use crate::built_info;
+use crate::client::mix_traffic::MixTrafficController;
+use crate::client::received_buffer::ReceivedMessagesBuffer;
+use crate::sockets::tcp;
+use crate::sockets::ws;
+use directory_client::presence::Topology;
+use futures::channel::mpsc;
+use futures::join;
+use log::*;
+use sfw_provider_requests::AuthToken;
+use sphinx::route::{Destination, DestinationAddressBytes};
+use std::net::SocketAddr;
+use tokio::runtime::Runtime;
+use topology::NymTopology;
+
+mod cover_traffic_stream;
+mod mix_traffic;
+mod provider_poller;
+mod real_traffic_stream;
+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;
+//  seconds;
+const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds;
+
+pub enum SocketType {
+    TCP,
+    WebSocket,
+    None,
+}
+
+pub struct NymClient {
+    // to be replaced by something else I guess
+    address: DestinationAddressBytes,
+
+    // to be used by "send" function or socket, etc
+    pub input_tx: mpsc::UnboundedSender<InputMessage>,
+
+    input_rx: mpsc::UnboundedReceiver<InputMessage>,
+    socket_listening_address: SocketAddr,
+    directory: String,
+    auth_token: Option<AuthToken>,
+    socket_type: SocketType,
+}
+
+// TODO: this will be moved into module responsible for refreshing topology
+#[derive(Debug)]
+enum TopologyError {
+    HealthCheckError,
+    NoValidPathsError,
+}
+
+#[derive(Debug)]
+pub struct InputMessage(pub Destination, pub Vec<u8>);
+
+impl NymClient {
+    pub fn new(
+        address: DestinationAddressBytes,
+        socket_listening_address: SocketAddr,
+        directory: String,
+        auth_token: Option<AuthToken>,
+        socket_type: SocketType,
+    ) -> Self {
+        let (input_tx, input_rx) = mpsc::unbounded::<InputMessage>();
+
+        NymClient {
+            address,
+            input_tx,
+            input_rx,
+            socket_listening_address,
+            directory,
+            auth_token,
+            socket_type,
+        }
+    }
+
+    // TODO: this will be moved into module responsible for refreshing topology
+    async fn get_compatible_topology(&self) -> Result<Topology, TopologyError> {
+        let score_threshold = 0.0;
+        info!("Trying to obtain valid, healthy, topology");
+
+        let full_topology = Topology::new(self.directory.clone());
+
+        // run a healthcheck to determine healthy-ish nodes:
+        // this is a temporary solution as the healthcheck will eventually be moved to validators
+        let healthcheck_config = healthcheck::config::HealthCheck {
+            directory_server: self.directory.clone(),
+            // those are literally unrelevant when running single check
+            interval: 100000.0,
+            resolution_timeout: 5.0,
+            num_test_packets: 2,
+        };
+        let healthcheck = healthcheck::HealthChecker::new(healthcheck_config);
+        let healthcheck_result = healthcheck.do_check().await;
+
+        let healthcheck_scores = match healthcheck_result {
+            Err(err) => {
+                error!("Error while performing the healtcheck: {:?}", err);
+                return Err(TopologyError::HealthCheckError);
+            }
+            Ok(scores) => scores,
+        };
+
+        let healthy_topology =
+            healthcheck_scores.filter_topology_by_score(&full_topology, score_threshold);
+
+        // for time being assume same versioning, i.e. if client is running X.Y.Z,
+        // we're expecting mixes, providers and coconodes to also be running X.Y.Z
+        let versioned_healthy_topology = healthy_topology.filter_node_versions(
+            built_info::PKG_VERSION,
+            built_info::PKG_VERSION,
+            built_info::PKG_VERSION,
+        );
+
+        // make sure you can still send a packet through the network:
+        if !versioned_healthy_topology.can_construct_path_through() {
+            return Err(TopologyError::NoValidPathsError);
+        }
+
+        Ok(versioned_healthy_topology)
+    }
+
+    pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
+        info!("Starting nym client");
+        let mut rt = Runtime::new()?;
+
+        // channels for inter-component communication
+
+        // mix_tx is the transmitter for any component generating sphinx packets that are to be sent to the mixnet
+        // they are used by cover traffic stream and real traffic stream
+        // mix_rx is the receiver used by MixTrafficController that sends the actual traffic
+        let (mix_tx, mix_rx) = mpsc::unbounded();
+
+        // poller_input_tx is the transmitter of messages fetched from the provider - used by ProviderPoller
+        // poller_input_rx is the receiver for said messages - used by ReceivedMessagesBuffer
+        let (poller_input_tx, poller_input_rx) = mpsc::unbounded();
+
+        // received_messages_buffer_output_tx is the transmitter for *REQUESTS* for messages contained in ReceivedMessagesBuffer - used by sockets
+        // the requests contain a oneshot channel to send a reply on
+        // received_messages_buffer_output_rx is the received for the said requests - used by ReceivedMessagesBuffer
+        let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) =
+            mpsc::unbounded();
+
+        // get initial topology; already filtered by health and version
+        let initial_topology = match rt.block_on(self.get_compatible_topology()) {
+            Ok(topology) => topology,
+            Err(err) => {
+                panic!("Failed to obtain initial network topology: {:?}", err);
+            }
+        };
+
+        // this is temporary and assumes there exists only a single provider.
+        let provider_client_listener_address: SocketAddr = initial_topology
+            .get_mix_provider_nodes()
+            .first()
+            .expect("Could not get a provider from the supplied network topology, are you using the right directory server?")
+            .client_listener;
+
+        let mut provider_poller = provider_poller::ProviderPoller::new(
+            poller_input_tx,
+            provider_client_listener_address,
+            self.address,
+            self.auth_token,
+        );
+
+        // registration
+        if let Err(err) = rt.block_on(provider_poller.perform_initial_registration()) {
+            panic!("Failed to perform initial registration: {:?}", err);
+        };
+
+        // setup all of futures for the components running on the client
+
+        // buffer controlling all messages fetched from provider
+        // required so that other components would be able to use them (say the websocket)
+        let received_messages_buffer_controllers_future = rt.spawn(
+            ReceivedMessagesBuffer::new()
+                .start_controllers(poller_input_rx, received_messages_buffer_output_rx),
+        );
+
+        // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
+        let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx));
+
+        // future constantly pumping loop cover traffic at some specified average rate
+        // the pumped traffic goes to the MixTrafficController
+        let loop_cover_traffic_future =
+            rt.spawn(cover_traffic_stream::start_loop_cover_traffic_stream(
+                mix_tx.clone(),
+                Destination::new(self.address, Default::default()),
+                initial_topology.clone(),
+            ));
+
+        // cloning arguments required by OutQueueControl; required due to move
+        let topology_clone = initial_topology.clone();
+        let self_address = self.address;
+        let input_rx = self.input_rx;
+
+        // future constantly pumping traffic at some specified average rate
+        // if a real message is available on 'input_rx' that might have been received from say
+        // the websocket, the real message is used, otherwise a loop cover message is generated
+        // the pumped traffic goes to the MixTrafficController
+        let out_queue_control_future = rt.spawn(async move {
+            real_traffic_stream::OutQueueControl::new(
+                mix_tx,
+                input_rx,
+                Destination::new(self_address, Default::default()),
+                topology_clone,
+            )
+            .run_out_queue_control()
+            .await
+        });
+
+        // future constantly trying to fetch any received messages from the provider
+        // the received messages are sent to ReceivedMessagesBuffer to be available to rest of the system
+        let provider_polling_future = rt.spawn(provider_poller.start_provider_polling());
+
+        // a temporary workaround for starting socket listener of specified type
+        // in the future the actual socket handler should start THIS client instead
+        match self.socket_type {
+            SocketType::WebSocket => {
+                rt.spawn(ws::start_websocket(
+                    self.socket_listening_address,
+                    self.input_tx,
+                    received_messages_buffer_output_tx,
+                    self.address,
+                    initial_topology,
+                ));
+            }
+            SocketType::TCP => {
+                rt.spawn(tcp::start_tcpsocket(
+                    self.socket_listening_address,
+                    self.input_tx,
+                    received_messages_buffer_output_tx,
+                    self.address,
+                    initial_topology,
+                ));
+            }
+            SocketType::None => (),
+        }
+
+        rt.block_on(async {
+            let future_results = join!(
+                received_messages_buffer_controllers_future,
+                mix_traffic_future,
+                loop_cover_traffic_future,
+                out_queue_control_future,
+                provider_polling_future,
+            );
+
+            assert!(
+                future_results.0.is_ok()
+                    && future_results.1.is_ok()
+                    && future_results.2.is_ok()
+                    && future_results.3.is_ok()
+                    && future_results.4.is_ok()
+            );
+        });
+
+        // this line in theory should never be reached as the runtime should be permanently blocked on traffic senders
+        eprintln!("The client went kaput...");
+        Ok(())
+    }
+}

nym-client/src/client/provider_poller.rs

@@ -0,0 +1,88 @@
+use crate::client::FETCH_MESSAGES_DELAY;
+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<Vec<Vec<u8>>>,
+}
+
+impl ProviderPoller {
+    pub(crate) fn new(
+        poller_tx: mpsc::UnboundedSender<Vec<Vec<u8>>>,
+        provider_client_listener_address: SocketAddr,
+        client_address: DestinationAddressBytes,
+        auth_token: Option<AuthToken>,
+    ) -> 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(self) {
+        info!("Starting provider poller");
+
+        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);
+        let extended_delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY * 10.0);
+
+        loop {
+            debug!("Polling provider...");
+
+            let messages = match self.provider_client.retrieve_messages().await {
+                Err(err) => {
+                    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;
+                }
+                Ok(messages) => messages,
+            };
+
+            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;
+        }
+    }
+}

nym-client/src/client/real_traffic_stream.rs

@@ -0,0 +1,111 @@
+use crate::client::mix_traffic::MixMessage;
+use crate::client::{InputMessage, MESSAGE_SENDING_AVERAGE_DELAY};
+use directory_client::presence::Topology;
+use futures::channel::mpsc;
+use futures::task::{Context, Poll};
+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;
+
+// have a rather low value for test sake
+const AVERAGE_PACKET_DELAY: f64 = 0.1;
+
+pub(crate) struct OutQueueControl {
+    delay: time::Delay,
+    mix_tx: mpsc::UnboundedSender<MixMessage>,
+    input_rx: mpsc::UnboundedReceiver<InputMessage>,
+    our_info: Destination,
+
+    // due to pinning, DerefMut trait, futures, etc its way easier to
+    // just have concrete implementation here rather than generic NymTopology
+    // considering that it will be replaced with refreshing topology within few days anyway
+    topology: Topology,
+}
+
+impl Stream for OutQueueControl {
+    type Item = (SocketAddr, SphinxPacket);
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // it is not yet time to return a message
+        if Future::poll(Pin::new(&mut self.delay), cx).is_pending() {
+            return Poll::Pending;
+        };
+
+        // we know it's time to send a message, so let's prepare delay for the next one
+        // Get the `now` by looking at the current `delay` deadline
+        let now = self.delay.deadline();
+
+        let next_poisson_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.
+        let next = now + next_poisson_delay;
+        self.delay.reset(next);
+
+        // decide what kind of message to send
+        match Stream::poll_next(Pin::new(&mut self.input_rx), cx) {
+            // in the case our real message channel stream was closed, we should also indicate we are closed
+            // (and whoever is using the stream should panic)
+            Poll::Ready(None) => Poll::Ready(None),
+
+            // if there's an actual message - return it
+            Poll::Ready(Some(real_message)) => {
+                trace!("real 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(mix_client::packet::loop_cover_message(
+                    self.our_info.address,
+                    self.our_info.identifier,
+                    &self.topology,
+                )))
+            }
+        }
+    }
+}
+
+impl OutQueueControl {
+    pub(crate) fn new(
+        mix_tx: mpsc::UnboundedSender<MixMessage>,
+        input_rx: mpsc::UnboundedReceiver<InputMessage>,
+        our_info: Destination,
+        topology: Topology,
+    ) -> Self {
+        let initial_delay = time::delay_for(Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY));
+        OutQueueControl {
+            delay: initial_delay,
+            mix_tx,
+            input_rx,
+            our_info,
+            topology,
+        }
+    }
+
+    pub(crate) async fn run_out_queue_control(mut self) {
+        info!("starting out queue controller");
+        while let Some(next_message) = self.next().await {
+            debug!("created new message");
+            // 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.mix_tx
+                .unbounded_send(MixMessage::new(next_message.0, next_message.1))
+                .unwrap();
+        }
+    }
+}

nym-client/src/client/received_buffer.rs

@@ -0,0 +1,89 @@
+use futures::channel::{mpsc, oneshot};
+use futures::lock::Mutex as FMutex;
+use futures::StreamExt;
+use log::{error, info, trace};
+use std::sync::Arc;
+
+pub type BufferResponse = oneshot::Sender<Vec<Vec<u8>>>;
+
+pub(crate) struct ReceivedMessagesBuffer {
+    inner: Arc<FMutex<Inner>>,
+}
+
+impl ReceivedMessagesBuffer {
+    pub(crate) fn new() -> Self {
+        ReceivedMessagesBuffer {
+            inner: Arc::new(FMutex::new(Inner::new())),
+        }
+    }
+
+    pub(crate) async fn start_controllers(
+        self,
+        poller_rx: mpsc::UnboundedReceiver<Vec<Vec<u8>>>, // to receive new messages
+        query_receiver: mpsc::UnboundedReceiver<BufferResponse>, // to receive requests to acquire all stored messages
+    ) {
+        let input_controller_future = tokio::spawn(Self::run_poller_input_controller(
+            self.inner.clone(),
+            poller_rx,
+        ));
+        let output_controller_future = tokio::spawn(Self::run_query_output_controller(
+            self.inner,
+            query_receiver,
+        ));
+
+        futures::future::select(input_controller_future, output_controller_future).await;
+        panic!("One of the received buffer controllers failed!")
+    }
+
+    pub(crate) async fn run_poller_input_controller(
+        buf: Arc<FMutex<Inner>>,
+        mut poller_rx: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
+    ) {
+        info!("Started Received Messages Buffer Input Controller");
+
+        while let Some(new_messages) = poller_rx.next().await {
+            Inner::add_new_messages(&*buf, new_messages).await;
+        }
+    }
+
+    pub(crate) async fn run_query_output_controller(
+        buf: Arc<FMutex<Inner>>,
+        mut query_receiver: mpsc::UnboundedReceiver<BufferResponse>,
+    ) {
+        info!("Started Received Messages Buffer Output Controller");
+
+        while let Some(request) = query_receiver.next().await {
+            let messages = Inner::acquire_and_empty(&*buf).await;
+            if let Err(failed_messages) = request.send(messages) {
+                error!(
+                    "Failed to send the messages to the requester. Adding them back to the buffer"
+                );
+                Inner::add_new_messages(&*buf, failed_messages).await;
+            }
+        }
+    }
+}
+
+pub(crate) struct Inner {
+    messages: Vec<Vec<u8>>,
+}
+
+impl Inner {
+    fn new() -> Self {
+        Inner {
+            messages: Vec::new(),
+        }
+    }
+
+    async fn add_new_messages(buf: &FMutex<Self>, msgs: Vec<Vec<u8>>) {
+        trace!("Adding new messages to the buffer! {:?}", msgs);
+        let mut unlocked = buf.lock().await;
+        unlocked.messages.extend(msgs);
+    }
+
+    async fn acquire_and_empty(buf: &FMutex<Self>) -> Vec<Vec<u8>> {
+        trace!("Emptying the buffer and returning all messages");
+        let mut unlocked = buf.lock().await;
+        std::mem::replace(&mut unlocked.messages, Vec::new())
+    }
+}

nym-client/src/clients/mod.rs

@@ -1,387 +0,0 @@
-use crate::built_info;
-use crate::sockets::tcp;
-use crate::sockets::ws;
-use crate::utils;
-use directory_client::presence::Topology;
-use futures::channel::{mpsc, oneshot};
-use futures::join;
-use futures::lock::Mutex as FMutex;
-use futures::select;
-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;
-
-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 {
-    messages: Vec<Vec<u8>>,
-}
-
-impl ReceivedMessagesBuffer {
-    fn add_arc_futures_mutex(self) -> Arc<FMutex<Self>> {
-        Arc::new(FMutex::new(self))
-    }
-
-    fn new() -> Self {
-        ReceivedMessagesBuffer {
-            messages: Vec::new(),
-        }
-    }
-
-    async fn add_new_messages(buf: Arc<FMutex<Self>>, msgs: Vec<Vec<u8>>) {
-        info!("Adding new messages to the buffer! {:?}", msgs);
-        let mut unlocked = buf.lock().await;
-        unlocked.messages.extend(msgs);
-    }
-
-    async fn run_poller_input_controller(
-        buf: Arc<FMutex<Self>>,
-        mut poller_rx: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
-    ) {
-        while let Some(new_messages) = poller_rx.next().await {
-            ReceivedMessagesBuffer::add_new_messages(buf.clone(), new_messages).await;
-        }
-    }
-
-    async fn acquire_and_empty(buf: Arc<FMutex<Self>>) -> Vec<Vec<u8>> {
-        let mut unlocked = buf.lock().await;
-        std::mem::replace(&mut unlocked.messages, Vec::new())
-    }
-
-    async fn run_query_output_controller(
-        buf: Arc<FMutex<Self>>,
-        mut query_receiver: mpsc::UnboundedReceiver<BufferResponse>,
-    ) {
-        while let Some(request) = query_receiver.next().await {
-            let messages = ReceivedMessagesBuffer::acquire_and_empty(buf.clone()).await;
-            // if this fails, the whole application needs to blow
-            // because currently only this thread would fail
-            request.send(messages).unwrap();
-        }
-    }
-}
-
-pub enum SocketType {
-    TCP,
-    WebSocket,
-    None,
-}
-
-pub struct NymClient {
-    // to be replaced by something else I guess
-    address: DestinationAddressBytes,
-    pub input_tx: mpsc::UnboundedSender<InputMessage>,
-    // to be used by "send" function or socket, etc
-    input_rx: mpsc::UnboundedReceiver<InputMessage>,
-    socket_listening_address: SocketAddr,
-    directory: String,
-    auth_token: Option<AuthToken>,
-    socket_type: SocketType,
-}
-
-#[derive(Debug)]
-pub struct InputMessage(pub Destination, pub Vec<u8>);
-
-impl NymClient {
-    pub fn new(
-        address: DestinationAddressBytes,
-        socket_listening_address: SocketAddr,
-        directory: String,
-        auth_token: Option<AuthToken>,
-        socket_type: SocketType,
-    ) -> Self {
-        let (input_tx, input_rx) = mpsc::unbounded::<InputMessage>();
-
-        NymClient {
-            address,
-            input_tx,
-            input_rx,
-            socket_listening_address,
-            directory,
-            auth_token,
-            socket_type,
-        }
-    }
-
-    async fn start_loop_cover_traffic_stream(
-        mut tx: mpsc::UnboundedSender<MixMessage>,
-        our_info: Destination,
-        topology: Topology,
-    ) {
-        loop {
-            info!("[LOOP COVER TRAFFIC STREAM] - next cover message!");
-            let delay = utils::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);
-            tx.send(MixMessage(cover_message.0, cover_message.1))
-                .await
-                .unwrap();
-        }
-    }
-
-    async fn control_out_queue(
-        mut mix_tx: mpsc::UnboundedSender<MixMessage>,
-        mut input_rx: mpsc::UnboundedReceiver<InputMessage>,
-        our_info: Destination,
-        topology: Topology,
-    ) {
-        loop {
-            info!("[OUT QUEUE] here I will be sending real traffic (or loop cover if nothing is available)");
-            // TODO: consider replacing select macro with our own proper future definition with polling
-            let traffic_message = select! {
-                real_message = input_rx.next() => {
-                    info!("[OUT QUEUE] - we got a real message!");
-                    if real_message.is_none() {
-                        error!("Unexpected 'None' real message!");
-                        std::process::exit(1);
-                    }
-                    let real_message = real_message.unwrap();
-                    println!("real: {:?}", real_message);
-                    utils::sphinx::encapsulate_message(real_message.0, real_message.1, &topology)
-                },
-
-                default => {
-                    info!("[OUT QUEUE] - no real message - going to send extra loop cover");
-                    utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology)
-                }
-            };
-
-            mix_tx
-                .send(MixMessage(traffic_message.0, traffic_message.1))
-                .await
-                .unwrap();
-
-            let delay_duration = Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY);
-            tokio::time::delay_for(delay_duration).await;
-        }
-    }
-
-    async fn start_provider_polling(
-        provider_client: provider_client::ProviderClient,
-        mut poller_tx: mpsc::UnboundedSender<Vec<Vec<u8>>>,
-    ) {
-        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<dyn std::error::Error>> {
-        let score_threshold = 0.0;
-        println!("Starting nym client");
-        let mut rt = Runtime::new()?;
-
-        println!("Trying to obtain valid, healthy, topology");
-        let full_topology = Topology::new(self.directory.clone());
-
-        // run a healthcheck to determine healthy-ish nodes:
-        // this is a temporary solution as the healthcheck will eventually be moved to validators
-        let healthcheck_config = healthcheck::config::HealthCheck {
-            directory_server: self.directory.clone(),
-            // those are literally unrelevant when running single check
-            interval: 100000.0,
-            resolution_timeout: 5.0,
-            num_test_packets: 2,
-        };
-        let healthcheck = healthcheck::HealthChecker::new(healthcheck_config);
-        let healthcheck_result = rt.block_on(healthcheck.do_check());
-
-        let healthcheck_scores = match healthcheck_result {
-            Err(err) => {
-                error!(
-                    "failed to perform healthcheck to determine healthy topology - {:?}",
-                    err
-                );
-                return Err(Box::new(err));
-            }
-            Ok(scores) => scores,
-        };
-
-        let healthy_topology =
-            healthcheck_scores.filter_topology_by_score(&full_topology, score_threshold);
-
-        // for time being assume same versioning, i.e. if client is running X.Y.Z,
-        // we're expecting mixes, providers and coconodes to also be running X.Y.Z
-        let versioned_healthy_topology = healthy_topology.filter_node_versions(
-            built_info::PKG_VERSION,
-            built_info::PKG_VERSION,
-            built_info::PKG_VERSION,
-        );
-
-        // make sure you can still send a packet through the network:
-        if !versioned_healthy_topology.can_construct_path_through() {
-            error!("No valid path exists in the topology");
-            // TODO: replace panic with proper return type
-            panic!("No valid path exists in the topology");
-        }
-
-        // this is temporary and assumes there exists only a single provider.
-        let provider_client_listener_address: SocketAddr = versioned_healthy_topology
-            .get_mix_provider_nodes()
-            .first()
-            .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(
-            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();
-
-        let received_messages_buffer = ReceivedMessagesBuffer::new().add_arc_futures_mutex();
-
-        let received_messages_buffer_input_controller_future =
-            rt.spawn(ReceivedMessagesBuffer::run_poller_input_controller(
-                received_messages_buffer.clone(),
-                poller_input_rx,
-            ));
-        let received_messages_buffer_output_controller_future =
-            rt.spawn(ReceivedMessagesBuffer::run_query_output_controller(
-                received_messages_buffer,
-                received_messages_buffer_output_rx,
-            ));
-
-        let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx));
-        let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream(
-            mix_tx.clone(),
-            Destination::new(self.address, Default::default()),
-            versioned_healthy_topology.clone(),
-        ));
-
-        let out_queue_control_future = rt.spawn(NymClient::control_out_queue(
-            mix_tx,
-            self.input_rx,
-            Destination::new(self.address, Default::default()),
-            versioned_healthy_topology.clone(),
-        ));
-
-        let provider_polling_future = rt.spawn(NymClient::start_provider_polling(
-            provider_client,
-            poller_input_tx,
-        ));
-
-        match self.socket_type {
-            SocketType::WebSocket => {
-                rt.spawn(ws::start_websocket(
-                    self.socket_listening_address,
-                    self.input_tx,
-                    received_messages_buffer_output_tx,
-                    self.address,
-                    versioned_healthy_topology,
-                ));
-            }
-            SocketType::TCP => {
-                rt.spawn(tcp::start_tcpsocket(
-                    self.socket_listening_address,
-                    self.input_tx,
-                    received_messages_buffer_output_tx,
-                    self.address,
-                    versioned_healthy_topology,
-                ));
-            }
-            SocketType::None => (),
-        }
-
-        rt.block_on(async {
-            let future_results = join!(
-                received_messages_buffer_input_controller_future,
-                received_messages_buffer_output_controller_future,
-                mix_traffic_future,
-                loop_cover_traffic_future,
-                out_queue_control_future,
-                provider_polling_future,
-            );
-
-            assert!(
-                future_results.0.is_ok()
-                    && future_results.1.is_ok()
-                    && future_results.2.is_ok()
-                    && future_results.3.is_ok()
-                    && future_results.4.is_ok()
-                    && future_results.5.is_ok()
-            );
-        });
-
-        // this line in theory should never be reached as the runtime should be permanently blocked on traffic senders
-        eprintln!("The client went kaput...");
-        Ok(())
-    }
-}

nym-client/src/commands/init.rs

@@ -1,13 +1,13 @@
-use crate::persistence::pathfinder::Pathfinder;
-use crate::persistence::pemstore::PemStore;
+use crate::config::persistance::pathfinder::ClientPathfinder;
 use clap::ArgMatches;
 use crypto::identity::MixnetIdentityKeyPair;
+use pemstore::pemstore::PemStore;
 
 pub fn execute(matches: &ArgMatches) {
     println!("Initialising client...");
 
     let id = matches.value_of("id").unwrap().to_string(); // required for now
-    let pathfinder = Pathfinder::new(id);
+    let pathfinder = ClientPathfinder::new(id);
 
     println!("Writing keypairs to {:?}...", pathfinder.config_dir);
     let mix_keys = crypto::identity::DummyMixIdentityKeyPair::new();

nym-client/src/commands/tcpsocket.rs

@@ -1,8 +1,8 @@
-use crate::clients::{NymClient, SocketType};
-use crate::persistence::pemstore;
-
+use crate::client::{NymClient, SocketType};
+use crate::config::persistance::pathfinder::ClientPathfinder;
 use clap::ArgMatches;
-use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPublicKey};
+use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey};
+use pemstore::pemstore::PemStore;
 use std::net::ToSocketAddrs;
 
 pub fn execute(matches: &ArgMatches) {
@@ -26,7 +26,8 @@ pub fn execute(matches: &ArgMatches) {
         .next()
         .expect("Failed to extract the socket address from the iterator");
 
-    let keypair = pemstore::read_mix_identity_keypair_from_disk(id);
+    // TODO: currently we know we are reading the 'DummyMixIdentityKeyPair', but how to properly assert the type?
+    let keypair: DummyMixIdentityKeyPair = PemStore::new(ClientPathfinder::new(id)).read_identity();
     // TODO: reading auth_token from disk (if exists);
 
     println!("Public key: {}", keypair.public_key.to_b64_string());

nym-client/src/commands/websocket.rs

@@ -1,8 +1,8 @@
-use crate::clients::{NymClient, SocketType};
-use crate::persistence::pemstore;
-
+use crate::client::{NymClient, SocketType};
+use crate::config::persistance::pathfinder::ClientPathfinder;
 use clap::ArgMatches;
-use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPublicKey};
+use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey};
+use pemstore::pemstore::PemStore;
 use std::net::ToSocketAddrs;
 
 pub fn execute(matches: &ArgMatches) {
@@ -26,7 +26,9 @@ pub fn execute(matches: &ArgMatches) {
         .next()
         .expect("Failed to extract the socket address from the iterator");
 
-    let keypair = pemstore::read_mix_identity_keypair_from_disk(id);
+    // TODO: currently we know we are reading the 'DummyMixIdentityKeyPair', but how to properly assert the type?
+    let keypair: DummyMixIdentityKeyPair = PemStore::new(ClientPathfinder::new(id)).read_identity();
+
     // TODO: reading auth_token from disk (if exists);
 
     println!("Public key: {}", keypair.public_key.to_b64_string());

nym-client/src/config/mod.rs

@@ -0,0 +1 @@
+pub mod persistance;

nym-client/src/config/persistance/mod.rs

@@ -0,0 +1 @@
+pub mod pathfinder;

nym-client/src/config/persistance/pathfinder.rs

@@ -1,21 +1,36 @@
+use pemstore::pathfinder::PathFinder;
 use std::path::PathBuf;
 
-pub struct Pathfinder {
+pub struct ClientPathfinder {
     pub config_dir: PathBuf,
     pub private_mix_key: PathBuf,
     pub public_mix_key: PathBuf,
 }
 
-impl Pathfinder {
-    pub fn new(id: String) -> Pathfinder {
+impl ClientPathfinder {
+    pub fn new(id: String) -> Self {
         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 private_mix_key = config_dir.join("private.pem");
         let public_mix_key = config_dir.join("public.pem");
-        Pathfinder {
+        ClientPathfinder {
             config_dir,
             private_mix_key,
             public_mix_key,
         }
     }
 }
+
+impl PathFinder for ClientPathfinder {
+    fn config_dir(&self) -> PathBuf {
+        self.config_dir.clone()
+    }
+
+    fn private_identity_key(&self) -> PathBuf {
+        self.private_mix_key.clone()
+    }
+
+    fn public_identity_key(&self) -> PathBuf {
+        self.public_mix_key.clone()
+    }
+}

nym-client/src/lib.rs

@@ -1,8 +1,5 @@
-#![recursion_limit = "256"]
-
 pub mod built_info;
-pub mod clients;
-pub mod commands;
-pub mod persistence;
-pub mod sockets;
-pub mod utils;
+pub mod client;
+mod commands;
+pub mod config;
+mod sockets;

nym-client/src/main.rs

@@ -1,13 +1,10 @@
-#![recursion_limit = "256"]
-
 use clap::{App, Arg, ArgMatches, SubCommand};
 
 pub mod built_info;
-pub mod clients;
+pub mod client;
 mod commands;
-mod persistence;
+pub mod config;
 mod sockets;
-pub mod utils;
 
 fn main() {
     env_logger::init();

nym-client/src/persistence/toml.rs

@@ -1 +0,0 @@
-// TODO: we can put all the TOML config templating code in here once we get to that.

nym-client/src/sockets/tcp.rs

@@ -1,5 +1,5 @@
-use crate::clients::BufferResponse;
-use crate::clients::InputMessage;
+use crate::client::received_buffer::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::received_buffer::BufferResponse;
+use crate::client::InputMessage;
 use directory_client::presence::Topology;
 use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
 use futures::channel::{mpsc, oneshot};

nym-client/src/utils/bytes.rs

@@ -1,63 +0,0 @@
-pub fn zero_pad_to_32(mut bytes: Vec<u8>) -> [u8; 32] {
-    assert!(bytes.len() <= 32);
-    if bytes.len() != 32 {
-        bytes.resize(32, 0);
-    }
-    let mut padded_bytes = [0; 32];
-    padded_bytes.copy_from_slice(&bytes[..]);
-    padded_bytes
-}
-
-#[cfg(test)]
-mod zero_padding_to_32_bytes {
-    use super::*;
-
-    #[cfg(test)]
-    mod with_empty_input {
-        use super::*;
-
-        #[test]
-        fn it_returns_32_zeros() {
-            let input = vec![];
-            let result = zero_pad_to_32(input);
-            assert_eq!([0u8; 32], result);
-        }
-    }
-
-    #[cfg(test)]
-    mod with_all_bytes_set_to_1 {
-        use super::*;
-        #[test]
-        fn it_returns_32_ones() {
-            let input = vec![1u8; 32];
-            let result = zero_pad_to_32(input);
-            assert_eq!([1u8; 32], result);
-        }
-    }
-
-    #[cfg(test)]
-    mod with_3_bytes_set {
-        use super::*;
-        #[test]
-        fn it_returns_input_zero_padded_to_32_bytes() {
-            let input = vec![1u8; 3];
-            let result = zero_pad_to_32(input);
-            let expected_content = vec![
-                1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
-                0, 0, 0, 0,
-            ];
-            assert_eq!(expected_content, result.to_vec());
-        }
-    }
-
-    #[cfg(test)]
-    mod with_oversized_input {
-        use super::*;
-        #[test]
-        #[should_panic]
-        fn it_panics() {
-            let input = vec![1u8; 33];
-            zero_pad_to_32(input);
-        }
-    }
-}

nym-client/src/utils/mod.rs

@@ -1,3 +0,0 @@
-pub mod bytes;
-pub mod poisson;
-pub mod sphinx;