From e6905cb6b2b2593f70cac5e2b6700040a9c5b3ed Mon Sep 17 00:00:00 2001 From: michael bailey Date: Wed, 3 Jan 2024 23:35:26 +0000 Subject: [PATCH] Adding client updates on addition and removal (#21) * adding user update support * Adding public key storage * replaced duplicate fields * changed messaegs to include a type property * added encryption example * Moved threads to tokio async * Moved threads to tokio async * Created global message support * created basic ui that counts * Created connection abstraction This abstracts a TcpStream await to use any serialisable types. * exposed server as lib +created lib output for server + added extra dependencies + added functions to retrieve the port of a server * updated connection + added manager trait + updated connection to use more idomatic rust code converted enums to Strigns rather that &str + * added network Manager tl;dr, as the title says * Update network_manager.rs Converted over to new messaging system. * meant to add this earlier * Right bare with me... This doesnt work for now but i have a system that works. After this commit i will be resuming the smaller changes. I needed to do this to find a new approach. God save my mind. * Impl IManager for ClientManager * Added client manager tests * Stripped server of functinoality This is temporary whilst the client manager is being tested. * Update client_manager.rs swapped lock type for tokio lock * Update client_manager.rs - Added type arguments to Client manager - deprecated start method * Update client_manager.rs - implemented handle channel method from old start method. - added notes to possible issues. * Update client_manager.rs + updated lifetime bounds. * Update client_manager.rs + created function to get the number of connected clients * Update client_manager.rs + created function to add/remove clients to the client manager. * Update server.rs - refactored server to simplify message passing * Update client.rs meant to add this earlier * Update messages.rs formatting * moved some messages around * Update server.rs - removed redundant messages * Update client_manager.rs bunch more changes yes i know i'm rubbish at this plz don't say anything :( * Added Message Broadcasting functionality * cleared up warnings * Update client.rs - removed redundant fields * Update client.rs added global broadcasting * Added the ability to disconnect from the server * Update network_manager.rs this protects the network manager from crashing when a erroneous message is sent * implemented message sending between users * Update client.rs added tokio, to protect client connections. * Added Lua dependency * Made server, Lua scriptable. * added basic scripting abilities to the server * Update client.rs fixed stray connection * made client Lua scriptable * Made client manager Lua scriptable * Adde fields to server to get client manager * Updated testing script * added client indexing to client manager * added basic callback support to ClientManager * moved Lua structs to separate module * added arbitrary self types * Created example plugin crate * Added libloading dependency * Created plugin trait and create function type * created basic PluginManager * Added plugin manager to server * added modules * Updated example plugin * added plugin manager to server * modified plugin module visibility * updated example plugin * updated plagin interface * created plugin interface * updated plugin init process * updated example plugin * cleaned up lib folder * fixed cloning issues with plugin manager * updated plugin trait implementation * updated module definitions * upadted rust fmt rules for imports * fixoing formatting and ther errors. This is a pain to look through. i'm sorry :( * minor fix * Fixed plugin not functioning with tokio * Added plugin lifecycle events added lifecycle events to the plugin entry. This allows plugins to be in a stopped paused or running state * Adjusted visibility modifiers * added basics of an event system * updated imports. * added function to get plugin entry as interface to plugin * updated plugin example * fixing linting errors * updated event architecture * renamed responder module * added documention * created a event result builder * renamed responder * updated example with interface setting * moved event system to foundation * added back tokio to foundation * modified visibility * added functions for IResponder * more module mess * made event generic * made responder generic * added basic plugin event propogation. * updated plugin event handler. * same because vscode? * made plugin event generic * RIGHT I'M MOVING TO ACTIX * Started work on actor conversion * implemented GetInfo for actix server * implementing clients using actors * added ability to add and remove clients * updated foundation with comments and better messages * Update Cargo.toml + added tokio stream * added auto connection closing to Connection * Update actix_server.rs + added ClientConnection handlers * added ability to get server updates * added basic messaging functonality * deleted old server version * renamed files to match std structure * reformatted project * renamed old files * renamed serverActor to server * added proper prelude * Update client.rs minor bug fixes * updated clap * made server configurable * added port configurability ~ moved components to seperate modules ~ added builder and config to network manager * way too much happened here + added scriptable version of the server, networkManager and clientManager + added lua engine creation +also added unfinished rhai support + also did some restructuring * made client manager and clients scriptable * added more scriptable objects * Created config manager with path read functionality * fixed network manager message result types * added bootstrapper actor * moved arg matcher funtion. * created singleton config manager using once pattern. * removed bootstrapper and created config builder * fixed panic on file not existing * ignoring config_file * added config manager to network manager * updated some docs * added config support to network manager, changed lifecycle a bit as well * removed redundant handlers * added configuration through args support to config manager and network manager. * added config manager support to server and removed old serverConfig references * performed cleanup of network manager * performed cleanup of server * updated config manager api to use optionals. This leads to pre-configuration and easier readability and understanding. * updated server and network manager to new style for setting config values * cleaned up prelude linting errors * cleaned up network listener * cleaned up connection initator * cleaned up connection actor * cleaned up lua manager * cleaned up rhai manager * cleaned up config manager * cleaned up client manager * cleaned up client actor * cleaned up scriptable interfaces * updated client uuid version * fixed client cargo toml * undone previous change * moved arg parser to folder as file * uhh... changes and formatting * created chat manager and added it to the client manager * added global message spport to client manager * refixed network manager (again again) * added debug messages * refactord client code * refactored messages and added new message types * refactord network manager module name * resolved warning in connection * refactored message into foundation, added get message support for clients * Fixed messages being sent to sender * fixed not implemented panic, and field misnaming * turned some references into weak variaties, to prevent memory leaks. * turned ref into weak ref * removing strong references from scriping managers * fixing lints * making moe things weak repferences * refactored connection initiator into spereate files * accedentally ran cargo fmt instead of on one file * removed the read loop, and replaced it with a recurrent messages. * added docker files and automated build scripts * Added client updates for addition and removal of clients --- server/src/client_management/client/actor.rs | 46 +++++++++++-------- .../src/client_management/client_manager.rs | 19 ++++++-- server/src/network/connection/actor.rs | 22 +++++---- server/src/network/connection/messages.rs | 5 +- .../src/network/connection_initiator/actor.rs | 5 +- server/src/network/listener/mod.rs | 2 + server/src/network/network_manager/actor.rs | 10 +++- 7 files changed, 67 insertions(+), 42 deletions(-) diff --git a/server/src/client_management/client/actor.rs b/server/src/client_management/client/actor.rs index c553c7a..d8675eb 100644 --- a/server/src/client_management/client/actor.rs +++ b/server/src/client_management/client/actor.rs @@ -83,7 +83,8 @@ impl Client { } } - pub(crate) fn error(&self, msg: String) { + + pub(crate) fn _error(&self, msg: String) { println!("[Client] sending error: {}", msg); use serde_json::to_string; use ConnectionMessage::SendData; @@ -219,29 +220,34 @@ impl Handler for Client { msg: ConnectionObservableOutput, ctx: &mut Self::Context, ) -> Self::Result { + use foundation::messages::client::ClientStreamIn::{ + Disconnect, + GetClients, + GetMessages, + SendGlobalMessage, + SendMessage, + }; + use serde_json::from_str; + use crate::network::ConnectionObservableOutput::RecvData; - if let RecvData(_sender, _addr, data) = msg { - use foundation::messages::client::ClientStreamIn::{ - Disconnect, - GetClients, - GetMessages, - SendGlobalMessage, - SendMessage, - }; - use serde_json::from_str; - if let Ok(msg) = from_str::(data.as_str()) { - match msg { - GetClients => self.get_clients(ctx), - GetMessages => self.get_messages(ctx), - SendMessage { to, content } => self.send_message(ctx, to, content), - SendGlobalMessage { content } => { - self.send_gloal_message(ctx, content) + + match msg { + RecvData(_sender, data) => { + if let Ok(msg) = from_str::(data.as_str()) { + match msg { + GetClients => self.get_clients(ctx), + GetMessages => self.get_messages(ctx), + SendMessage { to, content } => self.send_message(ctx, to, content), + SendGlobalMessage { content } => { + self.send_gloal_message(ctx, content) + } + Disconnect => self.disconnect(ctx), } - Disconnect => self.disconnect(ctx), } - } else { - self.error(format!("Failed to parse Message: {}", data)); } + ConnectionObservableOutput::ConnectionClosed(_) => self + .broadcast(ClientObservableMessage::Disconnecting(self.details.uuid)), + } } } diff --git a/server/src/client_management/client_manager.rs b/server/src/client_management/client_manager.rs index d3c25b4..2148a4e 100644 --- a/server/src/client_management/client_manager.rs +++ b/server/src/client_management/client_manager.rs @@ -58,7 +58,7 @@ impl ClientManager { } pub(crate) fn send_client_list( - &mut self, + &self, ctx: &mut Context, sender: WeakAddr, ) { @@ -66,7 +66,7 @@ impl ClientManager { use crate::client_management::client::ClientMessage::ClientList; if let Some(to_send) = sender.upgrade() { let client_addr: Vec> = - self.clients.iter().map(|(_, v)| v).cloned().collect(); + self.clients.values().cloned().collect(); let collection = tokio_stream::iter(client_addr) .then(|addr| addr.send(ClientDataMessage::Details)) @@ -115,7 +115,7 @@ impl ClientManager { ) { println!("[ClientManager] sending message to client"); let client_addr: Vec> = - self.clients.iter().map(|(_, v)| v).cloned().collect(); + self.clients.values().cloned().collect(); let collection = tokio_stream::iter(client_addr.clone()) .then(|addr| addr.send(ClientDataMessage::Details)) @@ -164,7 +164,7 @@ impl ClientManager { use crate::client_management::client::ClientMessage::GloballySentMessage; let client_addr: Vec> = - self.clients.iter().map(|(_, v)| v).cloned().collect(); + self.clients.values().cloned().collect(); if let Some(sender) = sender.upgrade() { let cm = self.chat_manager.clone(); @@ -227,16 +227,24 @@ impl ClientManager { println!("[ClientManager] sending subscribe message to client"); addr.do_send(Subscribe(recp.downgrade())); self.clients.insert(uuid, addr); + for (_k, v) in self.clients.clone() { + self.send_client_list(ctx, v.downgrade()) + } } fn remove_client(&mut self, ctx: &mut Context, uuid: Uuid) { println!("[ClientManager] removing client"); use crate::prelude::messages::ObservableMessage::Unsubscribe; let recp = ctx.address().recipient::(); - if let Some(addr) = self.clients.remove(&uuid) { + let addr = self.clients.remove(&uuid); + if let Some(addr) = addr { println!("[ClientManager] sending unsubscribe message to client"); addr.do_send(Unsubscribe(recp.downgrade())); } + println!("[ClientManager] sending client list to other clients"); + for (_k, v) in self.clients.iter() { + self.send_client_list(ctx, v.downgrade()) + } } fn disconnect_client( @@ -249,6 +257,7 @@ impl ClientManager { let recp = ctx.address().recipient::(); if let Some(addr) = self.clients.remove(&uuid) { addr.do_send(Unsubscribe(recp.downgrade())); + self.remove_client(ctx, uuid); } } } diff --git a/server/src/network/connection/actor.rs b/server/src/network/connection/actor.rs index 98dbeb1..a5b7819 100644 --- a/server/src/network/connection/actor.rs +++ b/server/src/network/connection/actor.rs @@ -5,15 +5,13 @@ use actix::{ fut::wrap_future, Actor, ActorContext, - ActorFutureExt, Addr, AsyncContext, Context, Handler, - SpawnHandle, WeakRecipient, }; -use futures::{future::join_all, stream::Buffered, Future, FutureExt}; +use futures::{future::join_all, Future, FutureExt}; use tokio::{ io::{split, AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf}, net::TcpStream, @@ -37,7 +35,7 @@ use crate::{ /// - loop_future: the future holding the receiving loop. pub struct Connection { write_half: Arc>>, - address: SocketAddr, + _address: SocketAddr, observers: Vec>, } @@ -49,7 +47,7 @@ impl Connection { let (read_half, write_half) = split(stream); let addr = Connection { write_half: Arc::new(Mutex::new(write_half)), - address, + _address: address, observers: Vec::new(), } .start(); @@ -88,7 +86,6 @@ impl Connection { ctx: &mut ::Context, mut buf_reader: BufReader>, ) { - let address = self.address; let weak_addr = ctx.address().downgrade(); let read_fut = async move { @@ -97,7 +94,6 @@ impl Connection { let read_fut = buf_reader.read_line(&mut buffer_string); let Ok(Ok(len)) = timeout(dur, read_fut).await else { - println!("[Connection] timeout reached"); if let Some(addr) = weak_addr.upgrade() { addr.do_send(ConnectionPrivateMessage::DoRead(buf_reader)); } @@ -106,6 +102,9 @@ impl Connection { if len == 0 { println!("[Connection] readline returned 0"); + if let Some(addr) = weak_addr.upgrade() { + addr.do_send(ConnectionPrivateMessage::Close); + } return; } @@ -114,7 +113,6 @@ impl Connection { .send(ConnectionPrivateMessage::Broadcast( ConnectionObservableOutput::RecvData( addr.downgrade(), - address, buffer_string.clone(), ), )) @@ -127,6 +125,11 @@ impl Connection { }; ctx.spawn(wrap_future(read_fut)); } + + fn close_connection(&self, ctx: &mut ::Context) { + use ConnectionObservableOutput::ConnectionClosed; + self.broadcast(ctx, ConnectionClosed(ctx.address().downgrade())) + } } impl Actor for Connection { @@ -135,7 +138,7 @@ impl Actor for Connection { /// runs when the actor is started. /// takes out eh read_half ad turns it into a buffered reader /// then eneters loop readling lines from the tcp stream - fn started(&mut self, ctx: &mut Self::Context) { + fn started(&mut self, _ctx: &mut Self::Context) { println!("[Connection] started"); } @@ -251,6 +254,7 @@ impl Handler for Connection { ConnectionPrivateMessage::DoRead(buf_reader) => { self.do_read(ctx, buf_reader) } + ConnectionPrivateMessage::Close => self.close_connection(ctx), }; } } diff --git a/server/src/network/connection/messages.rs b/server/src/network/connection/messages.rs index c8005ce..4876323 100644 --- a/server/src/network/connection/messages.rs +++ b/server/src/network/connection/messages.rs @@ -1,5 +1,3 @@ -use std::net::SocketAddr; - use actix::{Message, WeakAddr}; use tokio::{ io::{BufReader, ReadHalf}, @@ -19,7 +17,7 @@ pub(crate) enum ConnectionMessage { #[derive(Message, Clone)] #[rtype(result = "()")] pub(crate) enum ConnectionObservableOutput { - RecvData(WeakAddr, SocketAddr, String), + RecvData(WeakAddr, String), ConnectionClosed(WeakAddr), } @@ -28,4 +26,5 @@ pub(crate) enum ConnectionObservableOutput { pub(super) enum ConnectionPrivateMessage { Broadcast(ConnectionObservableOutput), DoRead(BufReader>), + Close, } diff --git a/server/src/network/connection_initiator/actor.rs b/server/src/network/connection_initiator/actor.rs index cb7e10c..b950792 100644 --- a/server/src/network/connection_initiator/actor.rs +++ b/server/src/network/connection_initiator/actor.rs @@ -58,7 +58,6 @@ impl ConnectionInitiator { &mut self, sender: WeakAddr, ctx: &mut ::Context, - _address: SocketAddr, data: String, ) { use InitiatorOutput::{ClientRequest, InfoRequest}; @@ -158,8 +157,8 @@ impl Handler for ConnectionInitiator { ) -> Self::Result { use ConnectionObservableOutput::RecvData; - if let RecvData(sender, addr, data) = msg { - self.handle_request(sender, ctx, addr, data) + if let RecvData(sender, data) = msg { + self.handle_request(sender, ctx, data) } } } diff --git a/server/src/network/listener/mod.rs b/server/src/network/listener/mod.rs index 7d254cf..8e047d7 100644 --- a/server/src/network/listener/mod.rs +++ b/server/src/network/listener/mod.rs @@ -57,6 +57,7 @@ impl NetworkListener { let delegate = self.delegate.clone(); ctx.spawn(wrap_future(async move { use ListenerOutput::NewConnection; + let listener = TcpListener::bind(addr).await.unwrap(); while let Ok((stream, addr)) = listener.accept().await { println!("[NetworkListener] accepted socket"); @@ -66,6 +67,7 @@ impl NetworkListener { break; }; + println!("[NetworkListener] sending connection to delegate"); delegate.do_send(NewConnection(conn)) } })); diff --git a/server/src/network/network_manager/actor.rs b/server/src/network/network_manager/actor.rs index 1ca6621..567ea3c 100644 --- a/server/src/network/network_manager/actor.rs +++ b/server/src/network/network_manager/actor.rs @@ -170,9 +170,12 @@ impl Actor for NetworkManager { }, ); ctx.spawn(fut); - println!("[NetworkManager] Finished Starting"); } } + + fn stopped(&mut self, ctx: &mut Self::Context) { + println!("[NetworkManager] network manager stopped"); + } } impl Handler for NetworkManager { @@ -215,7 +218,10 @@ impl Handler for NetworkManager { ) -> Self::Result { use ListenerOutput::NewConnection; match msg { - NewConnection(connection) => self.new_connection(ctx, connection), + NewConnection(connection) => { + println!("new connection"); + self.new_connection(ctx, connection) + } }; } }