diff --git a/soulseek-rs-lib/src/actor/mod.rs b/soulseek-rs-lib/src/actor/mod.rs index fee2a68..1f7bfb9 100644 --- a/soulseek-rs-lib/src/actor/mod.rs +++ b/soulseek-rs-lib/src/actor/mod.rs @@ -1,5 +1,5 @@ -use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; +use std::sync::mpsc::{Receiver, Sender, channel}; use std::time::{Duration, Instant}; use crate::trace; @@ -161,7 +161,10 @@ impl ActorSystem { } } } - trace!("[actor_system] run_actor_loop ENDED - processed {} messages, {} ticks", message_count, tick_count); + trace!( + "[actor_system] run_actor_loop ENDED - processed {} messages, {} ticks", + message_count, tick_count + ); } } diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index e98106b..b41032f 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -159,9 +159,12 @@ impl PeerActor { if let Some(ref handle) = self.self_handle && let Err(e) = handle.send(PeerMessage::SendMessage(transfer_response)) - { - error!("[peer:{}] Failed to send TransferResponse message: {}", username, e); - } + { + error!( + "[peer:{}] Failed to send TransferResponse message: {}", + username, e + ); + } } PeerMessage::TransferResponse { token, @@ -177,15 +180,15 @@ impl PeerActor { if !allowed { if let Some(reason_text) = reason { debug!( - "[peer:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...", - username, reason_text, token - ); + "[peer:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...", + username, reason_text, token + ); } } else { debug!( - "[peer:{}] Transfer allowed, ready to connect with token {:}", - username, token - ); + "[peer:{}] Transfer allowed, ready to connect with token {:}", + username, token + ); self.client_channel .send(ClientOperation::DownloadFromPeer( token, @@ -258,7 +261,9 @@ impl PeerActor { let username = self.peer.read().unwrap().username.clone(); error!( "[peer:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.", - username, e, e.kind() + username, + e, + e.kind() ); self.disconnect_with_error(e); return; diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index fc76e12..b6dfd2b 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -24,7 +24,7 @@ use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; -use crate::{debug, error, info, trace, warn, SoulseekRs}; +use crate::{SoulseekRs, debug, error, info, trace, warn}; #[derive(Debug, Clone)] pub struct PeerAddress { @@ -483,8 +483,8 @@ impl ServerActor { obfuscated_port, } => { debug!( - "[server] Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})" - , username, host, port, obfuscation_type, obfuscated_port + "[server] Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})", + username, host, port, obfuscation_type, obfuscated_port ); if let Err(e) = self.client_channel.send( @@ -496,7 +496,10 @@ impl ServerActor { obfuscated_port, }, ) { - error!("[server] Error forwarding GetPeerAddress response to client: {}", e); + error!( + "[server] Error forwarding GetPeerAddress response to client: {}", + e + ); } } ServerMessage::ProcessRead => { @@ -515,23 +518,27 @@ impl ServerActor { let timeout = Duration::from_secs(5); let context = self.context.clone(); - std::thread::spawn(move || loop { - if start.elapsed() >= timeout { - let _ = response.send(Err(SoulseekRs::Timeout)); - break; - } + std::thread::spawn(move || { + loop { + if start.elapsed() >= timeout { + let _ = response.send(Err(SoulseekRs::Timeout)); + break; + } - if let Some(logged_in) = context.read().unwrap().logged_in { - let result = if logged_in { - Ok(true) - } else { - Err(SoulseekRs::AuthenticationFailed) - }; - let _ = response.send(result); - break; - } + if let Some(logged_in) = + context.read().unwrap().logged_in + { + let result = if logged_in { + Ok(true) + } else { + Err(SoulseekRs::AuthenticationFailed) + }; + let _ = response.send(result); + break; + } - std::thread::sleep(Duration::from_millis(100)); + std::thread::sleep(Duration::from_millis(100)); + } }); } ServerMessage::FileSearch { token, query } => { @@ -560,7 +567,8 @@ impl ServerActor { Err(e) => { error!( "[server] Error reading from server: {} (kind: {:?}). Disconnecting.", - e, e.kind() + e, + e.kind() ); self.disconnect_with_error(e); return; diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index 71403f6..addd925 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -1,27 +1,27 @@ -use crate::actor::server_actor::{PeerAddress, ServerActor, ServerMessage}; use crate::actor::ActorHandle; +use crate::actor::server_actor::{PeerAddress, ServerActor, ServerMessage}; use crate::types::DownloadStatus; use crate::utils::logger; use crate::{ - actor::{peer_registry::PeerRegistry, ActorSystem}, + Transfer, + actor::{ActorSystem, peer_registry::PeerRegistry}, error::{Result, SoulseekRs}, - peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer}, + peer::{ConnectionType, DownloadPeer, NewPeer, Peer, listen::Listen}, types::{Download, Search, SearchResult}, utils::{md5, thread_pool::ThreadPool}, - Transfer, }; use std::{ collections::HashMap, net::TcpStream, sync::{ + RwLock, atomic::{AtomicBool, Ordering}, mpsc::{Receiver, Sender}, - RwLock, }, thread::{self, sleep}, }; use std::{ - sync::{mpsc, Arc}, + sync::{Arc, mpsc}, time::{Duration, Instant}, }; @@ -286,9 +286,10 @@ impl Client { pub fn remove_peer(&self, username: &str) { let context = self.context.read().unwrap(); if let Some(ref registry) = context.peer_registry - && let Some(handle) = registry.remove_peer(username) { - let _ = handle.stop(); - } + && let Some(handle) = registry.remove_peer(username) + { + let _ = handle.stop(); + } } pub fn search( @@ -334,10 +335,11 @@ impl Client { // Check if cancelled if let Some(ref flag) = cancel_flag - && flag.load(Ordering::Relaxed) { - info!("Search cancelled by user"); - break; - } + && flag.load(Ordering::Relaxed) + { + info!("Search cancelled by user"); + break; + } // Check if timeout reached if start.elapsed() >= timeout { @@ -470,225 +472,12 @@ impl Client { client_context: Arc>, own_username: String, ) { - thread::spawn(move || loop { - match reader.recv() { - Ok(operation) => { - match operation { - ClientOperation::ConnectToPeer(peer) => { - let client_context_clone = client_context.clone(); - let own_username_clone = own_username.clone(); - - thread::spawn(move || { - Self::connect_to_peer( - peer, - client_context_clone, - own_username_clone, - None, - ); - }); - } - ClientOperation::SearchResult(search_result) => { - trace!("[client] SearchResult {:?}", search_result); - let mut context = client_context.write().unwrap(); - let result_token = search_result.token; - - // Find the search with matching token - for search in context.searches.values_mut() { - if search.token == result_token { - search.results.push(search_result); - break; - } - } - } - ClientOperation::PeerDisconnected(username, error) => { - let context = client_context.read().unwrap(); - if let Some(ref registry) = context.peer_registry - && let Some(handle) = - registry.remove_peer(&username) - { - let _ = handle.stop(); - } - if let Some(error) = error { - warn!( - "[client] Peer {} disconnected with error: {:?}", - username, error - ); - Self::process_failed_uploads( - client_context.clone(), - &username, - None, - ); - } - } - ClientOperation::PierceFireWall(peer) => { - Self::pierce_firewall( - peer, - client_context.clone(), - own_username.clone(), - ); - } - ClientOperation::DownloadFromPeer( - token, - peer, - allowed, - ) => { - let maybe_download = { - let client_context = - client_context.read().unwrap(); - client_context - .get_download_by_token(token) - .cloned() - }; - let own_username = own_username.clone(); - let client_context_clone = client_context.clone(); - - trace!( - "[client] DownloadFromPeer token: {} peer: {:?}", - token, - peer - ); - match maybe_download { - Some(download) => { - thread::spawn(move || { - let download_peer = DownloadPeer::new( - download.username.clone(), - peer.host.clone(), - peer.port, - token, - allowed, - own_username, - ); - let filename: Option<&str> = download - .filename - .split('\\') - .next_back(); - match filename { - Some(filename) => { - match download_peer.download_file( - client_context_clone.clone(), - Some(download.clone()), - None - ) { - Ok((download, filename)) => { - let _ = download.sender.send(DownloadStatus::Completed); - client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Completed); - info!("Successfully downloaded {} bytes to {}", download.size, filename); - } - Err(e) => { - let _ = download.sender.send(DownloadStatus::Failed); - client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Failed); - error!( - "Failed to download file '{}' from {}:{} (token: {}) - Error: {}", - filename, peer.host, peer.port, download.token, e - ); - } - } - } - None => error!("Cant find filename to save download: {:?}", download.filename), - } - }); - } - None => { - error!( - "Can't find download with token {:?}", - token - ); - } - }; - } - ClientOperation::NewPeer(new_peer) => { - let peer_exists = client_context - .read() - .unwrap() - .peer_registry - .as_ref() - .map(|r| r.contains(&new_peer.username)) - .unwrap_or(false); - - if peer_exists { - debug!( - "Already connected to {}", - new_peer.username - ); - } else if let Some(server_sender) = - &client_context.read().unwrap().server_sender - { - server_sender - .send(ServerMessage::GetPeerAddress( - new_peer.username.clone(), - )) - .unwrap(); - } - - let addr = new_peer.tcp_stream.peer_addr().unwrap(); - let host = addr.ip().to_string(); - let port: u32 = addr.port().into(); - - let peer = Peer { - username: new_peer.username.clone(), - connection_type: new_peer.connection_type, - host, - port, - token: Some(new_peer.token), - privileged: None, - obfuscated_port: None, - unknown: None, - }; - - Self::connect_to_peer( - peer, - client_context.clone(), - own_username.clone(), - Some(new_peer.tcp_stream), - ); - } - ClientOperation::GetPeerAddressResponse { - username, - host, - port, - obfuscation_type, - obfuscated_port, - } => { - debug!( - "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", - username, host, port, obfuscation_type, obfuscated_port - ); - - let peer_exists = client_context - .read() - .unwrap() - .peer_registry - .as_ref() - .map(|r| r.contains(&username)) - .unwrap_or(false); - - 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, - // ); - } else { - let peer = Peer::new( - username, - ConnectionType::P, - host, - port, - None, - 0, - obfuscation_type.try_into().unwrap(), - obfuscated_port.try_into().unwrap(), - ); + thread::spawn(move || { + loop { + match reader.recv() { + Ok(operation) => { + match operation { + ClientOperation::ConnectToPeer(peer) => { let client_context_clone = client_context.clone(); let own_username_clone = own_username.clone(); @@ -702,62 +491,319 @@ impl Client { ); }); } - } - ClientOperation::UpdateDownloadTokens( - transfer, - username, - ) => { - let mut context = client_context.write().unwrap(); - - let download_to_update = - context.get_downloads().iter().find_map(|d| { - if d.username == username - && d.filename == transfer.filename - { - Some((d.token, d.clone())) - } else { - None - } - }); - - if let Some((old_token, download)) = - download_to_update - { + ClientOperation::SearchResult(search_result) => { trace!( - "[client] UpdateDownloadTokens found {old_token}, transfer: {:?}", - transfer - ); + "[client] SearchResult {:?}", + search_result + ); + let mut context = + client_context.write().unwrap(); + let result_token = search_result.token; - context.add_download(Download { - username: username.clone(), - filename: transfer.filename, - token: transfer.token, - size: transfer.size, - download_directory: download - .download_directory, - status: download.status.clone(), - sender: download.sender.clone(), - }); - context.remove_download(old_token); + // Find the search with matching token + for search in context.searches.values_mut() { + if search.token == result_token { + search.results.push(search_result); + break; + } + } + } + ClientOperation::PeerDisconnected( + username, + error, + ) => { + let context = client_context.read().unwrap(); + if let Some(ref registry) = + context.peer_registry + && let Some(handle) = + registry.remove_peer(&username) + { + let _ = handle.stop(); + } + if let Some(error) = error { + warn!( + "[client] Peer {} disconnected with error: {:?}", + username, error + ); + Self::process_failed_uploads( + client_context.clone(), + &username, + None, + ); + } + } + ClientOperation::PierceFireWall(peer) => { + Self::pierce_firewall( + peer, + client_context.clone(), + own_username.clone(), + ); + } + ClientOperation::DownloadFromPeer( + token, + peer, + allowed, + ) => { + let maybe_download = { + let client_context = + client_context.read().unwrap(); + client_context + .get_download_by_token(token) + .cloned() + }; + let own_username = own_username.clone(); + let client_context_clone = + client_context.clone(); + + trace!( + "[client] DownloadFromPeer token: {} peer: {:?}", + token, peer + ); + match maybe_download { + Some(download) => { + thread::spawn(move || { + let download_peer = + DownloadPeer::new( + download.username.clone(), + peer.host.clone(), + peer.port, + token, + allowed, + own_username, + ); + let filename: Option<&str> = + download + .filename + .split('\\') + .next_back(); + match filename { + Some(filename) => { + match download_peer + .download_file( + client_context_clone + .clone(), + Some(download.clone()), + None, + ) { + Ok(( + download, + filename, + )) => { + let _ = download.sender.send(DownloadStatus::Completed); + client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Completed); + info!( + "Successfully downloaded {} bytes to {}", + download.size, + filename + ); + } + Err(e) => { + let _ = download.sender.send(DownloadStatus::Failed); + client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Failed); + error!( + "Failed to download file '{}' from {}:{} (token: {}) - Error: {}", + filename, + peer.host, + peer.port, + download.token, + e + ); + } + } + } + None => error!( + "Cant find filename to save download: {:?}", + download.filename + ), + } + }); + } + None => { + error!( + "Can't find download with token {:?}", + token + ); + } + }; + } + ClientOperation::NewPeer(new_peer) => { + let peer_exists = client_context + .read() + .unwrap() + .peer_registry + .as_ref() + .map(|r| r.contains(&new_peer.username)) + .unwrap_or(false); + + if peer_exists { + debug!( + "Already connected to {}", + new_peer.username + ); + } else if let Some(server_sender) = + &client_context + .read() + .unwrap() + .server_sender + { + server_sender + .send(ServerMessage::GetPeerAddress( + new_peer.username.clone(), + )) + .unwrap(); + } + + let addr = + new_peer.tcp_stream.peer_addr().unwrap(); + let host = addr.ip().to_string(); + let port: u32 = addr.port().into(); + + let peer = Peer { + username: new_peer.username.clone(), + connection_type: new_peer.connection_type, + host, + port, + token: Some(new_peer.token), + privileged: None, + obfuscated_port: None, + unknown: None, + }; + + Self::connect_to_peer( + peer, + client_context.clone(), + own_username.clone(), + Some(new_peer.tcp_stream), + ); + } + ClientOperation::GetPeerAddressResponse { + username, + host, + port, + obfuscation_type, + obfuscated_port, + } => { + debug!( + "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", + username, + host, + port, + obfuscation_type, + obfuscated_port + ); + + let peer_exists = client_context + .read() + .unwrap() + .peer_registry + .as_ref() + .map(|r| r.contains(&username)) + .unwrap_or(false); + + 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, + // ); + } else { + let peer = Peer::new( + username, + ConnectionType::P, + host, + port, + None, + 0, + obfuscation_type.try_into().unwrap(), + obfuscated_port.try_into().unwrap(), + ); + let client_context_clone = + client_context.clone(); + let own_username_clone = + own_username.clone(); + + thread::spawn(move || { + Self::connect_to_peer( + peer, + client_context_clone, + own_username_clone, + None, + ); + }); + } + } + ClientOperation::UpdateDownloadTokens( + transfer, + username, + ) => { + let mut context = + client_context.write().unwrap(); + + let download_to_update = context + .get_downloads() + .iter() + .find_map(|d| { + if d.username == username + && d.filename == transfer.filename + { + Some((d.token, d.clone())) + } else { + None + } + }); + + if let Some((old_token, download)) = + download_to_update + { + trace!( + "[client] UpdateDownloadTokens found {old_token}, transfer: {:?}", + transfer + ); + + context.add_download(Download { + username: username.clone(), + filename: transfer.filename, + token: transfer.token, + size: transfer.size, + download_directory: download + .download_directory, + status: download.status.clone(), + sender: download.sender.clone(), + }); + context.remove_download(old_token); + } + } + ClientOperation::UploadFailed( + username, + filename, + ) => { + Self::process_failed_uploads( + client_context.clone(), + &username, + Some(&filename), + ); + } + ClientOperation::SetServerSender(sender) => { + client_context.write().unwrap().server_sender = + Some(sender); + debug!("[client] Server sender initialized"); } } - ClientOperation::UploadFailed(username, filename) => { - Self::process_failed_uploads( - client_context.clone(), - &username, - Some(&filename), - ); - } - ClientOperation::SetServerSender(sender) => { - client_context.write().unwrap().server_sender = - Some(sender); - debug!("[client] Server sender initialized"); - } } - } - Err(e) => { - error!("[client] Channel recv error: {:?}", e); - break; + Err(e) => { + error!("[client] Channel recv error: {:?}", e); + break; + } } } }); @@ -772,7 +818,10 @@ impl Client { let client_context = client_context.clone(); let peer_clone = peer.clone(); - trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); + trace!( + "[client] connecting to {}, with connection_type: {}, and token {:?}", + peer.username, peer.connection_type, peer.token + ); match peer.connection_type { ConnectionType::P => { let username = peer.username.clone(); @@ -784,8 +833,7 @@ impl Client { Err(e) => { trace!( "Failed to spawn peer actor for {:?}: {:?}", - username, - e + username, e ); } } @@ -797,8 +845,7 @@ impl Client { ConnectionType::F => { trace!( "[client] downloading from: {}, {:?}", - peer.username, - peer.token + peer.username, peer.token ); let download_peer = DownloadPeer::new( peer.username, @@ -817,8 +864,7 @@ impl Client { Ok((download, filename)) => { trace!( "[client] downloaded {} bytes {:?} ", - filename, - download.size + filename, download.size ); let _ = download.sender.send(DownloadStatus::Completed); client_context diff --git a/soulseek-rs-lib/src/dispatcher.rs b/soulseek-rs-lib/src/dispatcher.rs index a362bbb..d1b1c4f 100644 --- a/soulseek-rs-lib/src/dispatcher.rs +++ b/soulseek-rs-lib/src/dispatcher.rs @@ -1,4 +1,4 @@ -use crate::message::{handlers::Handlers, Message}; +use crate::message::{Message, handlers::Handlers}; use std::sync::mpsc::Sender; use crate::warn; diff --git a/soulseek-rs-lib/src/message/peer/get_share_file_list.rs b/soulseek-rs-lib/src/message/peer/get_share_file_list.rs index 683b94f..9f04870 100644 --- a/soulseek-rs-lib/src/message/peer/get_share_file_list.rs +++ b/soulseek-rs-lib/src/message/peer/get_share_file_list.rs @@ -1,5 +1,5 @@ use crate::{ - message::{server::MessageFactory, Message, MessageHandler}, + message::{Message, MessageHandler, server::MessageFactory}, peer::PeerMessage, }; use std::sync::mpsc::Sender; diff --git a/soulseek-rs-lib/src/message/peer/peer_init.rs b/soulseek-rs-lib/src/message/peer/peer_init.rs index ea1ff0a..1071150 100644 --- a/soulseek-rs-lib/src/message/peer/peer_init.rs +++ b/soulseek-rs-lib/src/message/peer/peer_init.rs @@ -20,9 +20,7 @@ impl MessageHandler for PeerInit { let token = message.read_int32(); trace!( "PeerInit: username: {}, connection_type: {}, token: {}", - username, - connection_type, - token + username, connection_type, token ); sender.send(PeerMessage::SetUsername(username)).unwrap(); diff --git a/soulseek-rs-lib/src/message/server/file_search.rs b/soulseek-rs-lib/src/message/server/file_search.rs index d64b922..5c1ebea 100644 --- a/soulseek-rs-lib/src/message/server/file_search.rs +++ b/soulseek-rs-lib/src/message/server/file_search.rs @@ -2,8 +2,8 @@ use crate::{debug, info}; use std::sync::mpsc::Sender; use crate::{ - actor::server_actor::ServerMessage, message::handlers::MessageHandler, - message::Message, + actor::server_actor::ServerMessage, message::Message, + message::handlers::MessageHandler, }; pub struct FileSearchHandler; diff --git a/soulseek-rs-lib/src/peer/download_peer.rs b/soulseek-rs-lib/src/peer/download_peer.rs index cd11dbc..914670e 100644 --- a/soulseek-rs-lib/src/peer/download_peer.rs +++ b/soulseek-rs-lib/src/peer/download_peer.rs @@ -195,8 +195,7 @@ impl DownloadPeer { ) -> Result<(), DownloadError> { trace!( "[download_peer:{}] performing handshake no_pierce: {}", - self.username, - self.no_pierce + self.username, self.no_pierce ); if !self.no_pierce { @@ -231,8 +230,7 @@ impl DownloadPeer { trace!( "[download_peer:{}] got token: {} from data chunk", - self.username, - token_u32 + self.username, token_u32 ); stream @@ -277,8 +275,7 @@ impl DownloadPeer { Ok(0) => { trace!( "[download_peer:{}] connection closed by peer. bytes read: {}", - self.username, - processor.total_bytes + self.username, processor.total_bytes ); break; } @@ -294,9 +291,7 @@ impl DownloadPeer { )?; trace!( "[download_peer:{}] got download info for token: {} - filename: {}", - self.username, - self.token, - new_download.filename + self.username, self.token, new_download.filename ); download = Some(new_download); processor.received = true; @@ -307,30 +302,30 @@ impl DownloadPeer { chunk_counter += 1; if let Some(ref dl) = download - && chunk_counter % PROGRESS_UPDATE_CHUNKS == 0 { - let elapsed = - last_update_time.elapsed().as_secs_f64(); - let bytes_since_last_update = - PROGRESS_UPDATE_CHUNKS * READ_BUFFER_SIZE; - let speed = if elapsed > 0.0 { - bytes_since_last_update as f64 / elapsed - } else { - 0.0 - }; + && chunk_counter % PROGRESS_UPDATE_CHUNKS == 0 + { + let elapsed = last_update_time.elapsed().as_secs_f64(); + let bytes_since_last_update = + PROGRESS_UPDATE_CHUNKS * READ_BUFFER_SIZE; + let speed = if elapsed > 0.0 { + bytes_since_last_update as f64 / elapsed + } else { + 0.0 + }; - let status = DownloadStatus::InProgress { - bytes_downloaded: processor.total_bytes as u64, - total_bytes: dl.size, - speed_bytes_per_sec: speed, - }; - let _ = dl.sender.send(status.clone()); - client_context - .write() - .unwrap() - .update_download_with_status(dl.token, status); + let status = DownloadStatus::InProgress { + bytes_downloaded: processor.total_bytes as u64, + total_bytes: dl.size, + speed_bytes_per_sec: speed, + }; + let _ = dl.sender.send(status.clone()); + client_context + .write() + .unwrap() + .update_download_with_status(dl.token, status); - last_update_time = Instant::now(); - } + last_update_time = Instant::now(); + } let expected_size = download .as_ref() @@ -486,7 +481,9 @@ mod tests { "file.mp3" ); assert_eq!( - FileManager::extract_filename_from_path("@@bhfrv\\Soulseek Downloads\\complete\\Beatport Top Deep House (2021)\\michel test file.mp3"), + FileManager::extract_filename_from_path( + "@@bhfrv\\Soulseek Downloads\\complete\\Beatport Top Deep House (2021)\\michel test file.mp3" + ), "michel test file.mp3" ); assert_eq!( diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 36cd4ef..692a260 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -9,7 +9,7 @@ use crate::client::{ClientContext, ClientOperation}; use crate::message::{Message, MessageReader}; use crate::peer::{ConnectionType, DownloadPeer, Peer}; use crate::types::Download; -use crate::{debug, error, info, trace, DownloadStatus}; +use crate::{DownloadStatus, debug, error, info, trace}; const PEER_INIT_MESSAGE_CODE: u8 = 1; @@ -79,8 +79,7 @@ fn extract_download_from_buffer( let token = parse_token_from_buffer(&buffer, username)?; trace!( "[listener:{}] got transfer_token: {} from data chunk", - username, - token + username, token ); let context = client_context.read().unwrap(); @@ -88,7 +87,10 @@ fn extract_download_from_buffer( if download.is_none() { let download_tokens = context.get_download_tokens(); - trace!("[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}", token, download_tokens); + trace!( + "[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}", + token, download_tokens + ); } download @@ -129,8 +131,7 @@ fn handle_file_connection( ) { trace!( "[client] DownloadFromPeer token: {} peer: {:?}", - token, - peer + token, peer ); let download = extract_download_from_buffer( @@ -198,7 +199,9 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { }; let Some(init_data) = parse_peer_init_message(message) else { - error!("[listener:{peer_ip}:{peer_port}] Invalid or unknown peer init message"); + error!( + "[listener:{peer_ip}:{peer_port}] Invalid or unknown peer init message" + ); return; }; @@ -241,7 +244,9 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { }); } ConnectionType::D => { - debug!("[listener:{peer_ip}:{peer_port}] connection type is D, not supported yet, closing connection. "); + debug!( + "[listener:{peer_ip}:{peer_port}] connection type is D, not supported yet, closing connection. " + ); } } } diff --git a/soulseek-rs-lib/src/utils/logger.rs b/soulseek-rs-lib/src/utils/logger.rs index a0feae6..8b362b0 100644 --- a/soulseek-rs-lib/src/utils/logger.rs +++ b/soulseek-rs-lib/src/utils/logger.rs @@ -2,8 +2,8 @@ use std::env; use std::fs::{File, OpenOptions}; use std::io::Write; use std::sync::{ - atomic::{AtomicBool, Ordering}, Mutex, Once, + atomic::{AtomicBool, Ordering}, }; #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] @@ -168,10 +168,11 @@ pub fn log(level: LogLevel, message: &str) { } // But ALSO write to file if configured (file logging bypasses buffering) if let Ok(mut log_file) = LOG_FILE.lock() - && let Some(file) = log_file.as_mut() { - let _ = writeln!(file, "{}", formatted_message_plain); - let _ = file.flush(); - } + && let Some(file) = log_file.as_mut() + { + let _ = writeln!(file, "{}", formatted_message_plain); + let _ = file.flush(); + } } else { // Write to file if configured, otherwise to stderr if let Ok(mut log_file) = LOG_FILE.lock() { diff --git a/soulseek-rs-lib/src/utils/md5.rs b/soulseek-rs-lib/src/utils/md5.rs index 76b8318..ba77ef7 100644 --- a/soulseek-rs-lib/src/utils/md5.rs +++ b/soulseek-rs-lib/src/utils/md5.rs @@ -369,7 +369,7 @@ fn bit_padding(input: &str) -> Vec { //check if bit length % 512 is 448 (64 less than 512) while (input_vector.len() * 8) % 512 != 448 { input_vector.push(0_u8); // push in another 8-bit 0 padded value until the correct - // result is reached; + // result is reached; } let length_bits_as_u8_array = split_u64_to_u8_array(bit_length); @@ -467,8 +467,7 @@ fn all_alphanumeric_correct_hash() { #[test] fn repeated_numbers_correct_hash() { - let repeated_number_str = - "12345678901234567890123456789012345678901234567890123456789012345678901234567890"; + let repeated_number_str = "12345678901234567890123456789012345678901234567890123456789012345678901234567890"; assert_eq!("57edf4a22be3c955ac49da2e2107b67a", md5(repeated_number_str)); } diff --git a/soulseek-rs-lib/src/utils/thread_pool.rs b/soulseek-rs-lib/src/utils/thread_pool.rs index 0fbf13c..1515898 100644 --- a/soulseek-rs-lib/src/utils/thread_pool.rs +++ b/soulseek-rs-lib/src/utils/thread_pool.rs @@ -1,5 +1,5 @@ use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{mpsc, Arc, Mutex}; +use std::sync::{Arc, Mutex, mpsc}; use std::thread; type Job = Box; @@ -89,37 +89,39 @@ impl Worker { _active_threads: Arc, _total_threads: usize, ) -> Worker { - let thread = thread::spawn(move || loop { - let message = receiver.lock().unwrap().recv(); - 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 - // ); + let thread = thread::spawn(move || { + loop { + let message = receiver.lock().unwrap().recv(); + 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(); + 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; + // 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; + } } } }); diff --git a/soulseek-rs/src/main.rs b/soulseek-rs/src/main.rs index 4c030ae..6bbcd32 100644 --- a/soulseek-rs/src/main.rs +++ b/soulseek-rs/src/main.rs @@ -4,16 +4,16 @@ mod models; mod ui; use clap::Parser; -use cli::{parse_server_address, Cli, Commands}; +use cli::{Cli, Commands, parse_server_address}; use color_eyre::Result; use config::SearchConfig; use soulseek_rs::{Client, ClientSettings, PeerAddress}; use std::{ env, - sync::{atomic::AtomicBool, Arc}, + sync::{Arc, atomic::AtomicBool}, time::Duration, }; -use ui::{launch_main_tui, show_multi_download_progress, FileSelector}; +use ui::{FileSelector, launch_main_tui, show_multi_download_progress}; fn main() -> Result<()> { color_eyre::install()?; @@ -33,14 +33,22 @@ fn main() -> Result<()> { // Set LOG_FILE env var if provided via CLI if let Some(log_file) = &cli.log_file { // SAFETY: Called before any threads are spawned - unsafe { env::set_var("LOG_FILE", log_file.to_string_lossy().to_string()) }; + unsafe { + env::set_var("LOG_FILE", log_file.to_string_lossy().to_string()) + }; } - let username = cli.username - .ok_or_else(|| color_eyre::eyre::eyre!("Username required: use --username or set SOULSEEK_USERNAME env var"))?; + let username = cli.username.ok_or_else(|| { + color_eyre::eyre::eyre!( + "Username required: use --username or set SOULSEEK_USERNAME env var" + ) + })?; - let password = cli.password - .ok_or_else(|| color_eyre::eyre::eyre!("Password required: use --password or set SOULSEEK_PASSWORD env var"))?; + let password = cli.password.ok_or_else(|| { + color_eyre::eyre::eyre!( + "Password required: use --password or set SOULSEEK_PASSWORD env var" + ) + })?; let (server_host, server_port) = parse_server_address(&cli.server)?; diff --git a/soulseek-rs/src/models/app_state.rs b/soulseek-rs/src/models/app_state.rs index ca0c880..2547f26 100644 --- a/soulseek-rs/src/models/app_state.rs +++ b/soulseek-rs/src/models/app_state.rs @@ -1,8 +1,8 @@ use crate::models::FileDisplayData; use ratatui::{layout::Rect, widgets::TableState}; -use soulseek_rs::{types::Download, DownloadStatus}; +use soulseek_rs::{DownloadStatus, types::Download}; use std::sync::atomic::AtomicBool; -use std::sync::{mpsc::Receiver, mpsc::Sender, Arc}; +use std::sync::{Arc, mpsc::Receiver, mpsc::Sender}; use std::time::Instant; #[derive(Debug, Clone, PartialEq)] diff --git a/soulseek-rs/src/ui/download_selector.rs b/soulseek-rs/src/ui/download_selector.rs index e60bede..b70bb02 100644 --- a/soulseek-rs/src/ui/download_selector.rs +++ b/soulseek-rs/src/ui/download_selector.rs @@ -1,27 +1,27 @@ use crate::models::FileDisplayData; use crate::ui::{ - border_style, border_type, format_bytes, format_shortcuts_styled, - get_bitrate, get_spinner_char, header_style, highlight_style, - primary_style, success_style, warning_style, HIGHLIGHT_SYMBOL, + HIGHLIGHT_SYMBOL, border_style, border_type, format_bytes, + format_shortcuts_styled, get_bitrate, get_spinner_char, header_style, + highlight_style, primary_style, success_style, warning_style, }; use color_eyre::Result; use ratatui::text::{Line, Span}; use ratatui::{ - crossterm::event::{self, poll, Event, KeyCode, KeyEvent, KeyEventKind}, + DefaultTerminal, Frame, + crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind, poll}, layout::{Alignment, Constraint, Layout}, style::Style, widgets::{ Block, Borders, Cell, HighlightSpacing, Paragraph, Row, StatefulWidget, Table, TableState, Wrap, }, - DefaultTerminal, Frame, }; use soulseek_rs::Client; use std::{ collections::HashSet, sync::{ - atomic::{AtomicBool, Ordering}, Arc, + atomic::{AtomicBool, Ordering}, }, time::{Duration, Instant}, }; @@ -114,9 +114,10 @@ impl FileSelector { }; if poll(timeout)? - && let Event::Key(key) = event::read()? { - self.handle_key(key); - } + && let Event::Key(key) = event::read()? + { + self.handle_key(key); + } } Ok((terminal, self.selected_indices.iter().copied().collect())) @@ -158,13 +159,13 @@ impl FileSelector { fn toggle_selection(&mut self) { if let Some(filtered_idx) = self.state.selected() && let Some(&original_idx) = self.filtered_indices.get(filtered_idx) - { - if self.selected_indices.contains(&original_idx) { - self.selected_indices.remove(&original_idx); - } else { - self.selected_indices.insert(original_idx); - } + { + if self.selected_indices.contains(&original_idx) { + self.selected_indices.remove(&original_idx); + } else { + self.selected_indices.insert(original_idx); } + } } fn handle_key(&mut self, key: KeyEvent) { @@ -194,11 +195,11 @@ impl FileSelector { // If nothing selected, select the current item under cursor if self.selected_indices.is_empty() && let Some(filtered_idx) = self.state.selected() - && let Some(&original_idx) = - self.filtered_indices.get(filtered_idx) - { - self.selected_indices.insert(original_idx); - } + && let Some(&original_idx) = + self.filtered_indices.get(filtered_idx) + { + self.selected_indices.insert(original_idx); + } self.should_exit = true; } KeyCode::Up | KeyCode::Down => { @@ -236,11 +237,11 @@ impl FileSelector { // If nothing selected, select the current item under cursor if self.selected_indices.is_empty() && let Some(filtered_idx) = self.state.selected() - && let Some(&original_idx) = - self.filtered_indices.get(filtered_idx) - { - self.selected_indices.insert(original_idx); - } + && let Some(&original_idx) = + self.filtered_indices.get(filtered_idx) + { + self.selected_indices.insert(original_idx); + } self.should_exit = true; } KeyCode::Esc | KeyCode::Char('q') => { diff --git a/soulseek-rs/src/ui/downloads.rs b/soulseek-rs/src/ui/downloads.rs index fa1d80f..27078b8 100644 --- a/soulseek-rs/src/ui/downloads.rs +++ b/soulseek-rs/src/ui/downloads.rs @@ -1,13 +1,14 @@ use crate::models::DownloadEntry; use crate::ui::{ - border_style, border_type, error_style, format_bytes_progress, - format_progress_bar, format_shortcuts_styled, format_speed, header_style, - highlight_style, inactive_style, primary_style, warning_style, - COLOR_PRIMARY, HIGHLIGHT_SYMBOL, + COLOR_PRIMARY, HIGHLIGHT_SYMBOL, border_style, border_type, error_style, + format_bytes_progress, format_progress_bar, format_shortcuts_styled, + format_speed, header_style, highlight_style, inactive_style, primary_style, + warning_style, }; use color_eyre::Result; use ratatui::{ - crossterm::event::{self, poll, Event, KeyCode, KeyEvent, KeyEventKind}, + DefaultTerminal, Frame, + crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind, poll}, layout::{Alignment, Constraint, Layout}, style::{Modifier, Style}, text::{Line, Span}, @@ -15,11 +16,10 @@ use ratatui::{ Block, Borders, Cell, HighlightSpacing, Paragraph, Row, StatefulWidget, Table, TableState, }, - DefaultTerminal, Frame, }; use soulseek_rs::{Client, DownloadStatus}; use std::{ - sync::{mpsc, mpsc::Receiver, Arc}, + sync::{Arc, mpsc, mpsc::Receiver}, thread, time::Duration, }; @@ -82,18 +82,18 @@ impl MultiDownloadProgress { let mut need_start_next = false; for download_entry in &mut self.downloads { if let Some(receiver) = &download_entry.receiver - && let Ok(status) = receiver.try_recv() { - let was_active = !download_entry.download.is_finished(); - download_entry.download.status = status; - let is_finished = download_entry.download.is_finished(); + && let Ok(status) = receiver.try_recv() + { + let was_active = !download_entry.download.is_finished(); + download_entry.download.status = status; + let is_finished = download_entry.download.is_finished(); - // If download just finished, decrement active count - if was_active && is_finished { - self.active_count = - self.active_count.saturating_sub(1); - need_start_next = true; - } + // If download just finished, decrement active count + if was_active && is_finished { + self.active_count = self.active_count.saturating_sub(1); + need_start_next = true; } + } } // Start next batch after iteration completes @@ -103,9 +103,10 @@ impl MultiDownloadProgress { // Handle keyboard input if poll(Duration::from_millis(100))? - && let Event::Key(key) = event::read()? { - self.handle_key(key); - } + && let Event::Key(key) = event::read()? + { + self.handle_key(key); + } // Exit when user cancels or all downloads finished if self.should_exit || self.all_finished() { diff --git a/soulseek-rs/src/ui/main_tui.rs b/soulseek-rs/src/ui/main_tui.rs index 74b2e0d..b7aa409 100644 --- a/soulseek-rs/src/ui/main_tui.rs +++ b/soulseek-rs/src/ui/main_tui.rs @@ -3,28 +3,28 @@ use crate::models::{ SearchStatus, }; use crate::ui::panes::{ - render_downloads_pane, render_results_pane, render_searches_pane, - ResultsPaneParams, + ResultsPaneParams, render_downloads_pane, render_results_pane, + render_searches_pane, }; use crate::ui::{ border_style, border_type, format_shortcuts_styled, render_download_stats, }; use color_eyre::Result; use ratatui::{ + DefaultTerminal, Frame, crossterm::event::{ - self, poll, Event, KeyCode, KeyEvent, KeyEventKind, MouseButton, - MouseEvent, MouseEventKind, + self, Event, KeyCode, KeyEvent, KeyEventKind, MouseButton, MouseEvent, + MouseEventKind, poll, }, layout::{Constraint, Layout}, widgets::{Block, Borders, Paragraph}, - DefaultTerminal, Frame, }; use soulseek_rs::{Client, DownloadStatus}; use std::{ sync::{ + Arc, atomic::AtomicBool, mpsc::{self}, - Arc, }, thread, time::{Duration, Instant}, @@ -521,35 +521,35 @@ impl MainTui { // Check if click is within searches pane if let Some(area) = self.state.searches_pane_area && col >= area.x - && col < area.x + area.width - && row >= area.y - && row < area.y + area.height - { - self.state.focused_pane = FocusedPane::Searches; - return Ok(()); - } + && col < area.x + area.width + && row >= area.y + && row < area.y + area.height + { + self.state.focused_pane = FocusedPane::Searches; + return Ok(()); + } // Check if click is within results pane if let Some(area) = self.state.results_pane_area && col >= area.x - && col < area.x + area.width - && row >= area.y - && row < area.y + area.height - { - self.state.focused_pane = FocusedPane::Results; - return Ok(()); - } + && col < area.x + area.width + && row >= area.y + && row < area.y + area.height + { + self.state.focused_pane = FocusedPane::Results; + return Ok(()); + } // Check if click is within downloads pane if let Some(area) = self.state.downloads_pane_area && col >= area.x - && col < area.x + area.width - && row >= area.y - && row < area.y + area.height - { - self.state.focused_pane = FocusedPane::Downloads; - return Ok(()); - } + && col < area.x + area.width + && row >= area.y + && row < area.y + area.height + { + self.state.focused_pane = FocusedPane::Downloads; + return Ok(()); + } } Ok(()) @@ -747,13 +747,14 @@ impl MainTui { // Update selected search if this is the active one if let Some(selected_idx) = selected_search_index - && selected_idx == idx { - self.state.results_items = search.results.clone(); - self.state.results_filtered_items = - search.results.clone(); - self.state.results_filtered_indices = - (0..search.results.len()).collect(); - } + && selected_idx == idx + { + self.state.results_items = search.results.clone(); + self.state.results_filtered_items = + search.results.clone(); + self.state.results_filtered_indices = + (0..search.results.len()).collect(); + } } // Mark as completed after timeout diff --git a/soulseek-rs/src/ui/panes/downloads_pane.rs b/soulseek-rs/src/ui/panes/downloads_pane.rs index 1ee676f..db1b074 100644 --- a/soulseek-rs/src/ui/panes/downloads_pane.rs +++ b/soulseek-rs/src/ui/panes/downloads_pane.rs @@ -1,16 +1,16 @@ use crate::models::DownloadEntry; use crate::ui::{ - border_style, border_type, error_style, format_bytes, format_speed, - header_style, highlight_style, inactive_style, success_style, - warning_style, HIGHLIGHT_SYMBOL, + HIGHLIGHT_SYMBOL, border_style, border_type, error_style, format_bytes, + format_speed, header_style, highlight_style, inactive_style, success_style, + warning_style, }; use ratatui::{ + Frame, layout::Rect, widgets::{ Block, Borders, Cell, HighlightSpacing, Paragraph, Row, Table, TableState, }, - Frame, }; use soulseek_rs::DownloadStatus; diff --git a/soulseek-rs/src/ui/panes/mod.rs b/soulseek-rs/src/ui/panes/mod.rs index 61521e0..0b81308 100644 --- a/soulseek-rs/src/ui/panes/mod.rs +++ b/soulseek-rs/src/ui/panes/mod.rs @@ -3,5 +3,5 @@ mod results_pane; mod searches_pane; pub use downloads_pane::render_downloads_pane; -pub use results_pane::{render_results_pane, ResultsPaneParams}; +pub use results_pane::{ResultsPaneParams, render_results_pane}; pub use searches_pane::render_searches_pane; diff --git a/soulseek-rs/src/ui/panes/results_pane.rs b/soulseek-rs/src/ui/panes/results_pane.rs index 9095f4b..1488fd7 100644 --- a/soulseek-rs/src/ui/panes/results_pane.rs +++ b/soulseek-rs/src/ui/panes/results_pane.rs @@ -1,15 +1,15 @@ use crate::models::FileDisplayData; use crate::ui::{ - border_style, border_type, format_bytes, header_style, highlight_style, - HIGHLIGHT_SYMBOL, + HIGHLIGHT_SYMBOL, border_style, border_type, format_bytes, header_style, + highlight_style, }; use ratatui::{ + Frame, layout::Rect, widgets::{ Block, Borders, Cell, HighlightSpacing, Paragraph, Row, Table, TableState, }, - Frame, }; use std::collections::HashSet; diff --git a/soulseek-rs/src/ui/panes/searches_pane.rs b/soulseek-rs/src/ui/panes/searches_pane.rs index aa45997..de24644 100644 --- a/soulseek-rs/src/ui/panes/searches_pane.rs +++ b/soulseek-rs/src/ui/panes/searches_pane.rs @@ -1,12 +1,12 @@ use crate::models::{SearchEntry, SearchStatus}; use crate::ui::{ - border_style, border_type, header_style, highlight_style, success_style, - warning_style, HIGHLIGHT_SYMBOL, + HIGHLIGHT_SYMBOL, border_style, border_type, header_style, highlight_style, + success_style, warning_style, }; use ratatui::{ + Frame, layout::Rect, widgets::{Block, Borders, Cell, HighlightSpacing, Row, Table, TableState}, - Frame, }; pub fn render_searches_pane(