diff --git a/soulseek-rs-lib/src/actor/mod.rs b/soulseek-rs-lib/src/actor/mod.rs index 1f7bfb9..a01a6ca 100644 --- a/soulseek-rs-lib/src/actor/mod.rs +++ b/soulseek-rs-lib/src/actor/mod.rs @@ -56,8 +56,6 @@ impl ActorHandle { pub(crate) enum ActorMessage { UserMessage(M), Stop, - #[allow(dead_code)] - Tick, } /// Actor system that manages actor lifecycle @@ -134,21 +132,9 @@ impl ActorSystem { ); break; } - Ok(ActorMessage::Tick) => { - tick_count += 1; - trace!( - "[actor_system] Received explicit Tick message #{}", - tick_count - ); - - actor.tick(); - } Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { if last_tick.elapsed() >= tick_interval { tick_count += 1; - // if tick_count % 10 == 0 { - // trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count); - // } actor.tick(); last_tick = Instant::now(); } diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index a98c51a..c22e964 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -53,26 +53,14 @@ impl std::fmt::Display for PeerAddress { } } -#[derive(Debug)] +#[derive(Debug, Default)] pub struct Context { pub logged_in: Option, - #[allow(dead_code)] - rooms: Rooms, -} - -impl Default for Context { - fn default() -> Self { - Self::new() - } } impl Context { pub fn new() -> Self { - Self { - #[allow(dead_code)] - rooms: Rooms::new(), - logged_in: Option::None, - } + Self::default() } } #[derive(Debug, Clone)] @@ -111,72 +99,6 @@ impl UserMessage { } } -#[allow(dead_code)] -#[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(), - } - } - - #[allow(dead_code)] - pub fn print(&self) { - info!("Public rooms ({}):", self.public_rooms.len()); - for room in &self.public_rooms { - room.print(); - } - info!("Owned private rooms ({}):", self.owned_private_rooms.len()); - for room in &self.owned_private_rooms { - room.print(); - } - info!("Private rooms ({}):", self.private_rooms.len()); - for room in &self.private_rooms { - room.print(); - } - info!( - "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, -} -impl Room { - #[allow(dead_code)] - pub fn new(name: String, number_of_users: i32) -> Self { - Self { - name, - number_of_users, - } - } - #[allow(dead_code)] - pub fn set_number_of_users(&mut self, number_of_users: i32) { - self.number_of_users = number_of_users; - } - pub fn print(&self) { - debug!( - "Room: {}, Number of users: {}", - self.name, self.number_of_users - ); - } -} - #[derive(Debug, Clone)] pub enum ServerMessage { ProcessRead, @@ -273,59 +195,37 @@ impl ServerActor { let socket_addr = socket_addrs.next(); - match socket_addr { - Some(addr) => { - if let Ok(stream) = TcpStream::connect(addr) { - if let Err(e) = stream.set_nonblocking(true) { - error!("[server] Failed to set non-blocking: {}", e); - self.disconnect_with_error(e); - return false; - } + let Some(addr) = socket_addr else { + let error_msg = + format!("No socket addresses found for {}:{}", host, port); + error!("[server] {}", error_msg); + self.disconnect_with_error(io::Error::new( + io::ErrorKind::InvalidInput, + error_msg, + )); + return false; + }; - stream.set_nodelay(true).ok(); + let stream = match TcpStream::connect(addr) { + Ok(s) => s, + Err(e) => { + self.disconnect_with_error(e); + return false; + } + }; - self.stream = Some(stream); - self.connection_state = ConnectionState::Connecting { - since: Instant::now(), - }; - true - } else { - match TcpStream::connect(addr) { - Ok(stream) => { - if let Err(e) = stream.set_nonblocking(true) { - error!( - "[server] Failed to set non-blocking: {}", - e - ); - self.disconnect_with_error(e); - return false; - } - stream.set_nodelay(true).ok(); - self.stream = Some(stream); - self.connection_state = - ConnectionState::Connecting { - since: Instant::now(), - }; - true - } - Err(e) => { - self.disconnect_with_error(e); - false - } - } - } - } - None => { - let error_msg = - format!("No socket addresses found for {}:{}", host, port); - error!("[server] {}", error_msg); - self.disconnect_with_error(io::Error::new( - io::ErrorKind::InvalidInput, - error_msg, - )); - false - } + if let Err(e) = stream.set_nonblocking(true) { + error!("[server] Failed to set non-blocking: {}", e); + self.disconnect_with_error(e); + return false; } + stream.set_nodelay(true).ok(); + + self.stream = Some(stream); + self.connection_state = ConnectionState::Connecting { + since: Instant::now(), + }; + true } pub fn set_self_handle(&mut self, handle: ActorHandle) { diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index b4d9d00..7abeb4a 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -969,21 +969,8 @@ impl Client { }; if peer_exists { - // don't know if i should update? and or reconnect the peer - // debug!( - // "existing peer: {:?}, new peer details: - // username: {}, - // host: {}, - // port: {} - // obfuscation_type: {} - // obfuscated_port: {}", - // peer, - // username, - // host, - // port, - // obfuscation_type, - // obfuscated_port, - // ); + // Existing peer: skip re-registration. Reconnect + // policy on conflict is intentionally undecided. } else { let peer = Peer::new( username, diff --git a/soulseek-rs-lib/src/message/message_reader.rs b/soulseek-rs-lib/src/message/message_reader.rs index 09947ba..25a54fe 100644 --- a/soulseek-rs-lib/src/message/message_reader.rs +++ b/soulseek-rs-lib/src/message/message_reader.rs @@ -3,25 +3,9 @@ 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 -// -// -// +// Soulseek messages are length-prefixed (u32 LE size, then payload). TCP gives us +// arbitrary-sized chunks, so we accumulate into a buffer and only emit a Message +// once size + 4 bytes are available. pub struct MessageReader { buffer: VecDeque, diff --git a/soulseek-rs-lib/src/message/mod.rs b/soulseek-rs-lib/src/message/mod.rs index 49aad94..7b29c31 100644 --- a/soulseek-rs-lib/src/message/mod.rs +++ b/soulseek-rs-lib/src/message/mod.rs @@ -93,50 +93,6 @@ impl Message { trace!(""); } } - // - // pub fn print_hex2(&self) -> String { - // let data = &self.data; - // let mut out = String::from(""); - // const BYTES_PER_LINE: usize = 16; - // - // let chunks = data.chunks(BYTES_PER_LINE); - // for (i, chunk) in chunks.enumerate() { - // // Print the offset - // // out += &format!("{:04x} ", i * BYTES_PER_LINE); - // - // // Print the hexadecimal part - // for j in 0..BYTES_PER_LINE { - // if j < chunk.len() { - // out += &format!("{:02x} ", chunk[j]); - // } else { - // // out += &format!(" "); - // } - // - // // Add extra space in the middle - // // if j == 7 { - // // out += &format!(" "); - // // } - // } - // - // // // Print the ASCII part - // // let mut i = 0; - // // for &byte in chunk { - // // i = i + 1; - // // if byte.is_ascii_graphic() || byte.is_ascii_whitespace() { - // // print!("{}", byte as char); - // // } else {messa - // // print!("."); - // // } - // // - // // if i == 8 { - // // print!(" "); - // // } - // // } - // // - // } - // println!("{:?}", out.trim()); - // return out.trim().into(); - // } #[allow(dead_code)] pub fn get_message_code_u32(&self) -> u32 { @@ -390,9 +346,6 @@ impl Message { }, } } - // pub fn decode(&self) { - // println!("{:?}", self.data); - // } } #[test] diff --git a/soulseek-rs-lib/src/message/server/login.rs b/soulseek-rs-lib/src/message/server/login.rs index 32d7524..8424709 100644 --- a/soulseek-rs-lib/src/message/server/login.rs +++ b/soulseek-rs-lib/src/message/server/login.rs @@ -26,16 +26,3 @@ impl MessageHandler for LoginHandler { sender.send(ServerMessage::LoginStatus(true)).unwrap(); } } - -// fn build_login_response_message() -> Message { -// return Message::new_with_data([ -// 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, -// ]); -// } -// #[test] -// fn test_can_handle() { -// assert_eq!(true, LoginHandler.can_handle(build_login_response_message()); -// ); -// } diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 1dcd412..debb36b 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -106,7 +106,6 @@ impl MessageFactory { .write_int32(41) .write_int32(transfer.token) .write_bool(true) - // .write_int64(transfer.size) .clone() } pub fn build_pierce_firewall_message(token: u32) -> Message { @@ -154,8 +153,6 @@ fn test_build_login_message() { ] .to_vec(); - // println!("{:?}", print_hex(message.get_data())); - // assert_eq!(expect, message.get_data()); assert_eq!(expect, message.get_data()) } diff --git a/soulseek-rs-lib/src/message/server/message_user.rs b/soulseek-rs-lib/src/message/server/message_user.rs index 1e2bce7..9fbe35c 100644 --- a/soulseek-rs-lib/src/message/server/message_user.rs +++ b/soulseek-rs-lib/src/message/server/message_user.rs @@ -29,37 +29,3 @@ impl MessageHandler for MessageUser { user_message.print() } } - -// #[cfg(test)] -// use std::sync::mpsc; -// #[cfg(test)] -// use std::sync::mpsc::{Receiver, Sender}; -// -// #[test] -// fn test_handle() { -// let (message_sender, _message_reader): (Sender, Receiver) = mpsc::channel(); -// -// let context = Arc::new(Mutex::new(Context::new(Arc::new(Mutex::new( -// message_sender, -// ))))); -// let mut message = Message::new_with_data(vec![ -// 254, 0, 0, 0, 22, 0, 0, 0, 153, 102, 47, 0, 58, 65, 228, 102, 6, 0, 0, 0, 115, 101, 114, -// 118, 101, 114, 227, 0, 0, 0, 89, 111, 117, 114, 32, 99, 111, 110, 110, 101, 99, 116, 105, -// 111, 110, 32, 105, 115, 32, 114, 101, 115, 116, 114, 105, 99, 116, 101, 100, 58, 32, 89, -// 111, 117, 32, 99, 97, 110, 110, 111, 116, 32, 115, 101, 97, 114, 99, 104, 32, 111, 114, 32, -// 99, 104, 97, 116, 46, 32, 89, 111, 117, 114, 32, 99, 108, 105, 101, 110, 116, 32, 118, 101, -// 114, 115, 105, 111, 110, 32, 105, 115, 32, 116, 111, 111, 32, 111, 108, 100, 46, 32, 89, -// 111, 117, 32, 110, 101, 101, 100, 32, 116, 111, 32, 117, 112, 103, 114, 97, 100, 101, 32, -// 116, 111, 32, 116, 104, 101, 32, 108, 97, 116, 101, 115, 116, 32, 118, 101, 114, 115, 105, -// 111, 110, 46, 32, 67, 108, 111, 115, 101, 32, 116, 104, 105, 115, 32, 99, 108, 105, 101, -// 110, 116, 44, 32, 100, 111, 119, 110, 108, 111, 97, 100, 32, 110, 101, 119, 32, 118, 101, -// 114, 115, 105, 111, 110, 32, 102, 114, 111, 109, 32, 104, 116, 116, 112, 58, 47, 47, 119, -// 119, 119, 46, 115, 108, 115, 107, 110, 101, 116, 46, 111, 114, 103, 44, 32, 105, 110, 115, -// 116, 97, 108, 108, 32, 105, 116, 32, 97, 110, 100, 32, 114, 101, 99, 111, 110, 110, 101, -// 99, 116, 46, 1, -// ]); -// message.set_pointer(8); -// MessageUser.handle(&mut message.clone(), context.clone()); -// assert_eq!(1, context.lock().unwrap().get_user_messages().len()); -// } -// // } diff --git a/soulseek-rs-lib/src/message/server/room_list.rs b/soulseek-rs-lib/src/message/server/room_list.rs index b34cd94..2920088 100644 --- a/soulseek-rs-lib/src/message/server/room_list.rs +++ b/soulseek-rs-lib/src/message/server/room_list.rs @@ -12,51 +12,7 @@ impl MessageHandler for RoomListHandler { } fn handle(&self, _message: &mut Message, _sender: Sender) { - // let num_public_rooms = message.read_int32(); - // for _ in 0..num_public_rooms { - // rooms.public_rooms.push(Room::new(message.read_string(), 0)); - // } - // - // 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()); - // } - // - // let num_owned_private_rooms = message.read_int32(); - // for _ in 0..num_owned_private_rooms { - // rooms - // .owned_private_rooms - // .push(Room::new(message.read_string(), 0)); - // } - // - // let num_owned_private_rooms = message.read_int32(); - // for i in 0..num_owned_private_rooms { - // rooms.owned_private_rooms[i as usize].set_number_of_users(message.read_int32()); - // } - // - // let num_private_rooms = message.read_int32(); - // for _ in 0..num_private_rooms { - // rooms - // .private_rooms - // .push(Room::new(message.read_string(), 0)); - // } - // - // let num_private_rooms = message.read_int32(); - // for i in 0..num_private_rooms { - // rooms.private_rooms[i as usize].set_number_of_users(message.read_int32()); - // } - // - // let num_operated_private_rooms = message.read_int32(); - // for _ in 0..num_operated_private_rooms { - // 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 { - // rooms.operated_private_rooms[i as usize].set_number_of_users(message.read_int32()); - // } - // cvat.notify_all(); - // rooms.print(); + // Room listing is not yet exposed to the client; this handler exists + // so the dispatcher can ack code 64 without logging an unknown message. } } diff --git a/soulseek-rs-lib/src/utils/md5.rs b/soulseek-rs-lib/src/utils/md5.rs index ba77ef7..dc0c3e9 100644 --- a/soulseek-rs-lib/src/utils/md5.rs +++ b/soulseek-rs-lib/src/utils/md5.rs @@ -361,7 +361,7 @@ fn compute_md5_digest(mut v: Vec) -> String { */ fn bit_padding(input: &str) -> Vec { let mut input_vector: Vec = convert_str_to_vec(input); - let bit_length: u64 = (input.len() as u64) * 8u64; // todo - add support for > 2^64 bit size + let bit_length: u64 = (input.len() as u64) * 8u64; // 128_u8 is the equivalent of padding 1 as an unsigned 8-bit integer // with lower-order bits first diff --git a/soulseek-rs-lib/src/utils/thread_pool.rs b/soulseek-rs-lib/src/utils/thread_pool.rs index 41e5bda..eafd0a1 100644 --- a/soulseek-rs-lib/src/utils/thread_pool.rs +++ b/soulseek-rs-lib/src/utils/thread_pool.rs @@ -1,4 +1,3 @@ -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::thread; @@ -14,7 +13,6 @@ enum Message { pub struct ThreadPool { workers: Vec, sender: Option>, - active_threads: Arc, } impl ThreadPool { @@ -23,22 +21,15 @@ impl ThreadPool { let (sender, receiver) = mpsc::channel(); let receiver = Arc::new(Mutex::new(receiver)); - let active_threads = Arc::new(AtomicUsize::new(0)); let mut workers = Vec::with_capacity(size); - for id in 0..size { - workers.push(Worker::new( - id, - Arc::clone(&receiver), - Arc::clone(&active_threads), - size, - )); + for _ in 0..size { + workers.push(Worker::new(Arc::clone(&receiver))); } ThreadPool { workers, sender: Some(sender), - active_threads, } } @@ -51,26 +42,16 @@ impl ThreadPool { sender.send(Message::NewJob(job)).unwrap(); } } - - pub fn thread_count(&self) -> usize { - self.workers.len() - } - - pub fn active_thread_count(&self) -> usize { - self.active_threads.load(Ordering::SeqCst) - } } impl Drop for ThreadPool { fn drop(&mut self) { - // Send termination message to all workers if let Some(sender) = self.sender.take() { for _ in &self.workers { sender.send(Message::Terminate).unwrap(); } } - // Wait for all workers to finish for worker in &mut self.workers { if let Some(thread) = worker.thread.take() { thread.join().unwrap(); @@ -80,17 +61,11 @@ impl Drop for ThreadPool { } struct Worker { - _id: usize, thread: Option>, } impl Worker { - fn new( - id: usize, - receiver: Arc>>, - _active_threads: Arc, - _total_threads: usize, - ) -> Worker { + fn new(receiver: Arc>>) -> Worker { let thread = thread::spawn(move || { loop { let message = match receiver.lock_safe() { @@ -98,41 +73,13 @@ impl Worker { Err(_) => break, }; match message { - Ok(Message::NewJob(job)) => { - // let active = - // active_threads.fetch_add(1, Ordering::SeqCst) + 1; - // trace!( - // "Thread {} started job (active: {}/{})", - // id, - // active, - // total_threads - // ); - - job(); - - // let active = - // active_threads.fetch_sub(1, Ordering::SeqCst) - 1; - // trace!( - // "Thread {} finished job (active: {}/{})", - // id, - // active, - // total_threads - // ); - } - Ok(Message::Terminate) => { - // Received termination signal - break; - } - Err(_) => { - // Channel disconnected - break; - } + Ok(Message::NewJob(job)) => job(), + Ok(Message::Terminate) | Err(_) => break, } } }); Worker { - _id: id, thread: Some(thread), } }