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

added core server funcs to ui

This commit is contained in:
Mitchell 2020-08-10 18:53:53 +01:00
parent 6d3ffc5ae5
commit fa57f25d8a
6 changed files with 505 additions and 230 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
Cargo.toml
View File

@ -9,13 +9,21 @@ edition = "2018"
[dependencies]
regex = "1"
crossbeam = "0.7"
parking_lot = "0.10"
crossbeam-channel = "0.4"
crossbeam-utils = "0.7"
crossbeam-queue = "0.2"
parking_lot = "0.10"
dashmap = "3.11.4"
async-std = "1.6.2"
lazy_static = "1.4.0"
rayon = "1.3.1"
diesel = { version = "1.4.5", features = ["sqlite"] }
zeroize = "1.1.0"
cursive = { version = "0.15.0", default-features = false, features = ["crossterm-backend"]}
crossterm = "0.17.7"
log = "0.4"
[profile.dev]
opt-level = 0

70
src/client_api/mod.rs Normal file
View File

@ -0,0 +1,70 @@
use std::{
net::TcpStream,
io::{Write, Read}
};
use crate::{
server::client::client_profile::Client,
commands::Commands,
};
use zeroize::Zeroize;
use std::time::Duration;
use async_std::net::SocketAddrV4;
use std::str::FromStr;
use std::net::SocketAddr;
pub struct ClientApi {
socket: TcpStream,
addr: String,
pub on_client_add_handle: fn(Client) -> (),
pub on_client_remove_handle: fn(String) -> (),
}
impl ClientApi {
pub fn new(addr: &str) -> Self {
let socket = TcpStream::connect(addr).expect("connection failed");
let on_add = |_client: Client| {println!("Client_api: Client added {:?}", _client)};
let on_remove = |_uuid: String| {println!("Client_api: Client removed {}", _uuid)};
Self {
socket,
addr: addr.to_string(),
on_client_add_handle: on_add,
on_client_remove_handle: on_remove,
}
}
pub fn set_on_client_add(&mut self, Fn: fn(Client) -> ()) {
self.on_client_add_handle = Fn;
}
pub fn set_on_client_removed(&mut self, Fn: fn(String) -> ()) {
self.on_client_remove_handle = Fn;
}
pub fn get_info(host: &str) -> Option<Commands> {
let mut buffer: [u8; 1024] = [0; 1024];
let addr = SocketAddr::from_str(host).ok()?;
let mut stream = TcpStream::connect_timeout(&addr, Duration::from_millis(500)).ok()?;
stream.read(&mut buffer).ok()?;
match Commands::from(&buffer) {
Commands::Request(None) => {
stream.write_all(Commands::Info(None).to_string().as_bytes()).unwrap();
stream.read(&mut buffer).ok()?;
Some(Commands::from(String::from(String::from_utf8_lossy(&buffer))))
},
_ => {
None
}
}
}
pub fn get_clients(&self) {
}
}

108
src/server/commands/mod.rs → src/commands/mod.rs
View File

