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

fixoing formatting and ther errors. This is a pain to look through. i'm sorry :(

This commit is contained in:
michael-bailey 2022-04-06 23:46:56 +01:00
parent c33cd7c9a1
commit 2595622b37
9 changed files with 380 additions and 399 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

180
client/src/managers/Network.rs
View File

@ -1,25 +1,25 @@
use async_trait::async_trait;
use std::io::{Error, ErrorKind};
use std::mem;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::sleep;
use async_trait::async_trait;
use std::sync::Arc;
use tokio::net::ToSocketAddrs;
use tokio::sync::mpsc::Sender;
use tokio::sync::Mutex;
use uuid::Uuid;
use crate::managers::NetworkManagerMessage;
use foundation::connection::Connection;
use foundation::messages::client::{ClientStreamIn, ClientStreamOut};
use foundation::messages::network::{NetworkSockIn, NetworkSockOut};
use foundation::prelude::IManager;
use crate::managers::NetworkManagerMessage;
pub struct NetworkManager<M>
where M: From<NetworkManagerMessage> {
where
M: From<NetworkManagerMessage>,
{
#[allow(unused)]
server_connection: Mutex<Option<Connection>>,
server_connection: Mutex<Option<Arc<Connection>>>,
#[allow(unused)]
cursive: Sender<M>,
@ -28,8 +28,9 @@ pub struct NetworkManager<M>
}
impl<M> NetworkManager<M>
where M: From<NetworkManagerMessage> {
where
M: From<NetworkManagerMessage>,
{
pub fn new(sender: Sender<M>) -> Arc<Self> {
Arc::new(NetworkManager {
server_connection: Mutex::new(None),
@ -39,8 +40,11 @@ impl<M> NetworkManager<M>
}
#[allow(unused)]
pub async fn info<T: ToSocketAddrs>(self: &Arc<Self>, host: T) -> Result<NetworkManagerMessage, Error> {
let connection= Connection::new();
pub async fn info<T: ToSocketAddrs>(
self: &Arc<Self>,
host: T,
) -> Result<NetworkManagerMessage, Error> {
let connection = Connection::new();
println!("Created connection");
connection.connect(host).await?;
let req = connection.read().await?;
@ -48,12 +52,15 @@ impl<M> NetworkManager<M>
println!("request: {:?}", req);
if let NetworkSockOut::Request = req {
connection.write::<NetworkSockIn>(NetworkSockIn::Info)
connection
.write::<NetworkSockIn>(NetworkSockIn::Info)
.await?;
return Ok(connection.read::<NetworkSockOut>()
.await?.into());
return Ok(connection.read::<NetworkSockOut>().await?.into());
} else {
Err(Error::new(ErrorKind::ConnectionAborted, "Request not received"))
Err(Error::new(
ErrorKind::ConnectionAborted,
"Request not received",
))
}
}
@ -63,9 +70,9 @@ impl<M> NetworkManager<M>
host: String,
uuid: Uuid,
username: String,
address: String
address: String,
) -> Result<(), Error> {
let connection= Connection::new();
let connection = Connection::new();
let _ = connection.connect(host).await?;
@ -76,10 +83,15 @@ impl<M> NetworkManager<M>
println!("read request");
return if let NetworkSockOut::Request = req {
println!("got request");
connection.write(NetworkSockIn::Connect {username, uuid: uuid.to_string(), address}).await?;
connection
.write(NetworkSockIn::Connect {
username,
uuid,
address,
})
.await?;
let res = connection.read().await?;
// switch over to ClientStreamOut
@ -88,12 +100,18 @@ impl<M> NetworkManager<M>
let _ = mem::replace(&mut *connection_lock, Some(connection));
Ok(())
} else {
Err(Error::new(ErrorKind::ConnectionRefused, format!("expected connecting received: {:?}", res)))
Err(Error::new(
ErrorKind::ConnectionRefused,
format!("expected connecting received: {:?}", res),
))
}
} else {
println!("request not found");
Err(Error::new(ErrorKind::ConnectionAborted, "Server did not send request"))
}
Err(Error::new(
ErrorKind::ConnectionAborted,
"Server did not send request",
))
};
}
#[allow(unused)]
@ -106,108 +124,90 @@ impl<M> NetworkManager<M>
return if let ClientStreamOut::Disconnected = connection.read().await? {
Ok(())
} else {
Err(Error::new(ErrorKind::InvalidData, "disconnect failed, forcing disconnect"))
}
}
#[allow(unused)]
pub async fn update() {
}
#[allow(unused)]
async fn start(self: Arc<Self>) {
let network_manager = self.clone();
tokio::spawn(async {
});
Err(Error::new(
ErrorKind::InvalidData,
"disconnect failed, forcing disconnect",
))
};
}
}
#[async_trait]
impl<M: 'static> IManager for NetworkManager<M>
where M: From<NetworkManagerMessage> + Send {
async fn run(self: Arc<Self>) {
// let networkManager = self.clone();
loop {
sleep(Duration::new(1,0)).await;
println!("networkManager tick")
}
}
async fn start(self: &Arc<Self>) {
let network_manager = self.clone();
tokio::spawn(
network_manager.run()
);
where
M: From<NetworkManagerMessage> + Send,
{
async fn run(self: &Arc<Self>) {
println!("networkManager tick")
}
}
#[cfg(test)]
mod test {
use crate::managers::network::NetworkManagerMessage;
use crate::managers::NetworkManager;
use serverlib::Server;
use std::future::Future;
use tokio::sync::mpsc::channel;
use uuid::Uuid;
use serverlib::Server;
use crate::managers::network::NetworkManagerMessage;
use crate::managers::NetworkManager;
async fn wrap_setup<T,F>(test: T)
where T: FnOnce(u16) -> F,
F: Future
async fn wrap_setup<T, F>(test: T)
where
T: FnOnce(u16) -> F,
F: Future,
{
let server = Server::new().await.unwrap();
let port = server.port();
tokio::spawn(
async move {
server.start().await;
}
);
let port = server.port().await;
tokio::spawn(async move {
server.start().await;
});
test(port).await;
}
#[tokio::test]
async fn test_fetch_server_info() {
use NetworkManagerMessage::Info;
#[allow(unused)]
let (tx,rx) =
channel::<NetworkManagerMessage>(16);
wrap_setup(|port| {
async move {
let network = NetworkManager::new(tx);
let info = network.info(format!("localhost:{}", port)).await.expect("Failed to fetch info");
assert_eq!(info, Info {
let (tx, rx) = channel::<NetworkManagerMessage>(16);
wrap_setup(|port| async move {
let network = NetworkManager::new(tx);
let info = network
.info(format!("localhost:{}", port))
.await
.expect("Failed to fetch info");
assert_eq!(
info,
Info {
server_name: "oof".to_string(),
server_owner: "michael".to_string()
});
}
}).await;
}
);
})
.await;
}
#[tokio::test]
async fn test_login_and_logout_to_server() {
#[allow(unused)]
let (tx,rx) =
channel::<NetworkManagerMessage>(16);
let (tx, rx) = channel::<NetworkManagerMessage>(16);
let network = NetworkManager::new(tx);
println!("created network manger");
wrap_setup(|port| {
async move {
network.login(
wrap_setup(|port| async move {
network
.login(
format!("localhost:{}", port),
Uuid::default(),
"user1".to_string(),
"localhost".to_string()
).await.expect("login failed");
"localhost".to_string(),
)
.await
.expect("login failed");
network.logout().await.expect("logout failed");
}
}).await;
network.logout().await.expect("logout failed");
})
.await;
}
}

180
client/src/managers/network.rs
View File

@ -1,25 +1,25 @@
use async_trait::async_trait;
use std::io::{Error, ErrorKind};
use std::mem;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::sleep;
use async_trait::async_trait;
use std::sync::Arc;
use tokio::net::ToSocketAddrs;
use tokio::sync::mpsc::Sender;
use tokio::sync::Mutex;
use uuid::Uuid;
use crate::managers::NetworkManagerMessage;
use foundation::connection::Connection;
use foundation::messages::client::{ClientStreamIn, ClientStreamOut};
use foundation::messages::network::{NetworkSockIn, NetworkSockOut};
use foundation::prelude::IManager;
use crate::managers::NetworkManagerMessage;
pub struct NetworkManager<M>
where M: From<NetworkManagerMessage> {
where
M: From<NetworkManagerMessage>,
{
#[allow(unused)]
server_connection: Mutex<Option<Connection>>,
server_connection: Mutex<Option<Arc<Connection>>>,
#[allow(unused)]
cursive: Sender<M>,
@ -28,8 +28,9 @@ pub struct NetworkManager<M>
}
impl<M> NetworkManager<M>
where M: From<NetworkManagerMessage> {
where
M: From<NetworkManagerMessage>,
{
pub fn new(sender: Sender<M>) -> Arc<Self> {
Arc::new(NetworkManager {
server_connection: Mutex::new(None),
@ -39,8 +40,11 @@ impl<M> NetworkManager<M>
}
#[allow(unused)]
pub async fn info<T: ToSocketAddrs>(self: &Arc<Self>, host: T) -> Result<NetworkManagerMessage, Error> {
let connection= Connection::new();
pub async fn info<T: ToSocketAddrs>(
self: &Arc<Self>,
host: T,
) -> Result<NetworkManagerMessage, Error> {
let connection = Connection::new();
println!("Created connection");
connection.connect(host).await?;
let req = connection.read().await?;
@ -48,12 +52,15 @@ impl<M> NetworkManager<M>
println!("request: {:?}", req);
if let NetworkSockOut::Request = req {
connection.write::<NetworkSockIn>(NetworkSockIn::Info)
connection
.write::<NetworkSockIn>(NetworkSockIn::Info)
.await?;
return Ok(connection.read::<NetworkSockOut>()
.await?.into());
return Ok(connection.read::<NetworkSockOut>().await?.into());
} else {
Err(Error::new(ErrorKind::ConnectionAborted, "Request not received"))
Err(Error::new(
ErrorKind::ConnectionAborted,
"Request not received",
))
}
}
@ -63,9 +70,9 @@ impl<M> NetworkManager<M>
host: String,
uuid: Uuid,
username: String,
address: String
address: String,
) -> Result<(), Error> {
let connection= Connection::new();
let connection = Connection::new();
let _ = connection.connect(host).await?;
@ -76,10 +83,15 @@ impl<M> NetworkManager<M>
println!("read request");
return if let NetworkSockOut::Request = req {
println!("got request");
connection.write(NetworkSockIn::Connect {username, uuid: uuid.to_string(), address}).await?;
connection
.write(NetworkSockIn::Connect {
username,
uuid,
address,
})
.await?;
let res = connection.read().await?;
// switch over to ClientStreamOut
@ -88,12 +100,18 @@ impl<M> NetworkManager<M>
let _ = mem::replace(&mut *connection_lock, Some(connection));
Ok(())
} else {
Err(Error::new(ErrorKind::ConnectionRefused, format!("expected connecting received: {:?}", res)))
Err(Error::new(
ErrorKind::ConnectionRefused,
format!("expected connecting received: {:?}", res),
))
}
} else {
println!("request not found");
Err(Error::new(ErrorKind::ConnectionAborted, "Server did not send request"))
}
Err(Error::new(
ErrorKind::ConnectionAborted,
"Server did not send request",
))
};
}
#[allow(unused)]
@ -106,108 +124,90 @@ impl<M> NetworkManager<M>
return if let ClientStreamOut::Disconnected = connection.read().await? {
Ok(())
} else {
Err(Error::new(ErrorKind::InvalidData, "disconnect failed, forcing disconnect"))
}
}
#[allow(unused)]
pub async fn update() {
}
#[allow(unused)]
async fn start(self: Arc<Self>) {
let network_manager = self.clone();
tokio::spawn(async {
});
Err(Error::new(
ErrorKind::InvalidData,
"disconnect failed, forcing disconnect",
))
};
}
}
#[async_trait]
impl<M: 'static> IManager for NetworkManager<M>
where M: From<NetworkManagerMessage> + Send {
async fn run(self: Arc<Self>) {
// let networkManager = self.clone();
loop {
sleep(Duration::new(1,0)).await;
println!("networkManager tick")
}
}
async fn start(self: &Arc<Self>) {
let network_manager = self.clone();
tokio::spawn(
network_manager.run()
);
where
M: From<NetworkManagerMessage> + Send,
{
async fn run(self: &Arc<Self>) {
println!("networkManager tick")
}
}
#[cfg(test)]
mod test {
use crate::managers::network::NetworkManagerMessage;
use crate::managers::NetworkManager;
use serverlib::Server;
use std::future::Future;
use tokio::sync::mpsc::channel;
use uuid::Uuid;
use serverlib::Server;
use crate::managers::network::NetworkManagerMessage;
use crate::managers::NetworkManager;
async fn wrap_setup<T,F>(test: T)
where T: FnOnce(u16) -> F,
F: Future
async fn wrap_setup<T, F>(test: T)
where
T: FnOnce(u16) -> F,
F: Future,
{
let server = Server::new().await.unwrap();
let port = server.port();
tokio::spawn(
async move {
server.start().await;
}
);
let port = server.port().await;
tokio::spawn(async move {
server.start().await;
});
test(port).await;
}
#[tokio::test]
async fn test_fetch_server_info() {
use NetworkManagerMessage::Info;
#[allow(unused)]
let (tx,rx) =
channel::<NetworkManagerMessage>(16);
wrap_setup(|port| {
async move {
let network = NetworkManager::new(tx);
let info = network.info(format!("localhost:{}", port)).await.expect("Failed to fetch info");
assert_eq!(info, Info {
let (tx, rx) = channel::<NetworkManagerMessage>(16);
wrap_setup(|port| async move {
let network = NetworkManager::new(tx);
let info = network
.info(format!("localhost:{}", port))
.await
.expect("Failed to fetch info");
assert_eq!(
info,
Info {
server_name: "oof".to_string(),
server_owner: "michael".to_string()
});
}
}).await;
}
);
})
.await;
}
#[tokio::test]
async fn test_login_and_logout_to_server() {
#[allow(unused)]
let (tx,rx) =
channel::<NetworkManagerMessage>(16);
let (tx, rx) = channel::<NetworkManagerMessage>(16);
let network = NetworkManager::new(tx);
println!("created network manger");
wrap_setup(|port| {
async move {
network.login(
wrap_setup(|port| async move {
network
.login(
format!("localhost:{}", port),
Uuid::default(),
"user1".to_string(),
"localhost".to_string()
).await.expect("login failed");
"localhost".to_string(),
)
.await
.expect("login failed");
network.logout().await.expect("logout failed");
}
}).await;
network.logout().await.expect("logout failed");
})
.await;
}
}

19
example_plugin/src/example.rs
View File

@ -1,18 +1,17 @@
use std::sync::Arc;
use serverlib::plugin::plugin::Plugin;
use serverlib::plugin::plugin_details::PluginDetails;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::sleep;
use serverlib::plugin::plugin::Plugin;
use serverlib::plugin::plugin_details::PluginDetails;
pub struct ExamplePlugin {
number: Mutex<u8>
number: Mutex<u8>,
}
impl Default for ExamplePlugin {
fn default() -> Self {
ExamplePlugin {
number: Mutex::new(0)
number: Mutex::new(0),
}
}
}
@ -24,19 +23,19 @@ impl Plugin for ExamplePlugin {
display_name: "ExamplePlugin",
id: "io.github.michael-bailey.ExamplePlugin",
version: "0.0.1",
contacts: vec!["bailey-michael1@outlook.com"]
contacts: vec!["bailey-michael1@outlook.com"],
}
}
fn init(self: &Arc<Self>) {
fn init(&self) {
println!("[ExamplePlugin]: example init")
}
async fn run(self: &Arc<Self>) {
sleep(Duration::new(1,0)).await;
async fn run(&self) {
sleep(Duration::new(1, 0)).await;
if let mut a = self.number.lock().await {
*a += 1;
println!("[ExamplePlugin]: example run");
}
}
}
}

4
foundation/src/prelude.rs
View File

@ -35,9 +35,7 @@ pub trait IManager {
loop {
sleep(Duration::new(1,0)).await;
if let Some(manager) =
Weak::upgrade(&weak_self)
{
if let Some(manager) = Weak::upgrade(&weak_self) {
manager.run().await
} else { () }
}

2
rustfmt.toml
View File

@ -1,5 +1,5 @@
hard_tabs = true
max_width = 90
max_width = 100
imports_indent = "Block"
imports_layout = "HorizontalVertical"
imports_granularity = "Crate"

118
server/src/client_manager.rs
View File

@ -2,18 +2,17 @@ use std::collections::HashMap;
use std::sync::Arc;
use futures::future::join_all;
use tokio::sync::Mutex;
use tokio::select;
use tokio::sync::Mutex;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use uuid::Uuid;
use async_trait::async_trait;
use mlua::{MetaMethod, Nil, ToLua, UserDataFields, UserDataMethods};
use foundation::prelude::IManager;
use foundation::connection::Connection;
use foundation::prelude::IManager;
use crate::client::Client;
use crate::messages::ClientMessage;
@ -22,45 +21,35 @@ use crate::messages::ClientMessage;
pub enum ClientMgrMessage {
#[allow(dead_code)]
Remove {
id: Uuid
id: Uuid,
},
#[allow(dead_code)]
SendClients {
to: Uuid
to: Uuid,
},
SendMessage {
from: Uuid,
to: Uuid,
content: String,
},
BroadcastGlobalMessage {from: Uuid, content: String},
BroadcastGlobalMessage {
from: Uuid,
content: String,
},
}
impl From<ClientMessage> for ClientMgrMessage {
fn from(msg: ClientMessage) -> Self {
use ClientMessage::{IncomingMessage,IncomingGlobalMessage,Disconnect};
use ClientMessage::{Disconnect, IncomingGlobalMessage, IncomingMessage};
match msg {
IncomingMessage {
from,
to,
content
} => ClientMgrMessage::SendMessage {
from,
to,
content
},
IncomingGlobalMessage{
from,
content
} => ClientMgrMessage::BroadcastGlobalMessage {
from,
content
},
Disconnect {id} => ClientMgrMessage::Remove {id},
_ => unimplemented!()
IncomingMessage { from, to, content } => ClientMgrMessage::SendMessage { from, to, content },
IncomingGlobalMessage { from, content } => {
ClientMgrMessage::BroadcastGlobalMessage { from, content }
}
Disconnect { id } => ClientMgrMessage::Remove { id },
_ => unimplemented!(),
}
}
}
@ -73,8 +62,8 @@ impl From<ClientMessage> for ClientMgrMessage {
/// - tx: the sender that clients will send their messages to.
/// - rx: the receiver where messages are sent to.
pub struct ClientManager<Out: 'static>
where
Out: From<ClientMgrMessage> + Send
where
Out: From<ClientMgrMessage> + Send,
{
pub clients: Mutex<HashMap<Uuid, Arc<Client<ClientMgrMessage>>>>,
@ -86,8 +75,8 @@ pub struct ClientManager<Out: 'static>
}
impl<Out> ClientManager<Out>
where
Out: From<ClientMgrMessage> + Send
where
Out: From<ClientMgrMessage> + Send,
{
pub fn new(out_channel: Sender<Out>) -> Arc<Self> {
let (tx, rx) = channel(1024);
@ -112,15 +101,9 @@ impl<Out> ClientManager<Out>
id: Uuid,
username: String,
address: String,
connection: Arc<Connection>
connection: Arc<Connection>,
) {
let client = Client::new(
id,
username,
address,
self.tx.clone(),
connection
);
let client = Client::new(id, username, address, self.tx.clone(), connection);
client.start();
let mut lock = self.clients.lock().await;
lock.insert(client.details.uuid, client);
@ -133,36 +116,38 @@ impl<Out> ClientManager<Out>
}
pub async fn handle_channel(&self, message: Option<ClientMgrMessage>) {
use ClientMgrMessage::{Remove, SendClients, BroadcastGlobalMessage, SendMessage};
use ClientMgrMessage::{BroadcastGlobalMessage, Remove, SendClients, SendMessage};
println!("Handling channel");
match message {
Some(Remove {id}) => {
Some(Remove { id }) => {
println!("[Client Manager]: removing client: {:?}", &id);
let mut lock = self.clients.lock().await;
lock.remove(&id);
},
Some(SendClients {to: _ }) => {
}
Some(SendClients { to: _ }) => {
let lock = self.clients.lock().await;
let futures = lock.iter().map(|(_,_)| async {
println!("Send message to Client")
});
let futures = lock
.iter()
.map(|(_, _)| async { println!("Send message to Client") });
join_all(futures).await;
}
Some(BroadcastGlobalMessage {from, content}) => {
Some(BroadcastGlobalMessage { from, content }) => {
let lock = self.clients.lock().await;
let futures = lock.iter()
.map(|(_,c)| (c.clone(),content.clone()))
.map(|(c,s)| async move {
c.broadcast_message(from, s).await.unwrap();
});
let futures =
lock
.iter()
.map(|(_, c)| (c.clone(), content.clone()))
.map(|(c, s)| async move {
c.broadcast_message(from, s).await.unwrap();
});
join_all(futures).await;
},
}
Some(SendMessage { from, to, content }) => {
let lock = self.clients.lock().await;
let client = lock.get(&to).unwrap();
let _ = client.user_message(from, content).await;
},
Some(Remove {id}) => {
}
Some(Remove { id }) => {
self.clients.lock().await.remove(&id);
}
_ => {
@ -174,12 +159,11 @@ impl<Out> ClientManager<Out>
#[async_trait]
impl<Out> IManager for ClientManager<Out>
where
Out: From<ClientMgrMessage> + Send
where
Out: From<ClientMgrMessage> + Send,
{
async fn run(self: &Arc<Self>) {
loop {
let mut receiver = self.rx.lock().await;
select! {
@ -193,18 +177,17 @@ impl<Out> IManager for ClientManager<Out>
#[cfg(test)]
mod test {
use std::io::Error;
use tokio::sync::mpsc::channel;
use uuid::Uuid;
use crate::client_manager::{ClientManager, ClientMgrMessage};
use foundation::messages::client::ClientStreamOut;
use foundation::prelude::IManager;
use foundation::test::create_connection_pair;
use crate::client_manager::{ClientManager, ClientMgrMessage};
use std::io::Error;
use tokio::sync::mpsc::channel;
use uuid::Uuid;
#[tokio::test]
async fn add_new_client_to_manager() -> Result<(), Error> {
let (sender, mut receiver) =
channel::<ClientMgrMessage>(1024);
let (sender, mut receiver) = channel::<ClientMgrMessage>(1024);
let (server, (client, addr)) = create_connection_pair().await?;
let client_manager = ClientManager::new(sender);
@ -213,12 +196,9 @@ mod test {
let id = Uuid::new_v4();
let username = "TestUser".to_string();
client_manager.add_client(
id,
username.clone(),
addr.to_string(),
server
).await;
client_manager
.add_client(id, username.clone(), addr.to_string(), server)
.await;
assert_eq!(client_manager.get_count().await, 1);
let msg = client.read::<ClientStreamOut>().await?;

34
server/src/messages.rs
View File

@ -1,8 +1,5 @@
use std::sync::{Arc};
use uuid::Uuid;
use foundation::connection::Connection;
/// # ClientMessage
///
/// These messages are send from the client to a receiver
@ -15,34 +12,41 @@ use foundation::connection::Connection;
///
#[derive(Debug)]
pub enum ClientMessage {
#[allow(dead_code)]
Connected,
#[allow(dead_code)]
IncomingMessage { from: Uuid, to: Uuid, content: String },
IncomingMessage {
from: Uuid,
to: Uuid,
content: String,
},
#[allow(dead_code)]
IncomingGlobalMessage { from: Uuid, content: String },
IncomingGlobalMessage {
from: Uuid,
content: String,
},
#[allow(dead_code)]
RequestedUpdate { from: Uuid },
RequestedUpdate {
from: Uuid,
},
Disconnect { id: Uuid },
Disconnect {
id: Uuid,
},
Error,
}
impl PartialEq for ClientMessage {
fn eq(&self, other: &Self) -> bool {
use ClientMessage::{Disconnect, Connected, Error};
use ClientMessage::{Connected, Disconnect, Error};
match (self,other) {
match (self, other) {
(Connected, Connected) => true,
(Error, Error) => true,
(Disconnect {id, .. }, Disconnect {id: other_id, .. }) => id == other_id,
_ => {
false
}
(Disconnect { id, .. }, Disconnect { id: other_id, .. }) => id == other_id,
_ => false,
}
}
}

145
server/src/network_manager.rs
View File

@ -6,8 +6,8 @@ use uuid::Uuid;
use async_trait::async_trait;
use tokio::net::TcpListener;
use tokio::select;
use tokio::sync::mpsc::Sender;
use tokio::{select};
use tokio::sync::Mutex;
use foundation::connection::Connection;
@ -22,7 +22,7 @@ pub enum NetworkManagerMessage {
address: String,
username: String,
connection: Arc<Connection>
connection: Arc<Connection>,
},
}
@ -31,15 +31,23 @@ impl PartialEq for NetworkManagerMessage {
use NetworkManagerMessage::ClientConnecting;
match (self, other) {
(ClientConnecting {uuid,address,username, .. },
(
ClientConnecting {
uuid,
address,
username,
..
},
ClientConnecting {
uuid: other_uuid,
address: other_address,
username: other_username, ..
}) => uuid == other_uuid && address == other_address && username == other_username,
username: other_username,
..
},
) => uuid == other_uuid && address == other_address && username == other_username,
#[allow(unreachable_patterns)]
_ => false
_ => false,
}
}
}
@ -53,22 +61,21 @@ impl PartialEq for NetworkManagerMessage {
/// - listener: the TcpListener that is receiving connections.
/// - out_channel: the channel that will be sent events from NetworkManager.
pub struct NetworkManager<Out>
where
Out: From<NetworkManagerMessage> + Send
where
Out: From<NetworkManagerMessage> + Send,
{
listener: Mutex<TcpListener>,
out_channel: Sender<Out>,
}
impl<Out> NetworkManager<Out>
where
Out: From<NetworkManagerMessage> + Send
where
Out: From<NetworkManagerMessage> + Send,
{
pub async fn new(
address: &str,
out_channel: Sender<Out>
out_channel: Sender<Out>,
) -> Result<Arc<NetworkManager<Out>>, Error> {
let listener = TcpListener::bind(address).await?;
Ok(Arc::new(NetworkManager {
@ -85,34 +92,58 @@ impl<Out> NetworkManager<Out>
/// This fetches the IP address from the NetworkManager
#[allow(dead_code)]
pub async fn address(&self) -> String {
self.listener.lock().await.local_addr().unwrap().ip().to_string()
self
.listener
.lock()
.await
.local_addr()
.unwrap()
.ip()
.to_string()
}
async fn handle_connection(&self, connection: Arc<Connection>) -> Result<(), Error>{
use NetworkSockIn::{Info, Connect};
use NetworkSockOut::{GotInfo, Request, Connecting};
async fn handle_connection(&self, connection: Arc<Connection>) -> Result<(), Error> {
use NetworkSockIn::{Connect, Info};
use NetworkSockOut::{Connecting, GotInfo, Request};
connection.write(Request).await?;
match connection.read().await? {
Info => connection.write(GotInfo {
server_name: "TestServer".into(),
server_owner: "Michael".into()
}).await?,
Connect { uuid, address, username } => {
Info => {
connection
.write(GotInfo {
server_name: "TestServer".into(),
server_owner: "Michael".into(),
})
.await?
}
Connect {
uuid,
address,
username,
} => {
connection.write(Connecting).await?;
let _ = self.out_channel.send(NetworkManagerMessage::ClientConnecting {
uuid,
address,
username,
let _ = self
.out_channel
.send(
NetworkManagerMessage::ClientConnecting {
uuid,
address,
username,
connection,
}.into()).await;
connection,
}
.into(),
)
.await;
}
#[allow(unreachable_patterns)]
_ => {
return Err(Error::new(ErrorKind::InvalidData, "Did not receive valid message"));
return Err(Error::new(
ErrorKind::InvalidData,
"Did not receive valid message",
));
}
}
Ok(())
@ -121,8 +152,8 @@ impl<Out> NetworkManager<Out>
#[async_trait]
impl<Out: 'static> IManager for NetworkManager<Out>
where
Out: From<NetworkManagerMessage> + Send
where
Out: From<NetworkManagerMessage> + Send,
{
async fn run(self: &Arc<Self>) {
let lock = self.listener.lock().await;
@ -142,23 +173,23 @@ impl<Out: 'static> IManager for NetworkManager<Out>
#[cfg(test)]
mod test {
use std::io::Error;
use tokio::sync::mpsc::channel;
use uuid::Uuid;
use crate::network_manager::{
NetworkManager, NetworkManagerMessage, NetworkManagerMessage::ClientConnecting,
};
use foundation::connection::Connection;
use foundation::messages::network::NetworkSockIn::{Connect, Info};
use foundation::messages::network::NetworkSockOut;
use foundation::messages::network::NetworkSockOut::{Connecting, GotInfo, Request};
use foundation::prelude::IManager;
use crate::network_manager::{NetworkManager, NetworkManagerMessage::{ClientConnecting}, NetworkManagerMessage};
use std::io::Error;
use tokio::sync::mpsc::channel;
use uuid::Uuid;
#[tokio::test]
async fn test_network_fetch_info() -> Result<(), Error> {
let (tx, _rx) = channel::<NetworkManagerMessage>(16);
let (tx,_rx) = channel::<NetworkManagerMessage>(16);
let network_manager =
NetworkManager::new("localhost:0",tx).await?;
let network_manager = NetworkManager::new("localhost:0", tx).await?;
network_manager.start();
let port = network_manager.port().await;
@ -171,7 +202,10 @@ mod test {
let out = client.read::<NetworkSockOut>().await?;
assert_eq!(
out,
GotInfo {server_owner: "Michael".into(), server_name: "TestServer".into()}
GotInfo {
server_owner: "Michael".into(),
server_name: "TestServer".into()
}
);
Ok(())
@ -180,8 +214,7 @@ mod test {
#[tokio::test]
async fn test_network_login() -> Result<(), Error> {
let (tx, mut rx) = channel::<NetworkManagerMessage>(16);
let network_manager =
NetworkManager::new("localhost:0",tx).await?;
let network_manager = NetworkManager::new("localhost:0", tx).await?;
network_manager.start();
let port = network_manager.port().await;
@ -190,17 +223,18 @@ mod test {
assert_eq!(client.read::<NetworkSockOut>().await?, Request);
// construct client data
let uuid = Uuid::new_v4();
let address = "localhost";
let username = "TestUser";
client.write(Connect {
uuid,
address: address.to_string(),
username: username.to_string()
}).await?;
client
.write(Connect {
uuid,
address: address.to_string(),
username: username.to_string(),
})
.await?;
let res: NetworkSockOut = client.read().await?;
@ -208,13 +242,16 @@ mod test {
let network_out = rx.recv().await.unwrap();
assert_eq!(network_out, ClientConnecting {
uuid,
address: address.to_string(),
username: username.to_string(),
connection: client
});
assert_eq!(
network_out,
ClientConnecting {
uuid,
address: address.to_string(),
username: username.to_string(),
connection: client
}
);
Ok(())
}
}
}

97
server/src/server.rs
View File

@ -3,73 +3,68 @@ use std::sync::Arc;
use uuid::Uuid;
use tokio::sync::{Mutex, mpsc::{channel, Receiver}};
use tokio::fs::{create_dir, DirBuilder, File, read_dir};
use mlua::{Lua, UserDataFields, UserDataMethods};
use mlua::prelude::LuaUserData;
use tokio::io::AsyncReadExt;
use tokio::join;
use tokio::sync::{
mpsc::{channel, Receiver},
Mutex,
};
use foundation::connection::Connection;
use foundation::prelude::IManager;
use crate::client_manager::{ClientManager, ClientMgrMessage};
use crate::lua::ServerLua;
use crate::network_manager::{NetworkManager, NetworkManagerMessage};
use crate::plugin_manager::PluginManager;
#[derive(Debug,Clone)]
#[derive(Debug, Clone)]
pub enum ServerMessage {
ClientConnected {
uuid: Uuid,
address: String,
username: String,
connection: Arc<Connection>
connection: Arc<Connection>,
},
BroadcastGlobalMessage {
from: Uuid,
content: String,
},
BroadcastGlobalMessage {from: Uuid, content: String},
}
impl From<NetworkManagerMessage> for ServerMessage {
fn from(msg: NetworkManagerMessage) -> Self {
use NetworkManagerMessage::{ClientConnecting};
use NetworkManagerMessage::ClientConnecting;
match msg {
ClientConnecting {
uuid,
address,
username,
connection
connection,
} => ServerMessage::ClientConnected {
uuid,
address,
username,
connection
connection,
},
#[allow(unreachable_patterns)]
_ => unimplemented!()
_ => unimplemented!(),
}
}
}
impl From<ClientMgrMessage> for ServerMessage {
fn from(msg: ClientMgrMessage) -> Self {
use ClientMgrMessage::{BroadcastGlobalMessage,};
use ClientMgrMessage::BroadcastGlobalMessage;
match msg {
BroadcastGlobalMessage {
from,
content,
} => ServerMessage::BroadcastGlobalMessage {
from,
content
},
_ => unimplemented!()
BroadcastGlobalMessage { from, content } => {
ServerMessage::BroadcastGlobalMessage { from, content }
}
_ => unimplemented!(),
}
}
}
/// # Server
/// authors: @michael-bailey, @Mitch161
/// This Represents a server instance.
@ -86,36 +81,26 @@ pub struct Server {
network_manager: Arc<NetworkManager<ServerMessage>>,
plugin_manager: Arc<PluginManager>,
receiver: Mutex<Receiver<ServerMessage>>,
lua: Arc<Mutex<Lua>>,
}
impl Server {
/// Create a new server object
pub async fn new() -> Result<Arc<Server>, Error> {
let (
sender,
receiver
) = channel(1024);
let (sender, receiver) = channel(1024);
let server = Arc::new(Server {
client_manager: ClientManager::new(sender.clone()),
network_manager: NetworkManager::new("0.0.0.0:5600", sender).await?,
plugin_manager: PluginManager::new(),
receiver: Mutex::new(receiver),
lua: Arc::new(Mutex::new(Lua::new())),
});
server.lua.lock().await.globals().set("Server", ServerLua::new(server.clone())).unwrap();
server.load_scripts().await?;
Ok(server)
}
pub async fn port(self: &Arc<Server>) -> u16 {
self.network_manager.port().await
}
pub async fn start(self: &Arc<Server>) {
// start client manager and network manager
self.network_manager.clone().start();
@ -135,16 +120,13 @@ impl Server {
uuid,
address,
username,
connection
connection,
} => {
server.client_manager
.add_client(
uuid,
username,
address,
connection
).await
},
server
.client_manager
.add_client(uuid, username, address, connection)
.await
}
ServerMessage::BroadcastGlobalMessage {
from: _,
content: _,
@ -157,30 +139,11 @@ impl Server {
// ).await
}
#[allow(unreachable_patterns)]
_ => {unimplemented!()}
_ => {
unimplemented!()
}
}
}
}
}
pub async fn load_scripts(self: &Arc<Server>) -> Result<(), Error>{
if let Ok( mut scripts) = read_dir("./scripts").await {
while let Some(child) = scripts.next_entry().await? {
let metadata = child.metadata().await?;
if metadata.is_file() && child.path().extension().unwrap() == "lua" {
let mut file = File::open(child.path()).await.unwrap();
let mut data = String::new();
file.read_to_string(&mut data).await.unwrap();
let server = self.clone();
println!("---| loaded script |---\n{}", data);
println!("---| script output |---");
server.clone().lua.lock().await.load(&data).exec().unwrap();
}
}
} else {
create_dir("./scripts").await?;
}
Ok(())
}
}