use crate::actor::server_actor::{PeerAddress, ServerActor, ServerMessage}; use crate::actor::ActorHandle; use crate::types::DownloadStatus; use crate::utils::logger; use crate::{ actor::{peer_registry::PeerRegistry, ActorSystem}, error::{Result, SoulseekRs}, peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer}, types::{Download, Search, SearchResult}, utils::{md5, thread_pool::ThreadPool}, Transfer, }; use std::{ collections::HashMap, net::TcpStream, sync::{ atomic::{AtomicBool, Ordering}, mpsc::{Receiver, Sender}, RwLock, }, thread::{self, sleep}, }; use std::{ sync::{mpsc, Arc}, time::{Duration, Instant}, }; use crate::{debug, error, info, trace, warn}; const DEFALT_LISTEN_PORT: u32 = 2234; #[derive(Debug, Clone)] pub struct ClientSettings { pub username: String, pub password: String, pub server_address: PeerAddress, pub enable_listen: bool, pub listen_port: u32, } impl ClientSettings { pub fn new( username: impl Into, password: impl Into, ) -> Self { Self { username: username.into(), password: password.into(), ..Default::default() } } } impl Default for ClientSettings { fn default() -> Self { Self { username: String::new(), password: String::new(), server_address: PeerAddress::new( "server.slsknet.org".to_string(), 2416, ), enable_listen: true, listen_port: DEFALT_LISTEN_PORT, } } } #[derive(Debug)] pub enum ClientOperation { NewPeer(NewPeer), ConnectToPeer(Peer), SearchResult(SearchResult), PeerDisconnected(String, Option), PierceFireWall(Peer), DownloadFromPeer(u32, Peer, bool), UpdateDownloadTokens(Transfer, String), GetPeerAddressResponse { username: String, host: String, port: u32, obfuscation_type: u32, obfuscated_port: u16, }, UploadFailed(String, String), SetServerSender(Sender), } pub struct ClientContext { pub peer_registry: Option, sender: Option>, server_sender: Option>, searches: HashMap, downloads: Vec, actor_system: Arc, } impl Default for ClientContext { fn default() -> Self { Self::new() } } impl ClientContext { pub fn add_download(&mut self, download: Download) { self.downloads.push(download); } pub fn remove_download(&mut self, token: u32) { self.downloads.retain(|d| d.token != token); } pub fn get_download_by_token(&self, token: u32) -> Option<&Download> { self.downloads.iter().find(|d| d.token == token) } pub fn get_download_by_token_mut( &mut self, token: u32, ) -> Option<&mut Download> { self.downloads.iter_mut().find(|d| d.token == token) } pub fn get_download_tokens(&self) -> Vec { self.downloads.iter().map(|d| d.token).collect() } pub fn get_downloads(&self) -> &Vec { &self.downloads } pub fn update_download_with_status( &mut self, token: u32, status: DownloadStatus, ) { if let Some(download) = self.get_download_by_token_mut(token) { download.status = status; } } } #[test] fn test_client_context_downloads() { let mut context = ClientContext::new(); let token = 123; let new_token = 1234; let download = Download { username: "test".to_string(), filename: "test.txt".to_string(), token, size: 100, download_directory: "test".to_string(), status: DownloadStatus::Queued, sender: mpsc::channel().0, }; context.add_download(download); assert!(context.get_download_by_token(123).is_some()); assert_eq!(context.get_download_tokens(), vec![123]); assert_eq!(context.get_downloads().len(), 1); if let Some(download) = context.get_download_by_token_mut(token) { assert_eq!(download.token, token); download.token = new_token } assert!(context.get_download_by_token(new_token).is_some()); assert_eq!(context.get_download_tokens(), vec![new_token]); context.remove_download(new_token); assert_eq!(context.get_downloads().len(), 0); assert!(context.get_download_by_token(1234).is_none()); } impl ClientContext { #[must_use] pub fn new() -> Self { let max_threads = thread::available_parallelism() .map(|n| n.get()) .unwrap_or(8); let thread_pool = Arc::new(ThreadPool::new(max_threads)); let actor_system = Arc::new(ActorSystem::new(thread_pool.clone())); Self { peer_registry: None, sender: None, server_sender: None, searches: HashMap::new(), downloads: Vec::new(), actor_system, } } } pub struct Client { enable_listen: bool, listen_port: u32, address: PeerAddress, username: String, password: String, server_handle: Option>, context: Arc>, } impl Client { pub fn new( username: impl Into, password: impl Into, ) -> Self { Self::with_settings(ClientSettings::new(username, password)) } pub fn with_settings(settings: ClientSettings) -> Self { logger::init(); Self { enable_listen: settings.enable_listen, listen_port: settings.listen_port, address: settings.server_address, username: settings.username, password: settings.password, context: Arc::new(RwLock::new(ClientContext::new())), server_handle: None, } } pub fn connect(&mut self) { let (sender, message_reader): ( Sender, Receiver, ) = mpsc::channel(); let mut ctx = self.context.write().unwrap(); ctx.sender = Some(sender.clone()); let peer_registry = PeerRegistry::new(ctx.actor_system.clone(), sender.clone()); ctx.peer_registry = Some(peer_registry); let listen_sender = sender.clone(); let server_actor = ServerActor::new( self.address.clone(), sender, self.listen_port, self.enable_listen, ); self.server_handle = Some(ctx.actor_system.spawn_with_handle( server_actor, |actor, handle| { actor.set_self_handle(handle); }, )); if self.enable_listen { let listen_port = self.listen_port; let client_sender = listen_sender; let context = self.context.clone(); let own_username = self.username.clone(); thread::spawn(move || { Listen::start( listen_port, client_sender, context, own_username, ); }); } Self::listen_to_client_operations( message_reader, self.context.clone(), self.username.clone(), ); } pub fn login(&self) -> Result { info!("Logging in as {}", self.username); if let Some(handle) = &self.server_handle { let (tx, rx) = std::sync::mpsc::channel(); let _ = handle.send(ServerMessage::Login { username: self.username.clone(), password: self.password.clone(), response: tx, }); match rx.recv() { Ok(result) => result, Err(_) => Err(SoulseekRs::Timeout), } } else { Err(SoulseekRs::NotConnected) } } #[allow(dead_code)] pub fn remove_peer(&self, username: &str) { let context = self.context.read().unwrap(); if let Some(ref registry) = context.peer_registry { if let Some(handle) = registry.remove_peer(username) { let _ = handle.stop(); } } } pub fn search( &self, query: &str, timeout: Duration, ) -> Result> { self.search_with_cancel(query, timeout, None) } pub fn search_with_cancel( &self, query: &str, timeout: Duration, cancel_flag: Option>, ) -> Result> { info!("Searching for {}", query); if let Some(handle) = &self.server_handle { let hash = md5::md5(query); let token = u32::from_str_radix(&hash[0..5], 16)?; // Initialize new search with query string as key self.context.write().unwrap().searches.insert( query.to_string(), Search { token, results: Vec::new(), }, ); let _ = handle.send(ServerMessage::FileSearch { token, query: query.to_string(), }); } else { return Err(SoulseekRs::NotConnected); } let start = Instant::now(); loop { sleep(Duration::from_millis(100)); // Check if cancelled if let Some(ref flag) = cancel_flag { if flag.load(Ordering::Relaxed) { info!("Search cancelled by user"); break; } } // Check if timeout reached if start.elapsed() >= timeout { break; } } Ok(self.get_search_results(query)) } pub fn get_search_results_count(&self, search_key: &str) -> usize { self.context .read() .unwrap() .searches .get(search_key) .map(|s| s.results.len()) .unwrap_or(0) } pub fn get_search_results(&self, search_key: &str) -> Vec { self.context .read() .unwrap() .searches .get(search_key) .map(|s| s.results.clone()) .unwrap_or_default() } pub fn get_all_searches(&self) -> HashMap { self.context.read().unwrap().searches.clone() } pub fn get_all_downloads(&self) -> Vec { self.context.read().unwrap().get_downloads().clone() } pub fn download( &self, filename: String, username: String, size: u64, download_directory: String, ) -> Result<(Download, Receiver)> { info!("[client] Downloading {} from {}", filename, username); let hash = md5::md5(&filename); let token = u32::from_str_radix(&hash[0..5], 16)?; let (download_sender, download_receiver): ( Sender, Receiver, ) = mpsc::channel(); let download = Download { username: username.clone(), filename: filename.clone(), token, size, download_directory, status: DownloadStatus::Queued, sender: download_sender, }; let mut context = self.context.write().unwrap(); context.add_download(download.clone()); let download_initiated = if let Some(ref registry) = context.peer_registry { registry .queue_upload(&username, download.filename.clone()) .is_ok() } else { false }; drop(context); if !download_initiated { let _ = download.sender.send(DownloadStatus::Failed); self.context .write() .unwrap() .update_download_with_status(token, DownloadStatus::Failed); } Ok((download, download_receiver)) } fn process_failed_uploads( client_context: Arc>, username: &str, filename: Option<&str>, ) { let context = client_context.read().unwrap(); let failed_downloads: Vec<_> = context .get_downloads() .iter() .filter(|download| { download.username == username && (filename.is_none_or(|f| download.filename == *f)) }) .map(|download| { let _ = download.sender.send(DownloadStatus::Failed); download.token }) .collect(); drop(context); failed_downloads.iter().for_each(|token| { let mut context = client_context.write().unwrap(); context.update_download_with_status(*token, DownloadStatus::Failed); context.remove_download(*token); }); } fn listen_to_client_operations( reader: Receiver, 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 { if 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"); } } } Err(e) => { error!("[client] Channel recv error: {:?}", e); break; } } }); } fn connect_to_peer( peer: Peer, client_context: Arc>, own_username: String, stream: Option, ) { 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); match peer.connection_type { ConnectionType::P => { let username = peer.username.clone(); let context = client_context.read().unwrap(); if let Some(ref registry) = context.peer_registry { match registry.register_peer(peer_clone, stream, None) { Ok(_) => (), Err(e) => { trace!( "Failed to spawn peer actor for {:?}: {:?}", username, e ); } } } else { trace!("PeerRegistry not initialized"); } } ConnectionType::F => { trace!( "[client] downloading from: {}, {:?}", peer.username, peer.token ); let download_peer = DownloadPeer::new( peer.username, peer.host, peer.port, peer.token.unwrap(), false, own_username.clone(), ); match download_peer.download_file( client_context.clone(), None, None, ) { Ok((download, filename)) => { trace!( "[client] downloaded {} bytes {:?} ", filename, download.size ); let _ = download.sender.send(DownloadStatus::Completed); client_context .write() .unwrap() .update_download_with_status( download.token, DownloadStatus::Completed, ); } Err(e) => { trace!("[client] failed to download: {}", e); } } } ConnectionType::D => { error!("ConnectionType::D not implemented") } } } fn pierce_firewall( peer: Peer, client_context: Arc>, own_username: String, ) { debug!("Piercing firewall for peer: {:?}", peer); let context = client_context.read().unwrap(); if let Some(server_sender) = &context.server_sender { if let Some(token) = peer.token { match server_sender.send(ServerMessage::PierceFirewall(token)) { Ok(_) => (), Err(e) => { error!("Failed to send PierceFirewall message: {}", e) } } } else { error!("No token available for PierceFirewall"); } } else { error!("No server sender available for PierceFirewall"); } drop(context); Self::connect_to_peer(peer, client_context, own_username, None); } }