@ -1,27 +1,18 @@
mod request;
mod info;
mod success;
mod error;
mod connect;
mod disconnect;
mod client_update;
mod client_info;
mod client;
mod test;
mod message;
use std::string::ToString;
use std::collections::HashMap;
use dashmap::DashMap;
use std::borrow::Borrow;
use regex::Regex;
use std::ops::Index;
use std::ops::{Index};
#[derive(Clone, Debug)]
// MARK: - command struct
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum Commands {
Request(Option<HashMap<String, String>>),
Info(Option<HashMap<String, String>>),
HeartBeat(Option<HashMap<String, String>>),
Connect(Option<HashMap<String, String>>),
Disconnect(Option<HashMap<String, String>>),
@ -34,53 +25,6 @@ pub enum Commands {
Error(Option<HashMap<String, String>>),
}
impl Commands {
fn compare_params(&self, params: &Option<HashMap<String, String>>, other_params: &Option<HashMap<String, String>>) -> bool {
match (params, other_params) {
(None, Some(_other_params)) => false,
(Some(_params), None) => false,
(None, None) => true,
(Some(params), Some(other_params)) => {
let mut result = false;
if params.len() == other_params.len() {
for (key, value) in params.iter() {
if let Some(other_value) = other_params.get(key) {
if value != other_value {
result = false;
break;
} else {
result = true;
}
}
}
}
result
},
}
}
}
impl PartialEq for Commands {
fn eq(&self, other: &Self) -> bool {
match (self, other) {
(Commands::Request(params), Commands::Request(other_params)) => self.compare_params(&params, &other_params),
(Commands::Info(params), Commands::Info(other_params)) => self.compare_params(&params, &other_params),
(Commands::Connect(params), Commands::Connect(other_params)) => self.compare_params(&params, &other_params),
(Commands::Disconnect(params), Commands::Disconnect(other_params)) => self.compare_params(&params, &other_params),
(Commands::ClientUpdate(params), Commands::ClientUpdate(other_params)) => self.compare_params(&params, &other_params),
(Commands::ClientInfo(params), Commands::ClientInfo(other_params)) => self.compare_params(&params, &other_params),
(Commands::ClientRemove(params), Commands::ClientRemove(other_params)) => self.compare_params(&params, &other_params),
(Commands::Client(params), Commands::Client(other_params)) => self.compare_params(&params, &other_params),
(Commands::Success(params), Commands::Success(other_params)) => self.compare_params(&params, &other_params),
(Commands::Error(params), Commands::Error(other_params)) => self.compare_params(&params, &other_params),
_ => false,
}
}
}
impl ToString for Commands {
fn to_string(&self) -> std::string::String {
@ -89,12 +33,13 @@ impl ToString for Commands {
let (command, parameters) = match self {
Commands::Request(arguments) => { ("!request:", arguments) },
Commands::Info(arguments) => { ("!info:", arguments) },
Commands::HeartBeat(arguments) => {("!heartbeat:", arguments)},
Commands::Connect(arguments) => { ("!connect:", arguments) },
Commands::Disconnect(arguments) => { ("!disconnect:", arguments) },
Commands::ClientUpdate(arguments) => { ("!clientUpdate:", arguments) },
Commands::ClientInfo(arguments) => { ("!clientInfo:", arguments) },
Commands::ClientRemove(arguments) => { ("!clientRemove", arguments) }
Commands::Client(arguments) => { ("!client:", arguments) },
Commands::Success(arguments) => { ("!success:", arguments) },
Commands::Error(arguments) => { ("!error:", arguments) },
_ => { ("!error:", &None) }
};
@ -107,19 +52,31 @@ impl ToString for Commands {
out_string.push_str(" ");
out_string.push_str(k.as_str());
out_string.push_str(":");
out_string.push_str(v.as_str())
if v.contains(":") {
out_string.push_str(format!("\"{}\"",v.as_str()).as_str())
} else {
out_string.push_str(v.as_str());
}
}
}
out_string
}
}
impl From<&str> for Commands {
fn from(data: &str) -> Self {
let regex = Regex::new(r###"(\?|!)([a-zA-z0-9]*):|([a-zA-z]*):([a-zA-Z0-9\-\+\[\]{}_=/]+|("(.*?)")+)"###).unwrap();
let regex = Regex::new(r###"(\?|!)([a-zA-z0-9]*):|([a-zA-z]*):([a-zA-Z0-9@\-\+\[\]{}_=/.]+|("(.*?)")+)"###).unwrap();
let mut iter = regex.find_iter(data);
let command = iter.next().unwrap().as_str();
let command_opt = iter.next();
if command_opt.is_none() {
return Commands::Error(None);
}
let command = command_opt.unwrap().as_str();
println!("command: {:?}", command);
@ -132,12 +89,14 @@ impl From<&str> for Commands {
map.insert(parts.index(0).to_string(), parts.index(1).to_string());
}
let params = if map.capacity() > 0 {Some(map)} else { None };
let params = if map.capacity() > 1 {Some(map)} else { None };
match command {
"!request:" => Commands::Request(params),
"!info:" => Commands::Info(params),
"!heartbeat:" => Commands::HeartBeat(params),
"!connect:" => Commands::Connect(params),
"!disconnect:" => Commands::Disconnect(params),
@ -163,17 +122,17 @@ impl From<String> for Commands {
impl From<&[u8; 1024]> for Commands {
fn from(data: &[u8; 1024]) -> Self {
let incoming_message = String::from(String::from_utf8_lossy(data));
Commands::from(incoming_message.as_str())
Commands::from(incoming_message)
}
}
#[cfg(test)]
mod test_commands_v2 {
#![feature(test)]
extern crate test;
use super::Commands;
use std::collections::HashMap;
use test::Bencher;
#[test]
fn test_creation_from_string() {
@ -193,4 +152,9 @@ mod test_commands_v2 {
println!("{:?}", command.to_string())
}
#[bench]
fn benchmark(b: &mut Bencher) {
b.iter(|| Commands::from("!connect: host:192.168.0.1 name:\"michael-bailey\" uuid:123456-1234-1234-123456"))
}
}

137
src/main.rs
View File

@ -1,23 +1,142 @@
#[macro_use]
extern crate lazy_static;
mod client_api;
mod commands;
mod server;
use crate::server::server_profile::Server;
fn main(){
use crate::server::server_profile::Server;
use client_api::ClientApi;
use crossterm::ErrorKind;
use cursive::{
Cursive,
menu::*,
event::Key,
views::{ Dialog, TextView, LinearLayout, ListView, ResizedView, Panel },
Rect,
CursiveExt,
align::{Align, HAlign},
view::SizeConstraint,
};
use std::sync::Arc;
use log::info;
fn main() -> Result<(), ErrorKind> {
lazy_static!{
static ref SERVER_NAME: &'static str = "Server-01";
static ref SERVER_ADDRESS: &'static str = "0.0.0.0:6000";
static ref SERVER_AUTHOR: &'static str = "noreply@email.com";
static ref SERVER: Server<'static> = Server::new(&SERVER_NAME, &SERVER_ADDRESS, &SERVER_AUTHOR);
}
/*
let server_name = String::from("Server-01");
let server_address = String::from("0.0.0.0:6000");
let server_author = String::from("noreply@email.com");
*/
//let server = Server::new("Server-01", "0.0.0.0:6000", "noreply@email.com");
/*server_name: &'static str = "Server-01";
server_address: &'static str = "0.0.0.0:6000";
server_author: &'static str = "noreply@email.com";
let server = Server::new(&server_name, &server_address, &server_author);*/
/*let server_arc = Arc::new(SERVER);
let s1 = server_arc.clone();
let s2 = s1.clone();*/
//let server = Server::new(server_name, server_address, server_author);
SERVER.start();
cursive::logger::init();
info!("Main: init Display");
let mut Display = Cursive::default();
info!("Main: setting up callbacks");
Display.add_global_callback(Key::Backspace, |s| s.quit());
Display.add_global_callback(Key::Tab, |s| s.toggle_debug_console());
Display.add_global_callback(Key::Esc, |s| s.select_menubar());
info!("Main: setting up menu bar");
let _ = Display.menubar()
.add_subtree("Server",
MenuTree::new()
.leaf("About",
|s| s.add_layer(About()))
.delimiter()
.leaf("quit", |s| s.quit()))
.add_subtree("File",
MenuTree::new()
.leaf("Start", move |s| {SERVER.start();})
.leaf("Stop", move |s| {SERVER.stop();})
.delimiter()
.leaf("Debug", |s| {s.toggle_debug_console();}));
info!("Main: entering loop");
Display.add_layer(Control_Panel());
Display.run();
Ok(())
}
fn About() -> Dialog {
Dialog::new()
.content(TextView::new("Rust-Chat-Server\nmade by\n Mitchell Hardie\nMichael Bailey\nMit Licence")
).button("Close", |s| {let _ = s.pop_layer(); s.add_layer(Control_Panel())} )
}
fn Launch_screen() -> Dialog {
Dialog::new()
.content(TextView::new("\
Server.
* press <ESC> for menu bar
* press <TAB> for debug (FIXME)
* press <DEL> to exit.
").align(Align::center()))
.button("ok", |s| {s.pop_layer();})
}
fn Control_Panel() -> ResizedView<Panel<LinearLayout>> {
let mut root = LinearLayout::horizontal();
let mut left = LinearLayout::vertical();
let mut right = ListView::new();
right.add_child("test", TextView::new(""));
right.add_child("test", TextView::new(""));
right.add_delimiter();
right.add_child("test", TextView::new(""));
right.add_child("test", TextView::new(""));
left.add_child(TextView::new("Hello world"));
root.add_child(ResizedView::new(SizeConstraint::Full, SizeConstraint::Full, Panel::new(left)));
root.add_child(ResizedView::new(SizeConstraint::Full, SizeConstraint::Full, Panel::new(right)));
ResizedView::new(SizeConstraint::Fixed(64), SizeConstraint::Fixed(20), Panel::new(root))
}
// MARK: - general testing zone
/*#[cfg(test)]
mod tests {
#![feature(test)]
use super::Server;
use crate::client_api::ClientApi;
use std::thread::spawn;
use std::collections::HashMap;
use crate::commands::Commands;
#[test]
fn test_server_info() {
// setup the server
let name = "Server-01";
let address = "0.0.0.0:6000";
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");
if api.is_some() {
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();
assert_eq!(api, expected);
} else {
return
}
}
}*/

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

@ -1,44 +1,60 @@
extern crate regex;
use crate::server::server_profile::Server;
use crate::server::commands::{Commands};
use std::{
sync::Arc,
sync::Mutex,
net::{Shutdown, TcpStream},
io::prelude::*,
time::Duration,
io::Error,
collections::HashMap,
};
use crossbeam::{Sender, Receiver, TryRecvError, unbounded};
use zeroize::Zeroize;
use crate::{
server::{
server_profile::Server,
server_profile::ServerMessages,
},
commands::Commands
};
use std::net::{Shutdown, TcpStream};
use std::sync::Arc;
use parking_lot::FairMutex;
use std::collections::HashMap;
use dashmap::DashMap;
use std::io::prelude::*;
use std::time::Duration;
use std::io::Error;
use crossbeam::{Sender, Receiver, TryRecvError};
use crossbeam_channel::unbounded;
#[derive(Debug)]
pub struct Client<'a> {
connected: bool,
stream: Arc<TcpStream>,
uuid: String,
username: String,
address: String,
stream: TcpStream,
pub uuid: String,
pub username: String,
pub address: String,
pub sender: Sender<Commands>,
receiver: Receiver<Commands>,
server: &'a Server<'a>,
tx_channel: Sender<Commands>,
rx_channel: Receiver<Commands>,
}
impl<'a> Client<'a> {
pub fn new(server: &'a Server<'static>, stream: Arc<TcpStream>, uuid: &String, username: &String, address: &String) -> Self{
let (tx_channel, rx_channel): (Sender<Commands>, Receiver<Commands>) = unbounded();
pub fn new(server: &'a Server<'static>, stream: TcpStream, uuid: &str, username: &str, address: &str) -> Self {
let (sender, receiver): (Sender<Commands>, Receiver<Commands>) = unbounded();
stream.set_read_timeout(Some(Duration::from_secs(1))).unwrap();
Client {
connected: true,
stream,
stream: stream,
uuid: uuid.to_string(),
username: username.to_string(),
address: address.to_string(),
sender,
receiver,
server,
tx_channel,
rx_channel,
}
}
@ -49,7 +65,7 @@ impl<'a> Client<'a> {
#[allow(dead_code)]
pub fn get_transmitter(&self) -> &Sender<Commands>{
&self.tx_channel
&self.sender
}
#[allow(dead_code)]
@ -68,11 +84,13 @@ impl<'a> Client<'a> {
}
pub fn handle_connection(&mut self){
self.stream.set_read_timeout(Some(Duration::from_millis(500))).unwrap();
//self.stream.set_nonblocking(true).expect("set_nonblocking call failed");
let mut buffer = [0; 1024];
while self.connected {
match self.rx_channel.try_recv() {
println!("{}: channel checks", self.uuid);
match self.receiver.try_recv() {
/*command is on the channel*/
Ok(command) => {
let a = command.clone();
@ -104,12 +122,12 @@ impl<'a> Client<'a> {
let command = Commands::Client(Some(params));
self.transmit_data(command.to_string().as_str());
self.confirm_success();
self.confirm_success(&mut buffer);
},
Commands::Client(Some(_params)) => {
self.transmit_data(a.to_string().as_str());
self.confirm_success();
self.confirm_success(&mut buffer);
},
Commands::Success(_params) => {
self.transmit_data(a.to_string().as_str());
@ -131,7 +149,7 @@ impl<'a> Client<'a> {
* one. Ethier make sure commands sent require a response before sending the next one
* or make a system to check for these issues.
*/
match self.read_data() {
match self.read_data(&mut buffer) {
Ok(command) => {
match command {
Commands::Info(None) => {
@ -156,7 +174,7 @@ impl<'a> Client<'a> {
self.transmit_data(command.to_string().as_str());
let data: HashMap<String, String> = [(String::from("uuid"), self.uuid.clone())].iter().cloned().collect();
command = Commands::ClientUpdate(Some(data));
let command = Commands::ClientUpdate(Some(data));
self.server.update_all_clients(self.uuid.as_str(), command);
},
@ -183,28 +201,26 @@ impl<'a> Client<'a> {
println!("---Thread Exit---");
}
pub fn transmit_data(&self, data: &str){
pub fn transmit_data(&mut self, data: &str){
println!("Transmitting...");
println!("{} data: {}", self.uuid, data);
self.get_stream().write(data.to_string().as_bytes()).unwrap();
self.get_stream().flush().unwrap();
self.stream.write(data.to_string().as_bytes()).unwrap();
self.stream.flush().unwrap();
}
fn read_data(&self) -> Result<Commands, Error> {
let mut buffer = [0; 1024];
self.get_stream().read(&mut buffer)?;
let command = Commands::from(&buffer);
fn read_data(&mut self, buffer: &mut [u8; 1024]) -> Result<Commands, Error> {
self.stream.read(buffer)?;
let command = Commands::from(buffer);
Ok(command)
}
fn confirm_success(&self){
fn confirm_success(&mut self, buffer: &mut [u8; 1024]){
//self.stream.set_nonblocking(false).expect("set_nonblocking call failed");
//self.stream.set_read_timeout(Some(Duration::from_millis(3000))).expect("set_read_timeout call failed");
match self.read_data() {
match self.read_data(buffer) {
Ok(command) => {
match command {
Commands::Success(_params) => {
@ -242,12 +258,11 @@ impl<'a> Client<'a> {
success_message.push_str(&" ".to_string());
success_message.push_str(&data.to_string());
}
self.transmit_data(&success_message);
}
#[deprecated(since="24.7.20", note="will be removed in future, please do not use!")]
#[allow(dead_code)]
fn transmit_error(&self, data: &String){
fn transmit_error(&mut self, data: &String){
let mut error_message = "!error:".to_string();
if !data.is_empty(){
error_message.push_str(&" ".to_string());

311
src/server/server_profile.rs
View File

@ -1,36 +1,76 @@
extern crate regex;
extern crate rayon;
use crate::server::client::client_profile::Client;
use crate::server::commands::{Commands};
use crate::{
server::{
client::client_profile::Client,
},
commands::Commands
};
use std::{
sync::{Arc, Mutex},
net::{TcpStream, TcpListener},
collections::HashMap,
io::prelude::*,
time::Duration,
io::Error,
io::prelude::*,
thread,
io
};
use log::info;
use crossbeam_channel::{Sender, Receiver, unbounded};
use rust_chat_server::ThreadPool;
use std::net::{TcpStream, TcpListener};
use std::sync::{Arc, Mutex};
use crossbeam_channel::Sender;
use zeroize::Zeroize;
use parking_lot::FairMutex;
use std::collections::HashMap;
use std::io::Error;
use dashmap::DashMap;
use std::io::prelude::*;
use regex::Regex;
#[derive(Debug)]
pub enum ServerMessages {
RequestUpdate(String),
#[allow(dead_code)]
RequestInfo(String, String),
#[allow(dead_code)]
RequestDisconnect(String),
#[allow(dead_code)]
Shutdown,
}
// MARK: - server struct
#[derive(Debug)]
pub struct Server<'z> {
name: &'z str,
address: &'z str,
author: &'z str,
connected_clients: Arc<Mutex<HashMap<String, Sender<Commands>>>>,
thread_pool: ThreadPool,
sender: Sender<ServerMessages>,
receiver: Receiver<ServerMessages>,
}
// MARK: - server implemetation
impl<'z> Server<'z> {
pub fn new(name: &'z str, address: &'z str, author: &'z str) -> Self {
let (sender, receiver) = unbounded();
Self {
name: name,
address: address,
author: author,
connected_clients: Arc::new(Mutex::new(HashMap::new())),
thread_pool: ThreadPool::new(16),
sender,
receiver,
}
}
@ -46,82 +86,98 @@ impl<'z> Server<'z> {
self.author.to_string()
}
pub fn start(&'static self) {
let listener = TcpListener::bind(self.get_address()).unwrap();
pub fn start(&'static self) -> Result<(), io::Error> {
info!("server: starting server...");
// clone elements for thread
let client_map = self.connected_clients.clone();
let receiver = self.receiver.clone();
loop {
if let Ok((mut stream, addr)) = listener.accept() {
println!("Server: new connection, {}", addr);
let request = Commands::Request(None);
self.transmit_data(&stream, &request.to_string().as_str());
match self.read_data(&stream) {
Ok(command) => {
match command {
Commands::Connect(Some(data)) => {
let uuid = data.get("uuid").unwrap();
let username = data.get("name").unwrap();
let address = data.get("host").unwrap();
let stream = Arc::new(stream);
let mut client = Client::new(self, stream, &uuid, &username, &address);
let tx = client.get_transmitter();
let mut clients_hashmap = self.connected_clients.lock().unwrap();
clients_hashmap.insert(uuid.to_string(), tx.clone());
std::mem::drop(clients_hashmap);
let success = Commands::Success(None);
tx.send(success).unwrap();
self.thread_pool.execute(move || {
client.handle_connection();
});
let params: HashMap<String, String> = [(String::from("name"), username.clone()), (String::from("host"), address.clone()), (String::from("uuid"), uuid.clone())].iter().cloned().collect();
let new_client = Commands::Client(Some(params));
self.update_all_clients(uuid.as_str(), new_client);
},
Commands::Info(None) => {
let params: HashMap<String, String> = [(String::from("name"), self.name.to_string().clone()), (String::from("owner"), self.author.to_string().clone())].iter().cloned().collect();
let command = Commands::Success(Some(params));
self.transmit_data(&stream, command.to_string().as_str());
},
_ => {
println!("Invalid command!");
self.transmit_data(&stream, Commands::Error(None).to_string().as_str());
},
}
},
Err(_) => println!("ERROR: stream closed"),
}
}
}
}
pub fn update_client(&self, uuid: &str, command: &Commands){
let clients = self.connected_clients.lock().unwrap();
let tx = clients.get(&uuid.to_string()).unwrap();
tx.send(command.clone()).unwrap();
}
pub fn update_all_clients(&self, uuid: &str, command: Commands){
let clients = self.connected_clients.lock().unwrap();
// set up listener and buffer
let listener = TcpListener::bind(self.get_address())?;
listener.set_nonblocking(true);
for (client_uuid, tx) in clients.iter() {
if uuid != client_uuid.to_string() {
tx.send(command.clone()).unwrap();
info!("server: spawning threads");
thread::Builder::new().name("Server Thread".to_string()).spawn(move || {
let mut buffer = [0; 1024];
'outer: loop {
// get messages from the servers channel.
info!("server: getting messages");
for i in receiver.try_iter() {
match i {
ServerMessages::Shutdown => {
// TODO: implement disconnecting all clients and shutting down the server
info!("server: shutting down...");
break 'outer;
},
_ => {},
}
}
info!("server: checking for new connections");
if let Ok((mut stream, addr)) = listener.accept() {
stream.set_read_timeout(Some(Duration::from_millis(10000))).unwrap();
let request = Commands::Request(None);
self.transmit_data(&stream, &request.to_string().as_str());
match self.read_data(&stream, &mut buffer) {
Ok(command) => {
match command {
Commands::Connect(Some(data)) => {
let uuid = data.get("uuid").unwrap();
let username = data.get("name").unwrap();
let address = data.get("host").unwrap();
info!("{}", format!("Server: new Client connection: addr = {}", address ));
let mut client = Client::new(self, stream, &uuid, &username, &address);
let tx = client.get_transmitter();
let mut clients_hashmap = self.connected_clients.lock().unwrap();
clients_hashmap.insert(uuid.to_string(), tx.clone());
std::mem::drop(clients_hashmap);
let success = Commands::Success(None);
tx.send(success).unwrap();
self.thread_pool.execute(move || {
client.handle_connection();
});
let params: HashMap<String, String> = [(String::from("name"), username.clone()), (String::from("host"), address.clone()), (String::from("uuid"), uuid.clone())].iter().cloned().collect();
let new_client = Commands::Client(Some(params));
self.update_all_clients(uuid.as_str(), new_client);
},
Commands::Info(None) => {
info!("Server: info requested");
let params: HashMap<String, String> = [(String::from("name"), self.name.to_string().clone()), (String::from("owner"), self.author.to_string().clone())].iter().cloned().collect();
let command = Commands::Success(Some(params));
self.transmit_data(&stream, command.to_string().as_str());
},
_ => {
info!("Server: Invalid command sent");
self.transmit_data(&stream, Commands::Error(None).to_string().as_str());
},
}
},
Err(_) => println!("ERROR: stream closed"),
}
}
}
}
info!("server: stopped")
});
info!("server: started");
Ok(())
}
pub fn remove_client(&self, uuid: &str){
let mut clients = self.connected_clients.lock().unwrap();
clients.remove(&uuid.to_string());
pub fn stop(&self) {
info!("server: sending stop message");
self.sender.send(ServerMessages::Shutdown);
}
fn transmit_data(&self, mut stream: &TcpStream, data: &str){
@ -137,15 +193,35 @@ impl<'z> Server<'z> {
stream.flush().unwrap();
}
fn read_data(&self, mut stream: &TcpStream) -> Result<Commands, Error> {
let mut buffer = [0; 1024];
stream.read(&mut buffer)?;
let command = Commands::from(&buffer);
fn read_data(&self, mut stream: &TcpStream, buffer: &mut [u8; 1024]) -> Result<Commands, Error> {
stream.read(buffer)?;
let command = Commands::from(buffer);
Ok(command)
}
pub fn update_client(&self, uuid: &str, command: &Commands){
let clients = self.connected_clients.lock().unwrap();
let sender = clients.get(&uuid.to_string()).unwrap();
sender.send(command.clone()).unwrap();
}
pub fn update_all_clients(&self, uuid: &str, command: Commands){
let clients = self.connected_clients.lock().unwrap();
for (client_uuid, sender) in clients.iter() {
if uuid != client_uuid.to_string() {
sender.send(command.clone()).unwrap();
}
}
}
pub fn remove_client(&self, uuid: &str){
let mut clients = self.connected_clients.lock().unwrap();
clients.remove(&uuid.to_string());
}
#[deprecated(since="24.7.20", note="will be removed in future, please do not use!")]
#[allow(dead_code)]
pub fn get_info(&self, tx: Sender<Commands>) {
@ -169,6 +245,20 @@ impl<'z> Server<'z> {
}
}
impl<'z> Drop for Server<'z> {
fn drop(&mut self) {
println!("server dropped");
let _ = self.sender.send(ServerMessages::Shutdown);
}
}
struct ServerDelegate {
}
#[cfg(test)]
mod tests{
use super::*;
@ -204,16 +294,18 @@ mod tests{
}
fn establish_client_connection(uuid: &str) -> TcpStream {
let mut buffer = [0; 1024];
let mut stream = TcpStream::connect("0.0.0.0:6000").unwrap();
let mut command = read_data(&stream);
let mut command = read_data(&stream, &mut buffer);
assert_eq!(command, Commands::Request(None));
let msg: String = format!("!connect: uuid:{uuid} name:\"{name}\" host:\"{host}\"", uuid=uuid, name="alice", host="127.0.0.1");
transmit_data(&stream, msg.as_str());
command = read_data(&stream);
command = read_data(&stream, &mut buffer);
assert_eq!(command, Commands::Success(None));
@ -225,11 +317,9 @@ mod tests{
stream.flush().unwrap();
}
fn read_data(mut stream: &TcpStream) -> Commands {
let mut buffer = [0; 1024];
match stream.read(&mut buffer) {
Ok(_) => Commands::from(&buffer),
fn read_data(mut stream: &TcpStream, buffer: &mut [u8; 1024]) -> Commands {
match stream.read(buffer) {
Ok(_) => Commands::from(buffer),
Err(_) => Commands::Error(None),
}
}
@ -248,7 +338,7 @@ mod tests{
let mut stream = TcpStream::connect("0.0.0.0:6000").unwrap();
stream.read(&mut buffer).unwrap();
let mut command = Commands::from(&buffer);
let mut command = Commands::from(&mut buffer);
assert_eq!(command, Commands::Request(None));
@ -256,7 +346,7 @@ mod tests{
stream.write(msg).unwrap();
stream.read(&mut buffer).unwrap();
command = Commands::from(&buffer);
command = Commands::from(&mut buffer);
assert_eq!(command, Commands::Success(None));
@ -275,14 +365,14 @@ mod tests{
let mut stream = TcpStream::connect("0.0.0.0:6000").unwrap();
let command = read_data(&stream);
let command = read_data(&stream, &mut buffer);
assert_eq!(command, Commands::Request(None));
let msg = "!info:";
transmit_data(&stream, msg);
let command = read_data(&stream);
let command = read_data(&stream, &mut buffer);
let params: HashMap<String, String> = [(String::from("name"), String::from("test")), (String::from("owner"), String::from("test"))].iter().cloned().collect();
assert_eq!(command, Commands::Success(Some(params)));
@ -290,6 +380,8 @@ mod tests{
#[test]
fn test_client_info(){
let mut buffer = [0; 1024];
setup_server();
let mut stream = establish_client_connection("1234-5542-2124-155");
@ -297,7 +389,7 @@ mod tests{
let msg = "!info:";
transmit_data(&stream, msg);
let command = read_data(&stream);
let command = read_data(&stream, &mut buffer);
let params: HashMap<String, String> = [(String::from("name"), String::from("test")), (String::from("owner"), String::from("test"))].iter().cloned().collect();
assert_eq!(command, Commands::Success(Some(params)));
@ -311,6 +403,8 @@ mod tests{
#[test]
fn test_clientUpdate_solo(){
let mut buffer = [0; 1024];
setup_server();
let mut stream = establish_client_connection("1222-555-6-7");
@ -318,7 +412,7 @@ mod tests{
let msg = "!clientUpdate:";
transmit_data(&stream, msg);
let command = read_data(&stream);
let command = read_data(&stream, &mut buffer);
assert_eq!(command, Commands::Success(None));
@ -332,6 +426,8 @@ mod tests{
#[test]
fn test_clientUpdate_multi(){
let mut buffer = [0; 1024];
setup_server();
let mut stream_one = establish_client_connection("0001-776-6-5");
@ -345,7 +441,7 @@ mod tests{
let mut user_3 = true;
for uuid in client_uuids.iter() {
let command = read_data(&stream_one);
let command = read_data(&stream_one, &mut buffer);
if *uuid == String::from("0010-776-6-5") && user_1 {
let params: HashMap<String, String> = [(String::from("uuid"), String::from("0010-776-6-5")), (String::from("name"), String::from("\"alice\"")), (String::from("host"), String::from("\"127.0.0.1\""))].iter().cloned().collect();
@ -375,7 +471,7 @@ mod tests{
let msg = "!clientUpdate:";
transmit_data(&stream_one, msg);
let command = read_data(&stream_one);
let command = read_data(&stream_one, &mut buffer);
match command.clone() {
Commands::Error(None) => println!("resending..."),
_ => {
@ -387,7 +483,7 @@ mod tests{
stream_one.set_read_timeout(None).unwrap();
for x in 0..3 {
let command = read_data(&stream_one);
let command = read_data(&stream_one, &mut buffer);
let command_clone = command.clone();
match command{
@ -429,12 +525,14 @@ mod tests{
#[test]
fn test_clientInfo(){
let mut buffer = [0; 1024];
setup_server();
let mut stream_one = establish_client_connection("0001-776-6-5");
let mut stream_two = establish_client_connection("\"0010-776-6-5\"");
let command = read_data(&stream_one);
let command = read_data(&stream_one, &mut buffer);
let params: HashMap<String, String> = [(String::from("uuid"), String::from("\"0010-776-6-5\"")), (String::from("name"), String::from("\"alice\"")), (String::from("host"), String::from("\"127.0.0.1\""))].iter().cloned().collect();
assert_eq!(command, Commands::Client(Some(params)));
@ -448,7 +546,7 @@ mod tests{
let msg = "!clientInfo: uuid:\"0010-776-6-5\"";
transmit_data(&stream_one, msg);
let command = read_data(&stream_one);
let command = read_data(&stream_one, &mut buffer);
match command.clone() {
Commands::Error(None) => println!("resending..."),
_ => {
@ -470,13 +568,14 @@ mod tests{
#[test]
fn test_client_disconnect(){
let mut tmp_buffer = [0; 1024];
let mut buffer = [0; 1024];
setup_server();
let mut stream_one = establish_client_connection("0001-776-6-5");
let mut stream_two = establish_client_connection("0010-776-6-5");
let command = read_data(&stream_one);
let command = read_data(&stream_one, &mut buffer);
let params: HashMap<String, String> = [(String::from("uuid"), String::from("0010-776-6-5")), (String::from("name"), String::from("\"alice\"")), (String::from("host"), String::from("\"127.0.0.1\""))].iter().cloned().collect();
assert_eq!(command, Commands::Client(Some(params)));
@ -486,7 +585,7 @@ mod tests{
let msg = "!disconnect:";
transmit_data(&stream_two, msg);
let command = read_data(&stream_one);
let command = read_data(&stream_one, &mut buffer);
let params: HashMap<String, String> = [(String::from("uuid"), String::from("0010-776-6-5"))].iter().cloned().collect();
assert_eq!(command, Commands::Client(Some(params)));
@ -494,7 +593,7 @@ mod tests{
transmit_data(&stream_one, msg);
stream_one.set_read_timeout(Some(Duration::from_millis(2000))).unwrap();
match stream_one.peek(&mut tmp_buffer) {
match stream_one.peek(&mut buffer) {
Ok(_) => assert!(false),
Err(_) => assert!(true),
}