diff --git a/Cargo.lock b/Cargo.lock index 233600748a..f33fb06525 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1016,6 +1016,8 @@ dependencies = [ "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", "mockito 0.22.0 (registry+https://github.com/rust-lang/crates.io-index)", "pem 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)", + "rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)", + "rand_distr 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)", "reqwest 0.9.22 (registry+https://github.com/rust-lang/crates.io-index)", "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", diff --git a/src/clients/mod.rs b/src/clients/mod.rs index bad816a794..076ec7d1ae 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -1,36 +1,68 @@ use sphinx::route::DestinationAddressBytes; use tokio::runtime::Runtime; +use futures::channel::mpsc; +use std::time::Duration; +use crate::utils; pub mod directory; pub mod mix; pub mod provider; pub mod validator; -struct Client { + +// 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 FETCH_MESSAGES_DELAY: f64 = 10.0; // assume 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 + +pub struct NymClient { // to be replaced by something else I guess address: DestinationAddressBytes } type TripleFutureResult = (Result<(), Box>, Result<(), Box>, Result<(), Box>); -impl Client { +impl NymClient { pub fn new(address: DestinationAddressBytes) -> Self { - Client { + NymClient { address } } async fn start_loop_cover_traffic_stream(&self) -> Result<(), Box> { - unimplemented!() + 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); + } + } async fn control_out_queue(&self) -> Result<(), Box> { - unimplemented!() + 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); + tokio::time::delay_for(delay_duration).await; + } } async fn start_provider_polling(&self) -> Result<(), Box> { - unimplemented!() + loop { + println!("here I will be polling provider for messages"); + let delay_duration = Duration::from_secs_f64(10.0); + tokio::time::delay_for(delay_duration).await; + } } @@ -39,8 +71,12 @@ impl Client { } pub fn start(&self) -> Result<(), Box> { + println!("starting nym client"); let mut rt = Runtime::new()?; +// let (out_queue_tx, out_queue_rx) = mpsc::unbounded(); + + rt.block_on(async { let future_results = self.start_traffic().await; assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); diff --git a/src/commands/run.rs b/src/commands/run.rs index 48474bc1a6..a73d83401b 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -4,6 +4,7 @@ use crate::clients::directory::requests::presence_topology_get::PresenceTopology use crate::clients::directory::DirectoryClient; use crate::clients::mix::MixClient; use crate::clients::provider::ProviderClient; +use crate::clients::NymClient; use crate::utils::bytes; use base64; use clap::ArgMatches; @@ -21,47 +22,52 @@ pub fn execute(matches: &ArgMatches) { custom_cfg ); - // Grab the network topology from the remote directory server - let topology = get_topology(); + // todo: to be taken from config or something + let my_address = [42u8; 32]; + let client = NymClient::new(my_address); + client.start().unwrap(); - // Create the runtime, probably later move it to Client struct itself? - let mut rt = Runtime::new().unwrap(); - - // Spawn the root task - rt.block_on(async { - let start = Instant::now() + Duration::from_nanos(1000); - let mut interval = interval_at(start, Duration::from_millis(1000)); - let mut i: usize = 0; - loop { - interval.tick().await; -// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); -// -// let route_len = 2; -// -// // data needed to generate a new Sphinx packet -// let route = route_from(&topology, route_len); -// let destination = get_destination(); -// let delays = sphinx::header::delays::generate(route_len); -// -// // build the packet -// let packet = -// sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap(); -// -// // send to mixnet -// let mix_client = MixClient::new(); -// let result = mix_client.send(packet, route.first().unwrap()).await; -// println!("packet sent: {:?}", i); - i += 1; - - // retrieve messages every now and then - if i % 3 == 0 { - interval.tick().await; - println!("going to retrieve messages!"); - let provider_client = ProviderClient::new(); - provider_client.retrieve_messages().await.unwrap(); - } - } - }) + // // Grab the network topology from the remote directory server + // let topology = get_topology(); + // + // // Create the runtime, probably later move it to Client struct itself? + // let mut rt = Runtime::new().unwrap(); + // + // // Spawn the root task + // rt.block_on(async { + // let start = Instant::now() + Duration::from_nanos(1000); + // let mut interval = interval_at(start, Duration::from_millis(1000)); + // let mut i: usize = 0; + // loop { + // interval.tick().await; + //// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); + //// + //// let route_len = 2; + //// + //// // data needed to generate a new Sphinx packet + //// let route = route_from(&topology, route_len); + //// let destination = get_destination(); + //// let delays = sphinx::header::delays::generate(route_len); + //// + //// // build the packet + //// let packet = + //// sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap(); + //// + //// // send to mixnet + //// let mix_client = MixClient::new(); + //// let result = mix_client.send(packet, route.first().unwrap()).await; + //// println!("packet sent: {:?}", i); + // i += 1; + // + // // retrieve messages every now and then + // if i % 3 == 0 { + // interval.tick().await; + // println!("going to retrieve messages!"); + // let provider_client = ProviderClient::new(); + // provider_client.retrieve_messages().await.unwrap(); + // } + // } + // }) } fn get_topology() -> Topology {