Ability to send mix messages from websocket

src/clients/mod.rs

@@ -9,7 +9,9 @@ use sphinx::route::{Destination, DestinationAddressBytes, NodeAddressBytes};
 use sphinx::SphinxPacket;
 use std::time::Duration;
 use tokio::runtime::Runtime;
-use futures::future::join4;
+use futures::future::join5;
+use crate::sockets::ws;
+use std::net::SocketAddr;
 
 pub mod directory;
 pub mod mix;
@@ -17,7 +19,7 @@ pub mod provider;
 pub mod validator;
 
 // TODO: put that in config once it exists
-const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5;
+const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0;
 // assume seconds
 const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0;
 // assume seconds;
@@ -134,30 +136,14 @@ impl NymClient {
         }
     }
 
-    pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
+    pub fn start(self, socket_address: SocketAddr) -> Result<(), Box<dyn std::error::Error>> {
         println!("starting nym client");
 
         let (mix_tx, mix_rx) = mpsc::unbounded();
         let mut rt = Runtime::new()?;
 
-
         let topology = get_topology(self.is_local);
 
-
-//        let mut input_channel = self.input_tx.clone();
-//        let bar = self.address.clone();
-//        rt.spawn(async move {
-//            loop {
-//                let foomp = Destination::new(bar, Default::default());
-//
-//                let test_message = b"foomp".to_vec();
-//                let recipient = foomp;
-//                let input_message = InputMessage(recipient, test_message);
-//                input_channel.send(input_message).await.unwrap();
-//                tokio::time::delay_for(Duration::from_secs(1)).await;
-//            }
-//        });
-
         let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx));
         let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream(
             mix_tx.clone(),
@@ -173,13 +159,12 @@ impl NymClient {
         ));
 
         let provider_polling_future = rt.spawn(NymClient::start_provider_polling());
-
-        // TODO: websocket handler future
+        let websocket_future = rt.spawn(ws::start_websocket(socket_address, self.input_tx));
 
         rt.block_on(async {
-            let future_results = join4(mix_traffic_future, loop_cover_traffic_future, out_queue_control_future, provider_polling_future).await;
+            let future_results = join5(mix_traffic_future, loop_cover_traffic_future, out_queue_control_future, provider_polling_future, websocket_future).await;
             assert!(
-                future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok() && future_results.3.is_ok()
+                future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok() && future_results.3.is_ok() && future_results.4.is_ok()
             );
         });
 

src/commands/run.rs

