chore: initial commit

semi working version

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",
+]

Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "soulseek_download"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+md5 = "0.7.0"

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),
+        }
+    }
+}

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<MessageType, Box<dyn MessageHandler + Send + Sync>>,
+    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<H: 'static + MessageHandler + Send + Sync>(
+        &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);
+    }
+}

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();
+    }
+}

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<u8> = [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<u8> = [
+        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())
+}

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<String> = 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);
+    }
+}

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");
+        }
+    }
+}

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;

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);
+    }
+}

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<Room> = Vec::new();
+        let mut owned_private_rooms: Vec<Room> = Vec::new();
+        let mut private_rooms: Vec<Room> = Vec::new();
+        let mut operated_private_rooms: Vec<Room> = 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);
+    }
+}

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<u8> 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<u8>,
+    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<u8>) -> 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<u8> {
+        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();
+}

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::*;

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<u8>,
+}
+
+impl MessageReader {
+    pub fn new() -> Self {
+        Self {
+            buffer: VecDeque::new(),
+        }
+    }
+
+    pub fn new_with_buffer(buffer: Vec<u8>) -> 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<Option<Message>> {
+        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<u8> = self.buffer.drain(..total_size).collect();
+        return Ok(Some(Message::new_with_data(message_buffer)));
+    }
+}
+
+#[test]
+fn test_extract_message() {
+    let buffer: Vec<u8> = [
+        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<u8> = buffered_reader
+        .buffer
+        .drain(..buffered_reader.buffer.len())
+        .collect();
+
+    assert!(buffered_reader.buffer.is_empty());
+    assert_eq!(vec![1, 2, 3], rest);
+}

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<Mutex<TcpStream>>,
+    message_queue: Arc<(Mutex<VecDeque<Vec<u8>>>, Condvar)>,
+}
+
+#[derive(Debug, Clone)]
+pub struct Context {
+    message_queue: Arc<(Mutex<VecDeque<Vec<u8>>>, Condvar)>,
+    rooms: Vec<Room>,
+}
+impl Context {
+    pub fn new(message_queue: Arc<(Mutex<VecDeque<Vec<u8>>>, Condvar)>) -> Self {
+        Self {
+            message_queue,
+            rooms: Vec::new(),
+        }
+    }
+
+    pub fn set_rooms(&mut self, rooms: Vec<Room>) {
+        self.rooms = rooms;
+    }
+
+    pub fn queue_message(&self, message: Vec<u8>) {
+        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<Self, io::Error> {
+        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<u8>) {
+        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(())
+    }
+}