diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 7486da6..ff8a6da 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -149,6 +149,13 @@ pub enum ServerMessage { token: u32, query: String, }, + /// A search the server distributed to us from another user; if it matches + /// our shares we reply with a FileSearchResponse. + FileSearchRequest { + username: String, + token: u32, + query: String, + }, #[allow(dead_code)] ConnectToPeer(Peer), PierceFirewall(u32), @@ -508,6 +515,21 @@ impl ServerActor { ServerMessage::FileSearch { token, query } => { self.file_search(token, &query); } + ServerMessage::FileSearchRequest { + username, + token, + query, + } => { + if let Err(e) = + self.client_channel.send(ClientOperation::IncomingSearch { + username, + token, + query, + }) + { + error!("[server] forward IncomingSearch: {}", e); + } + } } } diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index 43d3342..b953719 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -9,7 +9,11 @@ use crate::{ Transfer, actor::{ActorSystem, peer_registry::PeerRegistry}, error::{Result, SoulseekRs}, - peer::{ConnectionType, DownloadPeer, NewPeer, Peer, listen::Listen}, + message::peer::{FileEntry, build_file_search_response}, + peer::{ + ConnectionType, DownloadPeer, NewPeer, Peer, PeerMessage, + listen::Listen, + }, shares::Shares, types::{Download, Search, SearchResult}, utils::{lock::RwLockExt, md5, thread_pool::ThreadPool}, @@ -43,6 +47,35 @@ fn next_connect_token() -> u32 { NEXT_CONNECT_TOKEN.fetch_add(1, Ordering::Relaxed).max(1) } +/// Build a `FileSearchResponse` for `query` against `shares`, or `None` if +/// nothing matches. `own_username` is the name the searcher will download from. +fn build_search_response( + shares: &Shares, + own_username: &str, + token: u32, + query: &str, +) -> Option { + let matches = shares.search(query); + if matches.is_empty() { + return None; + } + let entries: Vec = matches + .iter() + .map(|f| FileEntry { + name: &f.virtual_path, + size: f.size, + attribs: &f.attributes, + }) + .collect(); + Some(build_file_search_response( + own_username, + token, + &entries, + 1, + 0, + )) +} + #[derive(Debug, Clone)] pub struct ClientSettings { pub username: String, @@ -110,6 +143,12 @@ pub enum ClientOperation { SetServerSender(Sender), PrivateMessageReceived(UserMessage), PeerConnected(String), + /// A search distributed to us by the server; reply if our shares match. + IncomingSearch { + username: String, + token: u32, + query: String, + }, /// A direct outbound connection to this peer failed before it was /// established — the peer is likely firewalled, so fall back to asking the /// server to broker the connection. Carries the reporting actor's id. @@ -127,6 +166,10 @@ pub struct ClientContext { pending_connect_tokens: HashMap, /// Files we share with peers (read-only after connect). pub shares: Arc, + /// Peer listen addresses learned from GetPeerAddress responses. + peer_addresses: HashMap, + /// Peer messages waiting for a control connection to that peer. + pending_peer_messages: HashMap>, actor_system: Arc, } impl Default for ClientContext { @@ -329,6 +372,29 @@ fn fail_queued_downloads_notifies_receiver_and_store() { )); } +#[test] +fn build_search_response_matches_shares_and_echoes_token() { + let dir = std::env::temp_dir() + .join(format!("soulseek-searchresp-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("probe_xyzzy.bin"), b"data").unwrap(); + let shares = Shares::scan(&dir).unwrap(); + + let response = build_search_response(&shares, "me", 99, "xyzzy") + .expect("a matching share yields a response"); + let mut decoded = + crate::message::Message::new_with_data(response.get_buffer()); + decoded.set_pointer(8); + let result = SearchResult::new_from_message(&mut decoded).unwrap(); + assert_eq!(result.username, "me"); + assert_eq!(result.token, 99); + assert!(result.files.iter().any(|f| f.name.contains("probe_xyzzy"))); + + assert!(build_search_response(&shares, "me", 1, "nomatch").is_none()); + let _ = std::fs::remove_dir_all(dir); +} + #[test] fn test_client_removes_only_queued_downloads() { let client = Client::new("test-user", "test-password"); @@ -389,11 +455,52 @@ impl ClientContext { private_messages: Vec::new(), pending_connect_tokens: HashMap::new(), shares: Arc::new(Shares::empty()), + peer_addresses: HashMap::new(), + pending_peer_messages: HashMap::new(), downloads: DownloadStore::new(), actor_system, } } + /// Cache a peer's listen address learned from a GetPeerAddress response. + pub fn cache_peer_address( + &mut self, + username: &str, + host: String, + port: u32, + ) { + self.peer_addresses + .insert(username.to_string(), (host, port)); + } + + /// The cached listen address for `username`, if known. + #[must_use] + pub fn peer_address(&self, username: &str) -> Option<(String, u32)> { + self.peer_addresses.get(username).cloned() + } + + /// Queue a peer message to send once a control connection to `username` is up. + pub fn queue_peer_message( + &mut self, + username: &str, + message: crate::message::Message, + ) { + self.pending_peer_messages + .entry(username.to_string()) + .or_default() + .push(message); + } + + /// Remove and return the messages queued for `username`. + pub fn take_peer_messages( + &mut self, + username: &str, + ) -> Vec { + self.pending_peer_messages + .remove(username) + .unwrap_or_default() + } + /// Remember that a server-brokered connection to `username` is pending under /// `token`; the peer will quote it back in a PierceFirewall. pub fn add_pending_connect(&mut self, token: u32, username: String) { @@ -1210,6 +1317,16 @@ impl Client { obfuscated_port ); + // Cache the address for the serve/search paths. + if let Ok(mut ctx) = client_context.write_safe() + { + ctx.cache_peer_address( + &username, + host.clone(), + port, + ); + } + let peer_exists = match client_context .read_safe() { @@ -1401,11 +1518,93 @@ impl Client { 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::PeerConnectFailed( diff --git a/soulseek-rs-lib/src/message/server/file_search.rs b/soulseek-rs-lib/src/message/server/file_search.rs index 5c1ebea..509f538 100644 --- a/soulseek-rs-lib/src/message/server/file_search.rs +++ b/soulseek-rs-lib/src/message/server/file_search.rs @@ -1,4 +1,4 @@ -use crate::{debug, info}; +use crate::trace; use std::sync::mpsc::Sender; use crate::{ @@ -12,14 +12,16 @@ impl MessageHandler for FileSearchHandler { fn get_code(&self) -> u8 { 26 } - fn handle(&self, message: &mut Message, _sender: Sender) { - debug!("Handling file search message"); + fn handle(&self, message: &mut Message, sender: Sender) { + // The server distributes another user's search to us: [user][token][query]. let username = message.read_string(); let token = message.read_int32(); let query = message.read_string(); - info!( - "Message search username:{}, token: {}, query: {}", - username, token, query - ); + trace!("[server] search from {}: {} ({})", username, query, token); + let _ = sender.send(ServerMessage::FileSearchRequest { + username, + token, + query, + }); } }