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

Uncommited work before switching websocket implementation

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2019-12-17 10:46:34 +00:00
parent 29106cef6f
commit a978e6db9b
4 changed files with 34 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/clients/directory/presence.rs
+5 -5
View File
@@ -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<MixProviderClient>,
}
#[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<CocoPresence>,
src/clients/mod.rs
+23 -16
View File
@@ -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<InputMessage>,
// to be used by "send" function or socket, etc
input_rx: mpsc::UnboundedReceiver<InputMessage>,
is_local: bool,
}
@@ -63,28 +67,29 @@ pub struct NymClient {
pub struct InputMessage(pub Destination, pub Vec<u8>);
impl NymClient {
pub fn new(address: DestinationAddressBytes) -> Self {
pub fn new(address: DestinationAddressBytes, is_local: bool) -> Self {
let (input_tx, input_rx) = mpsc::unbounded::<InputMessage>();
NymClient {
address,
input_tx,
input_rx,
is_local,
}
}
async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender<MixMessage>, our_info: Destination) -> Result<(), Box<dyn std::error::Error>> {
async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender<MixMessage>, our_info: Destination, topology: &Topology) -> Result<(), Box<dyn std::error::Error>> {
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<MixMessage>, mut input_rx: mpsc::UnboundedReceiver<InputMessage>, our_info: Destination) -> Result<(), Box<dyn std::error::Error>> {
async fn control_out_queue(mut mix_tx: mpsc::UnboundedSender<MixMessage>, mut input_rx: mpsc::UnboundedReceiver<InputMessage>, our_info: Destination, topology: &Topology) -> Result<(), Box<dyn std::error::Error>> {
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
src/commands/run.rs
+2 -1
View File
@@ -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);
src/utils/sphinx.rs
+4 -2
View File
@@ -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<u8>,
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(