From 16ddb5015910fa01fe257c2756c705ab10966c8d Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Dec 2019 14:55:23 +0000 Subject: [PATCH] Moved client-related functionalities to separate module --- src/provider/client_handling/mod.rs | 75 +++++++++++++++++++++++ src/provider/mod.rs | 94 +++++------------------------ 2 files changed, 91 insertions(+), 78 deletions(-) diff --git a/src/provider/client_handling/mod.rs b/src/provider/client_handling/mod.rs index e69de29bb2..d43f20142e 100644 --- a/src/provider/client_handling/mod.rs +++ b/src/provider/client_handling/mod.rs @@ -0,0 +1,75 @@ +use crate::provider::storage::{ClientStorage, StoreError}; +use sfw_provider_requests::requests::{ProviderRequestError, ProviderRequests, PullRequest}; +use sfw_provider_requests::responses::{ProviderResponse, PullResponse}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, RwLock}; + +#[derive(Debug)] +pub enum ClientProcessingError { + ClientDoesntExistError, + StoreError, + InvalidRequest, +} + +impl From for ClientProcessingError { + fn from(_: ProviderRequestError) -> Self { + use ClientProcessingError::*; + + InvalidRequest + } +} + +impl From for ClientProcessingError { + fn from(_: StoreError) -> Self { + use ClientProcessingError::*; + + StoreError + } +} + +#[derive(Debug, Clone)] +pub(crate) struct ClientProcessingData { + store_dir: PathBuf, +} + +impl ClientProcessingData { + pub(crate) fn new(store_dir: PathBuf) -> Self { + ClientProcessingData { store_dir } + } + + pub(crate) fn add_arc_rwlock(self) -> Arc> { + Arc::new(RwLock::new(self)) + } +} + +pub(crate) struct ClientRequestProcessor(()); + +impl ClientRequestProcessor { + pub(crate) fn process_client_request( + data: &[u8], + processing_data: &RwLock, + ) -> Result, ClientProcessingError> { + let client_request = ProviderRequests::from_bytes(&data)?; + println!("received the following request: {:?}", client_request); + match client_request { + ProviderRequests::Register(req) => unimplemented!(), + ProviderRequests::PullMessages(req) => { + Ok(ClientRequestProcessor::process_pull_messages_request( + req, + processing_data.read().unwrap().store_dir.as_path(), + )? + .to_bytes()) + } + } + } + + fn process_pull_messages_request( + req: PullRequest, + store_dir: &Path, + ) -> Result { + println!("processing pull!"); + let retrieved_messages = + ClientStorage::retrieve_client_files(req.destination_address, store_dir)?; + Ok(PullResponse::new(retrieved_messages)) + } +} diff --git a/src/provider/mod.rs b/src/provider/mod.rs index b06e9b472e..d85e0a44bc 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -1,19 +1,23 @@ use std::fs::{File, ReadDir}; use std::io::Write; -use std::net::{SocketAddr, Shutdown}; +use std::net::{Shutdown, SocketAddr}; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::RwLock; -use sfw_provider_requests::requests::*; -use sfw_provider_requests::responses::*; + use curve25519_dalek::scalar::Scalar; use rand::Rng; +use sfw_provider_requests::DUMMY_MESSAGE_CONTENT; +use sfw_provider_requests::requests::*; +use sfw_provider_requests::responses::*; use sphinx::{ProcessedPacket, SphinxPacket}; use sphinx::route::{DestinationAddressBytes, SURBIdentifier}; use tokio::prelude::*; use tokio::runtime::Runtime; -use sfw_provider_requests::DUMMY_MESSAGE_CONTENT; -use crate::provider::storage::{StoreData, ClientStorage, StoreError}; + +use crate::provider::storage::{ClientStorage, StoreData, StoreError}; +use crate::provider::client_handling::{ClientProcessingData, ClientRequestProcessor}; +use crate::provider::mix_handling::{MixProcessingData, MixPacketProcessor}; mod client_handling; mod mix_handling; @@ -22,72 +26,7 @@ mod storage; // TODO: if we ever create config file, this should go there const STORED_MESSAGE_FILENAME_LENGTH: usize = 16; -const MESSAGE_RETRIEVAL_LIMIT:usize = 2; - - - -#[derive(Debug)] -enum ClientProcessingError { - ClientDoesntExistError, - StoreError, - InvalidRequest, -} - -impl From for ClientProcessingError { - fn from(_: ProviderRequestError) -> Self { - use ClientProcessingError::*; - - InvalidRequest - } -} - -impl From for ClientProcessingError { - fn from(_: StoreError) -> Self { - use ClientProcessingError::*; - - StoreError - } -} - -#[derive(Debug, Clone)] -struct ClientProcessingData { - store_dir: PathBuf, -} - -impl ClientProcessingData { - fn new(store_dir: PathBuf) -> Self { - ClientProcessingData { - store_dir, - } - } - - fn add_arc_rwlock(self) -> Arc> { - Arc::new(RwLock::new(self)) - } -} - - - -struct ClientRequestProcessor(()); - -impl ClientRequestProcessor { - fn process_client_request(data: &[u8], processing_data: &RwLock) -> Result, ClientProcessingError> { - let client_request = ProviderRequests::from_bytes(&data)?; - println!("received the following request: {:?}", client_request); - match client_request { - ProviderRequests::Register(req) => unimplemented!(), - ProviderRequests::PullMessages(req) => { - Ok(ClientRequestProcessor::process_pull_messages_request(req, processing_data.read().unwrap().store_dir.as_path())?.to_bytes()) - } - } - } - - fn process_pull_messages_request(req: PullRequest, store_dir: &Path) -> Result{ - println!("processing pull!"); - let retrieved_messages = ClientStorage::retrieve_client_files(req.destination_address, store_dir)?; - Ok(PullResponse::new(retrieved_messages)) - } -} +const MESSAGE_RETRIEVAL_LIMIT: usize = 2; pub struct ServiceProvider { @@ -108,7 +47,7 @@ impl ServiceProvider { } - async fn process_mixnet_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc>) { + async fn process_mixnet_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc>) { let mut buf = [0u8; sphinx::PACKET_SIZE]; // In a loop, read data from the socket and write the data back. @@ -120,7 +59,7 @@ impl ServiceProvider { return; } Ok(_) => { - let store_data = match mix_handling::MixPacketProcessor::process_sphinx_data_packet(buf.as_ref(), processing_data.as_ref()) { + let store_data = match MixPacketProcessor::process_sphinx_data_packet(buf.as_ref(), processing_data.as_ref()) { Ok(sd) => sd, Err(e) => { eprintln!("failed to process sphinx packet; err = {:?}", e); @@ -156,7 +95,7 @@ impl ServiceProvider { } // TODO: FIGURE OUT HOW TO SET READ_DEADLINES IN TOKIO - async fn process_client_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc>) { + async fn process_client_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc>) { let mut buf = [0; 1024]; // TODO: restore the for loop once we go back to persistent tcp socket connection @@ -189,17 +128,17 @@ impl ServiceProvider { Ok(res) => { println!("should send this response! {:?}", res); ServiceProvider::send_response(socket, &res).await; - }, + } _ => { println!("we failed..."); ServiceProvider::send_response(socket, b"bad foomp").await; - }, + } } } async fn start_mixnet_listening(&self) -> Result<(), Box> { let mut listener = tokio::net::TcpListener::bind(self.mix_network_address).await?; - let processing_data = mix_handling::MixProcessingData::new(self.secret_key, self.store_dir.clone()).add_arc_rwlock(); + let processing_data = MixProcessingData::new(self.secret_key, self.store_dir.clone()).add_arc_rwlock(); loop { let (socket, _) = listener.accept().await?; @@ -225,7 +164,6 @@ impl ServiceProvider { ServiceProvider::process_client_socket_connection(socket, thread_processing_data).await }); } - } async fn start_listeners(&self) -> (Result<(), Box>, Result<(), Box>) {