@@ -17,14 +17,16 @@ use tokio::runtime::Runtime;
 use tokio::time::{interval_at, Instant};
 
 pub fn execute(matches: &ArgMatches) {
-    let is_local = matches.is_present("local");
-    println!("Starting client, local: {:?}", is_local);
+    unimplemented!() // currently the 'run' starts websocket!
 
-    // todo: to be taken from config or something
-    let my_address = [42u8; 32];
-    let is_local = true;
-    let client = NymClient::new(my_address, is_local);
-    client.start().unwrap();
+    //    let is_local = matches.is_present("local");
+    //    println!("Starting client, local: {:?}", is_local);
+    //
+    //    // todo: to be taken from config or something
+    //    let my_address = [42u8; 32];
+    //    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/commands/websocket.rs

@@ -1,5 +1,5 @@
 use crate::banner;
-use crate::sockets::ws;
+use crate::clients::NymClient;
 use clap::ArgMatches;
 use std::net::ToSocketAddrs;
 
@@ -19,5 +19,13 @@ pub fn execute(matches: &ArgMatches) {
         .next()
         .expect("Failed to extract the socket address from the iterator");
 
-    ws::start(socket_address);
+    let is_local = matches.is_present("local");
+    println!("Starting client, local: {:?}", is_local);
+
+    // todo: to be taken from config or something
+    let my_address = [42u8; 32];
+    let is_local = true;
+    let client = NymClient::new(my_address, is_local);
+
+    client.start(socket_address).unwrap();
 }

src/lib.rs

@@ -1,4 +1,5 @@
 pub mod clients;
 pub mod identity;
 pub mod persistence;
+pub mod sockets;
 pub mod utils;

src/sockets/ws.rs

@@ -1,98 +1,148 @@
 use crate::clients::InputMessage;
 use futures::channel::mpsc;
 use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
-use futures::stream::Stream;
-use futures::Future;
 use futures::{SinkExt, StreamExt};
-use hex::FromHexError;
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 use sphinx::route::Destination;
 use std::net::SocketAddr;
-use tokio::runtime::Runtime;
-use tokio_tungstenite::accept_async;
 use tungstenite::protocol::Message;
-use std::time::Duration;
+use std::io;
+use futures::io::Error;
 
 struct Connection {
     address: SocketAddr,
     rx: UnboundedReceiver<Message>,
     tx: UnboundedSender<Message>,
+
+    msg_input: mpsc::UnboundedSender<InputMessage>,
 }
 
 #[derive(Debug)]
-enum WebSocketError {
-    InvalidDestinationEncoding,
-    InvalidDestinationLength,
+pub enum WebSocketError {
+    FailedToStartSocketError,
+    UnknownSocketError,
 }
 
-impl From<hex::FromHexError> for WebSocketError {
-    fn from(_: FromHexError) -> Self {
+impl From<io::Error> for WebSocketError {
+    fn from(err: Error) -> Self {
         use WebSocketError::*;
+        match err.kind() {
+            io::ErrorKind::ConnectionRefused => FailedToStartSocketError,
+            io::ErrorKind::ConnectionReset => FailedToStartSocketError,
+            io::ErrorKind::ConnectionAborted => FailedToStartSocketError,
+            io::ErrorKind::NotConnected => FailedToStartSocketError,
 
-        InvalidDestinationEncoding
+            io::ErrorKind::AddrInUse => FailedToStartSocketError,
+            io::ErrorKind::AddrNotAvailable => FailedToStartSocketError,
+            _ => UnknownSocketError,
+        }
     }
 }
 
 
+
 #[derive(Serialize, Deserialize, Debug)]
-struct ClientMessageJSON {
-    message: String,
-    recipient_address: String,
+#[serde(tag = "type", rename_all = "camelCase")]
+enum ClientRequest {
+    Send { message: String, recipient_address: String },
+    Fetch,
+    GetClients,
+    OwnDetails,
+}
+
+impl From<Message> for ClientRequest {
+    fn from(msg: Message) -> Self {
+        let text_msg = match msg {
+            Message::Text(msg) => msg,
+            Message::Binary(_) => panic!("binary messages are not supported!"),
+            Message::Close(_) => panic!("todo: handle close!"),
+            _ => panic!("Other types of messages are also unsupported!"),
+        };
+        serde_json::from_str(&text_msg).unwrap()
+    }
+}
+
+impl ClientRequest {
+    async fn handle_send(msg: String, recipient_address: String, mut input_tx: mpsc::UnboundedSender<InputMessage>) -> ServerResponse {
+        let address_vec = match hex::decode(recipient_address) {
+            Err(e) => return ServerResponse::Error { message: e.to_string() },
+            Ok(hex) => hex,
+        };
+
+        if address_vec.len() != 32 {
+            return ServerResponse::Error { message: "InvalidDestinationLength".to_string() };
+        }
+
+        let mut address = [0; 32];
+        address.copy_from_slice(&address_vec);
+
+        let dummy_surb = [0; 16];
+
+        let input_msg = InputMessage(
+            Destination::new(address, dummy_surb),
+            msg.into_bytes(),
+        );
+
+        input_tx.send(input_msg).await.unwrap();
+
+        ServerResponse::Send
+    }
+
+    async fn handle_fetch() -> ServerResponse {
+        ServerResponse::Error { message: "NOT IMPLEMENTED".to_string() }
+    }
+
+    async fn handle_get_clients() -> ServerResponse {
+        ServerResponse::Error { message: "NOT IMPLEMENTED".to_string() }
+    }
+
+    async fn handle_own_details() -> ServerResponse {
+        ServerResponse::Error { message: "NOT IMPLEMENTED".to_string() }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(tag = "type", rename_all = "camelCase")]
+enum ServerResponse {
+    Send,
+    Fetch { messages: Vec<Vec<u8>> },
+    GetClients { clients: Vec<String> },
+    OwnDetails { address: String },
+    Error { message: String },
 }
 
 fn dummy_response() -> Message {
     Message::Text("foomp".to_string())
 }
 
+impl Into<Message> for ServerResponse {
+    fn into(self) -> Message {
+        dummy_response()
+    }
+}
+
+
 async fn handle_connection(conn: Connection) {
     let mut conn = conn;
     while let Some(msg) = conn.rx.next().await {
         println!("Received a message from {}: {}", conn.address, msg);
-        // TODO: currently only hardcoded sends are supported
-        let parsed_message = parse_message(msg);
-        println!("parsed: {:?}", parsed_message);
-        println!("test async pre wait");
+        let request: ClientRequest = msg.into();
 
-        tokio::time::delay_for(Duration::from_secs(2)).await;
+        let response = match request {
+            ClientRequest::Send { message, recipient_address } => ClientRequest::handle_send(message, recipient_address, conn.msg_input.clone()).await,
+            ClientRequest::Fetch => ClientRequest::handle_fetch().await,
+            ClientRequest::GetClients => ClientRequest::handle_get_clients().await,
+            ClientRequest::OwnDetails => ClientRequest::handle_own_details().await,
+        };
 
-        println!("test async post wait");
         conn
             .tx
-            .unbounded_send(dummy_response())
+            .unbounded_send(response.into())
             .expect("Failed to forward message");
     }
 }
 
-// Proves we can call Rust methods from the websocket listener. Re-route it to wherever JS puts
-// the `send_message` functionality.
-fn parse_message(msg: Message) -> Result<InputMessage, WebSocketError> {
-    let text_msg = match msg {
-        Message::Text(msg) => msg,
-        Message::Binary(_) => panic!("binary messages are not supported!"),
-        Message::Close(_) => panic!("todo: handle close!"),
-        _ => panic!("Other types of messages are also unsupported!"),
-    };
-
-    let raw_msg: ClientMessageJSON = serde_json::from_str(&text_msg).unwrap();
-    let address_vec = hex::decode(raw_msg.recipient_address)?;
-
-    if address_vec.len() != 32 {
-        return Err(WebSocketError::InvalidDestinationLength);
-    }
-
-    let mut address = [0; 32];
-    address.copy_from_slice(&address_vec);
-
-    let dummy_surb = [0; 16];
-
-    Ok(InputMessage(
-        Destination::new(address, dummy_surb),
-        raw_msg.message.into_bytes(),
-    ))
-}
-
-async fn accept_connection(stream: tokio::net::TcpStream) {
+async fn accept_connection(stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender<InputMessage>) {
     let address = stream
         .peer_addr()
         .expect("connected streams should have a peer address");
@@ -113,6 +163,8 @@ async fn accept_connection(stream: tokio::net::TcpStream) {
         address,
         rx: msg_rx,
         tx: response_tx,
+
+        msg_input,
     };
     tokio::spawn(handle_connection(conn));
 
@@ -127,21 +179,16 @@ async fn accept_connection(stream: tokio::net::TcpStream) {
     }
 }
 
+pub async fn start_websocket(address: SocketAddr, message_tx: mpsc::UnboundedSender<InputMessage>) -> Result<(), WebSocketError> {
+    let mut listener = tokio::net::TcpListener::bind(address).await?;
 
+    while let Ok((stream, _)) = listener.accept().await {
+        // it's fine to be cloning the channel on all new connection, because in principle
+        // this server should only EVER have a single client connected
+        tokio::spawn(accept_connection(stream, message_tx.clone()));
+    }
 
-pub fn start(address: SocketAddr) -> Result<(), Box<dyn std::error::Error>> {
-    let mut rt = Runtime::new()?;
-
-    rt.block_on(async {
-        let mut listener = tokio::net::TcpListener::bind(address).await?;
-
-        while let Ok((stream, _)) = listener.accept().await {
-            // TODO: should it rather be rt.spawn?
-            tokio::spawn(accept_connection(stream));
-        }
-
-        eprintln!("The websocket went kaput...");
-        Ok(())
-    })
+    eprintln!("The websocket went kaput...");
+    Ok(())
 }