diff --git a/src/dispatcher.rs b/src/dispatcher.rs index dc7847d..8e6cd1c 100644 --- a/src/dispatcher.rs +++ b/src/dispatcher.rs @@ -7,18 +7,19 @@ use crate::{ server::Context, }; use std::collections::HashMap; +use std::sync::{Arc, Mutex}; pub trait MessageHandler { - fn handle(&self, message: &mut Message, context: Context); + fn handle(&self, message: &mut Message, context: Arc>); } pub struct MessageDispatcher { handlers: HashMap>, - context: Context, + context: Arc>, } impl MessageDispatcher { - pub fn new(context: Context) -> Self { + pub fn new(context: Arc>) -> Self { let mut dispatcher = Self { handlers: HashMap::new(), context, diff --git a/src/message/handlers/excluded_search_phrases.rs b/src/message/handlers/excluded_search_phrases.rs index 94426dd..4e1fa63 100644 --- a/src/message/handlers/excluded_search_phrases.rs +++ b/src/message/handlers/excluded_search_phrases.rs @@ -1,8 +1,10 @@ -use crate::{dispatcher::MessageHandler, message::Message}; +use std::sync::{Arc, Mutex}; + +use crate::{dispatcher::MessageHandler, message::Message, server::Context}; pub struct ExcludedSearchPhrasesHandler; impl MessageHandler for ExcludedSearchPhrasesHandler { - fn handle(&self, message: &mut Message, _context: crate::server::Context) { + fn handle(&self, message: &mut Message, _context: Arc>) { let item_count = message.read_int32(); let mut exluded_phrases: Vec = Vec::new(); diff --git a/src/message/handlers/login.rs b/src/message/handlers/login.rs index dfd8a7b..aca7080 100644 --- a/src/message/handlers/login.rs +++ b/src/message/handlers/login.rs @@ -1,19 +1,22 @@ +use std::sync::{Arc, Mutex}; + use crate::{ dispatcher::MessageHandler, message::{factory::build_shared_folders_message, Message}, + server::Context, }; pub struct LoginHandler; impl MessageHandler for LoginHandler { - fn handle(&self, message: &mut Message, context: crate::server::Context) { + fn handle(&self, message: &mut Message, context: Arc>) { let response = message.read_int8(); if response == 1 { println!("Login successful"); let greeting = message.read_string(); println!("Server geeting: {:?}", greeting); // Build the shared folders message and queue it - let shared_message = build_shared_folders_message(1, 1).get_data(); - context.queue_message(shared_message); + let shared_message = build_shared_folders_message(1, 1); + context.lock().unwrap().queue_message(shared_message); } else { println!("Login failed"); } diff --git a/src/message/handlers/privileged_users.rs b/src/message/handlers/privileged_users.rs index 0160e0d..0b9a429 100644 --- a/src/message/handlers/privileged_users.rs +++ b/src/message/handlers/privileged_users.rs @@ -1,8 +1,10 @@ -use crate::{dispatcher::MessageHandler, message::Message}; +use std::sync::{Arc, Mutex}; + +use crate::{dispatcher::MessageHandler, message::Message, server::Context}; pub struct PrivilegedUsersHandler; impl MessageHandler for PrivilegedUsersHandler { - fn handle(&self, message: &mut Message, _context: crate::server::Context) { + fn handle(&self, message: &mut Message, _context: Arc>) { let number = message.read_int32(); println!("Number of privilaged users: {}", number); } diff --git a/src/message/handlers/room_list.rs b/src/message/handlers/room_list.rs index 4ee6524..852028e 100644 --- a/src/message/handlers/room_list.rs +++ b/src/message/handlers/room_list.rs @@ -1,65 +1,60 @@ -use crate::{dispatcher::MessageHandler, message::Message, server::Room}; +use crate::{ + dispatcher::MessageHandler, + message::Message, + server::{Context, Room}, +}; +use std::sync::{Arc, Mutex}; pub struct RoomListHandler; impl MessageHandler for RoomListHandler { - fn handle(&self, message: &mut Message, mut context: crate::server::Context) { - let mut rooms: Vec = Vec::new(); - let mut owned_private_rooms: Vec = Vec::new(); - let mut private_rooms: Vec = Vec::new(); - let mut operated_private_rooms: Vec = Vec::new(); + fn handle(&self, message: &mut Message, context: Arc>) { + let mut binding = context.lock().unwrap(); + let rooms = binding.get_rooms(); - // Number of rooms - let num_rooms = message.read_int32(); - for _ in 0..num_rooms { - rooms.push(Room::new(message.read_string(), 0)); + let num_public_rooms = message.read_int32(); + for _ in 0..num_public_rooms { + rooms.public_rooms.push(Room::new(message.read_string(), 0)); } - // Number of rooms - let num_rooms = message.read_int32(); - // get the number of users in each room - for i in 0..num_rooms { - rooms[i as usize].set_number_of_users(message.read_int32()); + let num_public_rooms = message.read_int32(); + for i in 0..num_public_rooms { + rooms.public_rooms[i as usize].set_number_of_users(message.read_int32()); } - // Number of owned private rooms let num_owned_private_rooms = message.read_int32(); for _ in 0..num_owned_private_rooms { - owned_private_rooms.push(Room::new(message.read_string(), 0)); + rooms + .owned_private_rooms + .push(Room::new(message.read_string(), 0)); } - // Number of owned private rooms let num_owned_private_rooms = message.read_int32(); for i in 0..num_owned_private_rooms { - owned_private_rooms[i as usize].set_number_of_users(message.read_int32()); + rooms.owned_private_rooms[i as usize].set_number_of_users(message.read_int32()); } - // Number of private rooms (except owned) let num_private_rooms = message.read_int32(); for _ in 0..num_private_rooms { - private_rooms.push(Room::new(message.read_string(), 0)); + rooms + .private_rooms + .push(Room::new(message.read_string(), 0)); } - // Number of private rooms (except owned) let num_private_rooms = message.read_int32(); for i in 0..num_private_rooms { - // Number of users in private rooms (except owned) - private_rooms[i as usize].set_number_of_users(message.read_int32()); + rooms.private_rooms[i as usize].set_number_of_users(message.read_int32()); } - // Number of operated private rooms let num_operated_private_rooms = message.read_int32(); for _ in 0..num_operated_private_rooms { - operated_private_rooms.push(Room::new(message.read_string(), 0)); + rooms + .operated_private_rooms + .push(Room::new(message.read_string(), 0)); } let num_operated_private_rooms = message.read_int32(); for i in 0..num_operated_private_rooms { - // Number of users in private rooms (except owned) - operated_private_rooms[i as usize].set_number_of_users(message.read_int32()); + rooms.operated_private_rooms[i as usize].set_number_of_users(message.read_int32()); } - println!("Rooms: {:?}", rooms.len()); - println!("Owned private rooms: {:?}", owned_private_rooms.len()); - println!("Private rooms: {:?}", private_rooms.len()); - println!("Operated private rooms: {:?}", operated_private_rooms.len()); - context.set_rooms(rooms); + rooms.print(); } } diff --git a/src/server.rs b/src/server.rs index 9cdd443..c685976 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,13 +1,13 @@ use crate::dispatcher::MessageDispatcher; use crate::message::factory::{build_init_message, build_login_message}; -use crate::message::MessageReader; -use std::collections::VecDeque; +use crate::message::{Message, MessageReader}; use std::io::{self, Write}; use std::net::{TcpStream, ToSocketAddrs}; -use std::sync::{Arc, Barrier, Condvar, Mutex}; +use std::sync::mpsc; +use std::sync::mpsc::{Receiver, Sender}; +use std::sync::{Arc, Barrier, Mutex}; use std::thread::{self}; use std::time::Duration; - #[derive(Debug, Clone)] pub struct ServerAddress { host: String, @@ -26,57 +26,69 @@ impl ServerAddress { pub fn get_port(&self) -> u16 { self.port } - // pub fn set_host(&mut self, host: String) { - // self.host = host; - // } - // - // pub fn set_port(&mut self, port: u16) { - // self.port = port; - // } - // - // pub fn set_host_port(&mut self, host: String, port: u16) { - // self.host = host; - // self.port = port; - // } - // - // pub fn get_host_port(&self) -> String { - // format!("{}:{}", self.host, self.port) - // } } -pub struct Server { - address: ServerAddress, - stream: Arc>, - message_queue: Arc<(Mutex>>, Condvar)>, -} - -#[derive(Debug, Clone)] +#[derive(Debug)] pub struct Context { - message_queue: Arc<(Mutex>>, Condvar)>, - rooms: Vec, + message_sender: Arc>>, + rooms: Rooms, } + impl Context { - pub fn new(message_queue: Arc<(Mutex>>, Condvar)>) -> Self { + pub fn new(message_sender: Arc>>) -> Self { Self { - message_queue, - rooms: Vec::new(), + message_sender, + rooms: Rooms::new(), } } - - pub fn set_rooms(&mut self, rooms: Vec) { - self.rooms = rooms; + pub fn get_rooms(&mut self) -> &mut Rooms { + &mut self.rooms } - pub fn queue_message(&self, message: Vec) { - let (lock, cvar) = &*self.message_queue; - let mut queue = lock.lock().unwrap(); - println!("Queueing message: {:?}", message); // Debug log before pushing to queue - queue.push_back(message); - cvar.notify_one(); + pub fn queue_message(&self, message: Message) { + self.message_sender.lock().unwrap().send(message).unwrap(); } } -#[derive(Debug, Clone)] +#[derive(Debug)] +pub struct Rooms { + pub public_rooms: Vec, + pub owned_private_rooms: Vec, + pub private_rooms: Vec, + pub operated_private_rooms: Vec, +} +impl Rooms { + fn new() -> Self { + Self { + public_rooms: Vec::new(), + owned_private_rooms: Vec::new(), + private_rooms: Vec::new(), + operated_private_rooms: Vec::new(), + } + } + pub fn print(&self) { + println!("Public rooms ({}):", self.public_rooms.len()); + for room in &self.public_rooms { + room.print(); + } + println!("Owned private rooms ({}):", self.owned_private_rooms.len()); + for room in &self.owned_private_rooms { + room.print(); + } + println!("Private rooms ({}):", self.private_rooms.len()); + for room in &self.private_rooms { + room.print(); + } + println!( + "Operated private rooms ({}):", + self.operated_private_rooms.len() + ); + for room in &self.operated_private_rooms { + room.print(); + } + } +} +#[derive(Debug)] pub struct Room { name: String, number_of_users: i32, @@ -91,34 +103,38 @@ impl Room { pub fn set_number_of_users(&mut self, number_of_users: i32) { self.number_of_users = number_of_users; } + pub fn print(&self) { + println!( + "Room: {}, Number of users: {}", + self.name, self.number_of_users + ); + } + + // pub fn get_name(&self) -> &str { + // &self.name + // } + // pub fn get_number_of_users(&self) -> i32 { + // self.number_of_users + // } +} + +pub struct Server { + address: ServerAddress, + context: Arc>, } impl Server { /// Create a new instance of Server, returning a Result pub fn new(address: ServerAddress) -> Result { - println!("Connecting to server at {}:{}", address.host, address.port); + let (message_sender, message_reader): (Sender, Receiver) = + mpsc::channel(); - // Convert the host and port into a socket address - let socket_address = format!("{}:{}", address.host, address.port) - .to_socket_addrs()? - .next() - .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid address"))?; + let context = Arc::new(Mutex::new(Context::new(Arc::new(Mutex::new( + message_sender, + ))))); - // Establish the connection with a timeout - let stream = TcpStream::connect_timeout(&socket_address, Duration::from_secs(10))?; + let server = Self { address, context }; - // Set the stream to non-blocking mode if needed or configure timeouts - stream.set_read_timeout(Some(Duration::from_secs(5)))?; - stream.set_write_timeout(Some(Duration::from_secs(5)))?; - - let message_queue = Arc::new((Mutex::new(VecDeque::new()), Condvar::new())); // Initialize the message queue with Condvar - - let server = Self { - address, - stream: Arc::new(Mutex::new(stream)), - message_queue, - }; - - server.start_read_write_loops(); + server.start_read_write_loops(message_reader)?; Ok(server) } @@ -128,25 +144,37 @@ impl Server { } /// Start reading and writing loops in separate threads - fn start_read_write_loops(&self) { - // Create a channel to send messages to the write thread - let read_stream = Arc::clone(&self.stream); - let write_stream = Arc::clone(&self.stream); - let message_queue = Arc::clone(&self.message_queue); + fn start_read_write_loops(&self, message_reader: Receiver) -> Result<(), io::Error> { + let socket_address = format!("{}:{}", self.address.host, self.address.port) + .to_socket_addrs()? + .next() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid address"))?; + + println!( + "Connecting to server at {}:{}", + self.address.host, self.address.port + ); + + let stream = TcpStream::connect_timeout(&socket_address, Duration::from_secs(10))?; + + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + stream.set_write_timeout(Some(Duration::from_secs(5)))?; + + let mut read_stream = stream.try_clone()?; + let mut write_stream = stream.try_clone()?; let barrier = Arc::new(Barrier::new(2)); let read_barrier = barrier.clone(); let write_barrier = barrier.clone(); - let mut buffered_reader = MessageReader::new(); - let mut dispatcher = MessageDispatcher::new(Context::new(Arc::clone(&message_queue))); + let dispatcher = MessageDispatcher::new(Arc::clone(&self.context)); - // Spawn a thread to handle reading from the server thread::spawn(move || { read_barrier.wait(); + let mut buffered_reader = MessageReader::new(); loop { - match buffered_reader.read_from_socket(&mut read_stream.lock().unwrap()) { + match buffered_reader.read_from_socket(&mut read_stream) { Ok(_) => {} Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, @@ -173,44 +201,31 @@ impl Server { } }); - // introduce a channel , get rid of the thread ? thread::spawn(move || { write_barrier.wait(); loop { - let (lock, cvar) = &*message_queue; - let mut queue = lock.lock().unwrap(); - - // Wait for a message to be added to the queue - while queue.is_empty() { - queue = cvar.wait(queue).unwrap(); - } - - // Pop and send the message - if let Some(message) = queue.pop_front() { - println!("Sending message from queue: {:?}", message); // Debug log when sending - match write_stream.lock().unwrap().write_all(&message) { - Ok(_) => println!("Sent buffered message: {:?}", message), - Err(e) => { - eprintln!("Failed to send buffered message: {}", e); - // Handle error appropriately - break; - } + let message = message_reader.recv().unwrap(); + println!("Sending message from queue: {:?}", message); // Debug log when sending + match write_stream.write_all(&message.get_data()) { + Ok(_) => println!("Sent buffered message: {:?}", message), + Err(e) => { + eprintln!("Failed to send buffered message: {}", e); + break; } } } }); + Ok(()) } - fn queue_message(&self, message: Vec) { - let context = Context::new(self.message_queue.clone()); - context.queue_message(message); + fn queue_message(&self, message: Message) { + self.context.lock().unwrap().queue_message(message); } pub fn login(&mut self, username: &str, password: &str) -> Result<(), std::io::Error> { - self.queue_message(build_init_message().get_data()); - self.queue_message(build_login_message(username, password).get_data()); - // self.queue_message(build_shared_folders_message(1, 2).get_data()); + self.queue_message(build_init_message()); + self.queue_message(build_login_message(username, password)); Ok(()) } }