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

fixed client not disconnecting issue

This commit is contained in:
michael-bailey 2021-03-30 07:20:32 +01:00
parent 2af0fb8bf3
commit 1decf77207
6 changed files with 108 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
foundation/src/prelude.rs
View File

@ -10,6 +10,6 @@ pub trait ICooperative {
}
pub trait IPreemptive {
fn run(arc: &Arc<Self>) {}
fn run(arc: &Arc<Self>);
fn start(arc: &Arc<Self>);
}

2
rustfmt.toml
View File

@ -1,2 +1,2 @@
hard_tabs = true
max_width = 80
max_width = 90

118
server/src/client.rs
View File

@ -4,7 +4,6 @@ use crate::messages::ServerMessage;
use foundation::prelude::IPreemptive;
use std::cmp::Ordering;
use std::io::BufRead;
use std::io::Read;
use std::io::Write;
use std::io::{BufReader, BufWriter};
use std::mem::replace;
@ -17,7 +16,7 @@ use serde::Serialize;
use uuid::Uuid;
use foundation::messages::client::{ClientStreamIn, ClientStreamOut};
use foundation::prelude::{ICooperative, IMessagable};
use foundation::prelude::IMessagable;
/// # Client
/// This struct represents a connected user.
@ -106,65 +105,74 @@ impl IPreemptive for Client {
let arc2 = arc.clone();
// read thread
std::thread::spawn(move || {
let arc = arc1.clone();
let mut buffer = String::new();
let mut reader_lock = arc.stream_reader.lock().unwrap();
let reader = reader_lock.as_mut().unwrap();
let _ = std::thread::Builder::new()
.name(format!("client thread recv [{:?}]", &arc.uuid))
.spawn(move || {
let arc = arc1;
while let Ok(size) = reader.read_line(&mut buffer) {
if size == 0 {
arc.send_message(Disconnect);
break;
}
let mut buffer = String::new();
let mut reader_lock = arc.stream_reader.lock().unwrap();
let reader = reader_lock.as_mut().unwrap();
let command =
serde_json::from_str::<ClientStreamIn>(buffer.as_str())
.unwrap();
'main: while let Ok(size) = reader.read_line(&mut buffer) {
if size == 0 {arc.send_message(Disconnect); break 'main;}
match command {
ClientStreamIn::Disconnect => arc.send_message(Disconnect),
_ => println!("[client]: command not found"),
}
}
});
// write thread
std::thread::spawn(move || {
let arc = arc2.clone();
let mut writer_lock = arc.stream_writer.lock().unwrap();
let writer = writer_lock.as_mut().unwrap();
let mut buffer: Vec<u8> = Vec::new();
writeln!(
buffer,
"{}",
serde_json::to_string(&ClientStreamOut::Connected).unwrap()
);
writer.write_all(&buffer).unwrap();
writer.flush().unwrap();
loop {
for message in arc.output.iter() {
match message {
Disconnect => {
arc.server_channel
.lock()
.unwrap()
.as_mut()
.unwrap()
.send(ServerMessage::ClientDisconnected(
arc.uuid,
))
.unwrap();
break;
let command = serde_json::from_str::<ClientStreamIn>(buffer.as_str());
match command {
Ok(ClientStreamIn::Disconnect) => {
println!("[Client {:?}]: Disconnect recieved", &arc.uuid);
arc.send_message(Disconnect);
break 'main;
}
_ => println!("[client]: message not implemented"),
_ => println!("[Client {:?}]: command not found", &arc.uuid),
}
}
}
});
println!("[Client {:?}] exited thread 1", &arc.uuid);
});
// write thread
let _ = std::thread::Builder::new()
.name(format!("client thread msg [{:?}]", &arc.uuid))
.spawn(move || {
let arc = arc2;
let mut writer_lock = arc.stream_writer.lock().unwrap();
let writer = writer_lock.as_mut().unwrap();
let mut buffer: Vec<u8> = Vec::new();
writeln!(
buffer,
"{}",
serde_json::to_string(&ClientStreamOut::Connected).unwrap()
);
let _ = writer.write_all(&buffer);
let _ = writer.flush();
'main: loop {
std::thread::sleep(std::time::Duration::from_secs(1));
println!("[Client {:?}]: thread 2 tick!", &arc.uuid);
for message in arc.output.iter() {
match message {
Disconnect => {
arc.server_channel
.lock()
.unwrap()
.as_mut()
.unwrap()
.send(ServerMessage::ClientDisconnected(arc.uuid))
.unwrap();
break 'main;
}
_ => println!(
"[Client {:?}]: message not implemented",
&arc.uuid
),
}
}
}
println!("[Client {:?}]: exited thread 2", &arc.uuid);
});
}
fn start(arc: &Arc<Self>) {

23
server/src/client_manager.rs
View File

@ -56,23 +56,27 @@ impl IPreemptive for ClientManager {
loop {
std::thread::sleep(std::time::Duration::from_secs(1));
println!("[client manager]: Tick!");
if !arc.receiver.is_empty() {
for message in arc.receiver.iter() {
for message in arc.receiver.try_iter() {
println!("[Client manager]: recieved message: {:?}", message);
use ClientMgrMessage::{Add, Remove, SendMessage};
match message {
Add(client) => {
println!("[Client Manager]: adding new client");
Client::start(&client);
arc.clients
.lock()
.unwrap()
.insert(client.uuid, client)
.unwrap_or_default();
let mut lock = arc.clients.lock().unwrap();
if lock.insert(client.uuid, client).is_none() {
println!("value is new");
}
}
Remove(uuid) => {
let _ = arc.clients.lock().unwrap().remove(&uuid);
println!("[Client Manager]: removing client: {:?}", &uuid);
if let Some(client) =
arc.clients.lock().unwrap().remove(&uuid)
{
client.send_message(ClientMessage::Disconnect);
}
}
SendMessage(to_uuid, from_uuid, content) => {
let lock = arc.clients.lock().unwrap();
@ -88,6 +92,7 @@ impl IPreemptive for ClientManager {
}
}
}
println!("client manager exited");
}
fn start(arc: &Arc<Self>) {

58
server/src/network_manager.rs
View File

@ -26,8 +26,7 @@ impl NetworkManager {
let mut address = "0.0.0.0:".to_string();
address.push_str(&port);
let listener =
TcpListener::bind(address).expect("Could not bind to address");
let listener = TcpListener::bind(address).expect("Could not bind to address");
Arc::new(NetworkManager {
listener,
@ -37,6 +36,9 @@ impl NetworkManager {
}
impl IPreemptive for NetworkManager {
fn run(_: &Arc<Self>) {}
fn start(arc: &Arc<Self>) {
let arc = arc.clone();
std::thread::spawn(move || {
@ -48,10 +50,8 @@ impl IPreemptive for NetworkManager {
let server_channel = arc.server_channel.clone();
// create readers
let mut reader =
BufReader::new(stream.try_clone().unwrap());
let mut writer =
BufWriter::new(stream.try_clone().unwrap());
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut writer = BufWriter::new(stream.try_clone().unwrap());
let _handle = thread::Builder::new()
.name("NetworkJoinThread".to_string())
@ -61,11 +61,9 @@ impl IPreemptive for NetworkManager {
// send request message to connection
writer
.write_all(
serde_json::to_string(
&NetworkSockOut::Request,
)
.unwrap()
.as_bytes(),
serde_json::to_string(&NetworkSockOut::Request)
.unwrap()
.as_bytes(),
)
.unwrap_or_default();
writer.write_all(b"\n").unwrap_or_default();
@ -78,21 +76,24 @@ impl IPreemptive for NetworkManager {
}
//match the response
if let Ok(request) = serde_json::from_str::<
NetworkSockIn,
>(&buffer)
if let Ok(request) =
serde_json::from_str::<NetworkSockIn>(&buffer)
{
match request {
NetworkSockIn::Info => {
// send back server info to the connection
writer.write_all(
serde_json::to_string(
&NetworkSockOut::GotInfo {
server_name: "oof",
server_owner: "michael"
}
).unwrap().as_bytes()
).unwrap();
writer
.write_all(
serde_json::to_string(
&NetworkSockOut::GotInfo {
server_name: "oof",
server_owner: "michael",
},
)
.unwrap()
.as_bytes(),
)
.unwrap();
writer.write_all(b"\n").unwrap();
writer.flush().unwrap();
}
@ -109,19 +110,18 @@ impl IPreemptive for NetworkManager {
stream.try_clone().unwrap(),
server_channel.clone(),
);
server_channel.send(
ServerMessage::ClientConnected(new_client)
).unwrap_or_default();
server_channel
.send(ServerMessage::ClientConnected(
new_client,
))
.unwrap_or_default();
}
}
}
});
}
Err(e) => {
println!(
"[Network manager]: error getting stream: {:?}",
e
);
println!("[Network manager]: error getting stream: {:?}", e);
continue;
}
}

1
server/src/server.rs
View File

@ -44,7 +44,6 @@ impl Server {
impl ICooperative for Server {
fn tick(&self) {
println!("[server]: Tick!");
use ClientMgrMessage::{Add, Remove};
// handle new messages loop