diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 63c8672ec0..b9595974fa 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -23,12 +23,9 @@ pub mod mix; pub mod provider; pub mod validator; -// 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; +const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; // seconds +const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 1.0; // seconds; +const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds; // provider-poller sends polls service provider; receives messages // provider-poller sends (TX) to ReceivedBufferController (RX) @@ -194,18 +191,24 @@ impl NymClient { provider_client: ProviderClient, mut poller_tx: mpsc::UnboundedSender>>, ) { + 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; println!("[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(messages).await.unwrap(); + poller_tx.send(good_messages).await.unwrap(); } } pub fn start(self) -> Result<(), Box> { - println!("starting nym client"); + println!("Starting nym client"); let mut rt = Runtime::new()?; let topology = get_topology(self.is_local); @@ -274,6 +277,8 @@ impl NymClient { self.socket_listening_address, self.input_tx, received_messages_buffer_output_tx, + self.address, + topology, )); rt.block_on(async { diff --git a/src/main.rs b/src/main.rs index 8167f91339..4065dedd16 100644 --- a/src/main.rs +++ b/src/main.rs @@ -75,7 +75,6 @@ fn main() { .long("port") .help("Port for websocket to listen on") .takes_value(true) - .required(true), ) .arg(Arg::with_name("local") .long("local") diff --git a/src/sockets/ws.rs b/src/sockets/ws.rs index 6afe2f9f90..4b211feb0e 100644 --- a/src/sockets/ws.rs +++ b/src/sockets/ws.rs @@ -1,3 +1,4 @@ +use crate::clients::directory::presence::Topology; use crate::clients::BufferResponse; use crate::clients::InputMessage; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; @@ -6,18 +7,19 @@ use futures::future::FutureExt; use futures::io::Error; use futures::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; -use sphinx::route::Destination; +use sphinx::route::{Destination, DestinationAddressBytes}; use std::io; use std::net::SocketAddr; use tungstenite::protocol::Message; struct Connection { address: SocketAddr, - rx: UnboundedReceiver, - tx: UnboundedSender, - msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, + rx: UnboundedReceiver, + self_address: DestinationAddressBytes, + topology: Topology, + tx: UnboundedSender, } #[derive(Debug)] @@ -72,7 +74,7 @@ impl ClientRequest { recipient_address: String, mut input_tx: mpsc::UnboundedSender, ) -> ServerResponse { - let address_vec = match hex::decode(recipient_address) { + let address_vec = match base64::decode_config(&recipient_address, base64::URL_SAFE) { Err(e) => { return ServerResponse::Error { message: e.to_string(), @@ -94,6 +96,7 @@ impl ClientRequest { let input_msg = InputMessage(Destination::new(address, dummy_surb), msg.into_bytes()); + println!("ALMOST ABOUT TO SOMEDAY SEND {:?}", input_msg); input_tx.send(input_msg).await.unwrap(); ServerResponse::Send @@ -116,19 +119,29 @@ impl ClientRequest { } let messages = messages.unwrap(); - - ServerResponse::Fetch { messages } - } - - async fn handle_get_clients() -> ServerResponse { - ServerResponse::Error { - message: "NOT IMPLEMENTED".to_string(), + let messages_as_b64 = messages + .iter() + .map(|message| base64::encode_config(message, base64::URL_SAFE)) + .collect(); + ServerResponse::Fetch { + messages: messages_as_b64, } } - async fn handle_own_details() -> ServerResponse { - ServerResponse::Error { - message: "NOT IMPLEMENTED".to_string(), + async fn handle_get_clients(topology: Topology) -> ServerResponse { + let clients = topology + .mix_provider_nodes + .into_iter() + .flat_map(|provider| provider.registered_clients.into_iter()) + .map(|client| client.pub_key) + .collect(); + ServerResponse::GetClients { clients } + } + + async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse { + let self_address = base64::encode_config(&self_address_bytes, base64::URL_SAFE); + ServerResponse::OwnDetails { + address: self_address, } } } @@ -137,7 +150,7 @@ impl ClientRequest { #[serde(tag = "type", rename_all = "camelCase")] enum ServerResponse { Send, - Fetch { messages: Vec> }, + Fetch { messages: Vec }, GetClients { clients: Vec }, OwnDetails { address: String }, Error { message: String }, @@ -164,8 +177,10 @@ async fn handle_connection(conn: Connection) { ClientRequest::handle_send(message, recipient_address, conn.msg_input.clone()).await } ClientRequest::Fetch => ClientRequest::handle_fetch(conn.msg_query.clone()).await, - ClientRequest::GetClients => ClientRequest::handle_get_clients().await, - ClientRequest::OwnDetails => ClientRequest::handle_own_details().await, + ClientRequest::GetClients => { + ClientRequest::handle_get_clients(conn.topology.clone()).await + } + ClientRequest::OwnDetails => ClientRequest::handle_own_details(conn.self_address).await, }; conn.tx @@ -178,6 +193,8 @@ async fn accept_connection( stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, + self_address: DestinationAddressBytes, + topology: Topology, ) { let address = stream .peer_addr() @@ -199,9 +216,10 @@ async fn accept_connection( address, rx: msg_rx, tx: response_tx, - + topology, msg_input, msg_query, + self_address, }; tokio::spawn(handle_connection(conn)); @@ -220,6 +238,8 @@ pub async fn start_websocket( address: SocketAddr, message_tx: mpsc::UnboundedSender, received_messages_query_tx: mpsc::UnboundedSender, + self_address: DestinationAddressBytes, + topology: Topology, ) -> Result<(), WebSocketError> { let mut listener = tokio::net::TcpListener::bind(address).await?; @@ -230,6 +250,8 @@ pub async fn start_websocket( stream, message_tx.clone(), received_messages_query_tx.clone(), + self_address, + topology.clone(), )); } diff --git a/src/utils/sphinx.rs b/src/utils/sphinx.rs index fffb3a2efe..57710f2f15 100644 --- a/src/utils/sphinx.rs +++ b/src/utils/sphinx.rs @@ -4,7 +4,7 @@ use curve25519_dalek::montgomery::MontgomeryPoint; use sphinx::route::{Destination, DestinationAddressBytes, Node, NodeAddressBytes, SURBIdentifier}; use sphinx::SphinxPacket; -const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!"; +pub const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!"; pub fn loop_cover_message( our_address: DestinationAddressBytes,