From a7372cc88ec493e282984d07a332083cd3a09aec Mon Sep 17 00:00:00 2001 From: michel Date: Sun, 19 Jul 2026 13:31:30 +0200 Subject: [PATCH] feat(lib): send and receive private messages Wire private messaging through the client: - MessageFactory::build_message_user (code 22) and build_message_acked (code 23); - the incoming MessageUser handler now acknowledges freshly delivered messages and forwards them to the client instead of only logging them; - ClientContext buffers received UserMessages; - Client::send_private_message and Client::take_private_messages expose the feature, and UserMessage (with accessors) is re-exported from the crate. Verified end to end against soulfind: a_private_message_is_delivered_between_users logs in two clients and asserts one receives the other's message. --- soulseek-rs-lib/src/actor/server_actor.rs | 47 ++++++++++++++ soulseek-rs-lib/src/client.rs | 62 ++++++++++++++++++- soulseek-rs-lib/src/lib.rs | 2 +- .../src/message/server/message_factory.rs | 36 +++++++++++ .../src/message/server/message_user.rs | 15 ++++- soulseek-rs-lib/tests/e2e.rs | 36 +++++++++++ 6 files changed, 195 insertions(+), 3 deletions(-) diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index df82e5f..e072ebc 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -102,6 +102,37 @@ impl UserMessage { self.message ); } + + /// The server-assigned id of this message (used to acknowledge it). + #[must_use] + pub const fn id(&self) -> u32 { + self.id + } + + /// Unix timestamp the server recorded for this message. + #[must_use] + pub const fn timestamp(&self) -> u32 { + self.timestamp + } + + /// The username of the sender. + #[must_use] + pub fn username(&self) -> &str { + &self.username + } + + /// The message body. + #[must_use] + pub fn message(&self) -> &str { + &self.message + } + + /// Whether the server flagged this as freshly delivered (as opposed to a + /// message replayed because it was queued while the recipient was offline). + #[must_use] + pub const fn is_new(&self) -> bool { + self.new_message + } } #[derive(Debug, Clone)] @@ -129,6 +160,7 @@ pub enum ServerMessage { obfuscation_type: u32, obfuscated_port: u16, }, + PrivateMessageReceived(UserMessage), } pub struct ServerActor { @@ -418,6 +450,21 @@ impl ServerActor { ); } } + ServerMessage::PrivateMessageReceived(user_message) => { + debug!( + "[server] Private message from {}", + user_message.username() + ); + if let Err(e) = self + .client_channel + .send(ClientOperation::PrivateMessageReceived(user_message)) + { + error!( + "[server] Error forwarding private message to client: {}", + e + ); + } + } ServerMessage::ProcessRead => { self.process_read(); } diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index 30a5a1a..30e3286 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -1,5 +1,7 @@ use crate::actor::ActorHandle; -use crate::actor::server_actor::{PeerAddress, ServerActor, ServerMessage}; +use crate::actor::server_actor::{ + PeerAddress, ServerActor, ServerMessage, UserMessage, +}; use crate::download_store::{DownloadStore, collect_failed_tokens}; use crate::types::{DownloadMetadata, DownloadStatus}; use crate::utils::logger; @@ -94,6 +96,7 @@ pub enum ClientOperation { place: u32, }, SetServerSender(Sender), + PrivateMessageReceived(UserMessage), } pub struct ClientContext { pub peer_registry: Option, @@ -101,6 +104,7 @@ pub struct ClientContext { sender: Option>, server_sender: Option>, searches: HashMap, + private_messages: Vec, actor_system: Arc, } impl Default for ClientContext { @@ -307,10 +311,21 @@ impl ClientContext { sender: None, server_sender: None, searches: HashMap::new(), + private_messages: Vec::new(), downloads: DownloadStore::new(), actor_system, } } + + /// Record a private message received from another user. + pub fn push_private_message(&mut self, message: UserMessage) { + self.private_messages.push(message); + } + + /// Remove and return all buffered private messages. + pub fn take_private_messages(&mut self) -> Vec { + std::mem::take(&mut self.private_messages) + } } pub struct Client { enable_listen: bool, @@ -424,6 +439,40 @@ impl Client { } } + /// Send a private message to another user via the server. + /// + /// # Errors + /// Returns [`SoulseekRs::NotConnected`] if the client is not connected. + pub fn send_private_message( + &self, + username: &str, + message: &str, + ) -> Result<()> { + let handle = self + .server_handle + .as_ref() + .ok_or(SoulseekRs::NotConnected)?; + let msg = crate::message::server::MessageFactory::build_message_user( + username, message, + ); + handle + .send(ServerMessage::SendMessage(msg)) + .map_err(|_| SoulseekRs::NotConnected)?; + Ok(()) + } + + /// Remove and return all private messages received since the last call. + #[must_use] + pub fn take_private_messages(&self) -> Vec { + match self.context.write_safe() { + Ok(mut ctx) => ctx.take_private_messages(), + Err(e) => { + error!("[client] take_private_messages: {}", e); + Vec::new() + } + } + } + #[allow(dead_code)] pub fn remove_peer(&self, username: &str) { let context = match self.context.read_safe() { @@ -1127,6 +1176,17 @@ impl Client { ), } } + 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 + ), + }, } } Err(e) => { diff --git a/soulseek-rs-lib/src/lib.rs b/soulseek-rs-lib/src/lib.rs index b35e2af..0575565 100644 --- a/soulseek-rs-lib/src/lib.rs +++ b/soulseek-rs-lib/src/lib.rs @@ -20,7 +20,7 @@ pub mod prelude { } // Re-export commonly used types -pub use actor::server_actor::PeerAddress; +pub use actor::server_actor::{PeerAddress, UserMessage}; pub use client::{Client, ClientSettings}; pub use error::{Result, SoulseekRs}; pub use types::{DownloadStatus, File, Search, SearchResult, Transfer}; diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 6df5742..9e041d8 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -63,6 +63,23 @@ impl MessageFactory { .write_string(query) .clone() } + /// Build a private message (server code 22) to send to another user. + #[must_use] + pub fn build_message_user(username: &str, message: &str) -> Message { + Message::new() + .write_int32(22) + .write_string(username) + .write_string(message) + .clone() + } + + /// Acknowledge a received private message (server code 23) so the server + /// stops re-delivering it. + #[must_use] + pub fn build_message_acked(id: u32) -> Message { + Message::new().write_int32(23).write_int32(id).clone() + } + #[must_use] pub fn build_set_status_message(status_code: u32) -> Message { Message::new() @@ -167,6 +184,25 @@ fn test_build_login_message() { assert_eq!(expect, message.get_data()); } +#[test] +fn test_build_message_user() { + let message = MessageFactory::build_message_user("bob", "hi"); + let expect: Vec = [ + 22, 0, 0, 0, // code + 3, 0, 0, 0, 98, 111, 98, // username "bob" + 2, 0, 0, 0, 104, 105, // message "hi" + ] + .to_vec(); + assert_eq!(expect, message.get_data()); +} + +#[test] +fn test_build_message_acked() { + let message = MessageFactory::build_message_acked(7); + let expect: Vec = [23, 0, 0, 0, 7, 0, 0, 0].to_vec(); + assert_eq!(expect, message.get_data()); +} + #[test] fn test_build_file_search_message() { let message = MessageFactory::build_file_search_message(12, "trance wax"); diff --git a/soulseek-rs-lib/src/message/server/message_user.rs b/soulseek-rs-lib/src/message/server/message_user.rs index 47ae544..ef3cb5f 100644 --- a/soulseek-rs-lib/src/message/server/message_user.rs +++ b/soulseek-rs-lib/src/message/server/message_user.rs @@ -1,5 +1,6 @@ use crate::actor::server_actor::{ServerMessage, UserMessage}; use crate::info; +use crate::message::server::MessageFactory; use crate::message::{Message, MessageHandler}; use std::sync::mpsc::Sender; @@ -11,7 +12,7 @@ impl MessageHandler for MessageUser { 22 } - fn handle(&self, message: &mut Message, _sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { let id = message.read_int32(); let timestamp = message.read_int32(); let username = message.read_string(); @@ -27,5 +28,17 @@ 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. + if new_message { + let _ = sender.send(ServerMessage::SendMessage( + MessageFactory::build_message_acked(id), + )); + } + + // Surface the message to the client so it can be read via the API. + let _ = + sender.send(ServerMessage::PrivateMessageReceived(user_message)); } } diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 029ca91..0f01896 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -251,6 +251,42 @@ fn a_search_is_forwarded_to_a_connected_peer() { assert!(receiver.get_all_searches().contains_key(probe)); } +#[test] +fn a_private_message_is_delivered_between_users() { + let server = server_or_skip!(); + + // Two logged-in users, one messages the other through the server. + let mut alice = + Client::with_settings(server.settings("e2e_alice_pm", "pw")); + let mut bob = Client::with_settings(server.settings("e2e_bob_pm", "pw")); + alice.connect().expect("alice connect"); + bob.connect().expect("bob connect"); + assert!(alice.login().expect("alice login")); + assert!(bob.login().expect("bob login")); + + let body = "hello bob, this is alice"; + alice + .send_private_message("e2e_bob_pm", body) + .expect("send private message"); + + // Delivery is asynchronous; poll Bob's inbox until the message arrives. + let deadline = Instant::now() + Duration::from_secs(5); + let mut received = Vec::new(); + while Instant::now() < deadline { + received.extend(bob.take_private_messages()); + if !received.is_empty() { + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + + let message = received + .iter() + .find(|m| m.message() == body) + .expect("bob should receive alice's message"); + assert_eq!(message.username(), "e2e_alice_pm"); +} + #[test] fn login_succeeds_with_listener_enabled() { let server = server_or_skip!();