From a978e6db9bffeab3fdc966440c1e997e60fa8978 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 17 Dec 2019 10:46:34 +0000 Subject: [PATCH] Uncommited work before switching websocket implementation --- src/clients/directory/presence.rs | 10 ++++---- src/clients/mod.rs | 39 ++++++++++++++++++------------- src/commands/run.rs | 3 ++- src/utils/sphinx.rs | 6 +++-- 4 files changed, 34 insertions(+), 24 deletions(-) diff --git a/src/clients/directory/presence.rs b/src/clients/directory/presence.rs index 62968eae19..20b666f171 100644 --- a/src/clients/directory/presence.rs +++ b/src/clients/directory/presence.rs @@ -1,6 +1,6 @@ use serde::{Deserialize, Serialize}; -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct CocoPresence { pub host: String, @@ -8,7 +8,7 @@ pub struct CocoPresence { pub last_seen: i64, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct MixNodePresence { pub host: String, @@ -17,7 +17,7 @@ pub struct MixNodePresence { pub last_seen: i64, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct MixProviderPresence { pub host: String, @@ -25,14 +25,14 @@ pub struct MixProviderPresence { pub registered_clients: Vec, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct MixProviderClient { pub pub_key: String, } // Topology shows us the current state of the overall Nym network -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct Topology { pub coco_nodes: Vec, diff --git a/src/clients/mod.rs b/src/clients/mod.rs index b07ca9a7d7..97b833f1b7 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -7,6 +7,8 @@ use futures::{StreamExt, SinkExt}; use crate::clients::mix::MixClient; use sphinx::SphinxPacket; use futures::select; +use crate::utils::topology::get_topology; +use crate::clients::directory::presence::Topology; pub mod directory; pub mod mix; @@ -56,6 +58,8 @@ pub struct NymClient { pub input_tx: mpsc::UnboundedSender, // to be used by "send" function or socket, etc input_rx: mpsc::UnboundedReceiver, + + is_local: bool, } @@ -63,28 +67,29 @@ pub struct NymClient { pub struct InputMessage(pub Destination, pub Vec); impl NymClient { - pub fn new(address: DestinationAddressBytes) -> Self { + pub fn new(address: DestinationAddressBytes, is_local: bool) -> Self { let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { address, input_tx, input_rx, + is_local, } } - async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender, our_info: Destination) -> Result<(), Box> { + async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender, our_info: Destination, topology: &Topology) -> Result<(), Box> { loop { println!("[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); + 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?; } } - async fn control_out_queue(mut mix_tx: mpsc::UnboundedSender, mut input_rx: mpsc::UnboundedReceiver, our_info: Destination) -> Result<(), Box> { + async fn control_out_queue(mut mix_tx: mpsc::UnboundedSender, mut input_rx: mpsc::UnboundedReceiver, our_info: Destination, topology: &Topology) -> Result<(), Box> { loop { println!("[OUT QUEUE] here I will be sending real traffic (or loop cover if nothing is available)"); select! { @@ -92,13 +97,13 @@ impl NymClient { println!("[OUT QUEUE] - we got a real message!"); let real_message = real_message.expect("The channel must have closed! - if the client hasn't crashed, it should have!"); println!("real: {:?}", real_message); - let encapsulated_message = utils::sphinx::encapsulate_message(real_message.0, real_message.1); + let encapsulated_message = utils::sphinx::encapsulate_message(real_message.0, real_message.1, topology); mix_tx.send(MixMessage(encapsulated_message.0, encapsulated_message.1)).await?; }, default => { println!("[OUT QUEUE] - no real message - going to send extra loop cover"); - let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier); + let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, topology); mix_tx.send(MixMessage(cover_message.0, cover_message.1)).await?; } } @@ -127,20 +132,22 @@ impl NymClient { let mut rt = Runtime::new()?; rt.spawn(MixTrafficController::run(mix_rx)); - let foomp = Destination::new(self.address, Default::default()); +// let foomp = Destination::new(self.address, Default::default()); +// +// let mut input_channel = self.input_tx.clone(); +// rt.spawn(async move { +// let test_message = b"foomp".to_vec(); +// let recipient = foomp; +// let input_message = InputMessage(recipient, test_message); +// input_channel.send(input_message).await.unwrap(); +// }); - let mut input_channel = self.input_tx.clone(); - rt.spawn(async move { - let test_message = b"foomp".to_vec(); - let recipient = foomp; - let input_message = InputMessage(recipient, test_message); - input_channel.send(input_message).await.unwrap(); - }); + let topology = get_topology(self.is_local); rt.block_on(async { let future_results = futures::future::join3( - NymClient::start_loop_cover_traffic_stream(mix_tx.clone(), Destination::new(self.address, Default::default())), - NymClient::control_out_queue(mix_tx, self.input_rx, Destination::new(self.address, Default::default())), + NymClient::start_loop_cover_traffic_stream(mix_tx.clone(), Destination::new(self.address, Default::default()), &topology.clone()), + NymClient::control_out_queue(mix_tx, self.input_rx, Destination::new(self.address, Default::default()), &topology.clone()), NymClient::start_provider_polling()).await; // start websocket handler here diff --git a/src/commands/run.rs b/src/commands/run.rs index d0b7ea3596..4e30fdad53 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -22,7 +22,8 @@ pub fn execute(matches: &ArgMatches) { // todo: to be taken from config or something let my_address = [42u8; 32]; - let client = NymClient::new(my_address); + let is_local = true; + let client = NymClient::new(my_address, is_local); client.start().unwrap(); // Grab the network topology from the remote directory server // let topology = get_topology(is_local); diff --git a/src/utils/sphinx.rs b/src/utils/sphinx.rs index 6ce8baa32d..2d0219b54b 100644 --- a/src/utils/sphinx.rs +++ b/src/utils/sphinx.rs @@ -1,3 +1,4 @@ +use crate::clients::directory::presence::Topology; use crate::utils::{bytes, topology}; use sphinx::route::{Destination, DestinationAddressBytes, Node, NodeAddressBytes, SURBIdentifier}; use sphinx::SphinxPacket; @@ -7,19 +8,20 @@ const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!"; pub fn loop_cover_message( our_address: DestinationAddressBytes, surb_id: SURBIdentifier, + topology: &Topology, ) -> (NodeAddressBytes, SphinxPacket) { let destination = Destination::new(our_address, surb_id); - encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec()) + encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), topology) } pub fn encapsulate_message( recipient: Destination, message: Vec, + topology: &Topology, ) -> (NodeAddressBytes, SphinxPacket) { // here we would be getting topology, etc - let topology = topology::get_topology(true); let mixes_route = topology::route_from(&topology, 1); let provider = Node::new(