GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Basic delay loop for cover traffic stream

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2019-12-13 11:36:02 +00:00
parent eee87a4d45
commit 59b931b919
3 changed files with 90 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -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)",
src/clients/mod.rs
+42 -6
View File
@@ -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<dyn std::error::Error>>, Result<(), Box<dyn std::error::Error>>, Result<(), Box<dyn std::error::Error>>);
impl Client {
impl NymClient {
pub fn new(address: DestinationAddressBytes) -> Self {
Client {
NymClient {
address
}
}
async fn start_loop_cover_traffic_stream(&self) -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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());
src/commands/run.rs
+46 -40
View File
@@ -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 {