From b706d36e65cb0fa2b641c27ea5401153079a0e49 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 17:10:35 +0000 Subject: [PATCH] Ability for the websocket to retrieve stored messages --- src/clients/mod.rs | 12 ++---------- src/sockets/ws.rs | 39 +++++++++++++++++++++++++++++---------- 2 files changed, 31 insertions(+), 20 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 3caaad2667..58bdb1a02c 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -61,7 +61,7 @@ impl MixTrafficController { } } -type BufferResponse = oneshot::Sender>>; +pub type BufferResponse = oneshot::Sender>>; struct ReceivedMessagesBuffer{ messages: Vec> @@ -247,7 +247,7 @@ impl NymClient { )); let provider_polling_future = rt.spawn(NymClient::start_provider_polling(provider_client, poller_input_tx)); - let websocket_future = rt.spawn(ws::start_websocket(self.socket_listening_address, self.input_tx)); + let websocket_future = rt.spawn(ws::start_websocket(self.socket_listening_address, self.input_tx, received_messages_buffer_output_tx)); rt.block_on(async { let future_results = join!( @@ -260,14 +260,6 @@ impl NymClient { websocket_future, ); -// 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() diff --git a/src/sockets/ws.rs b/src/sockets/ws.rs index b4778a15ae..edd2ac13dc 100644 --- a/src/sockets/ws.rs +++ b/src/sockets/ws.rs @@ -1,13 +1,15 @@ use crate::clients::InputMessage; -use futures::channel::mpsc; +use futures::channel::{mpsc, oneshot}; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; -use futures::{SinkExt, StreamExt}; +use futures::{SinkExt, StreamExt, AsyncReadExt}; use serde::{Deserialize, Serialize}; use sphinx::route::Destination; use std::net::SocketAddr; use tungstenite::protocol::Message; use std::io; use futures::io::Error; +use crate::clients::BufferResponse; +use futures::future::FutureExt; struct Connection { address: SocketAddr, @@ -15,6 +17,7 @@ struct Connection { tx: UnboundedSender, msg_input: mpsc::UnboundedSender, + msg_query: mpsc::UnboundedSender, } #[derive(Debug)] @@ -40,7 +43,6 @@ impl From for WebSocketError { } - #[derive(Serialize, Deserialize, Debug)] #[serde(tag = "type", rename_all = "camelCase")] enum ClientRequest { @@ -88,8 +90,23 @@ impl ClientRequest { ServerResponse::Send } - async fn handle_fetch() -> ServerResponse { - ServerResponse::Error { message: "NOT IMPLEMENTED".to_string() } + async fn handle_fetch(mut msg_query: mpsc::UnboundedSender) -> ServerResponse { + let (res_tx, res_rx) = oneshot::channel(); + if msg_query.send(res_tx).await.is_err() { + return ServerResponse::Error { message: "Server failed to receive messages".to_string() }; + } + + let messages = res_rx.map(|msg| { + msg + }).await; + + if messages.is_err() { + return ServerResponse::Error { message: "Server failed to receive messages".to_string() }; + } + + let messages = messages.unwrap(); + + ServerResponse::Fetch {messages} } async fn handle_get_clients() -> ServerResponse { @@ -117,7 +134,8 @@ fn dummy_response() -> Message { impl Into for ServerResponse { fn into(self) -> Message { - dummy_response() + let str_res = serde_json::to_string(&self).unwrap(); + Message::Text(str_res) } } @@ -130,7 +148,7 @@ async fn handle_connection(conn: Connection) { 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::Fetch => ClientRequest::handle_fetch(conn.msg_query.clone()).await, ClientRequest::GetClients => ClientRequest::handle_get_clients().await, ClientRequest::OwnDetails => ClientRequest::handle_own_details().await, }; @@ -142,7 +160,7 @@ async fn handle_connection(conn: Connection) { } } -async fn accept_connection(stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender) { +async fn accept_connection(stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender) { let address = stream .peer_addr() .expect("connected streams should have a peer address"); @@ -165,6 +183,7 @@ async fn accept_connection(stream: tokio::net::TcpStream, msg_input: mpsc::Unbou tx: response_tx, msg_input, + msg_query, }; tokio::spawn(handle_connection(conn)); @@ -179,13 +198,13 @@ async fn accept_connection(stream: tokio::net::TcpStream, msg_input: mpsc::Unbou } } -pub async fn start_websocket(address: SocketAddr, message_tx: mpsc::UnboundedSender) -> Result<(), WebSocketError> { +pub async fn start_websocket(address: SocketAddr, message_tx: mpsc::UnboundedSender, received_messages_query_tx: mpsc::UnboundedSender) -> 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())); + tokio::spawn(accept_connection(stream, message_tx.clone(), received_messages_query_tx.clone())); } eprintln!("The websocket went kaput...");