commit 5c5606696065228b61e62e7e0457f2f242389c82 Author: michel Date: Mon Sep 9 08:32:37 2024 +0200 chore: initial commit semi working version diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..894466a --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,16 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 3 + +[[package]] +name = "md5" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771" + +[[package]] +name = "soulseek_download" +version = "0.1.0" +dependencies = [ + "md5", +] diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..d8afbc7 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "soulseek_download" +version = "0.1.0" +edition = "2021" + +[dependencies] +md5 = "0.7.0" diff --git a/src/client.rs b/src/client.rs new file mode 100644 index 0000000..e49ed97 --- /dev/null +++ b/src/client.rs @@ -0,0 +1,37 @@ +use crate::server::{Server, ServerAddress}; + +pub struct Client { + address: ServerAddress, + username: String, + password: String, +} + +impl Client { + pub fn new(address: ServerAddress, username: String, password: String) -> Self { + Self { + address, + username, + password, + } + } + + pub fn connect(&self) { + // Attempt to create a server instance + match Server::new(self.address.clone()) { + Ok(mut server) => { + println!( + "Connected to server at {}:{}", + server.get_address().get_host(), + server.get_address().get_port() + ); + // Attempt to login + println!("Logging in as {}", self.username); + + if let Err(e) = server.login(&self.username, &self.password) { + eprintln!("Error during login: {}", e); + } + } + Err(e) => eprintln!("Error connecting to server: {}", e), + } + } +} diff --git a/src/dispatcher.rs b/src/dispatcher.rs new file mode 100644 index 0000000..dc7847d --- /dev/null +++ b/src/dispatcher.rs @@ -0,0 +1,58 @@ +use crate::message::handlers::excluded_search_phrases::ExcludedSearchPhrasesHandler; +use crate::message::handlers::login::LoginHandler; +use crate::message::handlers::privileged_users::PrivilegedUsersHandler; +use crate::message::handlers::room_list::RoomListHandler; +use crate::{ + message::{Message, MessageType}, + server::Context, +}; +use std::collections::HashMap; + +pub trait MessageHandler { + fn handle(&self, message: &mut Message, context: Context); +} + +pub struct MessageDispatcher { + handlers: HashMap>, + context: Context, +} + +impl MessageDispatcher { + pub fn new(context: Context) -> Self { + let mut dispatcher = Self { + handlers: HashMap::new(), + context, + }; + dispatcher.register_default_handlers(); + dispatcher + } + + pub fn register_handler( + &mut self, + msg_type: MessageType, + handler: H, + ) -> &mut Self { + self.handlers.insert(msg_type, Box::new(handler)); + self + } + + pub fn dispatch(&self, message: &mut Message) { + if let Some(handler) = self.handlers.get(&message.get_message_type()) { + message.set_pointer(8); + return handler.handle(message, self.context.clone()); + } + println!( + "No handler found for message type: {:?}", + message.get_message_type() + ); + } + pub fn register_default_handlers(&mut self) { + self.register_handler(MessageType::Login, LoginHandler); + self.register_handler(MessageType::PrivilegedUsers, PrivilegedUsersHandler); + self.register_handler( + MessageType::ExcludedSearchPhrases, + ExcludedSearchPhrasesHandler, + ); + self.register_handler(MessageType::RoomList, RoomListHandler); + } +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..d1c5b62 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,23 @@ +mod client; +mod dispatcher; +mod message; +mod server; + +use std::thread; + +use client::Client; +use server::ServerAddress; + +fn main() { + let client = Client::new( + // ServerAddress::new(String::from("localhost"), 2416), + ServerAddress::new(String::from("server.slsknet.org"), 2416), + String::from("insane_in_the_brain2"), + // String::from("invalid"), + String::from("13375137"), + ); + client.connect(); + loop { + thread::park(); + } +} diff --git a/src/message/factory.rs b/src/message/factory.rs new file mode 100644 index 0000000..52cf210 --- /dev/null +++ b/src/message/factory.rs @@ -0,0 +1,60 @@ +use super::Message; + +pub fn build_init_message() -> Message { + let mut message = Message::new(); + message.write_int32(81); // need to send as first message? + message +} + +pub fn build_login_message(username: &str, password: &str) -> Message { + let mut hasher = md5::Context::new(); + + hasher.consume(username); + hasher.consume(password); + + let hash = format!("{:x}", hasher.compute()); + let hash_str = hash.as_str(); + + let mut message = Message::new(); + + message + .write_int32(1) + .write_string(username) + .write_string(password) + .write_int32(160) // version + .write_string(hash_str) + .write_int32(17); + + message +} + +pub fn build_shared_folders_message(folder_count: i32, file_count: i32) -> Message { + let mut message = Message::new(); + message + .write_int32(35) + .write_int32(folder_count) + .write_int32(file_count); + + message +} + +#[test] +fn test_build_init_message() { + let message = build_init_message(); + let expect: Vec = [81, 0, 0, 0].to_vec(); + assert_eq!(expect, message.get_data()) +} + +#[test] +fn test_build_login_message() { + let message = build_login_message("username", "password"); + + let expect: Vec = [ + 1, 0, 0, 0, 8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101, 8, 0, 0, 0, 112, 97, 115, + 115, 119, 111, 114, 100, 160, 0, 0, 0, 32, 0, 0, 0, 100, 53, 49, 99, 57, 97, 55, 101, 57, + 51, 53, 51, 55, 52, 54, 97, 54, 48, 50, 48, 102, 57, 54, 48, 50, 100, 52, 53, 50, 57, 50, + 57, 17, 0, 0, 0, + ] + .to_vec(); + assert_eq!(expect, message.get_data()) +} diff --git a/src/message/handlers/excluded_search_phrases.rs b/src/message/handlers/excluded_search_phrases.rs new file mode 100644 index 0000000..94426dd --- /dev/null +++ b/src/message/handlers/excluded_search_phrases.rs @@ -0,0 +1,16 @@ +use crate::{dispatcher::MessageHandler, message::Message}; +pub struct ExcludedSearchPhrasesHandler; + +impl MessageHandler for ExcludedSearchPhrasesHandler { + fn handle(&self, message: &mut Message, _context: crate::server::Context) { + let item_count = message.read_int32(); + + let mut exluded_phrases: Vec = Vec::new(); + for _ in 0..item_count { + // Read the file name, size, and path (structure can vary) + let phrase = message.read_string(); + exluded_phrases.push(phrase); + } + println!("Excluded search phrases: {:?}", exluded_phrases); + } +} diff --git a/src/message/handlers/login.rs b/src/message/handlers/login.rs new file mode 100644 index 0000000..dfd8a7b --- /dev/null +++ b/src/message/handlers/login.rs @@ -0,0 +1,21 @@ +use crate::{ + dispatcher::MessageHandler, + message::{factory::build_shared_folders_message, Message}, +}; +pub struct LoginHandler; + +impl MessageHandler for LoginHandler { + fn handle(&self, message: &mut Message, context: crate::server::Context) { + 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); + } else { + println!("Login failed"); + } + } +} diff --git a/src/message/handlers/mod.rs b/src/message/handlers/mod.rs new file mode 100644 index 0000000..52526f6 --- /dev/null +++ b/src/message/handlers/mod.rs @@ -0,0 +1,4 @@ +pub mod excluded_search_phrases; +pub mod login; +pub mod privileged_users; +pub mod room_list; diff --git a/src/message/handlers/privileged_users.rs b/src/message/handlers/privileged_users.rs new file mode 100644 index 0000000..f92985c --- /dev/null +++ b/src/message/handlers/privileged_users.rs @@ -0,0 +1,12 @@ +use crate::{ + dispatcher::MessageHandler, + message::{factory::build_shared_folders_message, Message}, +}; +pub struct PrivilegedUsersHandler; + +impl MessageHandler for PrivilegedUsersHandler { + fn handle(&self, message: &mut Message, _context: crate::server::Context) { + 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 new file mode 100644 index 0000000..a986493 --- /dev/null +++ b/src/message/handlers/room_list.rs @@ -0,0 +1,74 @@ +use std::{ + collections::VecDeque, + sync::{Arc, Condvar, Mutex}, +}; + +use crate::{ + dispatcher::MessageHandler, + message::{factory::build_shared_folders_message, Message}, + server::Room, +}; +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(); + + // Number of rooms + let num_rooms = message.read_int32(); + for _ in 0..num_rooms { + 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()); + } + + // 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)); + } + + // 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()); + } + + // 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)); + } + + // 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()); + } + + // 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)); + } + 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()); + } + 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); + } +} diff --git a/src/message/message.rs b/src/message/message.rs new file mode 100644 index 0000000..aec5ccc --- /dev/null +++ b/src/message/message.rs @@ -0,0 +1,160 @@ +use std::str; + +#[derive(Debug, Clone, Eq, PartialEq, Hash)] +pub enum MessageType { + Login, + PrivilegedUsers, + ExcludedSearchPhrases, + RoomList, + Unknown(u8), +} +impl From for MessageType { + fn from(value: u8) -> Self { + match value { + 1 => MessageType::Login, + 69 => MessageType::PrivilegedUsers, + 160 => MessageType::ExcludedSearchPhrases, + 64 => MessageType::RoomList, + _ => MessageType::Unknown(value), + } + } +} + +#[derive(Debug, PartialEq, Clone)] +pub struct Message { + data: Vec, + pointer: usize, +} + +impl Message { + pub fn new() -> Self { + Self { + data: Vec::new(), + pointer: 0, + } + } + + pub fn get_message_type(&self) -> MessageType { + MessageType::from(self.data[4]) + } + + pub fn new_with_data(data: Vec) -> Self { + Self { data, pointer: 0 } + } + // pub fn reset_pointer(&mut self) { + // self.pointer = 0; + // } + pub fn set_pointer(&mut self, pointer: usize) { + self.pointer = pointer; + } + + pub fn get_data(&self) -> Vec { + self.data.clone() + } + + pub fn read_string(&mut self) -> String { + let size = u32::from_le_bytes([ + self.data[self.pointer], + self.data[self.pointer + 1], + self.data[self.pointer + 2], + self.data[self.pointer + 3], + ]) as usize; + + self.pointer += 4; + let data = &self.data[self.pointer..self.pointer + size]; + self.pointer += size; + + String::from_utf8(data.to_vec()).expect("Failed to read string") + } + + pub fn read_int8(&mut self) -> i8 { + let val = self.data[self.pointer] as i8; + self.pointer += 1; + val + } + + pub fn read_int64(&mut self) -> i64 { + let val = i64::from_le_bytes([ + self.data[self.pointer], + self.data[self.pointer + 1], + self.data[self.pointer + 2], + self.data[self.pointer + 3], + self.data[self.pointer + 4], + self.data[self.pointer + 5], + self.data[self.pointer + 6], + self.data[self.pointer + 7], + ]); + self.pointer += 8; + val + } + + pub fn read_int32(&mut self) -> i32 { + if self.pointer + 4 > self.data.len() { + return 0; + } + + let val = i32::from_le_bytes([ + self.data[self.pointer], + self.data[self.pointer + 1], + self.data[self.pointer + 2], + self.data[self.pointer + 3], + ]); + self.pointer += 4; + val + } + + pub fn write_string(&mut self, val: &str) -> &mut Self { + let length = val.len() as u32; + self.data.extend_from_slice(&length.to_le_bytes()); + self.data.extend_from_slice(val.as_bytes()); + self + } + + pub fn write_int32(&mut self, value: i32) -> &mut Self { + self.data.extend_from_slice(&value.to_le_bytes()); + self + } + + // pub fn decode(&self) { + // let size = + // u32::from_le_bytes([self.data[0], self.data[1], self.data[2], self.data[3]]) as usize; + // println!("Size: {}", size); + // if size >= 4 { + // let code = u32::from_le_bytes([self.data[4], self.data[5], self.data[6], self.data[7]]); + // println!("Code: {}", code); + // } + // } +} + +#[test] +fn test_read_string() { + let data = vec![ + 5, 0, 0, 0, // size = 5 + 72, 101, 108, 108, 111, // "Hello" + ]; + let mut test_data = Message::new_with_data(data); + assert_eq!(test_data.read_string(), "Hello"); +} + +#[test] +fn test_read_string_2() { + let data = vec![ + 50, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 81, 170, 162, 77, 32, 0, 0, 0, 101, 102, 99, 97, + 51, 52, 102, 99, 52, 99, 56, 98, 101, 56, 98, 55, 101, 102, 51, 56, 97, 102, 50, 54, 50, + 52, 100, 101, 53, 52, 54, 52, 0, + ]; + let mut test_data = Message::new_with_data(data); + test_data.set_pointer(9); + assert_eq!(test_data.read_string(), ""); +} + +#[test] +#[should_panic(expected = "Failed to read string")] +fn test_read_string_invalid_utf8() { + let data = vec![ + 3, 0, 0, 0, // size = 3 + 255, 255, 255, // invalid UTF-8 sequence + ]; + let mut test_data = Message::new_with_data(data); + test_data.read_string(); +} diff --git a/src/message/mod.rs b/src/message/mod.rs new file mode 100644 index 0000000..3fcba0e --- /dev/null +++ b/src/message/mod.rs @@ -0,0 +1,7 @@ +pub mod factory; +pub mod handlers; +pub mod message; +pub mod reader; + +pub use message::*; +pub use reader::*; diff --git a/src/message/reader.rs b/src/message/reader.rs new file mode 100644 index 0000000..0457252 --- /dev/null +++ b/src/message/reader.rs @@ -0,0 +1,109 @@ +use std::io::{self, Read}; +use std::{collections::VecDeque, net::TcpStream}; + +use crate::message::Message; + +// # why buffered message reader? +// the stream comes in chunks of bytes: 1024 for example +// the soulseek protocol has binary messages +// the first 4 bytes of the message represent the message size +// the rest of the bytes represent the message itself +// +// # how to implement the buffered message reader? +// read 4 bytes from the stream +// if the stream has less than 4 bytes, buffer then add the rest of the bytes to the buffered +// message reader buffer and return +// if the stream has 4 bytes, read the message size +// if the stream is as long as the message size + 4 bytes, read the message +// create a new message struct and read the message size from the buffer, thats a message +// the rest of the bytes are the next message so you call the read function again with the rest of the bytes +// else +// buffer the rest of the bytes and return +// +// +// + +pub struct MessageReader { + buffer: VecDeque, +} + +impl MessageReader { + pub fn new() -> Self { + Self { + buffer: VecDeque::new(), + } + } + + pub fn new_with_buffer(buffer: Vec) -> Self { + Self { + buffer: buffer.into(), + } + } + + pub fn read_from_socket(&mut self, stream: &mut TcpStream) -> io::Result<()> { + let mut temp_buffer = [0; 1024]; // Temporary buffer for reading from the socket + let bytes_read = stream.read(&mut temp_buffer)?; + // println!("Read {} bytes from the socket", bytes_read); + + // Add the read bytes to the internal buffer + self.buffer.extend(&temp_buffer[..bytes_read]); + + Ok(()) + } + + pub fn extract_message(&mut self) -> io::Result> { + let bytes_read = self.buffer.len(); + if bytes_read < 4 { + return Ok(None); + } + + let message_size = u32::from_le_bytes([ + self.buffer[0], + self.buffer[1], + self.buffer[2], + self.buffer[3], + ]) as usize; + + let total_size = message_size + 4; + + if bytes_read < total_size { + return Ok(None); + } + + let message_buffer: Vec = self.buffer.drain(..total_size).collect(); + return Ok(Some(Message::new_with_data(message_buffer))); + } +} + +#[test] +fn test_extract_message() { + let buffer: Vec = [ + 8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101, 8, 0, 0, 0, 112, 97, 115, 115, 119, 111, + 114, 100, 160, 0, 0, 0, 32, 0, 0, 0, 100, 53, 49, 99, 57, 97, 55, 101, 57, 51, 53, 51, 55, + 52, 54, 97, 54, 48, 50, 48, 102, 57, 54, 48, 50, 100, 52, 53, 50, 57, 50, 57, 17, 0, 0, 0, + ] + .to_vec(); + let mut buffered_reader = MessageReader::new_with_buffer(buffer); + let mut message = buffered_reader.extract_message().unwrap().unwrap(); + assert_eq!( + message.get_data(), + vec![8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101] + ); + assert_eq!(message.read_string(), "username"); +} +#[test] +fn test_extract_message_incomplete_message() { + let incomplete_buffer = vec![1, 2, 3]; + let mut buffered_reader = MessageReader::new_with_buffer(incomplete_buffer); + + let result = buffered_reader.extract_message(); + assert_eq!(None, result.unwrap()); + + let rest: Vec = buffered_reader + .buffer + .drain(..buffered_reader.buffer.len()) + .collect(); + + assert!(buffered_reader.buffer.is_empty()); + assert_eq!(vec![1, 2, 3], rest); +} diff --git a/src/server.rs b/src/server.rs new file mode 100644 index 0000000..3b4d61e --- /dev/null +++ b/src/server.rs @@ -0,0 +1,214 @@ +use crate::dispatcher::MessageDispatcher; +use crate::message::factory::{build_init_message, build_login_message}; +use crate::message::{MessageReader, MessageType}; +use std::collections::VecDeque; +use std::io::{self, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::sync::{Arc, Barrier, Condvar, Mutex}; +use std::thread::{self}; +use std::time::Duration; + +#[derive(Debug, Clone)] +pub struct ServerAddress { + host: String, + port: u16, +} + +impl ServerAddress { + pub fn new(host: String, port: u16) -> Self { + Self { host, port } + } + + pub fn get_host(&self) -> &str { + &self.host + } + + 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)] +pub struct Context { + message_queue: Arc<(Mutex>>, Condvar)>, + rooms: Vec, +} +impl Context { + pub fn new(message_queue: Arc<(Mutex>>, Condvar)>) -> Self { + Self { + message_queue, + rooms: Vec::new(), + } + } + + pub fn set_rooms(&mut self, rooms: Vec) { + self.rooms = 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(); + } +} + +#[derive(Debug, Clone)] +pub struct Room { + name: String, + number_of_users: i32, +} +impl Room { + pub fn new(name: String, number_of_users: i32) -> Self { + Self { + name, + number_of_users, + } + } + pub fn set_number_of_users(&mut self, number_of_users: i32) { + self.number_of_users = number_of_users; + } +} +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); + + // 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"))?; + + // Establish the connection with a timeout + let stream = TcpStream::connect_timeout(&socket_address, Duration::from_secs(10))?; + + // 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(); + + Ok(server) + } + + pub fn get_address(&self) -> &ServerAddress { + &self.address + } + + /// 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); + + 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))); + + // Spawn a thread to handle reading from the server + thread::spawn(move || { + read_barrier.wait(); + + loop { + match buffered_reader.read_from_socket(&mut read_stream.lock().unwrap()) { + Ok(_) => {} + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, + Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { + println!("Read operation timed out"); + continue; + } + Err(e) => { + eprintln!("Error reading from server: {}", e); + break; + } + } + + match buffered_reader.extract_message() { + Ok(Some(mut message)) => { + println!("Received message: {:?}", message.get_data()); + dispatcher.dispatch(&mut message) + } + Err(e) => { + println!("Error extracting message: {}", e) + } + Ok(None) => continue, + } + } + }); + 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; + } + } + } + } + }); + } + + fn queue_message(&self, message: Vec) { + let context = Context::new(self.message_queue.clone()); + context.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()); + Ok(()) + } +}