From e9e5c83ab358737f46c96b82678fbcbac2fb8ab2 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 21 Jan 2020 10:33:26 +0000 Subject: [PATCH] Moved ReceivedMessagesBuffer to seperate module --- nym-client/src/client/mod.rs | 56 ++---------------------- nym-client/src/client/received_buffer.rs | 55 +++++++++++++++++++++++ nym-client/src/sockets/tcp.rs | 2 +- nym-client/src/sockets/ws.rs | 2 +- 4 files changed, 60 insertions(+), 55 deletions(-) create mode 100644 nym-client/src/client/received_buffer.rs diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index bfd9e7198d..376e3a102c 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -4,21 +4,21 @@ use crate::sockets::tcp; use crate::sockets::ws; use crate::utils; use directory_client::presence::Topology; -use futures::channel::{mpsc, oneshot}; +use futures::channel::mpsc; use futures::join; -use futures::lock::Mutex as FMutex; use futures::select; use futures::{SinkExt, StreamExt}; use log::*; use sfw_provider_requests::AuthToken; use sphinx::route::{Destination, DestinationAddressBytes}; use std::net::SocketAddr; -use std::sync::Arc; use std::time::Duration; use tokio::runtime::Runtime; use topology::NymTopology; +use crate::client::received_buffer::ReceivedMessagesBuffer; mod mix_traffic; +pub mod received_buffer; const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5; // seconds @@ -26,56 +26,6 @@ const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 0.5; // seconds; const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds; -pub type BufferResponse = oneshot::Sender>>; - -struct ReceivedMessagesBuffer { - messages: Vec>, -} - -impl ReceivedMessagesBuffer { - fn add_arc_futures_mutex(self) -> Arc> { - Arc::new(FMutex::new(self)) - } - - fn new() -> Self { - ReceivedMessagesBuffer { - messages: Vec::new(), - } - } - - async fn add_new_messages(buf: Arc>, msgs: Vec>) { - info!("Adding new messages to the buffer! {:?}", msgs); - let mut unlocked = buf.lock().await; - unlocked.messages.extend(msgs); - } - - async fn run_poller_input_controller( - buf: Arc>, - mut poller_rx: mpsc::UnboundedReceiver>>, - ) { - while let Some(new_messages) = poller_rx.next().await { - ReceivedMessagesBuffer::add_new_messages(buf.clone(), new_messages).await; - } - } - - async fn acquire_and_empty(buf: Arc>) -> Vec> { - let mut unlocked = buf.lock().await; - std::mem::replace(&mut unlocked.messages, Vec::new()) - } - - async fn run_query_output_controller( - buf: Arc>, - mut query_receiver: mpsc::UnboundedReceiver, - ) { - while let Some(request) = query_receiver.next().await { - let messages = ReceivedMessagesBuffer::acquire_and_empty(buf.clone()).await; - // if this fails, the whole application needs to blow - // because currently only this thread would fail - request.send(messages).unwrap(); - } - } -} - pub enum SocketType { TCP, WebSocket, diff --git a/nym-client/src/client/received_buffer.rs b/nym-client/src/client/received_buffer.rs new file mode 100644 index 0000000000..4c6d460290 --- /dev/null +++ b/nym-client/src/client/received_buffer.rs @@ -0,0 +1,55 @@ +use futures::channel::{mpsc, oneshot}; +use futures::lock::Mutex as FMutex; +use futures::StreamExt; +use log::info; +use std::sync::Arc; + +pub type BufferResponse = oneshot::Sender>>; + +pub(crate) struct ReceivedMessagesBuffer { + messages: Vec>, +} + +impl ReceivedMessagesBuffer { + pub(crate) fn new() -> Self { + ReceivedMessagesBuffer { + messages: Vec::new(), + } + } + + pub(crate) fn add_arc_futures_mutex(self) -> Arc> { + Arc::new(FMutex::new(self)) + } + + pub(crate) async fn add_new_messages(buf: Arc>, msgs: Vec>) { + info!("Adding new messages to the buffer! {:?}", msgs); + let mut unlocked = buf.lock().await; + unlocked.messages.extend(msgs); + } + + pub(crate) async fn run_poller_input_controller( + buf: Arc>, + mut poller_rx: mpsc::UnboundedReceiver>>, + ) { + while let Some(new_messages) = poller_rx.next().await { + ReceivedMessagesBuffer::add_new_messages(buf.clone(), new_messages).await; + } + } + + pub(crate) async fn acquire_and_empty(buf: Arc>) -> Vec> { + let mut unlocked = buf.lock().await; + std::mem::replace(&mut unlocked.messages, Vec::new()) + } + + pub(crate) async fn run_query_output_controller( + buf: Arc>, + mut query_receiver: mpsc::UnboundedReceiver, + ) { + while let Some(request) = query_receiver.next().await { + let messages = ReceivedMessagesBuffer::acquire_and_empty(buf.clone()).await; + // if this fails, the whole application needs to blow + // because currently only this thread would fail + request.send(messages).unwrap(); + } + } +} diff --git a/nym-client/src/sockets/tcp.rs b/nym-client/src/sockets/tcp.rs index a6cf018bec..c063c7e4b8 100644 --- a/nym-client/src/sockets/tcp.rs +++ b/nym-client/src/sockets/tcp.rs @@ -1,4 +1,4 @@ -use crate::client::BufferResponse; +use crate::client::received_buffer::BufferResponse; use crate::client::InputMessage; use directory_client::presence::Topology; use futures::channel::{mpsc, oneshot}; diff --git a/nym-client/src/sockets/ws.rs b/nym-client/src/sockets/ws.rs index 685964c898..14aacdcb5d 100644 --- a/nym-client/src/sockets/ws.rs +++ b/nym-client/src/sockets/ws.rs @@ -1,4 +1,4 @@ -use crate::client::BufferResponse; +use crate::client::received_buffer::BufferResponse; use crate::client::InputMessage; use directory_client::presence::Topology; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};