British-Information-Technologies
/
ChatKit-server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ref-method and command merge #4

Merged
Mitch161 merged 39 commits from cmd--optimization into master 2020-08-19 21:29:44 +00:00
Conversation 0 Commits 39 Files Changed 19 +96 -90
4 changed files with 96 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit be167055e8 - Show all commits

28
src/client_api/mod.rs
View File

@ -16,19 +16,18 @@ pub struct ClientApi {
}
impl ClientApi {
pub fn new(addr: &str) -> Self {
let socket = TcpStream::connect(addr).expect("connection failed");
pub fn new(addr: &str) -> Result<Self, io::Error> {
let socket = TcpStream::connect(addr)?;
let on_add = |_client: Client| {println!("Client_api: Client added {:?}", _client)};
let on_remove = |_uuid: String| {println!("Client_api: Client removed {}", _uuid)};
Self {
let a = Self {
socket,
addr: addr.to_string(),
on_client_add_handle: on_add,
on_client_remove_handle: on_remove,
}
};
Ok(a)
}
pub fn set_on_client_add(&mut self, func: fn(Client) -> ()) {
@ -42,16 +41,23 @@ impl ClientApi {
pub fn get_info(host: &str) -> Result<Commands, io::Error> {
let mut buffer: [u8; 1024] = [0; 1024];
let addr = host.parse().unwrap();
let mut stream = TcpStream::connect_timeout(&addr, Duration::from_millis(10000))?;
let mut stream = TcpStream::connect_timeout(&addr, Duration::from_millis(1000))?;
let _ = stream.read(&mut buffer)?;
println!("data recieved: {:?}", &buffer[0..20]);
match Commands::from(&buffer) {
Commands::Request(None) => {
println!("zeroing");
buffer.zeroize();
stream.write_all(Commands::Info(None).to_string().as_bytes()).unwrap();
let a = stream.read(&mut buffer);
a?;
println!("writing");
let sending_command = Commands::Info(None).to_string();
println!("sending string: {:?} as_bytes: {:?}", &sending_command, &sending_command.as_bytes());
stream.write_all(sending_command.as_bytes())?;
stream.flush()?;
println!("reading");
let bytes = stream.read(&mut buffer)?;
println!("new buffer size: {:?} contents: {:?}", bytes, &buffer[0..20]);
println!("commanding");
Ok(Commands::from(String::from(String::from_utf8_lossy(&buffer))))
},
_ => {

31
src/main.rs
View File

@ -114,13 +114,10 @@ fn control_panel() -> ResizedView<Panel<LinearLayout>> {
// MARK: - general testing zone
#[cfg(test)]
mod tests {
#![feature(test)]
use super::Server;
use crate::server::server_profile::Server;
use crate::client_api::ClientApi;
use std::thread::spawn;
use std::collections::HashMap;
use crate::commands::Commands;
use log::info;
#[test]
fn test_server_info() {
@ -130,42 +127,36 @@ mod tests {
let owner = "noreply@email.com";
let server = Server::new(name, address, owner);
let _ = server.start().unwrap();
let result = server.start();
assert_eq!(result.is_ok(), true);
let api = ClientApi::get_info("127.0.0.1:6000");
assert_eq!(api.is_ok(), true);
if api.is_ok() {
if let Ok(api) = api {
println!("received: {:?}", api);
let mut map = HashMap::new();
map.insert("name".to_string(), name.to_string());
map.insert("owner".to_string(), owner.to_string());
let expected = Commands::Info(Some(map));
let api = api.unwrap();
println!("expected: {:?}", expected);
assert_eq!(api, expected);
} else {
return
}
}
#[test]
fn test_server_connect() {
let name = "Server-01";
let address = "0.0.0.0:6000";
let address = "0.0.0.0:6001";
let owner = "noreply@email.com";
let server = Server::new(name, address, owner);
let _ = server.start().unwrap();
let api = ClientApi::get_info("127.0.0.1:6000");
assert_eq!(api.is_ok(), true);
if let Commands::Success(Some(params)) = api.unwrap() {
let mut api = ClientApi::new(address);
api.on_client_add_handle = |s| info!("new clinet: {:?}", s);
api.on_client_remove_handle = |s| info!("removed clinet: {:?}", s);
let api_result = ClientApi::new(address);
assert_eq!(api_result.is_ok(), true);
if let Ok(api) = api_result {
std::thread::sleep(std::time::Duration::from_secs(2));
}
}
}

105
src/server/client/client_profile.rs
View File

@ -56,60 +56,14 @@ impl Client {
// TODO: - add heartbeat timer.
pub fn handle_connection(&self) {
info!("{}: handling connection", self.uuid);
println!("buffer");
let mut buffer = [0; 1024];
// test to see if there is anything for the client to receive from its channel
match self.receiver.try_recv() {
/*command is on the channel*/
Ok(Commands::ClientRemove(Some(params))) => {
let mut stream = self.stream_arc.lock().unwrap();
let mut retry: u8 = 3;
'retry_loop1: loop {
if retry < 1 {
self.transmit_data(Commands::Error(None).to_string().as_str());
break 'retry_loop1
}
self.transmit_data(Commands::ClientRemove(Some(params.clone())).to_string().as_str());
let _ = stream.read(&mut buffer);
let command = Commands::from(&buffer);
if command == Commands::Success(None) {
break 'retry_loop1;
} else {
retry -= 1;
}
}
},
Ok(Commands::Client(Some(params))) => {
let mut stream = self.stream_arc.lock().unwrap();
let mut retry: u8 = 3;
'retry_loop2: loop {
if retry < 1 {
stream.write_all(Commands::Error(None).to_string().as_bytes());
break 'retry_loop2;
}
stream.write_all(Commands::Client(Some(params.clone())).to_string().as_bytes());
let _ = stream.read(&mut buffer);
let command = Commands::from(&buffer);
if command == Commands::Success(None) {
break 'retry_loop2;
} else {
retry -= 1;
}
}
},
/*no data available yet*/
Err(TryRecvError::Empty) => {},
_ => {}
// TODO: - Check heartbeat
{
info!("heartbeat")
}
println!("socket");
info!("{}: handling connection", self.uuid);
if self.stream_arc.lock().unwrap().peek(&mut buffer).is_ok() {
let mut stream = self.stream_arc.lock().unwrap();
@ -136,6 +90,55 @@ impl Client {
}
}
}
println!("buffer");
// test to see if there is anything for the client to receive from its channel
match self.receiver.try_recv() {
/*command is on the channel*/
Ok(Commands::ClientRemove(Some(params))) => {
let mut stream = self.stream_arc.lock().unwrap();
let mut retry: u8 = 3;
'retry_loop1: loop {
if retry < 1 {
self.transmit_data(Commands::Error(None).to_string().as_str());
break 'retry_loop1
} else {
self.transmit_data(Commands::ClientRemove(Some(params.clone())).to_string().as_str());
let _ = stream.read(&mut buffer);
let command = Commands::from(&buffer);
if command == Commands::Success(None) {
break 'retry_loop1;
} else {
retry -= 1;
}
}
}
},
Ok(Commands::Client(Some(params))) => {
let mut stream = self.stream_arc.lock().unwrap();
let mut retry: u8 = 3;
'retry_loop2: loop {
if retry < 1 {
let _ = stream.write_all(Commands::Error(None).to_string().as_bytes());
break 'retry_loop2;
} else {
let _ = stream.write_all(Commands::Client(Some(params.clone())).to_string().as_bytes());
let _ = stream.read(&mut buffer);
let command = Commands::from(&buffer);
if command == Commands::Success(None) {
break 'retry_loop2;
} else {
retry -= 1;
}
}
}
},
/*no data available yet*/
Err(TryRecvError::Empty) => {},
_ => {}
}
println!("end");
}

22
src/server/server_profile.rs
View File

@ -60,8 +60,7 @@ impl Server {
address: address.to_string(),
author: author.to_string(),
connected_clients: Arc::new(Mutex::new(HashMap::new())),
thread_pool: ThreadPool::new(16),
thread_pool: ThreadPool::new(16),
sender,
receiver,
@ -72,6 +71,10 @@ impl Server {
self.address.to_string()
}
pub fn set_port(&mut self) {
}
pub fn start<'a>(&self) -> Result<(), io::Error>{
info!("server: starting server...");
// clone elements for thread
@ -118,7 +121,7 @@ impl Server {
info!("server: checking for new connections");
if let Ok((mut stream, _addr)) = listener.accept() {
stream.set_read_timeout(Some(Duration::from_millis(10000))).unwrap();
stream.set_read_timeout(Some(Duration::from_millis(1000))).unwrap();
let _ = stream.set_nonblocking(false);
let request = Commands::Request(None);
@ -129,7 +132,7 @@ impl Server {
if let Ok(size) = stream.read(&mut buffer) {
let incoming_message = String::from(String::from_utf8_lossy(&buffer));
let command = Commands::from(incoming_message);
info!("Server: new connection sent - {:?}", command);
println!("Server: new connection sent - {:?}", command);
// clears the buffer.
buffer.zeroize();
@ -153,18 +156,21 @@ impl Server {
// TODO: - correct connection reset error when getting info.
Commands::Info(None) => {
info!("Server: info requested");
println!("Server: info requested");
let mut params: HashMap<String, String> = HashMap::new();
params.insert(String::from("name"), server_details.0.clone());
params.insert(String::from("owner"), server_details.1.clone());
let command = Commands::Info(Some(params));
stream.write_all(command.to_string().as_bytes()).expect("Server -Info: writing failed");
stream.flush().expect("Server -Info: flushing errored");
let result = stream.write_all(command.to_string().as_bytes());
if let Err(error) = result {
println!("Server: error {:?}", error);
}
let _ = stream.flush();
},
_ => {
info!("Server: Invalid command sent");
println!("Server: Invalid command sent");
let _ = stream.write_all(Commands::Error(None).to_string().as_bytes());
let _ = stream.flush();
},