Merge branch 'network-client' of https://github.com/mitch161/rust-chat-server into network-client

foundation/src/messages/client.rs

@@ -4,12 +4,24 @@ use serde::{Serialize, Deserialize};
 /// # ClientMessage
 /// This enum defined the message that a client can receive from the server
 /// This uses the serde library to transform to and from json. 
+/// 
 #[derive(Serialize, Deserialize)]
 pub enum ClientStreamIn {
-  Disconnect {id: String},
+  Connected,
+
+  Update,
+  SendMessage {to_uuid: String, contents: String},
+  SendGlobalMessage {contents: String},
+
+  Disconnect,
 }
 
+#[derive(Serialize, Deserialize)]
 pub enum ClientStreamOut {
-  Message {from_uuid: String},
-  Disconnect,
+  Connected,
+
+  UserMessage {from_uuid: String, contents: String},
+  GlobalMessage {contents: String},
+
+  Disconnected,
 }

foundation/src/messages/network.rs

@@ -10,6 +10,7 @@ pub enum NetworkSockIn {
 #[derive(Serialize, Deserialize)]
 pub enum NetworkSockOut<'a> {
 	Request,
+  
 	GotInfo {server_name: &'a str, server_owner: &'a str},
   Connecting,
 }

server/src/client.rs

@@ -1,3 +1,4 @@
+use std::mem::replace;
 use crate::messages::ClientMessage;
 use crate::messages::ServerMessage;
 use std::cmp::Ordering;
