diff --git a/src/clients/mod.rs b/src/clients/mod.rs index a68b1d6728..2b92200057 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -274,6 +274,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/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(), )); }