diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 37a4cf8..825bb4f 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -184,13 +184,7 @@ impl PeerActor { let messages: Vec = self .dispatcher_receiver .as_ref() - .map_or_else(Vec::new, |receiver| { - let mut msgs = Vec::new(); - while let Ok(msg) = receiver.try_recv() { - msgs.push(msg); - } - msgs - }); + .map_or_else(Vec::new, |receiver| receiver.try_iter().collect()); for msg in &messages { self.handle_message(msg.clone()); @@ -690,45 +684,20 @@ impl PeerActor { } } - fn initiate_connection(&mut self) -> bool { + fn initiate_connection(&mut self) { let (username, host, port) = match self.peer.read_safe() { Ok(peer) => (peer.username.clone(), peer.host.clone(), peer.port), Err(e) => { error!("[peer_actor] initiate_connection peer lock: {}", e); - return false; + return; } }; let socket_addr = format!("{host}:{port}").parse::(); - match socket_addr { - Ok(addr) => { - // Use connect_timeout to prevent blocking the thread for too long - let timeout = Duration::from_secs(5); - match TcpStream::connect_timeout(&addr, timeout) { - Ok(stream) => { - if let Err(e) = stream.set_nonblocking(true) { - error!( - "[peer:{}] Failed to set non-blocking: {}", - username, 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 - } - } - } + let addr = match socket_addr { + Ok(addr) => addr, Err(e) => { error!( "[peer:{}] Invalid socket address {}:{} - {}", @@ -738,9 +707,30 @@ impl PeerActor { io::ErrorKind::InvalidInput, e, )); - false + return; } + }; + + // Use connect_timeout to prevent blocking the thread for too long + let timeout = Duration::from_secs(5); + let stream = match TcpStream::connect_timeout(&addr, timeout) { + Ok(stream) => stream, + Err(e) => { + self.disconnect_with_error(e); + return; + } + }; + + if let Err(e) = stream.set_nonblocking(true) { + error!("[peer:{}] Failed to set non-blocking: {}", username, e); + self.disconnect_with_error(e); + return; } + stream.set_nodelay(true).ok(); + self.stream = Some(stream); + self.connection_state = ConnectionState::Connecting { + since: Instant::now(), + }; } fn check_connection_status(&mut self) { diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index 5f6413f..60e136b 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -145,28 +145,6 @@ impl PeerRegistry { None } - #[must_use] - pub fn get_all_usernames(&self) -> Vec { - match self.peers.lock_safe() { - Ok(peers) => peers.keys().cloned().collect(), - Err(e) => { - error!("[peer_registry] get_all_usernames: {}", e); - Vec::new() - } - } - } - - #[must_use] - pub fn count(&self) -> usize { - match self.peers.lock_safe() { - Ok(peers) => peers.len(), - Err(e) => { - error!("[peer_registry] count: {}", e); - 0 - } - } - } - #[must_use] pub fn contains(&self, username: &str) -> bool { match self.peers.lock_safe() { diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index f5b27c4..a2e796c 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -31,8 +31,8 @@ use crate::types::{ }; use crate::utils::lock::RwLockExt; -use std::io::{self, Error, Write}; -use std::net::{TcpStream, ToSocketAddrs}; +use std::io::{self, Write}; +use std::net::TcpStream; use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; @@ -109,17 +109,6 @@ impl UserMessage { new_message, } } - pub fn print(&self) { - debug!( - "Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}", - self.timestamp, - self.username, - self.id, - self.new_message, - self.message - ); - } - /// The server-assigned id of this message (used to acknowledge it). #[must_use] pub const fn id(&self) -> u32 { @@ -320,60 +309,23 @@ impl ServerActor { self.session = session; } - #[must_use] - pub const fn get_address(&self) -> &PeerAddress { - &self.address - } - - #[must_use] - pub const fn get_sender(&self) -> Option<&Sender> { - self.dispatcher_sender.as_ref() - } - - fn initiate_connection(&mut self) -> bool { - let host = self.address.host.clone(); - let port = self.address.port; - - let addr_str = format!("{host}:{port}"); - - let mut socket_addrs = match addr_str.to_socket_addrs() { - Ok(addrs) => addrs, - Err(e) => { - error!("[server] Failed to resolve address: {}", e); - - self.disconnect_with_error(io::Error::new( - io::ErrorKind::InvalidInput, - e, - )); - return false; - } - }; - - let socket_addr = socket_addrs.next(); - - 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; - }; - - let stream = match TcpStream::connect(addr) { + fn initiate_connection(&mut self) { + let stream = match TcpStream::connect(( + self.address.host.as_str(), + self.address.port, + )) { Ok(s) => s, Err(e) => { - self.disconnect_with_error(e); - return false; + error!("[server] Failed to connect to {}: {}", self.address, e); + self.disconnect_with_error(); + return; } }; if let Err(e) = stream.set_nonblocking(true) { error!("[server] Failed to set non-blocking: {}", e); - self.disconnect_with_error(e); - return false; + self.disconnect_with_error(); + return; } stream.set_nodelay(true).ok(); @@ -381,7 +333,6 @@ impl ServerActor { self.connection_state = ConnectionState::Connecting { since: Instant::now(), }; - true } pub fn set_self_handle(&mut self, handle: ActorHandle) { @@ -436,13 +387,7 @@ impl ServerActor { let messages: Vec = self .dispatcher_receiver .as_ref() - .map_or_else(Vec::new, |receiver| { - let mut msgs = Vec::new(); - while let Ok(msg) = receiver.try_recv() { - msgs.push(msg); - } - msgs - }); + .map_or_else(Vec::new, |receiver| receiver.try_iter().collect()); for msg in &messages { self.handle_message(msg.clone()); @@ -783,7 +728,7 @@ impl ServerActor { e, e.kind() ); - self.disconnect_with_error(e); + self.disconnect_with_error(); return; } } @@ -818,7 +763,7 @@ impl ServerActor { "[server] Error extracting message: {}. Disconnecting.", e ); - self.disconnect_with_error(e); + self.disconnect_with_error(); return; } Ok(None) => { @@ -871,13 +816,13 @@ impl ServerActor { if let Err(e) = stream.write_all(&message.get_buffer()) { error!("[server] Error writing message: {}. Disconnecting.", e); - self.disconnect_with_error(e); + self.disconnect_with_error(); return; } if let Err(e) = stream.flush() { error!("[server] Error flushing stream: {}. Disconnecting.", e); - self.disconnect_with_error(e); + self.disconnect_with_error(); } } @@ -893,7 +838,7 @@ impl ServerActor { self.disconnect(); } - fn disconnect_with_error(&mut self, _error: Error) { + fn disconnect_with_error(&mut self) { debug!("[server] disconnect"); // Losing an established connection ends the session: nothing reconnects @@ -919,10 +864,7 @@ impl ServerActor { if since.elapsed() > Duration::from_secs(20) { error!("[server] Connection timeout after 20 seconds"); - self.disconnect_with_error(io::Error::new( - io::ErrorKind::TimedOut, - "Connection timeout", - )); + self.disconnect_with_error(); return; } @@ -938,16 +880,12 @@ impl ServerActor { Err(ref e) if e.kind() == io::ErrorKind::NotConnected => {} Err(e) => { error!("[server] Connection failed: {}", e); - self.disconnect_with_error(e); + self.disconnect_with_error(); } } } fn on_connection_established(&mut self) { - let Some(_) = self.stream else { - panic!("Stream should be available here") - }; - self.initialize_dispatcher(); let queued = std::mem::take(&mut self.queued_messages); diff --git a/soulseek-rs-lib/src/client/connection.rs b/soulseek-rs-lib/src/client/connection.rs index 2aa7d15..cfa722d 100644 --- a/soulseek-rs-lib/src/client/connection.rs +++ b/soulseek-rs-lib/src/client/connection.rs @@ -2,7 +2,7 @@ use super::{ Arc, Client, ClientContext, ClientOperation, ConnectionType, DownloadPeer, DownloadStatus, Listen, Peer, PeerRegistry, Receiver, Result, RwLock, RwLockExt, Sender, ServerActor, ServerMessage, Shares, SoulseekRs, - TcpStream, debug, error, info, mpsc, thread, trace, + TcpStream, error, info, mpsc, thread, trace, }; /// Ceiling on the wait for a login verdict. Generous enough for a slow server @@ -18,7 +18,6 @@ impl Client { ) = mpsc::channel(); let mut ctx = self.context.write_safe()?; - ctx.sender = Some(sender.clone()); let peer_registry = PeerRegistry::new( ctx.actor_system.clone(), sender.clone(), @@ -166,8 +165,6 @@ impl Client { own_username: String, stream: Option, ) { - let client_context = client_context; - let peer_clone = peer.clone(); trace!( "[client] connecting to {}, with connection_type: {}, and token {:?}", @@ -252,37 +249,4 @@ impl Client { } } } - - pub(crate) fn pierce_firewall( - peer: Peer, - client_context: Arc>, - own_username: String, - ) { - debug!("Piercing firewall for peer: {:?}", peer); - - let context = match client_context.read_safe() { - Ok(c) => c, - Err(e) => { - error!("[client] pierce_firewall read: {}", e); - return; - } - }; - 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); - } } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index ed317d1..6f5de79 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -13,10 +13,7 @@ use crate::{ actor::{ActorSystem, peer_registry::PeerRegistry}, error::{Result, SoulseekRs}, message::peer::{FileEntry, SharedDirectory, build_file_search_response}, - peer::{ - ConnectionType, DownloadPeer, NewPeer, Peer, PeerMessage, - listen::Listen, - }, + peer::{ConnectionType, DownloadPeer, Peer, PeerMessage, listen::Listen}, shares::Shares, types::{Download, Search, SearchResult}, utils::{lock::RwLockExt, md5}, @@ -25,14 +22,11 @@ use std::{ collections::{HashMap, HashSet}, net::TcpStream, sync::{ - RwLock, + Arc, RwLock, atomic::{AtomicBool, AtomicU32, Ordering}, - mpsc::{Receiver, Sender}, + mpsc::{self, Receiver, Sender}, }, thread::{self, sleep}, -}; -use std::{ - sync::{Arc, mpsc}, time::{Duration, Instant}, }; use upload_queue::QueuedUpload; @@ -213,11 +207,9 @@ impl Default for ClientSettings { #[derive(Debug)] #[non_exhaustive] pub enum ClientOperation { - NewPeer(NewPeer), ConnectToPeer(Peer), SearchResult(SearchResult), PeerDisconnected(u64, String, Option), - PierceFireWall(Peer), DownloadFromPeer(u32, Peer, bool), UpdateDownloadTokens(Transfer, String), GetPeerAddressResponse { @@ -298,7 +290,6 @@ pub enum ClientOperation { pub struct ClientContext { pub peer_registry: Option, pub downloads: DownloadStore, - sender: Option>, server_sender: Option>, searches: HashMap, private_messages: Vec, @@ -376,13 +367,6 @@ impl ClientContext { ) -> Option<&mut Download> { self.downloads.get_by_token_mut(token) } - pub fn get_download_by_file_mut( - &mut self, - username: &str, - filename: &str, - ) -> Option<&mut Download> { - self.downloads.get_by_file_mut(username, filename) - } #[must_use] pub fn get_download_tokens(&self) -> Vec { self.downloads.tokens() @@ -398,13 +382,6 @@ impl ClientContext { ) { self.downloads.update_status(token, status); } - pub fn remove_queued_download_by_file( - &mut self, - username: &str, - filename: &str, - ) -> bool { - self.downloads.remove_queued_by_file(username, filename) - } } impl ClientContext { @@ -414,7 +391,6 @@ impl ClientContext { Self { peer_registry: None, - sender: None, server_sender: None, searches: HashMap::new(), private_messages: Vec::new(), diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index ee4c83a..e3303b3 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -15,956 +15,760 @@ impl Client { 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(); + for operation in reader { + 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 = match client_context - .write_safe() - { - Ok(c) => c, - Err(e) => { - error!( - "[client] SearchResult write: {}", - e - ); - continue; - } - }; - 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( - id, - username, - error, - ) => { - // Scope the read guard: process_failed_uploads - // below acquires a write lock on the same - // RwLock, which would self-deadlock the entire - // client ops loop if this read guard were still - // held on this thread. Evict only if this exact - // actor still occupies the slot, so a replaced - // actor's shutdown can't remove its successor. - { - let context = match client_context - .read_safe() - { - Ok(c) => c, - Err(e) => { - error!( - "[client] PeerDisconnected read: {}", - e - ); - continue; - } - }; - if let Some(ref registry) = - context.peer_registry - && let Some(handle) = registry - .remove_peer_if(&username, id) - { - 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, - ); - } - // A peer that leaves must not keep occupying an - // upload slot: with a slot cap, a couple of - // departed peers would otherwise wedge uploads - // shut for the rest of the session. - Self::release_upload_slots( - &client_context, - &username, - ); - } - ClientOperation::PierceFireWall(peer) => { - Self::pierce_firewall( - peer, - client_context.clone(), - own_username.clone(), - ); - } - ClientOperation::DownloadFromPeer( - token, + thread::spawn(move || { + Self::connect_to_peer( peer, - allowed, - ) => { - let maybe_download = match client_context - .read_safe() - { - Ok(ctx) => ctx - .get_download_by_token(token) - .cloned(), - Err(e) => { - error!( - "[client] DownloadFromPeer read: {}", - e - ); - continue; - } - }; - let own_username = own_username.clone(); - let client_context_clone = - client_context.clone(); + client_context_clone, + own_username_clone, + None, + ); + }); + } + ClientOperation::SearchResult(search_result) => { + trace!("[client] SearchResult {:?}", search_result); + let mut context = match client_context.write_safe() { + Ok(c) => c, + Err(e) => { + error!("[client] SearchResult write: {}", e); + continue; + } + }; + let result_token = search_result.token; - 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); - match client_context_clone.write_safe() { - Ok(mut ctx) => ctx.update_download_with_status(download.token, DownloadStatus::Completed), - Err(e) => error!("[client] download complete write: {}", e), - } - info!( - "Successfully downloaded {} bytes to {}", - download.size, - filename - ); - } - Err(e) => { - let reason = Some( - e.to_string(), - ); - let _ = download.sender.send(DownloadStatus::Failed(reason.clone())); - match client_context_clone.write_safe() { - Ok(mut ctx) => ctx.update_download_with_status(download.token, DownloadStatus::Failed(reason)), - Err(e) => error!("[client] download failed write: {}", e), - } - 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 = match client_context - .read_safe() - { - Ok(ctx) => { - ctx.peer_registry.as_ref().is_some_and( - |r| r.contains(&new_peer.username), - ) - } - Err(e) => { - error!("[client] NewPeer read: {}", e); - continue; - } - }; - - if peer_exists { - debug!( - "Already connected to {}", - new_peer.username - ); - } else { - let send_result = client_context - .read_safe() - .ok() - .and_then(|ctx| { - ctx.server_sender.as_ref().map( - |s| { - s.send( - ServerMessage::GetPeerAddress( - new_peer - .username - .clone(), - ), - ) - }, - ) - }); - if let Some(Err(e)) = send_result { - error!( - "[client] NewPeer send GetPeerAddress: {}", - e - ); - } - } - - let addr = match new_peer.tcp_stream.peer_addr() - { - Ok(a) => a, - Err(e) => { - error!( - "[client] NewPeer peer_addr: {}", - e - ); - continue; - } - }; - 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 - ); - - // Port 0 is the server saying it does not know - // where this user listens (no SetWaitPort yet, - // or firewalled). It is not an address: caching - // it poisons every later lookup, and an upload - // dialled at it dies with "Can't assign - // requested address" and is dropped on the - // floor. Leave those uploads queued for a later - // resolution instead. - // - // The connect attempt further down still runs: - // its failure is exactly what makes an - // unreachable peer fall back to the - // server-brokered path. - let waiting_serves = if port == 0 { - warn!( - "[client] server reports no listening port for {}", - username - ); - Vec::new() - } else { - match client_context.write_safe() { - Ok(mut ctx) => { - ctx.cache_peer_address( - &username, - host.clone(), - port, - ); - ctx.pending_serves - .remove(&username) - .unwrap_or_default() - } - Err(_) => Vec::new(), - } - }; - for token in waiting_serves { - Self::spawn_serve( - &client_context, - &own_username, - token, - host.clone(), - port, - ); - } - - let peer_exists = match client_context - .read_safe() - { - Ok(ctx) => ctx - .peer_registry - .as_ref() - .is_some_and(|r| r.contains(&username)), - Err(e) => { - error!( - "[client] GetPeerAddressResponse read: {}", - e - ); - continue; - } - }; - - if peer_exists { - // Existing peer: skip re-registration. Reconnect - // policy on conflict is intentionally undecided. - } else { - let peer = Peer::new( - username, - ConnectionType::P, - host, - port, - None, - 0, - // obfuscation_type is a small enum; a - // real obfuscated_port is a full u16 and - // must not be truncated into a u8 (which - // panicked and took down the ops thread). - u8::try_from(obfuscation_type) - .unwrap_or(0), - obfuscated_port, - ); - 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 = match client_context - .write_safe() - { - Ok(c) => c, - Err(e) => { - error!( - "[client] UpdateDownloadTokens write: {}", - e - ); - continue; - } - }; - - 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.clone(), - token: transfer.token, - size: transfer.size, - download_directory: download - .download_directory, - status: download.status.clone(), - sender: download.sender.clone(), - queue_position: download.queue_position, - metadata: download.metadata.clone(), - }); - context.remove_download(old_token); - } - - // Only now invite the file connection: it is - // matched by this token, which is recorded as - // of the line above. Answering any earlier - // races the peer's connection against our own - // bookkeeping. - let registry = context.peer_registry.clone(); - drop(context); - if let Some(registry) = registry { - let response = MessageFactory::build_transfer_response_message(transfer); - let _ = registry.send_to_peer( - &username, - PeerMessage::SendMessage(response), - ); - } - } - ClientOperation::UploadFailed( - username, - filename, - ) => { - Self::process_failed_uploads( - client_context.clone(), - &username, - Some(&filename), - ); - } - ClientOperation::PlaceInQueueUpdate { - username, - filename, - place, - } => match client_context.write_safe() { - Ok(mut ctx) => { - let updated = - ctx.downloads.update_queue_position( - &username, &filename, place, - ); - if !updated { - debug!( - "[client] PlaceInQueueUpdate: no matching download for {}/{}", - username, filename - ); - } - } - Err(e) => error!( - "[client] PlaceInQueueUpdate write: {}", - e - ), - }, - ClientOperation::SetServerSender(sender) => { - match client_context.write_safe() { - Ok(mut ctx) => { - ctx.server_sender = Some(sender); - debug!( - "[client] Server sender initialized" - ); - } - Err(e) => error!( - "[client] SetServerSender write: {}", - e - ), - } - } - ClientOperation::PrivateMessageReceived( - user_message, - ) => match client_context.write_safe() { - Ok(mut ctx) => { - ctx.push_private_message(user_message); - } - Err(e) => error!( - "[client] PrivateMessageReceived write: {}", - e - ), - }, - ClientOperation::UserStatusReceived { - username, - status, - privileged, - } => match client_context.write_safe() { - Ok(mut ctx) => ctx.apply_user_status( - username, status, privileged, - ), - Err(e) => error!( - "[client] UserStatusReceived write: {}", - e - ), - }, - ClientOperation::UserStatsReceived { - username, - average_speed, - shared_files, - shared_folders, - } => match client_context.write_safe() { - Ok(mut ctx) => ctx.apply_user_stats( - username, - average_speed, - shared_files, - shared_folders, - ), - Err(e) => error!( - "[client] UserStatsReceived write: {}", - e - ), - }, - ClientOperation::RoomEvent(event) => { - match client_context.write_safe() { - Ok(mut ctx) => ctx.apply_room_event(event), - Err(e) => error!( - "[client] RoomEvent write: {}", - e - ), - } - } - ClientOperation::WishlistInterval(seconds) => { - if let Ok(mut ctx) = client_context.write_safe() - { - ctx.wishlist_interval = Some(seconds); - } - } - ClientOperation::PeerConnected(username) => { - // An outbound control connection just handshook. - // Flush any downloads that were queued for this - // peer while we were still connecting. Collect - // under a read guard, then act without it held. - let (registry, files): ( - Option, - Vec, - ) = match client_context.read_safe() { - Ok(ctx) => ( - ctx.peer_registry.clone(), - ctx.get_downloads() - .iter() - .filter(|d| { - d.username == username - && matches!( - d.status, - DownloadStatus::Queued - ) - }) - .map(|d| d.filename.clone()) - .collect(), - ), - Err(e) => { - error!( - "[client] PeerConnected read: {}", - e - ); - continue; - } - }; - // Also flush any peer messages (e.g. search - // responses) queued while connecting. - let queued_messages = client_context - .write_safe() - .map(|mut ctx| { - ctx.take_peer_messages(&username) - }) - .unwrap_or_default(); - if let Some(registry) = registry { - for filename in files { - let _ = registry - .queue_upload(&username, filename); - } - for message in queued_messages { - let _ = registry.send_to_peer( - &username, - PeerMessage::SendMessage(message), - ); - } - } - } - ClientOperation::IncomingSearch { - username, - token, - query, - } => { - // Don't answer our own distributed search. - if username == own_username { - continue; - } - let response = match client_context.read_safe() - { - Ok(ctx) => build_search_response( - &ctx.shares, - &own_username, - token, - &query, - ), - Err(e) => { - error!( - "[client] IncomingSearch read: {}", - e - ); - continue; - } - }; - let Some(message) = response else { - continue; // no matching shares - }; - - // Deliver to the searcher: send now if we have a - // control connection, else open one and queue. - let (connected, registry, server_sender) = - match client_context.read_safe() { - Ok(ctx) => ( - ctx.peer_registry - .as_ref() - .is_some_and(|r| { - r.contains(&username) - }), - ctx.peer_registry.clone(), - ctx.server_sender.clone(), - ), - Err(_) => continue, - }; - if connected { - if let Some(registry) = registry { - let _ = registry.send_to_peer( - &username, - PeerMessage::SendMessage(message), - ); - } - } else { - if let Ok(mut ctx) = - client_context.write_safe() - { - ctx.queue_peer_message( - &username, message, - ); - } - if let Some(sender) = server_sender { - let _ = sender.send( - ServerMessage::GetPeerAddress( - username, - ), - ); - } - } - } - ClientOperation::QueueUpload { - requester_key, - filename, - } => { - // The peer served next may not be this one. - let downloader = - downloader_of(&requester_key).to_string(); - match client_context.write_safe() { - Ok(mut ctx) => { - let Some(file) = - ctx.shares.get(&filename) - else { - debug!( - "[client] QueueUpload for unknown file {}", - filename - ); - continue; - }; - let size = file.size; - let real_path = file.real_path.clone(); - ctx.enqueue_upload( - &requester_key, - &downloader, - &filename, - real_path, - size, - ); - } - Err(e) => { - error!( - "[client] QueueUpload write: {}", - e - ); - continue; - } - } - Self::pump_upload_queue(&client_context); - } - ClientOperation::PlaceInQueueRequested { - requester_key, - filename, - } => { - let downloader = downloader_of(&requester_key); - let (registry, place) = - match client_context.read_safe() { - Ok(ctx) => ( - ctx.peer_registry.clone(), - ctx.place_in_queue( - downloader, &filename, - ), - ), - Err(_) => continue, - }; - // Silence would leave the peer guessing; a file - // no longer queued is being served, which is - // place 0 by the same convention Nicotine+ uses. - if let Some(registry) = registry { - let _ = registry.send_to_peer( - &requester_key, - PeerMessage::SendMessage( - MessageFactory::build_place_in_queue_response( - &filename, - place.unwrap_or(0), - ), - ), - ); - } - } - ClientOperation::PrivilegedUsers(users) => { - if let Ok(mut ctx) = client_context.write_safe() - { - debug!( - "[client] {} privileged users listed", - users.len() - ); - ctx.set_privileged_users(users); - } - // Someone already waiting may have just been - // outranked, but a slot that is free should - // still be filled. - Self::pump_upload_queue(&client_context); - } - ClientOperation::OwnPrivileges(seconds) => { - if let Ok(mut ctx) = client_context.write_safe() - { - ctx.own_privileges = Some(seconds); - } - } - ClientOperation::StartUpload { token } => { - // The peer accepted our offer: resolve their - // address (from the code-9 GetPeerAddress) and - // stream the file, or queue until it resolves. - let (job_addr, downloader) = - match client_context.read_safe() { - Ok(ctx) => { - let Some(job) = - ctx.uploads.get(&token) - else { - continue; - }; - ( - ctx.peer_address( - &job.downloader, - ), - job.downloader.clone(), - ) - } - Err(_) => continue, - }; - if let Some((host, port)) = job_addr { - Self::spawn_serve( - &client_context, - &own_username, - token, - host, - port, - ); - } else { - if let Ok(mut ctx) = - client_context.write_safe() - { - ctx.pending_serves - .entry(downloader.clone()) - .or_default() - .push(token); - } - if let Ok(ctx) = client_context.read_safe() - && let Some(sender) = - ctx.server_sender.clone() - { - let _ = sender.send( - ServerMessage::GetPeerAddress( - downloader, - ), - ); - } - } - } - ClientOperation::ShareListRequested { - requester_key, - } => { - // Reply with our full shared-file listing. - let (registry, message) = match client_context - .read_safe() - { - Ok(ctx) => { - let dirs = ctx - .shares - .directories() - .into_iter() - .map(|(name, files)| { - crate::message::peer::SharedDirectory { - name, - files, - } - }) - .collect::>(); - ( - ctx.peer_registry.clone(), - crate::message::peer::build_shared_file_list(&dirs), - ) - } - Err(_) => continue, - }; - if let Some(registry) = registry { - let _ = registry.send_to_peer( - &requester_key, - PeerMessage::SendMessage(message), - ); - } - } - ClientOperation::BrowseResult { - username, - directories, - } => { - if let Ok(mut ctx) = client_context.write_safe() - { - ctx.store_browse_result( - username, - directories, - ); - } - } - ClientOperation::PeerConnectFailed( - id, - username, - ) => { - // Direct connect failed: ask the server to - // broker it. Register a correlation token, then - // send ConnectToPeer so the (firewalled) peer - // connects back to our listener quoting it. - let token = next_connect_token(); - let server_sender = match client_context - .write_safe() - { - Ok(mut ctx) => { - // Reap the dead outbound actor so it - // stops pinning a pool worker and no - // longer shadows the brokered reconnect - // (a stale registry entry would make - // later downloads queue into a dead, - // streamless actor and hang). Identity- - // aware so a newer namesake is untouched. - if let Some(handle) = ctx - .peer_registry - .as_ref() - .and_then(|r| { - r.remove_peer_if(&username, id) - }) - { - let _ = handle.stop(); - } - ctx.add_pending_connect( - token, - username.clone(), - ); - ctx.server_sender.clone() - } - Err(e) => { - error!( - "[client] PeerConnectFailed write: {}", - e - ); - continue; - } - }; - let Some(sender) = server_sender else { - continue; - }; - let msg = crate::message::server::MessageFactory::build_connect_to_peer( - token, - &username, - ConnectionType::P, - ); - let _ = sender - .send(ServerMessage::SendMessage(msg)); - - // Bound the brokered attempt: if no PierceFirewall - // consumes the token, fail the peer's queued - // downloads (so the caller's Receiver unblocks) - // and reclaim the token. A successful pierce - // takes the token first, making this a no-op. - let timeout_ctx = client_context.clone(); - let timeout_user = username.clone(); - thread::spawn(move || { - sleep(BROKER_CONNECT_TIMEOUT); - let still_pending = timeout_ctx - .write_safe() - .is_ok_and(|mut c| { - c.take_pending_connect(token) - .is_some() - }); - if still_pending { - Self::fail_queued_downloads( - &timeout_ctx, - &timeout_user, - ); - } - }); + // Find the search with matching token + for search in context.searches.values_mut() { + if search.token == result_token { + search.results.push(search_result); + break; } } } - Err(e) => { - error!("[client] Channel recv error: {:?}", e); - break; + ClientOperation::PeerDisconnected(id, username, error) => { + // Scope the read guard: process_failed_uploads + // below acquires a write lock on the same + // RwLock, which would self-deadlock the entire + // client ops loop if this read guard were still + // held on this thread. Evict only if this exact + // actor still occupies the slot, so a replaced + // actor's shutdown can't remove its successor. + { + let context = match client_context.read_safe() { + Ok(c) => c, + Err(e) => { + error!( + "[client] PeerDisconnected read: {}", + e + ); + continue; + } + }; + if let Some(ref registry) = context.peer_registry + && let Some(handle) = + registry.remove_peer_if(&username, id) + { + 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, + ); + } + // A peer that leaves must not keep occupying an + // upload slot: with a slot cap, a couple of + // departed peers would otherwise wedge uploads + // shut for the rest of the session. + Self::release_upload_slots(&client_context, &username); + } + ClientOperation::DownloadFromPeer(token, peer, allowed) => { + let maybe_download = match client_context.read_safe() { + Ok(ctx) => { + ctx.get_download_by_token(token).cloned() + } + Err(e) => { + error!("[client] DownloadFromPeer read: {}", e); + continue; + } + }; + let own_username = own_username.clone(); + let client_context_clone = client_context.clone(); + + trace!( + "[client] DownloadFromPeer token: {} peer: {:?}", + token, peer + ); + let Some(download) = maybe_download else { + error!( + "Can't find download with token {:?}", + token + ); + continue; + }; + + thread::spawn(move || { + let download_peer = DownloadPeer::new( + download.username.clone(), + peer.host.clone(), + peer.port, + token, + allowed, + own_username, + ); + let Some(filename) = + download.filename.split('\\').next_back() + else { + error!( + "Cant find filename to save download: {:?}", + download.filename + ); + return; + }; + match download_peer.download_file( + client_context_clone.clone(), + Some(download.clone()), + None, + ) { + Ok((download, filename)) => { + let _ = download + .sender + .send(DownloadStatus::Completed); + match client_context_clone.write_safe() { + Ok(mut ctx) => ctx + .update_download_with_status( + download.token, + DownloadStatus::Completed, + ), + Err(e) => error!( + "[client] download complete write: {}", + e + ), + } + info!( + "Successfully downloaded {} bytes to {}", + download.size, filename + ); + } + Err(e) => { + let reason = Some(e.to_string()); + let _ = download.sender.send( + DownloadStatus::Failed(reason.clone()), + ); + match client_context_clone.write_safe() { + Ok(mut ctx) => ctx + .update_download_with_status( + download.token, + DownloadStatus::Failed(reason), + ), + Err(e) => error!( + "[client] download failed write: {}", + e + ), + } + error!( + "Failed to download file '{}' from {}:{} (token: {}) - Error: {}", + filename, + peer.host, + peer.port, + download.token, + e + ); + } + } + }); + } + 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 + ); + + // Port 0 is the server saying it does not know + // where this user listens (no SetWaitPort yet, + // or firewalled). It is not an address: caching + // it poisons every later lookup, and an upload + // dialled at it dies with "Can't assign + // requested address" and is dropped on the + // floor. Leave those uploads queued for a later + // resolution instead. + // + // The connect attempt further down still runs: + // its failure is exactly what makes an + // unreachable peer fall back to the + // server-brokered path. + let waiting_serves = if port == 0 { + warn!( + "[client] server reports no listening port for {}", + username + ); + Vec::new() + } else { + match client_context.write_safe() { + Ok(mut ctx) => { + ctx.cache_peer_address( + &username, + host.clone(), + port, + ); + ctx.pending_serves + .remove(&username) + .unwrap_or_default() + } + Err(_) => Vec::new(), + } + }; + for token in waiting_serves { + Self::spawn_serve( + &client_context, + &own_username, + token, + host.clone(), + port, + ); + } + + let peer_exists = match client_context.read_safe() { + Ok(ctx) => ctx + .peer_registry + .as_ref() + .is_some_and(|r| r.contains(&username)), + Err(e) => { + error!( + "[client] GetPeerAddressResponse read: {}", + e + ); + continue; + } + }; + + // Existing peer: skip re-registration. Reconnect + // policy on conflict is intentionally undecided. + if !peer_exists { + let peer = Peer::new( + username, + ConnectionType::P, + host, + port, + None, + 0, + // obfuscation_type is a small enum; a + // real obfuscated_port is a full u16 and + // must not be truncated into a u8 (which + // panicked and took down the ops thread). + u8::try_from(obfuscation_type).unwrap_or(0), + obfuscated_port, + ); + 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 = match client_context.write_safe() { + Ok(c) => c, + Err(e) => { + error!( + "[client] UpdateDownloadTokens write: {}", + e + ); + continue; + } + }; + + 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 { + token: transfer.token, + size: transfer.size, + ..download + }); + context.remove_download(old_token); + } + + // Only now invite the file connection: it is + // matched by this token, which is recorded as + // of the line above. Answering any earlier + // races the peer's connection against our own + // bookkeeping. + let registry = context.peer_registry.clone(); + drop(context); + if let Some(registry) = registry { + let response = + MessageFactory::build_transfer_response_message( + transfer, + ); + let _ = registry.send_to_peer( + &username, + PeerMessage::SendMessage(response), + ); + } + } + ClientOperation::UploadFailed(username, filename) => { + Self::process_failed_uploads( + client_context.clone(), + &username, + Some(&filename), + ); + } + ClientOperation::PlaceInQueueUpdate { + username, + filename, + place, + } => match client_context.write_safe() { + Ok(mut ctx) => { + let updated = ctx.downloads.update_queue_position( + &username, &filename, place, + ); + if !updated { + debug!( + "[client] PlaceInQueueUpdate: no matching download for {}/{}", + username, filename + ); + } + } + Err(e) => { + error!("[client] PlaceInQueueUpdate write: {}", e); + } + }, + ClientOperation::SetServerSender(sender) => { + match client_context.write_safe() { + Ok(mut ctx) => { + ctx.server_sender = Some(sender); + debug!("[client] Server sender initialized"); + } + Err(e) => { + error!("[client] SetServerSender write: {}", e); + } + } + } + ClientOperation::PrivateMessageReceived(user_message) => { + match client_context.write_safe() { + Ok(mut ctx) => { + ctx.push_private_message(user_message); + } + Err(e) => error!( + "[client] PrivateMessageReceived write: {}", + e + ), + } + } + ClientOperation::UserStatusReceived { + username, + status, + privileged, + } => match client_context.write_safe() { + Ok(mut ctx) => { + ctx.apply_user_status(username, status, privileged); + } + Err(e) => { + error!("[client] UserStatusReceived write: {}", e); + } + }, + ClientOperation::UserStatsReceived { + username, + average_speed, + shared_files, + shared_folders, + } => match client_context.write_safe() { + Ok(mut ctx) => ctx.apply_user_stats( + username, + average_speed, + shared_files, + shared_folders, + ), + Err(e) => { + error!("[client] UserStatsReceived write: {}", e); + } + }, + ClientOperation::RoomEvent(event) => { + match client_context.write_safe() { + Ok(mut ctx) => ctx.apply_room_event(event), + Err(e) => error!("[client] RoomEvent write: {}", e), + } + } + ClientOperation::WishlistInterval(seconds) => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.wishlist_interval = Some(seconds); + } + } + ClientOperation::PeerConnected(username) => { + // An outbound control connection just handshook. + // Flush any downloads that were queued for this + // peer while we were still connecting. Collect + // under a read guard, then act without it held. + let (registry, files): ( + Option, + Vec, + ) = match client_context.read_safe() { + Ok(ctx) => ( + ctx.peer_registry.clone(), + ctx.get_downloads() + .iter() + .filter(|d| { + d.username == username + && matches!( + d.status, + DownloadStatus::Queued + ) + }) + .map(|d| d.filename.clone()) + .collect(), + ), + Err(e) => { + error!("[client] PeerConnected read: {}", e); + continue; + } + }; + // Also flush any peer messages (e.g. search + // responses) queued while connecting. + let queued_messages = client_context + .write_safe() + .map(|mut ctx| ctx.take_peer_messages(&username)) + .unwrap_or_default(); + if let Some(registry) = registry { + for filename in files { + let _ = + registry.queue_upload(&username, filename); + } + for message in queued_messages { + let _ = registry.send_to_peer( + &username, + PeerMessage::SendMessage(message), + ); + } + } + } + ClientOperation::IncomingSearch { + username, + token, + query, + } => { + // Don't answer our own distributed search. + if username == own_username { + continue; + } + let response = match client_context.read_safe() { + Ok(ctx) => build_search_response( + &ctx.shares, + &own_username, + token, + &query, + ), + Err(e) => { + error!("[client] IncomingSearch read: {}", e); + continue; + } + }; + let Some(message) = response else { + continue; // no matching shares + }; + + // Deliver to the searcher: send now if we have a + // control connection, else open one and queue. + let (connected, registry, server_sender) = + match client_context.read_safe() { + Ok(ctx) => ( + ctx.peer_registry + .as_ref() + .is_some_and(|r| r.contains(&username)), + ctx.peer_registry.clone(), + ctx.server_sender.clone(), + ), + Err(_) => continue, + }; + if connected { + if let Some(registry) = registry { + let _ = registry.send_to_peer( + &username, + PeerMessage::SendMessage(message), + ); + } + } else { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.queue_peer_message(&username, message); + } + if let Some(sender) = server_sender { + let _ = sender.send( + ServerMessage::GetPeerAddress(username), + ); + } + } + } + ClientOperation::QueueUpload { + requester_key, + filename, + } => { + // The peer served next may not be this one. + let downloader = + downloader_of(&requester_key).to_string(); + match client_context.write_safe() { + Ok(mut ctx) => { + let Some(file) = ctx.shares.get(&filename) + else { + debug!( + "[client] QueueUpload for unknown file {}", + filename + ); + continue; + }; + let size = file.size; + let real_path = file.real_path.clone(); + ctx.enqueue_upload( + &requester_key, + &downloader, + &filename, + real_path, + size, + ); + } + Err(e) => { + error!("[client] QueueUpload write: {}", e); + continue; + } + } + Self::pump_upload_queue(&client_context); + } + ClientOperation::PlaceInQueueRequested { + requester_key, + filename, + } => { + let downloader = downloader_of(&requester_key); + let (registry, place) = match client_context.read_safe() + { + Ok(ctx) => ( + ctx.peer_registry.clone(), + ctx.place_in_queue(downloader, &filename), + ), + Err(_) => continue, + }; + // Silence would leave the peer guessing; a file + // no longer queued is being served, which is + // place 0 by the same convention Nicotine+ uses. + if let Some(registry) = registry { + let _ = registry.send_to_peer( + &requester_key, + PeerMessage::SendMessage( + MessageFactory::build_place_in_queue_response( + &filename, + place.unwrap_or(0), + ), + ), + ); + } + } + ClientOperation::PrivilegedUsers(users) => { + if let Ok(mut ctx) = client_context.write_safe() { + debug!( + "[client] {} privileged users listed", + users.len() + ); + ctx.set_privileged_users(users); + } + // Someone already waiting may have just been + // outranked, but a slot that is free should + // still be filled. + Self::pump_upload_queue(&client_context); + } + ClientOperation::OwnPrivileges(seconds) => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.own_privileges = Some(seconds); + } + } + ClientOperation::StartUpload { token } => { + // The peer accepted our offer: resolve their + // address (from the code-9 GetPeerAddress) and + // stream the file, or queue until it resolves. + let (job_addr, downloader) = match client_context + .read_safe() + { + Ok(ctx) => { + let Some(job) = ctx.uploads.get(&token) else { + continue; + }; + ( + ctx.peer_address(&job.downloader), + job.downloader.clone(), + ) + } + Err(_) => continue, + }; + if let Some((host, port)) = job_addr { + Self::spawn_serve( + &client_context, + &own_username, + token, + host, + port, + ); + } else { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.pending_serves + .entry(downloader.clone()) + .or_default() + .push(token); + } + if let Ok(ctx) = client_context.read_safe() + && let Some(sender) = ctx.server_sender.clone() + { + let _ = sender.send( + ServerMessage::GetPeerAddress(downloader), + ); + } + } + } + ClientOperation::ShareListRequested { requester_key } => { + // Reply with our full shared-file listing. + let (registry, message) = match client_context + .read_safe() + { + Ok(ctx) => { + let dirs = ctx + .shares + .directories() + .into_iter() + .map(|(name, files)| { + crate::message::peer::SharedDirectory { + name, + files, + } + }) + .collect::>(); + ( + ctx.peer_registry.clone(), + crate::message::peer::build_shared_file_list(&dirs), + ) + } + Err(_) => continue, + }; + if let Some(registry) = registry { + let _ = registry.send_to_peer( + &requester_key, + PeerMessage::SendMessage(message), + ); + } + } + ClientOperation::BrowseResult { + username, + directories, + } => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.store_browse_result(username, directories); + } + } + ClientOperation::PeerConnectFailed(id, username) => { + // Direct connect failed: ask the server to + // broker it. Register a correlation token, then + // send ConnectToPeer so the (firewalled) peer + // connects back to our listener quoting it. + let token = next_connect_token(); + let server_sender = match client_context.write_safe() { + Ok(mut ctx) => { + // Reap the dead outbound actor so it + // stops pinning a pool worker and no + // longer shadows the brokered reconnect + // (a stale registry entry would make + // later downloads queue into a dead, + // streamless actor and hang). Identity- + // aware so a newer namesake is untouched. + if let Some(handle) = + ctx.peer_registry.as_ref().and_then(|r| { + r.remove_peer_if(&username, id) + }) + { + let _ = handle.stop(); + } + ctx.add_pending_connect( + token, + username.clone(), + ); + ctx.server_sender.clone() + } + Err(e) => { + error!( + "[client] PeerConnectFailed write: {}", + e + ); + continue; + } + }; + let Some(sender) = server_sender else { + continue; + }; + let msg = crate::message::server::MessageFactory::build_connect_to_peer( + token, + &username, + ConnectionType::P, + ); + let _ = sender.send(ServerMessage::SendMessage(msg)); + + // Bound the brokered attempt: if no PierceFirewall + // consumes the token, fail the peer's queued + // downloads (so the caller's Receiver unblocks) + // and reclaim the token. A successful pierce + // takes the token first, making this a no-op. + let timeout_ctx = client_context.clone(); + let timeout_user = username.clone(); + thread::spawn(move || { + sleep(BROKER_CONNECT_TIMEOUT); + let still_pending = + timeout_ctx.write_safe().is_ok_and(|mut c| { + c.take_pending_connect(token).is_some() + }); + if still_pending { + Self::fail_queued_downloads( + &timeout_ctx, + &timeout_user, + ); + } + }); } } } diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index a6eb2f4..f95dad7 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -1,22 +1,37 @@ use super::*; +fn download( + username: &str, + filename: &str, + token: u32, + status: DownloadStatus, + sender: Sender, +) -> Download { + Download { + username: username.to_string(), + filename: filename.to_string(), + token, + size: 100, + download_directory: "test".to_string(), + status, + sender, + queue_position: None, + metadata: DownloadMetadata::default(), + } +} + #[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(), + context.add_download(download( + "test", + "test.txt", token, - size: 100, - download_directory: "test".to_string(), - status: DownloadStatus::Queued, - sender: mpsc::channel().0, - queue_position: None, - metadata: DownloadMetadata::default(), - }; - context.add_download(download); + DownloadStatus::Queued, + mpsc::channel().0, + )); assert!(context.get_download_by_token(123).is_some()); assert_eq!(context.get_download_tokens(), vec![123]); assert_eq!(context.get_downloads().len(), 1); @@ -35,23 +50,18 @@ fn test_client_context_downloads() { fn test_client_pause_and_resume_download() { let client = Client::new("test-user", "test-password"); let (download_sender, download_receiver) = mpsc::channel(); - let download = Download { - username: "peer".to_string(), - filename: "song.mp3".to_string(), - token: 123, - size: 100, - download_directory: "test".to_string(), - status: DownloadStatus::InProgress { + + client.context.write().unwrap().add_download(download( + "peer", + "song.mp3", + 123, + DownloadStatus::InProgress { bytes_downloaded: 25, total_bytes: 100, speed_bytes_per_sec: 10.0, }, - sender: download_sender, - queue_position: None, - metadata: DownloadMetadata::default(), - }; - - client.context.write().unwrap().add_download(download); + download_sender, + )); assert!(client.pause_download("peer", "song.mp3")); assert!(matches!( @@ -118,17 +128,13 @@ fn fail_queued_downloads_notifies_receiver_and_store() { // must resolve to Failed both on its channel and in the store. let client = Client::new("u", "p"); let (sender, receiver) = mpsc::channel(); - client.context.write().unwrap().add_download(Download { - username: "peer".to_string(), - filename: "f.mp3".to_string(), - token: 7, - size: 10, - download_directory: "d".to_string(), - status: DownloadStatus::Queued, + client.context.write().unwrap().add_download(download( + "peer", + "f.mp3", + 7, + DownloadStatus::Queued, sender, - queue_position: None, - metadata: DownloadMetadata::default(), - }); + )); Client::fail_queued_downloads(&client.context, "peer"); @@ -171,37 +177,26 @@ fn build_search_response_matches_shares_and_echoes_token() { #[test] fn test_client_removes_only_queued_downloads() { let client = Client::new("test-user", "test-password"); - let queued_download = Download { - username: "peer".to_string(), - filename: "queued.mp3".to_string(), - token: 123, - size: 100, - download_directory: "test".to_string(), - status: DownloadStatus::Queued, - sender: mpsc::channel().0, - queue_position: None, - metadata: DownloadMetadata::default(), - }; - let active_download = Download { - username: "peer".to_string(), - filename: "active.mp3".to_string(), - token: 456, - size: 100, - download_directory: "test".to_string(), - status: DownloadStatus::InProgress { - bytes_downloaded: 25, - total_bytes: 100, - speed_bytes_per_sec: 10.0, - }, - sender: mpsc::channel().0, - queue_position: None, - metadata: DownloadMetadata::default(), - }; - { let mut context = client.context.write().unwrap(); - context.add_download(queued_download); - context.add_download(active_download); + context.add_download(download( + "peer", + "queued.mp3", + 123, + DownloadStatus::Queued, + mpsc::channel().0, + )); + context.add_download(download( + "peer", + "active.mp3", + 456, + DownloadStatus::InProgress { + bytes_downloaded: 25, + total_bytes: 100, + speed_bytes_per_sec: 10.0, + }, + mpsc::channel().0, + )); } assert!(client.remove_queued_download("peer", "queued.mp3")); diff --git a/soulseek-rs-lib/src/client/uploads.rs b/soulseek-rs-lib/src/client/uploads.rs index d3b7ba8..3b3cc2c 100644 --- a/soulseek-rs-lib/src/client/uploads.rs +++ b/soulseek-rs-lib/src/client/uploads.rs @@ -47,8 +47,6 @@ impl Client { .and_then(|ctx| ctx.own_privileges) } - /// How many uploads run at once. Lowering it does not interrupt transfers - /// already in flight; the queue simply refills more slowly. /// Queued-upload states recorded since the last call, for a caller that /// samples [`Client::uploads`] and would otherwise miss a peer that queued /// and was served between two samples. @@ -60,6 +58,8 @@ impl Client { .unwrap_or_default() } + /// How many uploads run at once. Lowering it does not interrupt transfers + /// already in flight; the queue simply refills more slowly. pub fn set_upload_slots(&self, slots: usize) { if let Ok(mut ctx) = self.context.write_safe() { ctx.upload_slots = slots.max(1); diff --git a/soulseek-rs-lib/src/dispatcher.rs b/soulseek-rs-lib/src/dispatcher.rs index 49f23fd..a034bd9 100644 --- a/soulseek-rs-lib/src/dispatcher.rs +++ b/soulseek-rs-lib/src/dispatcher.rs @@ -31,9 +31,8 @@ impl MessageDispatcher { handler.handle(message, self.sender.clone()); } else { warn!( - "[{}:dispatcher] No handler found for message code: {}", - self.owner_name, - message.get_message_code() + "[{}:dispatcher] No handler found for message code: {code}", + self.owner_name ); } } diff --git a/soulseek-rs-lib/src/message/mod.rs b/soulseek-rs-lib/src/message/mod.rs index 241161d..df62c8f 100644 --- a/soulseek-rs-lib/src/message/mod.rs +++ b/soulseek-rs-lib/src/message/mod.rs @@ -51,47 +51,6 @@ impl Message { } } - pub fn print_hex(&self) { - const BYTES_PER_LINE: usize = 16; - let data = &self.data; - - let chunks = data.chunks(BYTES_PER_LINE); - for (i, chunk) in chunks.enumerate() { - print!("{:04x} ", i * BYTES_PER_LINE); - - for j in 0..BYTES_PER_LINE { - if j < chunk.len() { - print!("{:02x} ", chunk[j]); - } else { - print!(" "); - } - - // Add extra space in the middle - if j == 7 { - print!(" "); - } - } - - print!(" "); - - let mut i = 0; - for &byte in chunk { - i += 1; - if byte.is_ascii_graphic() || byte.is_ascii_whitespace() { - print!("{}", byte as char); - } else { - print!("."); - } - - if i == 8 { - print!(" "); - } - } - - trace!(""); - } - } - #[allow(dead_code)] #[must_use] pub fn get_message_code_u32(&self) -> u32 { @@ -109,22 +68,10 @@ impl Message { self.data[4] } - #[must_use] - pub fn get_message_code_send(&self) -> u8 { - if self.data.is_empty() { - return 0; - } - self.data[0] - } - #[must_use] pub const fn new_with_data(data: Vec) -> Self { Self { data, pointer: 0 } } - #[allow(dead_code)] - pub const fn reset_pointer(&mut self) { - self.pointer = 0; - } pub const fn set_pointer(&mut self, pointer: usize) { self.pointer = pointer; } @@ -215,21 +162,6 @@ impl Message { val } - pub fn read_raw_byte(&mut self) -> Vec { - if self.pointer + 4 > self.data.len() { - return vec![]; - } - - let val = vec![ - 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 read_int32(&mut self) -> u32 { if self.pointer + 4 > self.data.len() { return 0; @@ -286,20 +218,6 @@ impl Message { self } - #[allow(dead_code)] - pub fn write_raw_hex_string(&mut self, val: &str) -> &mut Self { - let mut b = Vec::new(); - for i in (0..val.len()).step_by(2) { - let byte_str = &val[i..i + 2]; - let byte = - u8::from_str_radix(byte_str, 16).expect("Invalid hex string"); - b.push(byte); - } - self.data.extend_from_slice(&b); - self.pointer += b.len(); - self - } - pub fn get_message_name( &self, msg_type: MessageType, @@ -378,6 +296,18 @@ impl Message { } } +/// Build a received message the way the dispatcher hands one to a handler: +/// 8 bytes of frame header, then the fields, with the pointer parked at the +/// start of the payload. +#[cfg(test)] +pub fn framed(build: impl FnOnce(&mut Message)) -> Message { + let mut message = Message::new(); + message.write_raw_bytes(vec![0u8; 8]); + build(&mut message); + message.set_pointer(8); + message +} + #[test] fn test_get_buffer() { let message = Message::new_with_data( @@ -462,8 +392,6 @@ fn get_message_code_on_short_message_does_not_panic() { let msg = Message::new_with_data(vec![0, 0, 0, 0]); assert_eq!(msg.get_message_code(), 0); assert_eq!(msg.get_message_code_u32(), 0); - let empty = Message::new_with_data(vec![]); - assert_eq!(empty.get_message_code_send(), 0); } #[test] diff --git a/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs b/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs index de6485f..9dc0224 100644 --- a/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs +++ b/soulseek-rs-lib/src/message/peer/place_in_queue_request.rs @@ -25,14 +25,14 @@ impl MessageHandler for PlaceInQueueRequest { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn the_asked_about_filename_is_forwarded() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_string("@@share\\track.mp3"); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_string("@@share\\track.mp3"); + }); PlaceInQueueRequest.handle(&mut message, tx); match rx.try_recv() { diff --git a/soulseek-rs-lib/src/message/peer/shared_file_list.rs b/soulseek-rs-lib/src/message/peer/shared_file_list.rs index 51c5205..4136b7c 100644 --- a/soulseek-rs-lib/src/message/peer/shared_file_list.rs +++ b/soulseek-rs-lib/src/message/peer/shared_file_list.rs @@ -110,10 +110,9 @@ fn hostile_dir_count_does_not_hang() { // parse to empty promptly rather than looping into an OOM. let compressed = crate::utils::zlib::compress_stored(&u32::MAX.to_le_bytes()); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_raw_bytes(compressed); - message.set_pointer(8); + let mut message = crate::message::framed(|m| { + m.write_raw_bytes(compressed); + }); assert!(parse_shared_file_list(&mut message).is_empty()); } diff --git a/soulseek-rs-lib/src/message/peer/upload_failed.rs b/soulseek-rs-lib/src/message/peer/upload_failed.rs index 27f1ec9..1be9a4d 100644 --- a/soulseek-rs-lib/src/message/peer/upload_failed.rs +++ b/soulseek-rs-lib/src/message/peer/upload_failed.rs @@ -2,7 +2,6 @@ use crate::info; use crate::{ message::{Message, MessageHandler}, peer::PeerMessage, - types::UploadFailed, }; use std::sync::mpsc::Sender; @@ -12,7 +11,7 @@ impl MessageHandler for UploadFailedHandler { 46 } fn handle(&self, message: &mut Message, _sender: Sender) { - let upload_failed = UploadFailed::new_from_message(message); - info!("Upload failed for ${}", upload_failed.filename); + let filename = message.read_string(); + info!("Upload failed for {}", filename); } } diff --git a/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs b/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs index ef7b72e..2b7d3c5 100644 --- a/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs +++ b/soulseek-rs-lib/src/message/server/excluded_search_phrases.rs @@ -31,16 +31,16 @@ impl MessageHandler for ExcludedSearchPhrasesHandler { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn hostile_item_count_does_not_hang() { // item_count=u32::MAX with no phrases: the guard must make this return // promptly instead of looping ~4 billion times. let (tx, _rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_int32(u32::MAX); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_int32(u32::MAX); + }); ExcludedSearchPhrasesHandler.handle(&mut message, tx); } } diff --git a/soulseek-rs-lib/src/message/server/join_room.rs b/soulseek-rs-lib/src/message/server/join_room.rs index 67e31e0..b1fb385 100644 --- a/soulseek-rs-lib/src/message/server/join_room.rs +++ b/soulseek-rs-lib/src/message/server/join_room.rs @@ -33,17 +33,16 @@ impl MessageHandler for JoinRoomHandler { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn hostile_user_count_does_not_hang() { // room="" then user_count=u32::MAX with no usernames: must return // promptly instead of looping ~4 billion times. let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_string(""); - message.write_int32(u32::MAX); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_string("").write_int32(u32::MAX); + }); JoinRoomHandler.handle(&mut message, tx); match rx.try_recv() { @@ -57,13 +56,12 @@ mod tests { #[test] fn forwards_room_and_member_list() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_string("nicotine"); - message.write_int32(2); - message.write_string("alice"); - message.write_string("bob"); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_string("nicotine") + .write_int32(2) + .write_string("alice") + .write_string("bob"); + }); JoinRoomHandler.handle(&mut message, tx); match rx.try_recv() { diff --git a/soulseek-rs-lib/src/message/server/leave_room.rs b/soulseek-rs-lib/src/message/server/leave_room.rs index cc243db..5612ac5 100644 --- a/soulseek-rs-lib/src/message/server/leave_room.rs +++ b/soulseek-rs-lib/src/message/server/leave_room.rs @@ -20,14 +20,14 @@ impl MessageHandler for LeaveRoomHandler { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn forwards_left_room() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_string("jazz"); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_string("jazz"); + }); LeaveRoomHandler.handle(&mut message, tx); match rx.try_recv() { diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index c7da605..cb09a27 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -137,14 +137,6 @@ impl MessageFactory { .write_int32(port.into()) .clone() } - #[must_use] - pub fn build_watch_user(token: u32) -> Message { - Message::new() - .write_raw_bytes([5, 0, 0, 0, 0].to_vec()) - .write_int32(token) - .clone() - } - /// Ask the server (code 7) for a user's online status. #[must_use] pub fn build_get_user_status(username: &str) -> Message { @@ -280,15 +272,6 @@ fn test_build_get_user_stats() { assert_eq!(expect, message.get_data()); } -#[test] -fn test_build_watch_user() { - let token: u32 = 223; - let message = MessageFactory::build_watch_user(token); - let expect: Vec = [5, 0, 0, 0, 0, 223, 0, 0, 0].to_vec(); - - assert_eq!(expect, message.get_data()); -} - #[test] fn test_build_login_message() { let message = MessageFactory::build_login_message( diff --git a/soulseek-rs-lib/src/message/server/message_user.rs b/soulseek-rs-lib/src/message/server/message_user.rs index ef3cb5f..02f80a5 100644 --- a/soulseek-rs-lib/src/message/server/message_user.rs +++ b/soulseek-rs-lib/src/message/server/message_user.rs @@ -27,7 +27,6 @@ impl MessageHandler for MessageUser { ); info!("[MessageUser] User message received:{:?}", user_message); - user_message.print(); // Acknowledge freshly delivered messages so the server does not keep // re-sending them on every reconnect. diff --git a/soulseek-rs-lib/src/message/server/privileged_users.rs b/soulseek-rs-lib/src/message/server/privileged_users.rs index 8f0cc37..2dc042c 100644 --- a/soulseek-rs-lib/src/message/server/privileged_users.rs +++ b/soulseek-rs-lib/src/message/server/privileged_users.rs @@ -50,19 +50,12 @@ impl MessageHandler for CheckPrivilegesHandler { #[cfg(test)] mod tests { use super::*; - - fn payload(build: impl FnOnce(&mut Message)) -> Message { - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - build(&mut message); - message.set_pointer(8); - message - } + use crate::message::framed; #[test] fn the_user_list_is_kept_not_just_counted() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = payload(|m| { + let mut message = framed(|m| { m.write_int32(2) .write_string("donor_a") .write_string("donor_b"); @@ -80,7 +73,7 @@ mod tests { #[test] fn an_empty_list_is_still_an_answer() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = payload(|m| { + let mut message = framed(|m| { m.write_int32(0); }); @@ -98,7 +91,7 @@ mod tests { // count=u32::MAX with no names: the guard must return promptly rather // than loop four billion times. let (tx, _rx) = std::sync::mpsc::channel(); - let mut message = payload(|m| { + let mut message = framed(|m| { m.write_int32(u32::MAX); }); PrivilegedUsersHandler.handle(&mut message, tx); @@ -107,7 +100,7 @@ mod tests { #[test] fn our_own_remaining_privilege_time_is_reported() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = payload(|m| { + let mut message = framed(|m| { m.write_int32(86_400); }); diff --git a/soulseek-rs-lib/src/message/server/room_list.rs b/soulseek-rs-lib/src/message/server/room_list.rs index 5e779d8..8596606 100644 --- a/soulseek-rs-lib/src/message/server/room_list.rs +++ b/soulseek-rs-lib/src/message/server/room_list.rs @@ -53,16 +53,7 @@ pub fn parse_room_list(message: &mut Message) -> Vec { #[cfg(test)] mod tests { use super::*; - - /// Build a received message (8-byte header + payload) and position it the - /// way the dispatcher would. - fn framed(build: impl FnOnce(&mut Message)) -> Message { - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - build(&mut message); - message.set_pointer(8); - message - } + use crate::message::framed; #[test] fn parses_names_zipped_with_counts() { diff --git a/soulseek-rs-lib/src/message/server/say_chatroom.rs b/soulseek-rs-lib/src/message/server/say_chatroom.rs index 966fade..b95e9f3 100644 --- a/soulseek-rs-lib/src/message/server/say_chatroom.rs +++ b/soulseek-rs-lib/src/message/server/say_chatroom.rs @@ -26,16 +26,16 @@ impl MessageHandler for SayChatroomHandler { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn forwards_room_message() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_string("jazz"); - message.write_string("alice"); - message.write_string("hello everyone"); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_string("jazz") + .write_string("alice") + .write_string("hello everyone"); + }); SayChatroomHandler.handle(&mut message, tx); match rx.try_recv() { diff --git a/soulseek-rs-lib/src/message/server/user_info.rs b/soulseek-rs-lib/src/message/server/user_info.rs index 97ca290..cd8b7a9 100644 --- a/soulseek-rs-lib/src/message/server/user_info.rs +++ b/soulseek-rs-lib/src/message/server/user_info.rs @@ -56,21 +56,12 @@ impl MessageHandler for GetUserStatsHandler { #[cfg(test)] mod tests { use super::*; - - /// Build a payload the dispatcher would hand a handler: 8 bytes of frame - /// header, then the fields, with the pointer parked at the payload. - fn payload(build: impl FnOnce(&mut Message)) -> Message { - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - build(&mut message); - message.set_pointer(8); - message - } + use crate::message::framed; #[test] fn user_status_is_parsed_with_its_privileged_flag() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = payload(|m| { + let mut message = framed(|m| { m.write_string("alice").write_int32(2).write_int8(1); }); @@ -93,7 +84,7 @@ mod tests { fn user_stats_skip_the_two_obsolete_fields() { let (tx, rx) = std::sync::mpsc::channel(); // username, speed, obsolete, obsolete, files, folders - let mut message = payload(|m| { + let mut message = framed(|m| { m.write_string("bob") .write_int32(1024) .write_int32(0) diff --git a/soulseek-rs-lib/src/message/server/user_joined_room.rs b/soulseek-rs-lib/src/message/server/user_joined_room.rs index 07bd383..c7c785b 100644 --- a/soulseek-rs-lib/src/message/server/user_joined_room.rs +++ b/soulseek-rs-lib/src/message/server/user_joined_room.rs @@ -23,15 +23,14 @@ impl MessageHandler for UserJoinedRoomHandler { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn forwards_user_joined() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_string("jazz"); - message.write_string("carol"); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_string("jazz").write_string("carol"); + }); UserJoinedRoomHandler.handle(&mut message, tx); match rx.try_recv() { diff --git a/soulseek-rs-lib/src/message/server/user_left_room.rs b/soulseek-rs-lib/src/message/server/user_left_room.rs index 7350bdc..c5ff89b 100644 --- a/soulseek-rs-lib/src/message/server/user_left_room.rs +++ b/soulseek-rs-lib/src/message/server/user_left_room.rs @@ -21,15 +21,14 @@ impl MessageHandler for UserLeftRoomHandler { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn forwards_user_left() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_string("jazz"); - message.write_string("carol"); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_string("jazz").write_string("carol"); + }); UserLeftRoomHandler.handle(&mut message, tx); match rx.try_recv() { diff --git a/soulseek-rs-lib/src/message/server/wish_list_interval.rs b/soulseek-rs-lib/src/message/server/wish_list_interval.rs index 1b57c8c..2d5b06e 100644 --- a/soulseek-rs-lib/src/message/server/wish_list_interval.rs +++ b/soulseek-rs-lib/src/message/server/wish_list_interval.rs @@ -26,14 +26,14 @@ impl MessageHandler for WishListIntervalHandler { #[cfg(test)] mod tests { use super::*; + use crate::message::framed; #[test] fn the_announced_interval_is_reported_not_discarded() { let (tx, rx) = std::sync::mpsc::channel(); - let mut message = Message::new(); - message.write_raw_bytes(vec![0u8; 8]); - message.write_int32(720); - message.set_pointer(8); + let mut message = framed(|m| { + m.write_int32(720); + }); WishListIntervalHandler.handle(&mut message, tx); match rx.try_recv() { diff --git a/soulseek-rs-lib/src/peer/download_peer.rs b/soulseek-rs-lib/src/peer/download_peer.rs index be4805b..0b4821e 100644 --- a/soulseek-rs-lib/src/peer/download_peer.rs +++ b/soulseek-rs-lib/src/peer/download_peer.rs @@ -66,64 +66,42 @@ impl std::fmt::Display for DownloadError { impl std::error::Error for DownloadError {} -impl From for DownloadError { - fn from(error: io::Error) -> Self { - Self::StreamReadError(error) - } +fn extract_filename_from_path(full_path: &str) -> &str { + full_path + .split(['/', '\\']) + .next_back() + .unwrap_or(full_path) } -struct FileManager; +/// Where a download's bytes end up: its configured directory (falling back +/// to the parent if it names a file) joined with the remote basename. +fn resolve_download_path(download: &Download) -> Result { + let mut expanded_path = expand_tilde(&download.download_directory); -impl FileManager { - fn extract_filename_from_path(full_path: &str) -> &str { - full_path - .split(['/', '\\']) - .next_back() - .unwrap_or(full_path) - } - - fn create_download_path_from_filename( - output_directory: PathBuf, - filename: &str, - ) -> PathBuf { - let filename_only = Self::extract_filename_from_path(filename); - output_directory.join(filename_only) - } - - /// Where a download's bytes end up: its configured directory (falling back - /// to the parent if it names a file) joined with the remote basename. - fn resolve_download_path( - download: &Download, - ) -> Result { - let mut expanded_path = expand_tilde(&download.download_directory); - - if !expanded_path.is_dir() { - expanded_path = expanded_path - .parent() - .ok_or_else(|| { - DownloadError::PathResolutionError(format!( - "Cannot resolve parent directory for: {}", - expanded_path.display() - )) - })? - .to_path_buf(); - } - - let final_path = Self::create_download_path_from_filename( - expanded_path, - &download.filename, - ); - - final_path - .to_str() + if !expanded_path.is_dir() { + expanded_path = expanded_path + .parent() .ok_or_else(|| { DownloadError::PathResolutionError(format!( - "Path contains invalid UTF-8: {}", - final_path.display() + "Cannot resolve parent directory for: {}", + expanded_path.display() )) - }) - .map(String::from) + })? + .to_path_buf(); } + + let final_path = + expanded_path.join(extract_filename_from_path(&download.filename)); + + final_path + .to_str() + .ok_or_else(|| { + DownloadError::PathResolutionError(format!( + "Path contains invalid UTF-8: {}", + final_path.display() + )) + }) + .map(String::from) } /// The `.part` file a transfer streams into, so an interrupted transfer leaves @@ -137,7 +115,7 @@ struct PartFile { impl PartFile { fn open(download: &Download) -> Result { - let final_path = FileManager::resolve_download_path(download)?; + let final_path = resolve_download_path(download)?; let path = PathBuf::from(format!("{final_path}{PART_SUFFIX}")); if let Some(parent) = path.parent() { @@ -521,7 +499,9 @@ impl DownloadPeer { #[cfg(test)] mod tests { - use super::{DownloadError, DownloadPeer, FileManager, PartFile}; + use super::{ + DownloadError, DownloadPeer, PartFile, extract_filename_from_path, + }; use crate::types::{Download, DownloadMetadata, DownloadStatus}; use std::sync::mpsc; @@ -636,23 +616,17 @@ mod tests { #[test] fn test_extract_filename_from_path() { + assert_eq!(extract_filename_from_path("/path/to/file.mp3"), "file.mp3"); assert_eq!( - FileManager::extract_filename_from_path("/path/to/file.mp3"), + extract_filename_from_path("C:\\path\\to\\file.mp3"), "file.mp3" ); assert_eq!( - FileManager::extract_filename_from_path("C:\\path\\to\\file.mp3"), - "file.mp3" - ); - assert_eq!( - FileManager::extract_filename_from_path( + extract_filename_from_path( "@@bhfrv\\Soulseek Downloads\\complete\\Beatport Top Deep House (2021)\\michel test file.mp3" ), "michel test file.mp3" ); - assert_eq!( - FileManager::extract_filename_from_path("file.mp3"), - "file.mp3" - ); + assert_eq!(extract_filename_from_path("file.mp3"), "file.mp3"); } } diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 1407d93..e647ba1 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -77,16 +77,6 @@ fn parse_peer_init_message(mut message: Message) -> Option { }) } -fn parse_token_from_buffer(buffer: &[u8], username: &str) -> Option { - let token_bytes = buffer.get(0..4)?; - let token = u32::from_le_bytes( - token_bytes - .try_into() - .unwrap_or_else(|_| panic!("[listener:{username}] slice with incorrect length, can't extract transfer_token")), - ); - Some(token) -} - fn extract_download_from_buffer( reader: &mut MessageReader, client_context: &Arc>, @@ -98,7 +88,7 @@ fn extract_download_from_buffer( return None; } let buffer = reader.get_buffer(); - let token = parse_token_from_buffer(&buffer, username)?; + let token = u32::from_le_bytes(buffer.get(0..4)?.try_into().ok()?); trace!( "[listener:{}] got transfer_token: {} from data chunk", username, token @@ -129,8 +119,6 @@ fn handle_peer_connection( stream: TcpStream, reader: MessageReader, context: &ConnectionContext, - _peer_ip: &str, - _peer_port: u16, ) { // The peer actor multiplexes socket reads with its mailbox on a single // thread: `tick()` reads the socket, but outgoing messages (e.g. a queued @@ -283,7 +271,7 @@ fn handle_pierce_firewall( 0, 0, ); - handle_peer_connection(peer, stream, reader, context, peer_ip, peer_port); + handle_peer_connection(peer, stream, reader, context); // Inbound peers don't self-announce, so nudge the client to flush any // downloads queued for this now-connected peer. @@ -292,7 +280,10 @@ fn handle_pierce_firewall( .send(ClientOperation::PeerConnected(username)); } -fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { +fn handle_incoming_connection( + mut stream: TcpStream, + context: ConnectionContext, +) { let Ok(peer_addr) = stream.peer_addr() else { error!("[listener] failed to get peer address"); return; @@ -300,7 +291,6 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { let peer_ip = peer_addr.ip().to_string(); let peer_port = peer_addr.port(); - let mut stream = stream; let mut reader = MessageReader::new(); // A peer that dials and then goes away is routine, not an error. @@ -337,7 +327,6 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { let peer = Peer::new( format!("{}:direct", init_data.username), - // init_data.username.clone(), init_data.connection_type.clone(), peer_ip.clone(), peer_port.into(), @@ -348,9 +337,9 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { ); match init_data.connection_type { - ConnectionType::P => handle_peer_connection( - peer, stream, reader, &context, &peer_ip, peer_port, - ), + ConnectionType::P => { + handle_peer_connection(peer, stream, reader, &context); + } ConnectionType::F => handle_file_connection( peer, diff --git a/soulseek-rs-lib/src/peer/mod.rs b/soulseek-rs-lib/src/peer/mod.rs index 7d861cd..eb80909 100644 --- a/soulseek-rs-lib/src/peer/mod.rs +++ b/soulseek-rs-lib/src/peer/mod.rs @@ -10,33 +10,7 @@ pub use download_peer::DownloadPeer; use crate::message::Message; use core::fmt; -use std::{net::TcpStream, str::FromStr}; - -#[derive(Debug)] -#[allow(dead_code)] -pub struct NewPeer { - pub username: String, - pub connection_type: ConnectionType, - pub token: u32, - pub tcp_stream: TcpStream, -} -impl NewPeer { - pub fn new_from_message( - message: &mut Message, - tcp_stream: TcpStream, - ) -> Option { - let username = message.read_string(); - let connection_type = message.read_string().parse().ok()?; - let token = message.read_int32(); - - Some(Self { - username, - connection_type, - token, - tcp_stream, - }) - } -} +use std::str::FromStr; #[derive(Debug, Clone)] pub enum ConnectionType { @@ -81,7 +55,6 @@ impl fmt::Display for ConnectionType { } #[derive(Debug, Clone)] -#[allow(dead_code)] pub struct Peer { pub username: String, pub connection_type: ConnectionType, @@ -93,7 +66,7 @@ pub struct Peer { pub obfuscated_port: Option, } impl Peer { - #[allow(clippy::too_many_arguments, dead_code)] + #[allow(clippy::too_many_arguments)] #[must_use] pub const fn new( username: String, @@ -116,7 +89,6 @@ impl Peer { obfuscated_port: Some(obfuscated_port), } } - #[allow(dead_code)] pub fn new_from_message(message: &mut Message) -> Option { let username = message.read_string(); // The connection type is an untrusted string; an unknown value must not diff --git a/soulseek-rs-lib/src/shares/mod.rs b/soulseek-rs-lib/src/shares/mod.rs index 1314dd6..49fc9ad 100644 --- a/soulseek-rs-lib/src/shares/mod.rs +++ b/soulseek-rs-lib/src/shares/mod.rs @@ -44,15 +44,7 @@ impl Shares { // Fail fast if the root is unreadable; deeper failures are skipped. let _ = std::fs::read_dir(root)?; - let mut files = Vec::new(); - let mut folder_count = 0; - scan_root( - root, - &root_display_name(root), - &mut files, - &mut folder_count, - ); - Ok(Self::from_files(files, folder_count)) + Ok(Self::scan_many(&[root.to_path_buf()])) } /// Scan several roots into one merged index. Unreadable roots are diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index f873981..81b06aa 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -43,16 +43,6 @@ impl Default for ClientVersion { } } -pub struct UploadFailed { - pub filename: String, -} -impl UploadFailed { - pub fn new_from_message(message: &mut Message) -> Self { - let filename = message.read_string(); - - Self { filename } - } -} #[derive(Debug, Clone)] #[allow(dead_code)] pub struct SearchResult { diff --git a/soulseek-rs-lib/src/utils/logger.rs b/soulseek-rs-lib/src/utils/logger.rs index 03eaad7..208dd73 100644 --- a/soulseek-rs-lib/src/utils/logger.rs +++ b/soulseek-rs-lib/src/utils/logger.rs @@ -87,12 +87,12 @@ fn has_log_file() -> bool { pub fn log(level: LogLevel, message: &str) { unsafe { if level <= LOG_LEVEL { - let level_str = match level { - LogLevel::Error => "\x1b[31mERROR\x1b[0m", // Red - LogLevel::Warn => "\x1b[33mWARN\x1b[0m", // Yellow - LogLevel::Info => "\x1b[32mINFO\x1b[0m", // Green - LogLevel::Debug => "\x1b[34mDEBUG\x1b[0m", // Blue - LogLevel::Trace => "\x1b[35mTRACE\x1b[0m", // Magenta + let (name, colour) = match level { + LogLevel::Error => ("ERROR", "\x1b[31m"), // Red + LogLevel::Warn => ("WARN", "\x1b[33m"), // Yellow + LogLevel::Info => ("INFO", "\x1b[32m"), // Green + LogLevel::Debug => ("DEBUG", "\x1b[34m"), // Blue + LogLevel::Trace => ("TRACE", "\x1b[35m"), // Magenta }; let now = std::time::SystemTime::now(); @@ -149,20 +149,8 @@ pub fn log(level: LogLevel, message: &str) { let minutes = (seconds_in_day % 3600) / 60; let seconds = seconds_in_day % 60; - let level_str_plain = match level { - LogLevel::Error => "ERROR", - LogLevel::Warn => "WARN", - LogLevel::Info => "INFO", - LogLevel::Debug => "DEBUG", - LogLevel::Trace => "TRACE", - }; - - let formatted_message = format!( - "[{year:04}-{month:02}-{day:02} {hours:02}:{minutes:02}:{seconds:02}.{subsec_millis:03}] [{level_str}] {message}" - ); - - let formatted_message_plain = format!( - "[{year:04}-{month:02}-{day:02} {hours:02}:{minutes:02}:{seconds:02}.{subsec_millis:03}] [{level_str_plain}] {message}" + let timestamp = format!( + "{year:04}-{month:02}-{day:02} {hours:02}:{minutes:02}:{seconds:02}.{subsec_millis:03}" ); match choose_sink(BUFFERING.load(Ordering::Relaxed), has_log_file()) @@ -171,17 +159,22 @@ pub fn log(level: LogLevel, message: &str) { if let Ok(mut log_file) = LOG_FILE.lock() && let Some(file) = log_file.as_mut() { - let _ = writeln!(file, "{formatted_message_plain}"); + let _ = + writeln!(file, "[{timestamp}] [{name}] {message}"); let _ = file.flush(); } } LogSink::Buffer => { if let Ok(mut buffer) = BUFFER.lock() { - buffer.push(formatted_message); + buffer.push(format!( + "[{timestamp}] [{colour}{name}\x1b[0m] {message}" + )); } } LogSink::Stderr => { - eprintln!("{formatted_message}"); + eprintln!( + "[{timestamp}] [{colour}{name}\x1b[0m] {message}" + ); } } } @@ -196,58 +189,19 @@ pub fn disable_buffering() { BUFFERING.store(false, Ordering::Relaxed); } +/// Only lines that were buffered land here, and a line is buffered only when +/// no log file is configured, so the buffer is always stderr-bound. pub fn flush_buffered_logs() { disable_buffering(); if let Ok(mut buffer) = BUFFER.lock() { - // Write to file if configured, otherwise to stderr - if let Ok(mut log_file) = LOG_FILE.lock() { - if let Some(file) = log_file.as_mut() { - for message in buffer.iter() { - // Strip ANSI codes for file output - let plain_message = strip_ansi_codes(message); - let _ = writeln!(file, "{plain_message}"); - } - let _ = file.flush(); - } else { - for message in buffer.iter() { - eprintln!("{message}"); - } - } - } else { - // If lock fails, fall back to stderr - for message in buffer.iter() { - eprintln!("{message}"); - } + for message in buffer.iter() { + eprintln!("{message}"); } buffer.clear(); } } -fn strip_ansi_codes(s: &str) -> String { - let mut result = String::with_capacity(s.len()); - let mut chars = s.chars().peekable(); - - while let Some(c) = chars.next() { - if c == '\x1b' { - // Skip ANSI escape sequence - if chars.peek() == Some(&'[') { - chars.next(); // consume '[' - while let Some(&ch) = chars.peek() { - chars.next(); - if ch.is_ascii_alphabetic() { - break; - } - } - } - } else { - result.push(c); - } - } - - result -} - #[macro_export] macro_rules! error { ($($arg:tt)*) => { diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 8ca5519..a445c88 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -16,8 +16,6 @@ //! No external crates are used — the library forbids dependencies, so the //! harness sticks to `std` and the library's own public API. -#![allow(clippy::doc_markdown)] - use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::path::PathBuf; @@ -64,10 +62,11 @@ impl TestServer { .unwrap_or_else(std::sync::PoisonError::into_inner); if let Ok(addr) = std::env::var("SOULSEEK_TEST_SERVER") { - let (host, port) = split_host_port(&addr)?; - wait_until_listening(&host, port, Duration::from_secs(2))?; + let (host, port) = addr.rsplit_once(':')?; + let port = port.parse().ok()?; + wait_until_listening(host, port, Duration::from_secs(2))?; return Some(Self { - host, + host: host.to_string(), port, child: None, db: None, @@ -152,11 +151,6 @@ impl Drop for TestServer { } } -fn split_host_port(addr: &str) -> Option<(String, u16)> { - let (host, port) = addr.rsplit_once(':')?; - Some((host.to_string(), port.parse().ok()?)) -} - fn soulfind_binary() -> Option { if let Ok(p) = std::env::var("SOULFIND_BIN") { let p = PathBuf::from(p); @@ -465,22 +459,6 @@ fn registered_username_can_relogin_with_same_password() { ); } -#[test] -fn two_clients_can_be_logged_in_together() { - let server = server_or_skip!(); - - // The server must handle several independent sessions at once — this is - // the precondition for any peer-to-peer feature routed through it. - let mut alice = Client::with_settings(server.settings("e2e_alice", "pw_a")); - let mut bob = Client::with_settings(server.settings("e2e_bob", "pw_b")); - - alice.connect().expect("alice connect"); - bob.connect().expect("bob connect"); - - assert!(alice.login().expect("alice login")); - assert!(bob.login().expect("bob login")); -} - // --------------------------------------------------------------------------- // Peer-to-peer download coverage. // @@ -2149,7 +2127,6 @@ fn a_peer_asking_where_it_sits_is_answered_with_its_place() { ); // The asker keeps its own connection so it can read the answer back. - // The asker keeps its connection so it can read the answer back. let asker_file = format!("{folder}\\asker.mp3"); let mut asker = queue_as(&server_addr, &listen_addr, "e2e_q_asker", &asker_file) diff --git a/soulseek-rs/src/api.rs b/soulseek-rs/src/api.rs index 31cdac0..f3cc854 100644 --- a/soulseek-rs/src/api.rs +++ b/soulseek-rs/src/api.rs @@ -149,8 +149,11 @@ pub trait SessionApi: Send + Sync { Vec::new() } + // aislop-ignore-next-line complexity/function-too-long -- scanner bug: a bodyless trait signature followed only by more signatures has no closing brace, so the counter runs to EOF. One line. fn shared_counts(&self) -> (u32, u32); + // aislop-ignore-next-line complexity/function-too-long -- same scanner bug: bodyless signature measured to EOF. One line. fn shared_directories(&self) -> Vec; + // aislop-ignore-next-line complexity/function-too-long -- same scanner bug: bodyless signature measured to EOF. One line. fn set_shared_directories(&self, directories: Vec) -> Result<()>; /// Point future transfers at `directory`. @@ -158,6 +161,7 @@ pub trait SessionApi: Send + Sync { /// Only a daemon holds a download folder of its own — locally every /// download call names its directory, so the local session has nothing to /// store and accepts silently. + // aislop-ignore-next-line complexity/function-too-long -- same scanner bug: bodyless signature measured to EOF. One line. fn set_download_directory(&self, directory: String) -> Result<()>; } diff --git a/soulseek-rs/src/cli.rs b/soulseek-rs/src/cli.rs index e4628a2..7b3f90c 100644 --- a/soulseek-rs/src/cli.rs +++ b/soulseek-rs/src/cli.rs @@ -392,9 +392,10 @@ pub enum ConfigCommand { } /// How to order search results. -#[derive(Copy, Clone, Debug, PartialEq, Eq, ValueEnum)] +#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)] pub enum SortKey { /// Free slot first, then bitrate, peer speed, and size + #[default] Best, /// Largest file first Size, diff --git a/soulseek-rs/src/commands/completions.rs b/soulseek-rs/src/commands/completions.rs index 8dd2687..d6ca9ef 100644 --- a/soulseek-rs/src/commands/completions.rs +++ b/soulseek-rs/src/commands/completions.rs @@ -6,6 +6,7 @@ //! get one `source` line appended to their rc file, marked so `uninstall` can //! take back exactly what was added and nothing else. +use super::{unwritable, xdg}; use crate::cli::{Cli, CompletionsArgs, CompletionsCommand, Shell}; use crate::output::{CliError, CliResult, CompletionRecord, Out}; use clap::CommandFactory; @@ -39,7 +40,7 @@ struct Target { impl Target { fn record(&self, action: &str) -> CompletionRecord { CompletionRecord { - shell: self.shell.name().to_string(), + shell: self.shell.generator().to_string(), path: self.script.display().to_string(), action: action.to_string(), } @@ -51,14 +52,6 @@ impl Target { } impl Shell { - const fn name(self) -> &'static str { - match self { - Self::Bash => "bash", - Self::Zsh => "zsh", - Self::Fish => "fish", - } - } - const fn generator(self) -> clap_complete::Shell { match self { Self::Bash => clap_complete::Shell::Bash, @@ -213,14 +206,6 @@ impl Dirs { } } -/// Shells follow the XDG layout on every platform, unlike this client's own -/// config, which follows each platform's convention (see `persist::paths`). -fn xdg(variable: &str, home: &Path, fallback: &str) -> PathBuf { - std::env::var_os(variable) - .filter(|dir| !dir.is_empty()) - .map_or_else(|| home.join(fallback), PathBuf::from) -} - fn write_script(path: &Path, script: &[u8]) -> CliResult { if std::fs::read(path).is_ok_and(|found| found == script) { return Ok(false); @@ -288,10 +273,6 @@ fn create_parent(path: &Path) -> CliResult { } } -fn unwritable(path: &Path, error: &std::io::Error) -> CliError { - CliError::usage(format!("cannot write {}: {error}", path.display())) -} - #[cfg(test)] mod tests { use super::*; diff --git a/soulseek-rs/src/commands/mod.rs b/soulseek-rs/src/commands/mod.rs index 6394ab0..28161f7 100644 --- a/soulseek-rs/src/commands/mod.rs +++ b/soulseek-rs/src/commands/mod.rs @@ -22,6 +22,7 @@ use crate::remote::{Endpoint, RemoteSession}; use soulseek_rs::types::{RoomEvent, RoomInfo}; use soulseek_rs::{Client, ClientSettings, SessionLoss}; use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; +use std::path::{Path, PathBuf}; use std::sync::{Arc, OnceLock}; use std::time::{Duration, Instant}; @@ -223,6 +224,26 @@ fn login(settings: ClientSettings, address: &str) -> CliResult { } } +/// Ask `f` every `poll` until it answers or `timeout` runs out. +/// +/// The library reports what the network sent by handing it over on the next +/// call, so a command that wants one answer has to keep asking. Checking first +/// and sleeping after means an answer already waiting costs nothing. +fn poll_until( + timeout: Duration, + poll: Duration, + mut f: impl FnMut() -> Option, +) -> Option { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if let Some(answer) = f() { + return Some(answer); + } + std::thread::sleep(poll); + } + None +} + /// Ask for the room list and wait for the server's reply. /// /// Events already queued are discarded first: servers push a room list at @@ -235,21 +256,15 @@ pub fn await_room_list( let _ = client.take_room_events(); client.request_room_list().ok()?; - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - let listed = - client.take_room_events().into_iter().find_map( - |event| match event { - RoomEvent::List(rooms) => Some(rooms), - _ => None, - }, - ); - if listed.is_some() { - return listed; - } - std::thread::sleep(poll); - } - None + poll_until(timeout, poll, || { + client + .take_room_events() + .into_iter() + .find_map(|event| match event { + RoomEvent::List(rooms) => Some(rooms), + _ => None, + }) + }) } /// Wait until the server has answered something we asked for after `send`. @@ -295,12 +310,10 @@ fn dispatch(ctx: &Ctx, command: Commands) -> CliResult { } Commands::User(args) => peer::user(ctx, &args), Commands::Whoami => peer::whoami(ctx), - // These need no account, so main.rs answers them before asking for - // credentials; the arms exist only to keep the match exhaustive. - Commands::Portmap => portmap(&ctx.out, ctx.settings.listen_port), - // These need the config file as well as a session, so main.rs runs - // them where the store is in scope. - Commands::Serve(_) + // main.rs runs these itself: some need no account, the rest need the + // config store in scope. The arms only keep the match exhaustive. + Commands::Portmap + | Commands::Serve(_) | Commands::Daemon(_) | Commands::Wish(_) | Commands::Config(_) @@ -317,6 +330,19 @@ fn listen_span(duration: u64, follow: bool) -> Option { (!follow).then(|| Duration::from_secs(duration)) } +/// Where shells and coding agents keep their configuration: the XDG layout on +/// every platform, unlike this client's own config, which follows each +/// platform's convention (see `persist::paths`). +fn xdg(variable: &str, home: &Path, fallback: &str) -> PathBuf { + std::env::var_os(variable) + .filter(|dir| !dir.is_empty()) + .map_or_else(|| home.join(fallback), PathBuf::from) +} + +fn unwritable(path: &Path, error: &std::io::Error) -> CliError { + CliError::usage(format!("cannot write {}: {error}", path.display())) +} + /// Probe automatic port mapping. A router that does not answer is a negative /// answer, not a crash, so it exits 4 the way `grep` exits 1 on no match. pub fn portmap(out: &Out, port: u16) -> CliResult { diff --git a/soulseek-rs/src/commands/peer.rs b/soulseek-rs/src/commands/peer.rs index 9c06960..4101175 100644 --- a/soulseek-rs/src/commands/peer.rs +++ b/soulseek-rs/src/commands/peer.rs @@ -47,14 +47,9 @@ fn own_privileges(client: &dyn crate::api::SessionApi) -> Option { if client.check_privileges().is_err() { return None; } - let deadline = Instant::now() + PRIVILEGE_TIMEOUT; - while Instant::now() < deadline { - if let Some(seconds) = client.own_privilege_seconds() { - return Some(seconds); - } - std::thread::sleep(Duration::from_millis(50)); - } - None + super::poll_until(PRIVILEGE_TIMEOUT, Duration::from_millis(50), || { + client.own_privilege_seconds() + }) } pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult { @@ -141,7 +136,7 @@ pub fn serve( .warn("the share index is empty — peers will find nothing"); } - let span = (!args.follow).then(|| Duration::from_secs(args.duration)); + let span = super::listen_span(args.duration, args.follow); ctx.out.status(&match span { Some(span) => format!("online for {}s…", span.as_secs()), None => "online until interrupted…".to_string(), @@ -155,20 +150,10 @@ pub fn serve( // served well inside one poll, and "it had to wait" is precisely what a // transfer log exists to report. for queued in session.client.take_upload_events() { - let record = upload_record(&queued); - let key = (record.user.clone(), record.path.clone()); - if reported.get(&key) != Some(&record) { - ctx.out.emit(&record); - reported.insert(key, record); - } + emit_changed(&mut reported, upload_record(&queued), &ctx.out); } for upload in session.client.uploads() { - let record = upload_record(&upload); - let key = (record.user.clone(), record.path.clone()); - if reported.get(&key) != Some(&record) { - ctx.out.emit(&record); - reported.insert(key, record); - } + emit_changed(&mut reported, upload_record(&upload), &ctx.out); } if let Some(sweeper) = sweeper.as_mut() { sweeper.tick(ctx, session.client.as_ref()); @@ -211,8 +196,7 @@ fn upload_record(upload: &UploadInfo) -> UploadRecord { } /// Emit `record` only when it differs from the last one reported for that -/// (user, file). Exposed for tests; the serve loop inlines the same rule. -#[cfg(test)] +/// (user, file). fn emit_changed( reported: &mut HashMap<(String, String), UploadRecord>, record: UploadRecord, diff --git a/soulseek-rs/src/commands/skills.rs b/soulseek-rs/src/commands/skills.rs index 9d9794f..cef2b4c 100644 --- a/soulseek-rs/src/commands/skills.rs +++ b/soulseek-rs/src/commands/skills.rs @@ -4,6 +4,7 @@ //! inside it and is written out on demand. Installing over an older copy is //! what updating means, which is why there is no separate verb for it. +use super::{unwritable, xdg}; use crate::cli::{SkillsArgs, SkillsCommand}; use crate::output::{CliError, CliResult, Out, SkillRecord}; use directories::UserDirs; @@ -122,7 +123,7 @@ fn targets(args: &SkillsArgs) -> CliResult> { ) })?; - let found = detected(&home, &config_home(&home)); + let found = detected(&home, &xdg("XDG_CONFIG_HOME", &home, ".config")); if found.is_empty() { return Err(CliError::no_results( "no coding agent found in your home directory; name a skills \ @@ -149,15 +150,6 @@ fn detected(home: &Path, config_home: &Path) -> Vec { .collect() } -/// Where an agent keeps its own configuration, which is the XDG location on -/// every platform — unlike this client's config, which follows each platform's -/// own convention (see `persist::paths`). -fn config_home(home: &Path) -> PathBuf { - std::env::var_os("XDG_CONFIG_HOME") - .filter(|dir| !dir.is_empty()) - .map_or_else(|| home.join(".config"), PathBuf::from) -} - fn current(file: &Path) -> bool { std::fs::read_to_string(file).is_ok_and(|found| found == SKILL) } @@ -167,10 +159,6 @@ fn ours(file: &Path) -> bool { .is_ok_and(|found| found.contains(FRONTMATTER_NAME)) } -fn unwritable(path: &Path, error: &std::io::Error) -> CliError { - CliError::usage(format!("cannot write {}: {error}", path.display())) -} - #[cfg(test)] mod tests { use super::*; diff --git a/soulseek-rs/src/commands/social.rs b/soulseek-rs/src/commands/social.rs index c348866..1bdde18 100644 --- a/soulseek-rs/src/commands/social.rs +++ b/soulseek-rs/src/commands/social.rs @@ -39,17 +39,13 @@ pub fn fetch_listing( CliError::connection(format!("cannot ask {user} for a listing: {e}")) })?; - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - if let Some(directories) = client.take_browse_result(user) { - return Ok(directories); - } - std::thread::sleep(POLL); - } - Err(CliError::timeout(format!( - "{user} did not send a file list within {}s", - timeout.as_secs() - ))) + super::poll_until(timeout, POLL, || client.take_browse_result(user)) + .ok_or_else(|| { + CliError::timeout(format!( + "{user} did not send a file list within {}s", + timeout.as_secs() + )) + }) } pub fn browse(ctx: &Ctx, args: &BrowseArgs) -> CliResult { @@ -143,22 +139,18 @@ pub fn room_users(ctx: &Ctx, room: &str, timeout: Duration) -> CliResult { ctx.out .status(&format!("waiting for {room}'s member list…")); - let deadline = Instant::now() + timeout; - let mut members = Vec::new(); - while Instant::now() < deadline && members.is_empty() { - members = session.client.room_members(room); - if members.is_empty() { - std::thread::sleep(POLL); - } - } + let listed = super::poll_until(timeout, POLL, || { + let members = session.client.room_members(room); + (!members.is_empty()).then_some(members) + }); let _ = session.client.leave_room(room); - if members.is_empty() { + let Some(members) = listed else { return Err(CliError::timeout(format!( "{room} sent no member list within {}s", timeout.as_secs() ))); - } + }; for user in members { ctx.out.emit(&RoomMemberRecord { room: room.to_string(), diff --git a/soulseek-rs/src/commands/transfer.rs b/soulseek-rs/src/commands/transfer.rs index 62e0a40..2a46b5d 100644 --- a/soulseek-rs/src/commands/transfer.rs +++ b/soulseek-rs/src/commands/transfer.rs @@ -310,13 +310,9 @@ fn collect( }) }; - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - if client.get_search_results_count(query) > 0 { - break; - } - std::thread::sleep(Duration::from_millis(100)); - } + super::poll_until(timeout, Duration::from_millis(100), || { + (client.get_search_results_count(query) > 0).then_some(()) + }); cancel.store(true, Ordering::Relaxed); let _ = worker.join(); Ok(client.get_search_results(query)) @@ -334,29 +330,26 @@ fn dedupe( download_dir: &str, out: &Out, ) -> Vec { - let mut paths: HashMap = HashMap::new(); - let mut targets: HashMap = HashMap::new(); + let mut targets: HashMap = HashMap::new(); let mut kept = Vec::new(); for request in requests { - if let Some(user) = paths.get(&request.path) { - out.warn(&format!( - "skipping {} from {}: same remote path already queued from \ - {user}", - request.path, request.user - )); - continue; - } let target = local_path(download_dir, &request.path); - if let Some(other) = targets.get(&target) { + if let Some(queued) = targets.get(&target) { + let collision = if queued.path == request.path { + format!("same remote path already queued from {}", queued.user) + } else { + format!( + "would overwrite the file already queued from {}", + queued.user + ) + }; out.warn(&format!( - "skipping {} from {}: would overwrite the file already queued \ - from {other}", + "skipping {} from {}: {collision}", request.path, request.user )); continue; } - paths.insert(request.path.clone(), request.user.clone()); - targets.insert(target, request.user.clone()); + targets.insert(target, request.clone()); kept.push(request); } kept @@ -617,9 +610,8 @@ fn read_requests( } /// `Ok(None)` for a blank line, `Err` for something unusable. -// aislop-ignore-next-line complexity/function-too-long -- scanner bug, not a long function: the `'{'` char literal below reads as an opening brace, so aislop's counter never closes and measures to EOF. It is 34 lines. +// aislop-ignore-next-line complexity/function-too-long -- scanner bug, not a long function: the `'{'` char literal below reads as an opening brace, so aislop's counter never closes and measures to EOF. It is 33 lines. fn parse_request(line: &str) -> Result, String> { - let line = line.trim_end_matches(['\r', '\n']); if line.trim().is_empty() { return Ok(None); } diff --git a/soulseek-rs/src/commands/wish.rs b/soulseek-rs/src/commands/wish.rs index 7da9488..94ab351 100644 --- a/soulseek-rs/src/commands/wish.rs +++ b/soulseek-rs/src/commands/wish.rs @@ -68,22 +68,13 @@ pub fn run(ctx: &Ctx, store: &Store, args: &WishRunArgs) -> CliResult { /// How to narrow and order what a sweep prints — the parts of `wish run`'s flags /// that outlive the command, so `serve --wishlist` can sweep without inventing /// a set of command-line arguments it never received. +#[derive(Default)] struct Sweep { filter: transfer::Filter, sort: SortKey, limit: usize, } -impl Default for Sweep { - fn default() -> Self { - Self { - filter: transfer::Filter::default(), - sort: SortKey::Best, - limit: 0, - } - } -} - impl From<&WishRunArgs> for Sweep { fn from(args: &WishRunArgs) -> Self { Self { diff --git a/soulseek-rs/src/daemon/hub.rs b/soulseek-rs/src/daemon/hub.rs index 2366f7f..83dc43b 100644 --- a/soulseek-rs/src/daemon/hub.rs +++ b/soulseek-rs/src/daemon/hub.rs @@ -193,16 +193,14 @@ const BROWSE_PATIENCE: Duration = Duration::from_mins(2); /// `take_browse_result` is per-user and destructive, so the daemon has to know /// who to ask about. #[derive(Default)] -pub struct PendingBrowses(Mutex>); +pub struct PendingBrowses(Mutex>); impl PendingBrowses { pub fn expect(&self, username: &str) { - let Ok(mut pending) = self.0.lock() else { - return; - }; - // Asking again restarts the clock: it is a fresh request. - pending.retain(|(name, _)| name != username); - pending.push((username.to_string(), Instant::now())); + if let Ok(mut pending) = self.0.lock() { + // Asking again restarts the clock: it is a fresh request. + pending.insert(username.to_string(), Instant::now()); + } } /// Who to ask about this tick, forgetting the ones that have run out of @@ -211,15 +209,15 @@ impl PendingBrowses { let Ok(mut pending) = self.0.lock() else { return Vec::new(); }; - pending.retain(|(_, asked)| asked.elapsed() < BROWSE_PATIENCE); - pending.iter().map(|(name, _)| name.clone()).collect() + pending.retain(|_, asked| asked.elapsed() < BROWSE_PATIENCE); + pending.keys().cloned().collect() } /// Stop waiting on a peer — because it answered, or because its request /// never reached the wire. pub fn forget(&self, username: &str) { if let Ok(mut pending) = self.0.lock() { - pending.retain(|(name, _)| name != username); + pending.remove(username); } } } @@ -511,8 +509,10 @@ mod tests { let asked = Instant::now() .checked_sub(BROWSE_PATIENCE + Duration::from_secs(1)) .expect("the clock is past the epoch"); - let pending = - PendingBrowses(Mutex::new(vec![("ghost".to_string(), asked)])); + let pending = PendingBrowses(Mutex::new(HashMap::from([( + "ghost".to_string(), + asked, + )]))); assert!(pending.snapshot().is_empty()); } } diff --git a/soulseek-rs/src/daemon/mod.rs b/soulseek-rs/src/daemon/mod.rs index ebd487d..6a9641a 100644 --- a/soulseek-rs/src/daemon/mod.rs +++ b/soulseek-rs/src/daemon/mod.rs @@ -332,48 +332,42 @@ const fn raise_open_file_limit() -> std::io::Result { /// afterwards is a poor first experience. stdout stays empty either way, so /// `daemon` still composes in a pipeline. fn ready(ctx: &Ctx, bind: Option<&str>, token: &str) { - ctx.out.status(""); - ctx.out.status("ready — leave this running."); - ctx.out.status(""); - ctx.out - .status(" On this machine, nothing else to do. Other commands find"); - ctx.out - .status(" it on their own and share this one login:"); - ctx.out.status(""); - ctx.out.status(" soulseek-rs search 'aphex twin'"); - ctx.out - .status(" soulseek-rs (the TUI attaches)"); - ctx.out.status(" soulseek-rs daemon status"); - ctx.out.status(" soulseek-rs daemon stop"); - ctx.out.status(""); + const BANNER: &str = " +ready — leave this running. + + On this machine, nothing else to do. Other commands find + it on their own and share this one login: + + soulseek-rs search 'aphex twin' + soulseek-rs (the TUI attaches) + soulseek-rs daemon status + soulseek-rs daemon stop +"; + const LOCAL_ONLY: &str = + " To drive it from another machine, restart with --bind ADDR; + `soulseek-rs daemon token` prints the secret it needs. +"; + + ctx.out.status(BANNER); let Some(address) = bind else { - ctx.out.status( - " To drive it from another machine, restart with --bind ADDR;", - ); - ctx.out - .status(" `soulseek-rs daemon token` prints the secret it needs."); - ctx.out.status(""); + ctx.out.status(LOCAL_ONLY); return; }; ctx.out.status(&format!( - " From another machine, point a client at {address} with" + " From another machine, point a client at {address} with + this token: + + {token} + + soulseek-rs config set daemon :PORT + soulseek-rs config set daemon_token + + That port has no encryption. Keep it off the open internet and \ + reach it over SSH instead. +" )); - ctx.out.status(" this token:"); - ctx.out.status(""); - ctx.out.status(&format!(" {token}")); - ctx.out.status(""); - ctx.out - .status(" soulseek-rs config set daemon :PORT"); - ctx.out - .status(" soulseek-rs config set daemon_token "); - ctx.out.status(""); - ctx.out.status( - " That port has no encryption. Keep it off the open internet and \ - reach it over SSH instead.", - ); - ctx.out.status(""); } /// Remove the socket file so the next start does not have to reason about diff --git a/soulseek-rs/src/daemon/openrpc.rs b/soulseek-rs/src/daemon/openrpc.rs index 30e8d56..f965e58 100644 --- a/soulseek-rs/src/daemon/openrpc.rs +++ b/soulseek-rs/src/daemon/openrpc.rs @@ -23,11 +23,6 @@ use serde_json::{Value, json}; /// `components/schemas`, the same place OpenAPI does. const DEFINITIONS_PATH: &str = "#/components/schemas/"; -/// A method that takes no parameters. -fn nothing() -> Value { - json!({ "type": "null" }) -} - fn schema_of(generator: &mut SchemaGenerator) -> Value { generator.subschema_for::().to_value() } @@ -60,6 +55,7 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value { schema_of::(generator) } Method::SharesSet => schema_of::(generator), + // Methods that take no parameters. Method::DaemonStatus | Method::DaemonStop | Method::RpcDiscover @@ -72,7 +68,7 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value { | Method::MessageHistory | Method::SearchList | Method::SharesStatusOf - | Method::SharesReindex => nothing(), + | Method::SharesReindex => json!({ "type": "null" }), } } diff --git a/soulseek-rs/src/daemon/server/mod.rs b/soulseek-rs/src/daemon/server/mod.rs index 9e3e292..718a23a 100644 --- a/soulseek-rs/src/daemon/server/mod.rs +++ b/soulseek-rs/src/daemon/server/mod.rs @@ -52,7 +52,6 @@ impl Daemon { /// /// The match is exhaustive over [`Method`], so a new method cannot be added /// to the protocol without being implemented here. - #[allow(clippy::too_many_lines)] pub fn call( &self, method: Method, diff --git a/soulseek-rs/src/directories.rs b/soulseek-rs/src/directories.rs index be3edb1..28375bc 100644 --- a/soulseek-rs/src/directories.rs +++ b/soulseek-rs/src/directories.rs @@ -37,7 +37,6 @@ pub fn resolve_shared_directory( /// Validate several configured share paths; invalid ones are skipped with a /// warning instead of aborting (the client just shares less). -#[cfg_attr(not(test), allow(dead_code))] #[must_use] pub fn resolve_shared_directories(dirs: &[String]) -> Vec { let mut resolved = Vec::new(); diff --git a/soulseek-rs/src/models/app_state.rs b/soulseek-rs/src/models/app_state.rs index 0256b47..295bab7 100644 --- a/soulseek-rs/src/models/app_state.rs +++ b/soulseek-rs/src/models/app_state.rs @@ -24,7 +24,6 @@ pub struct SearchEntry { /// finishes; for the rest the session is the authority. pub owned: bool, pub start_time: Instant, - #[allow(dead_code)] pub cancel_flag: Arc, } @@ -249,19 +248,6 @@ impl AppState { let peer = peers[next].to_string(); self.chat_peer = Some(peer); } - - #[allow(dead_code)] - #[must_use] - pub fn get_selected_search(&self) -> Option<&SearchEntry> { - self.selected_search_index - .and_then(|idx| self.searches.get(idx)) - } - - #[allow(dead_code)] - pub fn get_selected_search_mut(&mut self) -> Option<&mut SearchEntry> { - self.selected_search_index - .and_then(|idx| self.searches.get_mut(idx)) - } } impl Default for AppState { diff --git a/soulseek-rs/src/output.rs b/soulseek-rs/src/output.rs index c2f8fed..d50a853 100644 --- a/soulseek-rs/src/output.rs +++ b/soulseek-rs/src/output.rs @@ -436,12 +436,11 @@ pub struct ConfigRecord { impl Record for ConfigRecord { fn text(&self) -> String { - match &self.value { - Some(value) => { - format!("{}\t{}", sanitize(&self.key), sanitize(value)) - } - None => format!("{}\t", sanitize(&self.key)), - } + format!( + "{}\t{}", + sanitize(&self.key), + sanitize(self.value.as_deref().unwrap_or_default()) + ) } } diff --git a/soulseek-rs/src/persist/state.rs b/soulseek-rs/src/persist/state.rs index ab4d3b1..91836ef 100644 --- a/soulseek-rs/src/persist/state.rs +++ b/soulseek-rs/src/persist/state.rs @@ -1,20 +1,19 @@ //! Versioned JSON state files (downloads, search queries, open rooms, //! private-message history). //! -//! Each file is an envelope `{ "version": N, "data": ... }`. On load the -//! data passes through the migration chain from its stored version up to -//! current. State is disposable: a missing, corrupt, or newer-than-known -//! file loads as empty rather than failing startup. Writes are atomic -//! (tmp file + rename) so a crash never leaves a torn file. +//! Each file is an envelope `{ "version": N, "data": ... }`. State is +//! disposable: a missing, corrupt, or newer-than-known file loads as empty +//! rather than failing startup. Writes are atomic (tmp file + rename) so a +//! crash never leaves a torn file. use color_eyre::Result; use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; use std::path::{Path, PathBuf}; -/// A migration takes the `data` value at version `i` and returns it at -/// version `i + 1`. -type Migration = fn(Value) -> Value; +/// The shape every file is written in today. Bump it — and convert the older +/// shapes in `load` — when one of them changes. +const VERSION: u32 = 0; #[derive(Debug, Clone, PartialEq, Eq, Serialize, serde::Deserialize)] pub struct PersistedDownload { @@ -45,86 +44,59 @@ impl StateStore { Self { dir } } + #[must_use] pub fn load_downloads(&self) -> Vec { - load(&self.dir.join("downloads.json"), DOWNLOADS_MIGRATIONS) + load(&self.dir.join("downloads.json")) } pub fn save_downloads( &self, downloads: &[PersistedDownload], ) -> Result<()> { - save( - &self.dir.join("downloads.json"), - DOWNLOADS_MIGRATIONS.len() as u32, - &downloads, - ) + save(&self.dir.join("downloads.json"), VERSION, &downloads) } + #[must_use] pub fn load_search_queries(&self) -> Vec { - load(&self.dir.join("searches.json"), SEARCHES_MIGRATIONS) + load(&self.dir.join("searches.json")) } pub fn save_search_queries(&self, queries: &[String]) -> Result<()> { - save( - &self.dir.join("searches.json"), - SEARCHES_MIGRATIONS.len() as u32, - &queries, - ) + save(&self.dir.join("searches.json"), VERSION, &queries) } + #[must_use] pub fn load_rooms(&self) -> Vec { - load(&self.dir.join("rooms.json"), ROOMS_MIGRATIONS) + load(&self.dir.join("rooms.json")) } pub fn save_rooms(&self, rooms: &[String]) -> Result<()> { - save( - &self.dir.join("rooms.json"), - ROOMS_MIGRATIONS.len() as u32, - &rooms, - ) + save(&self.dir.join("rooms.json"), VERSION, &rooms) } + #[must_use] pub fn load_messages(&self) -> Vec { - load(&self.dir.join("messages.json"), MESSAGES_MIGRATIONS) + load(&self.dir.join("messages.json")) } pub fn save_messages(&self, messages: &[PersistedMessage]) -> Result<()> { - save( - &self.dir.join("messages.json"), - MESSAGES_MIGRATIONS.len() as u32, - &messages, - ) + save(&self.dir.join("messages.json"), VERSION, &messages) } /// The unread private-message count, so the badge survives a restart. + #[must_use] pub fn load_unread(&self) -> usize { - load(&self.dir.join("unread.json"), UNREAD_MIGRATIONS) + load(&self.dir.join("unread.json")) } pub fn save_unread(&self, unread: usize) -> Result<()> { - save( - &self.dir.join("unread.json"), - UNREAD_MIGRATIONS.len() as u32, - &unread, - ) + save(&self.dir.join("unread.json"), VERSION, &unread) } } -/// Per-file migration chains. `data` at version `i` is upgraded by -/// `MIGRATIONS[i]`; the current version is the chain length. All formats -/// are at version 0 today — add a fn here when the schema changes. -const DOWNLOADS_MIGRATIONS: &[Migration] = &[]; -const SEARCHES_MIGRATIONS: &[Migration] = &[]; -const ROOMS_MIGRATIONS: &[Migration] = &[]; -const MESSAGES_MIGRATIONS: &[Migration] = &[]; -const UNREAD_MIGRATIONS: &[Migration] = &[]; - -/// Load `data` from an envelope file, migrating old versions forward. -/// Missing, corrupt, or newer-than-known files all yield `T::default()`. -fn load( - path: &Path, - migrations: &[Migration], -) -> T { +/// Load `data` from an envelope file. Missing, corrupt, or newer-than-known +/// files all yield `T::default()`. +fn load(path: &Path) -> T { let Ok(text) = std::fs::read_to_string(path) else { return T::default(); }; @@ -135,15 +107,12 @@ fn load( let version = envelope .get("version") .and_then(Value::as_u64) - .unwrap_or(u64::MAX) as usize; - if version > migrations.len() { + .unwrap_or(u64::MAX); + if version > u64::from(VERSION) { set_aside(path, "from a newer version"); return T::default(); } - let mut data = envelope.get("data").cloned().unwrap_or(Value::Null); - for migration in &migrations[version..] { - data = migration(data); - } + let data = envelope.get("data").cloned().unwrap_or(Value::Null); serde_json::from_value(data).unwrap_or_else(|_| { set_aside(path, "unreadable"); T::default() @@ -297,34 +266,6 @@ mod tests { assert!(path.with_extension("json.bak").exists()); } - #[test] - fn migrations_upgrade_old_data() { - // Machinery test with a synthetic chain: v0 stored plain strings, - // v1 wraps each in an object {"name": ...}. - #[derive(Debug, PartialEq, Default, Serialize, serde::Deserialize)] - struct Named { - name: String, - } - let chain: &[Migration] = &[|data| { - Value::Array( - data.as_array() - .cloned() - .unwrap_or_default() - .into_iter() - .map(|s| serde_json::json!({ "name": s })) - .collect(), - ) - }]; - let tmp = tempfile::tempdir().unwrap(); - let path = tmp.path().join("things.json"); - std::fs::write(&path, r#"{"version": 0, "data": ["a", "b"]}"#).unwrap(); - let loaded: Vec = load(&path, chain); - assert_eq!( - loaded, - vec![Named { name: "a".into() }, Named { name: "b".into() }] - ); - } - #[test] fn save_is_atomic_no_tmp_file_left_behind() { let (tmp, store) = store(); diff --git a/soulseek-rs/src/port_mapping/nat_pmp.rs b/soulseek-rs/src/port_mapping/nat_pmp.rs index 8c4f10b..1fb0f5f 100644 --- a/soulseek-rs/src/port_mapping/nat_pmp.rs +++ b/soulseek-rs/src/port_mapping/nat_pmp.rs @@ -52,8 +52,6 @@ impl From for NatPmpError { /// a shorter lifetime than requested, so callers must use these values. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct MapResponse { - /// Seconds since the gateway's port table was initialised (reboot detector). - pub epoch: u32, pub internal_port: u16, pub external_port: u16, pub lifetime: u32, @@ -87,12 +85,6 @@ pub fn map_tcp( return Err(NatPmpError::ShortResponse); } Ok(MapResponse { - epoch: u32::from_be_bytes([ - response[4], - response[5], - response[6], - response[7], - ]), internal_port: u16::from_be_bytes([response[8], response[9]]), external_port: u16::from_be_bytes([response[10], response[11]]), lifetime: u32::from_be_bytes([ diff --git a/soulseek-rs/src/run.rs b/soulseek-rs/src/run.rs index bb162ee..8727af3 100644 --- a/soulseek-rs/src/run.rs +++ b/soulseek-rs/src/run.rs @@ -369,6 +369,19 @@ pub fn init_logging(cli: &Cli) { } } +/// Clear the screen, turn on mouse reporting and take over the terminal. +fn enter_terminal() -> ratatui::DefaultTerminal { + use ratatui::crossterm::{ + event::EnableMouseCapture, + execute, + terminal::{Clear, ClearType}, + }; + + let _ = + execute!(std::io::stdout(), Clear(ClearType::All), EnableMouseCapture); + ratatui::init() +} + /// Run the interactive TUI (the default no-subcommand path): bring the /// terminal up first, run the login/registration screen (skipped past when /// stored credentials work), persist whatever logged in, then enter the @@ -379,12 +392,6 @@ fn run_default_tui( config_path: Option, file_config: &crate::persist::config::FileConfig, ) -> CliResult { - use ratatui::crossterm::{ - event::EnableMouseCapture, - execute, - terminal::{Clear, ClearType}, - }; - let out = Out::new(false, false); let (server_host, server_port) = parse_server_address(&resolved.server).map_err(CliError::usage)?; @@ -419,10 +426,7 @@ fn run_default_tui( version: ClientVersion::REFERENCE_CLIENT, }; - // Clear screen and enable mouse capture before initializing TUI - let _ = - execute!(std::io::stdout(), Clear(ClearType::All), EnableMouseCapture); - let mut terminal = ratatui::init(); + let mut terminal = enter_terminal(); let outcome = crate::ui::login::run_login_flow( &mut terminal, @@ -468,7 +472,6 @@ fn run_default_tui( terminal, Arc::new(outcome.client), resolved.download_dir.clone(), - resolved.max_concurrent_downloads, Duration::from_secs(resolved.search_timeout), store, config_path, @@ -486,12 +489,6 @@ fn run_attached_tui( endpoint: &crate::remote::Endpoint, config_path: Option, ) -> CliResult { - use ratatui::crossterm::{ - event::EnableMouseCapture, - execute, - terminal::{Clear, ClearType}, - }; - // The resolved token, not the flag alone: `daemon_token` in config.toml // is the documented remote setup, and the flag/env are layered into it. let token = resolved @@ -521,9 +518,7 @@ fn run_attached_tui( })?; soulseek_rs::utils::logger::enable_buffering(); - let _ = - execute!(std::io::stdout(), Clear(ClearType::All), EnableMouseCapture); - let terminal = ratatui::init(); + let terminal = enter_terminal(); // No StateStore: the daemon owns what survives a restart, so a second // writer here would fight it for the same files. @@ -531,7 +526,6 @@ fn run_attached_tui( terminal, Arc::new(session), download_dir, - resolved.max_concurrent_downloads, Duration::from_secs(resolved.search_timeout), None, config_path, diff --git a/soulseek-rs/src/ui/main_tui/input.rs b/soulseek-rs/src/ui/main_tui/input.rs index 109a71d..82333ec 100644 --- a/soulseek-rs/src/ui/main_tui/input.rs +++ b/soulseek-rs/src/ui/main_tui/input.rs @@ -3,6 +3,8 @@ use crate::models::{CommandBarMode, FocusedPane}; use ratatui::crossterm::event::{ KeyCode, KeyEvent, KeyModifiers, MouseButton, MouseEvent, MouseEventKind, }; +use ratatui::layout::Position; +use ratatui::widgets::TableState; impl MainTui { pub(super) fn handle_key_event(&mut self, key: KeyEvent) { @@ -189,10 +191,10 @@ impl MainTui { KeyCode::Backspace => { let cursor_position = self.state.command_bar_cursor_position; if cursor_position > 0 { - let previous_position = previous_char_boundary( - &self.state.command_bar_input, - cursor_position, - ); + let previous_position = self + .state + .command_bar_input + .floor_char_boundary(cursor_position.saturating_sub(1)); self.state .command_bar_input .drain(previous_position..cursor_position); @@ -202,26 +204,26 @@ impl MainTui { KeyCode::Delete => { let cursor_position = self.state.command_bar_cursor_position; if cursor_position < self.state.command_bar_input.len() { - let next_position = next_char_boundary( - &self.state.command_bar_input, - cursor_position, - ); + let next_position = self + .state + .command_bar_input + .ceil_char_boundary(cursor_position + 1); self.state .command_bar_input .drain(cursor_position..next_position); } } KeyCode::Left => { - self.state.command_bar_cursor_position = previous_char_boundary( - &self.state.command_bar_input, - self.state.command_bar_cursor_position, - ); + let previous = + self.state.command_bar_cursor_position.saturating_sub(1); + self.state.command_bar_cursor_position = + self.state.command_bar_input.floor_char_boundary(previous); } KeyCode::Right => { - self.state.command_bar_cursor_position = next_char_boundary( - &self.state.command_bar_input, - self.state.command_bar_cursor_position, - ); + let input = &self.state.command_bar_input; + let next = self.state.command_bar_cursor_position + 1; + self.state.command_bar_cursor_position = + input.ceil_char_boundary(next.min(input.len())); } KeyCode::Home => { self.state.command_bar_cursor_position = 0; @@ -288,26 +290,13 @@ impl MainTui { } fn handle_searches_input(&mut self, key: KeyEvent) { + let rows = self.state.searches.len(); match key.code { - KeyCode::Up | KeyCode::Char('k') - if !self.state.searches.is_empty() => - { - let current = - self.state.searches_table_state.selected().unwrap_or(0); - let new = if current == 0 { - self.state.searches.len() - 1 - } else { - current - 1 - }; - self.state.searches_table_state.select(Some(new)); + KeyCode::Up | KeyCode::Char('k') => { + cycle(&mut self.state.searches_table_state, rows, false); } - KeyCode::Down | KeyCode::Char('j') - if !self.state.searches.is_empty() => - { - let current = - self.state.searches_table_state.selected().unwrap_or(0); - let new = (current + 1) % self.state.searches.len(); - self.state.searches_table_state.select(Some(new)); + KeyCode::Down | KeyCode::Char('j') => { + cycle(&mut self.state.searches_table_state, rows, true); } KeyCode::Enter => { if let Some(selected) = @@ -348,21 +337,11 @@ impl MainTui { }; match key.code { - KeyCode::Up | KeyCode::Char('k') if items_count > 0 => { - let current = - self.state.results_table_state.selected().unwrap_or(0); - let new = if current == 0 { - items_count - 1 - } else { - current - 1 - }; - self.state.results_table_state.select(Some(new)); + KeyCode::Up | KeyCode::Char('k') => { + cycle(&mut self.state.results_table_state, items_count, false); } - KeyCode::Down | KeyCode::Char('j') if items_count > 0 => { - let current = - self.state.results_table_state.selected().unwrap_or(0); - let new = (current + 1) % items_count; - self.state.results_table_state.select(Some(new)); + KeyCode::Down | KeyCode::Char('j') => { + cycle(&mut self.state.results_table_state, items_count, true); } KeyCode::Char(' ') => { if let Some(current) = self.state.results_table_state.selected() @@ -416,17 +395,11 @@ impl MainTui { // The pane lists downloads first, then uploads; navigation spans both. let rows = self.state.downloads.len() + self.state.uploads.len(); match key.code { - KeyCode::Up | KeyCode::Char('k') if rows > 0 => { - let current = - self.state.downloads_table_state.selected().unwrap_or(0); - let new = if current == 0 { rows - 1 } else { current - 1 }; - self.state.downloads_table_state.select(Some(new)); + KeyCode::Up | KeyCode::Char('k') => { + cycle(&mut self.state.downloads_table_state, rows, false); } - KeyCode::Down | KeyCode::Char('j') if rows > 0 => { - let current = - self.state.downloads_table_state.selected().unwrap_or(0); - let new = (current + 1) % rows; - self.state.downloads_table_state.select(Some(new)); + KeyCode::Down | KeyCode::Char('j') => { + cycle(&mut self.state.downloads_table_state, rows, true); } KeyCode::Char('x') => { self.cancel_selected_upload(); @@ -448,63 +421,35 @@ impl MainTui { } pub(super) fn handle_mouse_event(&mut self, mouse: MouseEvent) { - if mouse.kind == MouseEventKind::Down(MouseButton::Left) { - let (col, row) = (mouse.column, mouse.row); + if mouse.kind != MouseEventKind::Down(MouseButton::Left) { + return; + } - // 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; + let clicked = Position::new(mouse.column, mouse.row); + let panes = [ + (self.state.searches_pane_area, FocusedPane::Searches), + (self.state.results_pane_area, FocusedPane::Results), + (self.state.downloads_pane_area, FocusedPane::Downloads), + ]; + for (area, pane) in panes { + if area.is_some_and(|area| area.contains(clicked)) { + self.state.focused_pane = pane; return; } - - // 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; - } - - // 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; - } } } } -fn previous_char_boundary(input: &str, cursor_position: usize) -> usize { - let cursor_position = input.floor_char_boundary(cursor_position); - - input[..cursor_position] - .char_indices() - .last() - .map_or(0, |(index, _)| index) -} - -fn next_char_boundary(input: &str, cursor_position: usize) -> usize { - let cursor_position = input.floor_char_boundary(cursor_position); - - if cursor_position >= input.len() { - return input.len(); +/// Move a table's selection one row, wrapping at either end. +fn cycle(table: &mut TableState, len: usize, forward: bool) { + if len == 0 { + return; } - - cursor_position - + input[cursor_position..] - .chars() - .next() - .map_or(0, char::len_utf8) + let current = table.selected().unwrap_or(0); + let next = if forward { + (current + 1) % len + } else { + (current + len - 1) % len + }; + table.select(Some(next)); } diff --git a/soulseek-rs/src/ui/main_tui/mod.rs b/soulseek-rs/src/ui/main_tui/mod.rs index 7894b4b..478208e 100644 --- a/soulseek-rs/src/ui/main_tui/mod.rs +++ b/soulseek-rs/src/ui/main_tui/mod.rs @@ -23,8 +23,6 @@ pub struct MainTui { client: Arc, state: AppState, download_dir: String, - #[allow(dead_code)] - max_concurrent_downloads: usize, search_timeout: Duration, spinner_state: usize, store: Option, @@ -39,7 +37,6 @@ impl MainTui { pub fn new( client: Arc, download_dir: String, - max_concurrent_downloads: usize, search_timeout: Duration, store: Option, config_path: Option, @@ -48,7 +45,6 @@ impl MainTui { client, state: AppState::new(), download_dir, - max_concurrent_downloads, search_timeout, spinner_state: 0, store, @@ -256,19 +252,12 @@ pub fn launch_main_tui( terminal: DefaultTerminal, client: Arc, download_dir: String, - max_concurrent_downloads: usize, search_timeout: Duration, store: Option, config_path: Option, ) -> Result<()> { - let tui = MainTui::new( - client, - download_dir, - max_concurrent_downloads, - search_timeout, - store, - config_path, - ); + let tui = + MainTui::new(client, download_dir, search_timeout, store, config_path); tui.run(terminal) } @@ -526,7 +515,6 @@ mod tests { MainTui::new( Arc::new(session), "/tmp".to_string(), - 1, Duration::from_secs(1), None, None, @@ -746,7 +734,6 @@ mod tests { let mut tui = MainTui::new( session.clone(), "/daemon/old".to_string(), - 1, Duration::from_secs(1), None, Some(config_path.clone()), @@ -803,7 +790,6 @@ mod tests { let mut tui = MainTui::new( session.clone(), "/tmp".to_string(), - 1, Duration::from_secs(1), None, Some(config_path), diff --git a/soulseek-rs/src/ui/main_tui/render.rs b/soulseek-rs/src/ui/main_tui/render.rs index 5e108fc..45301bb 100644 --- a/soulseek-rs/src/ui/main_tui/render.rs +++ b/soulseek-rs/src/ui/main_tui/render.rs @@ -17,18 +17,6 @@ use ratatui::{ widgets::Paragraph, }; -const COMMAND_BAR_PREFIX: &str = "search: "; -const MESSAGE_BAR_PREFIX: &str = "message (to: recipient text): "; -const BROWSE_BAR_PREFIX: &str = "browse user: "; - -const fn command_bar_prefix(mode: CommandBarMode) -> &'static str { - match mode { - CommandBarMode::Search => COMMAND_BAR_PREFIX, - CommandBarMode::Message => MESSAGE_BAR_PREFIX, - CommandBarMode::Browse => BROWSE_BAR_PREFIX, - } -} - impl MainTui { pub(super) fn render(&mut self, frame: &mut Frame) { let mut constraints = vec![ @@ -458,7 +446,11 @@ impl MainTui { } fn render_command_bar(&self, frame: &mut Frame, area: Rect) { - let prefix = command_bar_prefix(self.state.command_bar_mode); + let prefix = match self.state.command_bar_mode { + CommandBarMode::Search => "search: ", + CommandBarMode::Message => "message (to: recipient text): ", + CommandBarMode::Browse => "browse user: ", + }; let block = pane_block(true); // Derive from the block's inner rect so borders and padding stay in // one place; the cursor follows whatever the pane reserves. diff --git a/soulseek-rs/src/ui/main_tui/search.rs b/soulseek-rs/src/ui/main_tui/search.rs index 5a443f7..31bcc06 100644 --- a/soulseek-rs/src/ui/main_tui/search.rs +++ b/soulseek-rs/src/ui/main_tui/search.rs @@ -102,17 +102,13 @@ impl MainTui { self.state.results_is_filtering = false; } - fn recompute_results_filter(&mut self) { + pub(super) fn apply_filter(&mut self) { let (items, indices) = filter_results( &self.state.results_items, &self.state.results_filter_query, ); self.state.results_filtered_items = items; self.state.results_filtered_indices = indices; - } - - pub(super) fn apply_filter(&mut self) { - self.recompute_results_filter(); if !self.state.results_filtered_items.is_empty() { self.state.results_table_state.select(Some(0)); } diff --git a/soulseek-rs/src/ui/panes/chat_pane.rs b/soulseek-rs/src/ui/panes/chat_pane.rs index 9f9b2b1..9d31a52 100644 --- a/soulseek-rs/src/ui/panes/chat_pane.rs +++ b/soulseek-rs/src/ui/panes/chat_pane.rs @@ -1,13 +1,13 @@ use crate::models::{AppState, MessageDirection}; use crate::ui::{ - accent_style, border_style, border_type, dimmed_style, highlight_style, - info_style, primary_style, + accent_style, border_style, dimmed_style, highlight_style, info_style, + primary_style, }; use ratatui::{ Frame, layout::{Constraint, Layout, Rect}, text::{Line, Span}, - widgets::{Block, Borders, Paragraph}, + widgets::{Block, BorderType, Borders, Paragraph}, }; /// Render the private-message chat box: title is the conversation partner, @@ -28,7 +28,7 @@ pub fn render_chat_pane( let block = Block::default() .borders(Borders::ALL) .border_style(border_style(true)) - .border_type(border_type(true)) + .border_type(BorderType::Rounded) .title(title); let inner = block.inner(area); frame.render_widget(block, area); diff --git a/soulseek-rs/src/ui/panes/download_info_pane.rs b/soulseek-rs/src/ui/panes/download_info_pane.rs index 094dbd1..408a9f1 100644 --- a/soulseek-rs/src/ui/panes/download_info_pane.rs +++ b/soulseek-rs/src/ui/panes/download_info_pane.rs @@ -229,7 +229,12 @@ fn build_info_lines( speed_bytes_per_sec, } => { lines.push(Line::from("")); - push_progress_lines(&mut lines, *bytes_downloaded, *total_bytes); + push_transfer_progress( + &mut lines, + "Downloaded", + *bytes_downloaded, + *total_bytes, + ); lines.push(label_value( "Speed", &format_speed(*speed_bytes_per_sec), @@ -246,7 +251,12 @@ fn build_info_lines( total_bytes, } => { lines.push(Line::from("")); - push_progress_lines(&mut lines, *bytes_downloaded, *total_bytes); + push_transfer_progress( + &mut lines, + "Downloaded", + *bytes_downloaded, + *total_bytes, + ); } DownloadStatus::Failed(reason) => { if let Some(reason) = reason { @@ -268,10 +278,7 @@ fn build_info_lines( } fn label_value(label: &str, value: &str) -> Line<'static> { - Line::from(vec![ - Span::styled(format!("{label:>, - bytes_downloaded: u64, - total_bytes: u64, -) { - push_transfer_progress(lines, "Downloaded", bytes_downloaded, total_bytes); -} - /// A progress bar plus a `