@@ -34,7 +35,7 @@ pub struct Client {
 
 	// non serializable
 	#[serde(skip)]
-  server_channel: Option<Sender<ServerMessage>>,
+  server_channel: Mutex<Option<Sender<ServerMessage>>>,
 
   #[serde(skip)]
   input: Sender<ClientMessage>,
@@ -72,7 +73,7 @@ impl Client {
       uuid: Uuid::parse_str(&uuid).expect("invalid id"),
       address,
 
-      server_channel: Some(server_channel),
+      server_channel: Mutex::new(Some(server_channel)),
 
       input: sender,
       output: receiver,
@@ -95,31 +96,51 @@ impl IMessagable<ClientMessage, Sender<ServerMessage>> for Client{
 		self.input.send(msg).expect("failed to send message to client.");
 	}
   fn set_sender(&self, sender: Sender<ServerMessage>) {
-
+    let mut server_lock = self.server_channel.lock().unwrap();
+    let _ = replace(&mut *server_lock, Some(sender));
   }
 }
 
 // cooperative multitasking implementation
 impl ICooperative for Client {
 	fn tick(&self) {
-    // aquire locks (so value isn't dropped)
-    let mut reader_lock = self.stream_reader.lock().unwrap();
-    let mut writer_lock = self.stream_writer.lock().unwrap();
+    println!("[client]: Tick!");
+    {
+      // aquire locks (so value isn't dropped)
+      let mut reader_lock = self.stream_reader.lock().unwrap();
+      let mut writer_lock = self.stream_writer.lock().unwrap();
 
-    // aquiring mutable buffers
-    let reader = reader_lock.as_mut().unwrap();
-    let _writer = writer_lock.as_mut().unwrap();
+      // aquiring mutable buffers
+      let reader = reader_lock.as_mut().unwrap();
+      let _writer = writer_lock.as_mut().unwrap();
 
-    // create buffer
-    let mut buffer = String::new();
+      // create buffer
+      let mut buffer = String::new();
 
-    // loop over all lines that have been sent.
-    while let Ok(_size) = reader.read_line(&mut buffer) {
-      let command = serde_json::from_str::<ClientStreamIn>(buffer.as_str()).unwrap();
+      // loop over all lines that have been sent.
+      while let Ok(_size) = reader.read_line(&mut buffer) {
+        let command = serde_json::from_str::<ClientStreamIn>(buffer.as_str()).unwrap();
 
-      match command {
-        ClientStreamIn::Disconnect {id} => println!("got Disconnect from id: {:?}", id),
-        _ => println!("New command found"),
+        match command {
+          ClientStreamIn::Disconnect => println!("got Disconnect"),
+          _ => println!("New command found"),
+        }
+      }
+    }
+
+    {
+      for message in self.output.iter() {
+        use ClientMessage::{Disconnect};
+        match message {
+          Disconnect => {
+            let lock = self.server_channel.lock().unwrap();
+            
+            if let Some(sender) = lock.as_ref() {
+              sender.send(ServerMessage::ClientDisconnected(self.uuid)).unwrap();
+            }
+          },
+          _ => println!("command not implemneted yet"),
+        }
       }
     }
 
@@ -141,7 +162,7 @@ impl Default for Client {
 		  output: reciever,
 			input: sender,
 
-      server_channel: None,
+      server_channel: Mutex::new(None),
 
       stream: Mutex::new(None),
 

server/src/client_manager.rs

@@ -1,33 +1,29 @@
 // use crate::lib::server::ServerMessages;
-use crate::messages::ServerMessage;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::collections::HashMap;
+use std::mem::replace;
 
 use crossbeam_channel::{unbounded, Receiver, Sender};
 use uuid::Uuid;
 
 use crate::client::Client;
+use crate::messages::ClientMgrMessage;
+use crate::messages::ServerMessage;
+use crate::messages::ClientMessage;
 use foundation::prelude::IMessagable;
 use foundation::prelude::ICooperative;
 
-enum ClientManagerMessage {
-  #[allow(dead_code)]
-  DropAll,
-  #[allow(dead_code)]
-  MessageClient,
-}
-
 /// # ClientManager
 /// This struct manages all connected users
 #[derive(Debug)]
 pub struct ClientManager {
   clients: Mutex<HashMap<Uuid, Arc<Client>>>,
 
-	server_channel: Sender<ServerMessage>,
+	server_channel: Mutex<Sender<ServerMessage>>,
 
-  sender: Sender<ClientManagerMessage>,
-  receiver: Receiver<ClientManagerMessage>,
+  sender: Sender<ClientMgrMessage>,
+  receiver: Receiver<ClientMgrMessage>,
 }
 
 impl ClientManager {
@@ -38,7 +34,7 @@ impl ClientManager {
     Arc::new(ClientManager {
       clients: Mutex::default(),
 
-			server_channel,
+			server_channel: Mutex::new(server_channel),
 
       sender,
       receiver,
@@ -46,40 +42,40 @@ impl ClientManager {
   }
 }
 
-// impl TClientManager<Client, ClientMessage> for ClientManager {
-//   fn add_client(&self, client: std::sync::Arc<Client>) {
-//     self.clients.lock().unwrap().insert(client.uuid, client);
-//   }
-
-//   fn remove_client(&self, uuid: Uuid) {
-//     let _ = self.clients.lock().unwrap().remove(&uuid);
-//   }
-
-//   fn send_message_to_client(&self, uuid: Uuid, msg: ClientMessage) {
-//     let clients = self.clients.lock().unwrap();
-//     let client = clients.get(&uuid).unwrap();
-//     client.send_message(msg);
-//   }
-// }
-
-impl IMessagable<ClientManagerMessage, Sender<ServerMessage>> for ClientManager {
-	fn send_message(&self, msg: ClientManagerMessage) {
-
+impl IMessagable<ClientMgrMessage, Sender<ServerMessage>> for ClientManager {
+	fn send_message(&self, msg: ClientMgrMessage) {
+    self.sender.send(msg).unwrap();
   }
   fn set_sender(&self, sender: Sender<ServerMessage>) {
-
+    let mut server_lock = self.server_channel.lock().unwrap();
+    let _ = replace(&mut *server_lock, sender);
   }
 }
 
 impl ICooperative for ClientManager {
   fn tick(&self) {
+    println!("[client manager]: Tick!");
 
-    for message in self.receiver.iter() {
-      match message {
-        ClientManagerMessage::DropAll => {
-          println!("cannot drop all clients yet")
+    if !self.receiver.is_empty() {
+      for message in self.receiver.iter() {
+        use ClientMgrMessage::{Add, Remove, SendMessage};
+
+        match message {
+          Add(client) => {
+            self.clients.lock().unwrap().insert(client.uuid, client).unwrap_or_default();
+          },
+          Remove(uuid) => {
+            let _ = self.clients.lock().unwrap().remove(&uuid);
+          },
+          SendMessage(to_uuid, from_uuid, content) => {
+            let lock = self.clients.lock().unwrap();
+            if let Some(client) = lock.get(&to_uuid) {
+              client.send_message(ClientMessage::Message(from_uuid, content))
+            }
+          },
+          #[allow(unreachable_patterns)]
+          _ => println!("[Client manager]: not implemented")
         }
-        _ => println!("[Client Manager]: method not implemented")
       }
     }
 
@@ -87,34 +83,4 @@ impl ICooperative for ClientManager {
     let clients = self.clients.lock().unwrap();
     let _ = clients.iter().map(|(_uuid, client)| client.tick());
   }
-}
-
-
-#[cfg(test)]
-mod test {
-  // use super::ClientManager;
-  // use std::sync::Arc;
-	// use crate::lib::Foundation::{IOwner};
-
-  #[test]
-  fn test_get_ref() {
-    // let client_manager = ClientManager::new();
-    // let _cm_ref = client_manager.get_ref();
-    // assert_eq!(Arc::weak_count(&client_manager), 2);
-  }
-
-  #[test]
-  fn test_add_client() {
-    todo!()
-  }
-
-  #[test]
-  fn test_remove_client() {
-    todo!()
-  }
-
-  #[test]
-  fn test_remove_all_clients() {
-    todo!()
-  }
-}
+}

server/src/messages.rs

@@ -3,10 +3,21 @@ use std::sync::Arc;
 
 use crate::client::Client;
 
+#[derive(Debug)]
 pub enum ClientMessage {
-  Disconnect
+  Message(Uuid, String),
+
+  Disconnect,
 }
 
+#[derive(Debug)]
+pub enum ClientMgrMessage {
+  Remove(Uuid),
+  Add(Arc<Client>),
+  SendMessage(Uuid, Uuid, String),
+}
+
+#[derive(Debug)]
 pub enum ServerMessage {
   ClientConnected(Arc<Client>),
   ClientDisconnected(Uuid)

server/src/network_manager.rs

@@ -3,7 +3,6 @@ use crate::messages::ServerMessage;
 use std::io::BufWriter;
 use std::io::BufReader;
 use std::sync::Arc;
-use crate::server::ServerMessages;
 use std::net::TcpListener;
 use std::io::Write;
 use std::io::BufRead;
@@ -28,6 +27,7 @@ impl NetworkManager {
 
 		let listener = TcpListener::bind(address)
       .expect("Could not bind to address");
+    listener.set_nonblocking(true).unwrap();
 
     Arc::new(NetworkManager {
       listener,
@@ -38,61 +38,74 @@ impl NetworkManager {
 
 impl ICooperative for NetworkManager {
   fn tick(&self) {
+    println!("[NetworkManager]: Tick!");
+
+
     // get all new connections
 		// handle each request
+    println!("[NetworkManager]: handling new connections!");
 		for connection in self.listener.incoming() {
-			if let Ok(stream) = connection {
+      match connection {
+        Ok(stream) => {
+          stream.set_nonblocking(false).expect("[NetworkManager]: cant set non-blocking on connection");
 
-				// create buffered writers
-				let mut reader = BufReader::new(stream.try_clone().unwrap());
-				let mut writer = BufWriter::new(stream.try_clone().unwrap());
 
-				let mut buffer = String::new();
+          // create buffered writers
+          let mut reader = BufReader::new(stream.try_clone().unwrap());
+          let mut writer = BufWriter::new(stream.try_clone().unwrap());
 
-				// send request message to connection
-				writer.write_all(
-					serde_json::to_string(&NetworkSockOut::Request).unwrap().as_bytes()
-				).unwrap_or_default();
-				writer.write_all(b"\n").unwrap_or_default();
-				writer.flush().unwrap_or_default();
+          let mut buffer = String::new();
 
-				// read the new request into a buffer
-				let res = reader.read_line(&mut buffer);
+          // send request message to connection
+          writer.write_all(
+            serde_json::to_string(&NetworkSockOut::Request).unwrap().as_bytes()
+          ).unwrap_or_default();
+          writer.write_all(b"\n").unwrap_or_default();
+          writer.flush().unwrap_or_default();
 
-        // if reading caused an error skip the connection
-				if res.is_err() {continue;}
+          // read the new request into a buffer
+          let res = reader.read_line(&mut buffer);
 
-				// turn into enum and perform pattern matching
-				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.flush().unwrap();
-						}
-						NetworkSockIn::Connect { uuid, username, address } => {
-              // create client and send to server
-							let new_client = Client::new(
-                uuid,
-								username,
-								address,
-								stream.try_clone().unwrap(),
-								self.server_channel.clone()
-							);
-							self.server_channel.send(
-								ServerMessage::ClientConnected(new_client)
-							).unwrap_or_default();
-						}
-					}
-				}
-			}
+          // if reading caused an error skip the connection
+          if res.is_err() {continue;}
+
+          // turn into enum and perform pattern matching
+          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.flush().unwrap();
+              }
+              NetworkSockIn::Connect { uuid, username, address } => {
+                // create client and send to server
+                let new_client = Client::new(
+                  uuid,
+                  username,
+                  address,
+                  stream.try_clone().unwrap(),
+                  self.server_channel.clone()
+                );
+                self.server_channel.send(
+                  ServerMessage::ClientConnected(new_client)
+                ).unwrap_or_default();
+              }
+            }
+          }
+        }
+        Err(e) => {
+          println!("[NetworkManager]: got error {:?}", e);
+          break;
+        }
+      }
 		}
+    println!("[NetworkManager], ending tick!")
   }
 }

server/src/server.rs

@@ -1,12 +1,14 @@
-use crate::messages::ServerMessage;
-use uuid::Uuid;
-
 use std::sync::Arc;
+
+use uuid::Uuid;
 use crossbeam_channel::{Receiver, unbounded};
 
 use foundation::prelude::ICooperative;
+use foundation::prelude::IMessagable;
 use crate::client_manager::ClientManager;
 use crate::network_manager::NetworkManager;
+use crate::messages::ClientMgrMessage;
+use crate::messages::ServerMessage;
 
 /// # ServerMessages
 /// This is used internally 
@@ -32,7 +34,7 @@ impl Server {
 		Arc::new(Server {
 			client_manager: ClientManager::new(sender.clone()),
 
-      network_manager: NetworkManager::new("5600".to_string(), sender.clone()),
+      network_manager: NetworkManager::new("5600".to_string(), sender),
 			receiver,
 		})
 	}
@@ -40,20 +42,32 @@ impl Server {
 
 impl ICooperative for Server{
 	fn tick(&self) {
+    println!("[server]: Tick!");
+    use ClientMgrMessage::{Remove, Add};
+
+
 
 		// handle new messages loop
-		for message in self.receiver.try_iter() {
-			match message {
-				ServerMessage::ClientConnected(client) => {
-				},
-				ServerMessage::ClientDisconnected(uuid) => {
-				}
-			}
-		}
+    
+    if !self.receiver.is_empty() {
+      println!("[server]: entering loop!");
+      for message in self.receiver.try_iter() {
+        println!("[server]: received message {:?}", &message);
+        match message {
+          ServerMessage::ClientConnected(client) => {
+            self.client_manager.send_message(Add(client))
+          },
+          ServerMessage::ClientDisconnected(uuid) => {
+            println!("disconnecting client {:?}", uuid);
+            self.client_manager.send_message(Remove(uuid));
+          }
+        }
+      }
+    }
 
 		// alocate time for other components
+    println!("[server]: allocating time for others");
 		self.network_manager.tick();
 		self.client_manager.tick();
-
 	}
 }