diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index bf63e3a..74b9e21 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -729,8 +729,20 @@ impl PeerActor { // unreachable (likely firewalled): signal a connect failure so the // client can fall back to server-brokered connect. Anything else is a // normal disconnect. + // A dial the server asked for carries the token the far peer quoted; + // it is what a CantConnectToPeer must echo back, so the peer stops + // waiting for a connection that is never coming. + let brokered_token = self + .peer + .read_safe() + .ok() + .and_then(|p| if p.brokered { p.token } else { None }); let op = if self.outbound && !self.established { - ClientOperation::PeerConnectFailed(self.id, username) + ClientOperation::PeerConnectFailed( + self.id, + username, + brokered_token, + ) } else { ClientOperation::PeerDisconnected( self.id, diff --git a/soulseek-rs-lib/src/actor/peer_actor/tests.rs b/soulseek-rs-lib/src/actor/peer_actor/tests.rs index de8d7c8..c7d8325 100644 --- a/soulseek-rs-lib/src/actor/peer_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/peer_actor/tests.rs @@ -50,7 +50,7 @@ fn a_timed_out_connect_parks_the_actor_in_disconnected() { "a timed-out connect must leave Connecting" ); match rx.try_recv() { - Ok(ClientOperation::PeerConnectFailed(7, username)) => { + Ok(ClientOperation::PeerConnectFailed(7, username, _)) => { assert_eq!(username, "bob"); } other => panic!("expected PeerConnectFailed, got {other:?}"), diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index f4b7007..02d1ce0 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -427,7 +427,7 @@ mod tests { // Play the client ops loop: take the terminal outcome, evict by id. match rx.recv_timeout(std::time::Duration::from_secs(10)) { - Ok(ClientOperation::PeerConnectFailed(id, username)) => { + Ok(ClientOperation::PeerConnectFailed(id, username, _)) => { assert_eq!(username, "ghost"); if let Some(handle) = registry.remove_peer_if(&username, id) { let _ = handle.stop(); diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index ecab497..6fb2636 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -2,12 +2,14 @@ use crate::actor::{Actor, ActorHandle, ConnectionState}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::server::AdminMessageHandler; +use crate::message::server::CantConnectToPeerHandler; use crate::message::server::CheckPrivilegesHandler; use crate::message::server::ConnectToPeerHandler; use crate::message::server::EmbeddedMessageHandler; use crate::message::server::ExcludedSearchPhrasesHandler; use crate::message::server::FileSearchHandler; use crate::message::server::GetPeerAddressHandler; +use crate::message::server::GlobalRoomMessageHandler; use crate::message::server::JoinRoomHandler; use crate::message::server::LeaveRoomHandler; use crate::message::server::LoginHandler; @@ -27,13 +29,21 @@ use crate::message::server::WishListIntervalHandler; use crate::message::server::{ GetUserStatsHandler, GetUserStatusHandler, RoomListHandler, }; +use crate::message::server::{ + GlobalRecommendationsHandler, ItemRecommendationsHandler, + ItemSimilarUsersHandler, RecommendationsHandler, SimilarUsersHandler, + UserInterestsHandler, +}; +use crate::message::server::{ + RoomTickerAddedHandler, RoomTickerRemovedHandler, RoomTickersHandler, +}; use crate::message::{Handlers, MessageType}; use crate::message::{Message, MessageReader}; use crate::peer::ConnectionType; use crate::peer::Peer; use crate::types::{ - ClientVersion, RoomEvent, RoomInfo, RoomUserStats, SessionLoss, - SessionWatch, + ClientVersion, Recommendation, RoomEvent, RoomInfo, RoomUserStats, + SessionLoss, SessionWatch, SimilarUser, UserInterests, }; use crate::utils::lock::RwLockExt; @@ -164,6 +174,59 @@ pub enum ServerMessage { PossibleParents(Vec<(String, String, u16)>), /// The server asks us to drop our parent and start over. ResetDistributed, + /// The ticker board of a room we just joined (code 113). + RoomTickers { + room: String, + tickers: Vec, + }, + /// One member set their ticker (code 114). + RoomTickerAdded { + room: String, + username: String, + ticker: String, + }, + /// One member cleared their ticker (code 115). + RoomTickerRemoved { + room: String, + username: String, + }, + /// A message from the global room feed (code 152). + GlobalRoomMessageReceived { + room: String, + username: String, + message: String, + }, + /// Recommendations from our own interests (code 54). + RecommendationsReceived { + recommended: Vec, + unrecommended: Vec, + }, + /// Server-wide recommendations (code 56). + GlobalRecommendationsReceived { + recommended: Vec, + unrecommended: Vec, + }, + /// Recommendations for one item (code 111). + ItemRecommendationsReceived { + item: String, + recommendations: Vec, + }, + /// Users similar to us (code 110). + SimilarUsersReceived { + users: Vec, + }, + /// Users who like one item (code 112). + ItemSimilarUsersReceived { + item: String, + usernames: Vec, + }, + /// What another user likes and hates (code 57). + UserInterestsReceived(UserInterests), + /// A peer could not connect to us after we asked the server to broker + /// (code 1001); the token is the one we quoted. + CantConnectToPeer { + token: u32, + }, } pub struct ServerActor { @@ -320,6 +383,17 @@ impl ServerActor { handlers.register_handler(FileSearchHandler); handlers.register_handler(GetPeerAddressHandler); handlers.register_handler(ConnectToPeerHandler); + handlers.register_handler(CantConnectToPeerHandler); + handlers.register_handler(RoomTickersHandler); + handlers.register_handler(RoomTickerAddedHandler); + handlers.register_handler(RoomTickerRemovedHandler); + handlers.register_handler(GlobalRoomMessageHandler); + handlers.register_handler(RecommendationsHandler); + handlers.register_handler(GlobalRecommendationsHandler); + handlers.register_handler(ItemRecommendationsHandler); + handlers.register_handler(SimilarUsersHandler); + handlers.register_handler(ItemSimilarUsersHandler); + handlers.register_handler(UserInterestsHandler); self.dispatcher = Some(MessageDispatcher::new( "server".into(), @@ -501,6 +575,37 @@ impl ServerActor { ServerMessage::RoomUserLeft { room, username } => { self.forward_room_event(RoomEvent::UserLeft { room, username }); } + ServerMessage::RoomTickers { room, tickers } => { + self.forward_room_event(RoomEvent::Tickers { room, tickers }); + } + ServerMessage::RoomTickerAdded { + room, + username, + ticker, + } => { + self.forward_room_event(RoomEvent::TickerAdded { + room, + username, + ticker, + }); + } + ServerMessage::RoomTickerRemoved { room, username } => { + self.forward_room_event(RoomEvent::TickerRemoved { + room, + username, + }); + } + ServerMessage::GlobalRoomMessageReceived { + room, + username, + message, + } => { + self.forward_room_event(RoomEvent::GlobalMessage { + room, + username, + message, + }); + } other => self.handle_standing_message(other), } } @@ -539,6 +644,54 @@ impl ServerActor { ServerMessage::CheckPrivileges => { self.queue_message(MessageFactory::build_check_privileges()); } + ServerMessage::RecommendationsReceived { + recommended, + unrecommended, + } => { + self.forward_to_client(ClientOperation::Recommendations { + global: false, + recommended, + unrecommended, + }); + } + ServerMessage::GlobalRecommendationsReceived { + recommended, + unrecommended, + } => { + self.forward_to_client(ClientOperation::Recommendations { + global: true, + recommended, + unrecommended, + }); + } + ServerMessage::ItemRecommendationsReceived { + item, + recommendations, + } => { + self.forward_to_client(ClientOperation::ItemRecommendations { + item, + recommendations, + }); + } + ServerMessage::SimilarUsersReceived { users } => { + self.forward_to_client(ClientOperation::SimilarUsers(users)); + } + ServerMessage::ItemSimilarUsersReceived { item, usernames } => { + self.forward_to_client(ClientOperation::ItemSimilarUsers { + item, + usernames, + }); + } + ServerMessage::UserInterestsReceived(interests) => { + self.forward_to_client(ClientOperation::UserInterests( + interests, + )); + } + ServerMessage::CantConnectToPeer { token } => { + self.forward_to_client(ClientOperation::CantConnectToPeer { + token, + }); + } other => { error!("[server] unroutable message: {:?}", other); } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 5dc3a34..75dd60c 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -4,9 +4,9 @@ use crate::actor::server_actor::{ }; use crate::download_store::{DownloadStore, collect_failed_tokens}; use crate::types::{ - ClientVersion, DownloadMetadata, DownloadStatus, RoomEvent, RoomInfo, - RoomUserStats, SessionLoss, SessionWatch, UserInfo, UserPresence, - UserStats, UserStatus, + ClientVersion, DownloadMetadata, DownloadStatus, Recommendation, RoomEvent, + RoomInfo, RoomTicker, RoomUserStats, SessionLoss, SessionWatch, + SimilarUser, UserInfo, UserInterests, UserPresence, UserStats, UserStatus, }; use crate::utils::logger; use crate::{ @@ -313,8 +313,11 @@ pub enum ClientOperation { }, /// 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. - PeerConnectFailed(u64, String), + /// server to broker the connection. Carries the reporting actor's id and, + /// when the dial was one the server asked us to make, the token that peer + /// quoted: that dial is answered with a `CantConnectToPeer` instead of a + /// broker request, since the peer is already waiting on the server. + PeerConnectFailed(u64, String, Option), /// Something happened in the chat-room subsystem (list refreshed, a room /// joined/left, a message said, a member joined/left). RoomEvent(RoomEvent), @@ -340,6 +343,31 @@ pub enum ClientOperation { PossibleParents(Vec<(String, String, u16)>), /// The server told us to drop our parent. ResetDistributed, + /// Recommendations from the server: either from our own interests + /// (`global` false, code 54) or server-wide (`global` true, code 56). + Recommendations { + global: bool, + recommended: Vec, + unrecommended: Vec, + }, + /// Recommendations for one item (code 111). + ItemRecommendations { + item: String, + recommendations: Vec, + }, + /// Users the server considers similar to us (code 110). + SimilarUsers(Vec), + /// Users who like one item (code 112). + ItemSimilarUsers { + item: String, + usernames: Vec, + }, + /// What another user likes and hates (code 57). + UserInterests(UserInterests), + /// A peer we asked the server to broker gave up reaching us (code 1001). + CantConnectToPeer { + token: u32, + }, /// A parent candidate told us how deep it sits. `link` says which dial /// to that user is talking. ParentBranchLevel { @@ -410,6 +438,22 @@ pub struct ClientContext { /// Per-member statistics for each joined room, from the stat vectors the /// server sends alongside the membership list. room_member_stats: HashMap>, + /// The ticker board of each joined room, kept current from the board sent + /// on join (code 113) and the later add/remove events (114/115). + room_tickers: HashMap>, + /// The latest recommendations from our own interests (code 54), as + /// (recommended, recommended-against). + recommendations: Option<(Vec, Vec)>, + /// The latest server-wide recommendations (code 56). + global_recommendations: Option<(Vec, Vec)>, + /// Per-item recommendations (code 111), keyed by the item asked about. + item_recommendations: HashMap>, + /// The latest similar-user answer (code 110). + similar_users: Vec, + /// Who likes an item (code 112), keyed by the item asked about. + item_similar_users: HashMap>, + /// Interests of other users (code 57), keyed by username. + user_interests: HashMap, /// What the server has told us about other users, merged across the /// separate status and statistics replies. user_info: HashMap, @@ -519,6 +563,13 @@ impl ClientContext { room_events: Vec::new(), room_members: HashMap::new(), room_member_stats: HashMap::new(), + room_tickers: HashMap::new(), + recommendations: None, + global_recommendations: None, + item_recommendations: HashMap::new(), + similar_users: Vec::new(), + item_similar_users: HashMap::new(), + user_interests: HashMap::new(), user_info: HashMap::new(), watched_users: HashSet::new(), wishlist_interval: None, @@ -562,11 +613,124 @@ impl ClientContext { members.remove(at); } } - RoomEvent::Message { .. } => {} + RoomEvent::Tickers { room, tickers } => { + self.room_tickers.insert(room.clone(), tickers.clone()); + } + RoomEvent::TickerAdded { + room, + username, + ticker, + } => { + // A user has at most one ticker: an added one replaces theirs + // rather than stacking, which is how the server treats it. + let board = self.room_tickers.entry(room.clone()).or_default(); + board.retain(|t| &t.username != username); + board.push(RoomTicker { + username: username.clone(), + ticker: ticker.clone(), + }); + } + RoomEvent::TickerRemoved { room, username } => { + if let Some(board) = self.room_tickers.get_mut(room) { + board.retain(|t| &t.username != username); + } + } + RoomEvent::Message { .. } | RoomEvent::GlobalMessage { .. } => {} } self.room_events.push(event); } + /// The ticker board of `room` as last reported by the server. + #[must_use] + pub fn room_tickers(&self, room: &str) -> Vec { + self.room_tickers.get(room).cloned().unwrap_or_default() + } + + /// Record a recommendations reply; `global` selects the server-wide set. + pub fn apply_recommendations( + &mut self, + global: bool, + recommended: Vec, + unrecommended: Vec, + ) { + let slot = if global { + &mut self.global_recommendations + } else { + &mut self.recommendations + }; + *slot = Some((recommended, unrecommended)); + } + + /// The last recommendations reply, or `None` until one arrives. + #[must_use] + pub fn recommendations( + &self, + global: bool, + ) -> Option<(Vec, Vec)> { + if global { + self.global_recommendations.clone() + } else { + self.recommendations.clone() + } + } + + pub fn apply_item_recommendations( + &mut self, + item: String, + recommendations: Vec, + ) { + self.item_recommendations.insert(item, recommendations); + } + + #[must_use] + pub fn item_recommendations(&self, item: &str) -> Vec { + self.item_recommendations + .get(item) + .cloned() + .unwrap_or_default() + } + + pub fn apply_similar_users(&mut self, users: Vec) { + self.similar_users = users; + } + + #[must_use] + pub fn similar_users(&self) -> Vec { + self.similar_users.clone() + } + + pub fn apply_item_similar_users( + &mut self, + item: String, + usernames: Vec, + ) { + self.item_similar_users.insert(item, usernames); + } + + #[must_use] + pub fn item_similar_users(&self, item: &str) -> Vec { + self.item_similar_users + .get(item) + .cloned() + .unwrap_or_default() + } + + pub fn apply_user_interests(&mut self, interests: UserInterests) { + self.user_interests + .insert(interests.username.clone(), interests); + } + + #[must_use] + pub fn user_interests(&self, username: &str) -> Option { + self.user_interests.get(username).cloned() + } + + /// Forget any interests cached for `username`, so a poll after a fresh + /// request cannot mistake the previous answer for this one. + pub fn invalidate_user_interests(&mut self, username: &str) { + self.user_interests.remove(username); + } + /// Record a `GetUserStatus` reply, merging it with any statistics already /// received for that user. pub fn apply_user_status( @@ -1050,6 +1214,7 @@ mod downloads; mod operations; mod rooms; mod search; +mod social; mod upload_queue; mod uploads; diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 81dcd8e..d53c33e 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -777,6 +777,71 @@ impl Client { ctx.own_privileges = Some(seconds); } } + ClientOperation::Recommendations { + global, + recommended, + unrecommended, + } => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.apply_recommendations( + global, + recommended, + unrecommended, + ); + } + } + ClientOperation::ItemRecommendations { + item, + recommendations, + } => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.apply_item_recommendations( + item, + recommendations, + ); + } + } + ClientOperation::SimilarUsers(users) => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.apply_similar_users(users); + } + } + ClientOperation::ItemSimilarUsers { item, usernames } => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.apply_item_similar_users(item, usernames); + } + } + ClientOperation::UserInterests(interests) => { + if let Ok(mut ctx) = client_context.write_safe() { + ctx.apply_user_interests(interests); + } + } + ClientOperation::CantConnectToPeer { token } => { + // The peer we asked the server to broker gave up. The + // token is the correlation we sent, so it names the + // peer: end the wait now instead of letting every + // queued download for them sit until the timeout. + let username = match client_context.write_safe() { + Ok(mut ctx) => ctx.take_pending_connect(token), + Err(e) => { + error!( + "[client] CantConnectToPeer write: {}", + e + ); + None + } + }; + if let Some(username) = username { + debug!( + "[client] {} cannot connect back (token {})", + username, token + ); + Self::fail_queued_downloads( + &client_context, + &username, + ); + } + } ClientOperation::StartUpload { token } => { // The peer accepted our offer: resolve their // address (from the code-9 GetPeerAddress) and @@ -875,7 +940,58 @@ impl Client { ctx.store_browse_result(username, directories); } } - ClientOperation::PeerConnectFailed(id, username) => { + ClientOperation::PeerConnectFailed( + id, + username, + brokered_token, + ) => { + // Direct connect failed. A dial the server asked + // us to make is answered with CantConnectToPeer + // quoting the peer's own token: that peer is + // already waiting on the server, and asking it to + // broker the same connection back only trades one + // unreachable direction for the other. Every other + // dial falls back to the broker below. + if let Some(peer_token) = brokered_token { + let server_sender = match client_context + .write_safe() + { + Ok(ctx) => { + if let Some(handle) = + ctx.peer_registry.as_ref().and_then( + |r| r.remove_peer_if(&username, id), + ) + { + let _ = handle.stop(); + } + ctx.server_sender.clone() + } + Err(e) => { + error!( + "[client] PeerConnectFailed write: {}", + e + ); + continue; + } + }; + if let Some(sender) = server_sender { + let msg = crate::message::server::MessageFactory::build_cant_connect_to_peer( + peer_token, + &username, + ); + let _ = sender + .send(ServerMessage::SendMessage(msg)); + } + // Nothing else is coming from this peer, so + // anything queued for it fails now rather than + // waiting out its timeout. + Self::fail_queued_downloads( + &client_context, + &username, + ); + continue; + } + // Direct connect failed: ask the server to // broker it. Register a correlation token, then // send ConnectToPeer so the (firewalled) peer diff --git a/soulseek-rs-lib/src/client/search.rs b/soulseek-rs-lib/src/client/search.rs index 215deb5..9e54136 100644 --- a/soulseek-rs-lib/src/client/search.rs +++ b/soulseek-rs-lib/src/client/search.rs @@ -81,6 +81,55 @@ impl Client { Ok(()) } + /// Search one user's shares (server code 42) and return at once. + /// + /// Results arrive from that user alone and accumulate under `query`, read + /// back with [`Client::get_search_results`] — the same place a plain + /// search's results land, since the replies are ordinary + /// `FileSearchResponse` frames. + /// + /// # Errors + /// [`SoulseekRs::NotConnected`] when there is no server connection. + pub fn search_user(&self, username: &str, query: &str) -> Result<()> { + let token = self.register_search(query)?; + self.send_server_message( + crate::message::server::MessageFactory::build_user_search( + username, token, query, + ), + ) + } + + /// Search the shares of everyone in `room` (server code 120), returning at + /// once. Results accumulate under `query` exactly as above. + /// + /// # Errors + /// [`SoulseekRs::NotConnected`] when there is no server connection. + pub fn search_room(&self, room: &str, query: &str) -> Result<()> { + let token = self.register_search(query)?; + self.send_server_message( + crate::message::server::MessageFactory::build_room_search( + room, token, query, + ), + ) + } + + /// Take a token for `query` and register it, so replies quoting that token + /// are filed under `query` when they arrive. + fn register_search(&self, query: &str) -> Result { + if self.server_handle.is_none() { + return Err(SoulseekRs::NotConnected); + } + let token = next_search_token(); + self.context.write_safe()?.searches.insert( + query.to_string(), + Search { + token, + results: Vec::new(), + }, + ); + Ok(token) + } + /// Let responses accumulate for `timeout`, or until cancelled. /// /// Peers answer a search over their own connections whenever they get round diff --git a/soulseek-rs-lib/src/client/social.rs b/soulseek-rs-lib/src/client/social.rs new file mode 100644 index 0000000..5d402e4 --- /dev/null +++ b/soulseek-rs-lib/src/client/social.rs @@ -0,0 +1,264 @@ +//! The social side of the protocol: room tickers (codes 113-116), the global +//! room feed (150-152), interests and the recommendation queries built on them +//! (51/52/54/56/57/110/111/112/117/118), and the odds and ends a client owes +//! the server — a keepalive ping (32), gifting privileges (123) and changing +//! our password (142). + +use super::{ + Client, Recommendation, Result, RoomTicker, RwLockExt, SimilarUser, + UserInterests, error, +}; +use crate::message::server::MessageFactory; + +impl Client { + /// Send a keepalive ping (server code 32). The server does not reply; the + /// point is that a connection with nothing to say still has traffic on it, + /// so a NAT or the server's idle reaper does not drop it. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn ping_server(&self) -> Result<()> { + self.send_server_message(MessageFactory::build_server_ping()) + } + + /// Set our ticker in `room` — the one-line message other members see. + /// Passing an empty `ticker` clears ours. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn set_room_ticker(&self, room: &str, ticker: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_set_room_ticker( + room, ticker, + )) + } + + /// The ticker board of `room`: what the server sent on join (code 113), + /// kept current by the later add/remove events. + #[must_use] + pub fn room_tickers(&self, room: &str) -> Vec { + match self.context.read_safe() { + Ok(ctx) => ctx.room_tickers(room), + Err(e) => { + error!("[client] room_tickers: {}", e); + Vec::new() + } + } + } + + /// Subscribe to the global room feed (code 150): every public room message + /// on the server arrives as a [`crate::RoomEvent::GlobalMessage`]. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn join_global_room(&self) -> Result<()> { + self.send_server_message(MessageFactory::build_join_global_room()) + } + + /// Stop the global room feed (code 151). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn leave_global_room(&self) -> Result<()> { + self.send_server_message(MessageFactory::build_leave_global_room()) + } + + /// Add an interest (code 51). Interests are what the recommendation and + /// similar-user queries below are computed from. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn add_interest(&self, item: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_add_thing_i_like(item)) + } + + /// Drop an interest (code 52). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn remove_interest(&self, item: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_remove_thing_i_like( + item, + )) + } + + /// Add a dislike (code 117). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn add_dislike(&self, item: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_add_thing_i_hate(item)) + } + + /// Drop a dislike (code 118). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn remove_dislike(&self, item: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_remove_thing_i_hate( + item, + )) + } + + /// Ask for recommendations drawn from our own interests (code 54). The + /// reply arrives asynchronously; read it with + /// [`Client::recommendations`]. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn request_recommendations(&self) -> Result<()> { + self.send_server_message(MessageFactory::build_get_recommendations()) + } + + /// Ask for the server-wide recommendations (code 56); read the reply with + /// [`Client::global_recommendations`]. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn request_global_recommendations(&self) -> Result<()> { + self.send_server_message(MessageFactory::build_global_recommendations()) + } + + /// The last reply to [`Client::request_recommendations`] as + /// (recommended, recommended-against), or `None` until one arrives. + #[must_use] + pub fn recommendations( + &self, + ) -> Option<(Vec, Vec)> { + self.context + .read_safe() + .ok() + .and_then(|ctx| ctx.recommendations(false)) + } + + /// The last reply to [`Client::request_global_recommendations`]. + #[must_use] + pub fn global_recommendations( + &self, + ) -> Option<(Vec, Vec)> { + self.context + .read_safe() + .ok() + .and_then(|ctx| ctx.recommendations(true)) + } + + /// Ask what else people who like `item` like (code 111). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn request_item_recommendations(&self, item: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_item_recommendations( + item, + )) + } + + /// The last per-item recommendations received for `item`. + #[must_use] + pub fn item_recommendations(&self, item: &str) -> Vec { + match self.context.read_safe() { + Ok(ctx) => ctx.item_recommendations(item), + Err(e) => { + error!("[client] item_recommendations: {}", e); + Vec::new() + } + } + } + + /// Ask which users share our interests (code 110). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn request_similar_users(&self) -> Result<()> { + self.send_server_message(MessageFactory::build_similar_users()) + } + + /// The last similar-user answer, weighted by how much we have in common. + #[must_use] + pub fn similar_users(&self) -> Vec { + match self.context.read_safe() { + Ok(ctx) => ctx.similar_users(), + Err(e) => { + error!("[client] similar_users: {}", e); + Vec::new() + } + } + } + + /// Ask which users like `item` (code 112). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn request_item_similar_users(&self, item: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_item_similar_users(item)) + } + + /// The last answer to [`Client::request_item_similar_users`] for `item`. + #[must_use] + pub fn item_similar_users(&self, item: &str) -> Vec { + match self.context.read_safe() { + Ok(ctx) => ctx.item_similar_users(item), + Err(e) => { + error!("[client] item_similar_users: {}", e); + Vec::new() + } + } + } + + /// Ask what `username` likes and hates (code 57). + /// + /// Any previous answer for `username` is dropped first, so + /// [`Client::user_interests`] reports the reply to *this* request rather + /// than a stale one. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn request_user_interests(&self, username: &str) -> Result<()> { + self.context + .write_safe()? + .invalidate_user_interests(username); + self.send_server_message(MessageFactory::build_user_interests(username)) + } + + /// What the server last told us `username` likes and hates. + #[must_use] + pub fn user_interests(&self, username: &str) -> Option { + self.context + .read_safe() + .ok() + .and_then(|ctx| ctx.user_interests(username)) + } + + /// Give `days` of our own privileges to `username` (code 123). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn give_privileges(&self, username: &str, days: u32) -> Result<()> { + self.send_server_message(MessageFactory::build_give_privileges( + username, days, + )) + } + + /// Change our account password (code 142). + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn change_password(&self, password: &str) -> Result<()> { + self.send_server_message(MessageFactory::build_change_password( + password, + )) + } + + /// Send one private message to several users at once (code 149). Each + /// recipient receives it as an ordinary private message. + /// + /// # Errors + /// [`crate::SoulseekRs::NotConnected`] when there is no server connection. + pub fn send_private_message_to_many( + &self, + usernames: &[String], + message: &str, + ) -> Result<()> { + self.send_server_message(MessageFactory::build_message_users( + usernames, message, + )) + } +} diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index fb97d6c..b1b8285 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -838,3 +838,116 @@ fn a_peer_message_nobody_could_deliver_expires() { ctx.expire_pending_peer_messages(Instant::now() + PENDING_PEER_TTL); assert!(ctx.take_peer_messages("ghost").is_empty(), "expired"); } + +#[test] +fn a_new_ticker_replaces_that_users_previous_one() { + // The server treats a user's ticker as singular: a second one from the + // same user supersedes the first rather than stacking beside it. + let mut ctx = ClientContext::new(); + ctx.apply_room_event(RoomEvent::Tickers { + room: "jazz".into(), + tickers: vec![ + RoomTicker { + username: "alice".into(), + ticker: "first".into(), + }, + RoomTicker { + username: "bob".into(), + ticker: "bobs".into(), + }, + ], + }); + ctx.apply_room_event(RoomEvent::TickerAdded { + room: "jazz".into(), + username: "alice".into(), + ticker: "second".into(), + }); + + let board = ctx.room_tickers("jazz"); + assert_eq!(board.len(), 2); + assert_eq!( + board.iter().find(|t| t.username == "alice").unwrap().ticker, + "second" + ); +} + +#[test] +fn a_removed_ticker_leaves_the_rest_of_the_board() { + let mut ctx = ClientContext::new(); + ctx.apply_room_event(RoomEvent::Tickers { + room: "jazz".into(), + tickers: vec![ + RoomTicker { + username: "alice".into(), + ticker: "a".into(), + }, + RoomTicker { + username: "bob".into(), + ticker: "b".into(), + }, + ], + }); + ctx.apply_room_event(RoomEvent::TickerRemoved { + room: "jazz".into(), + username: "alice".into(), + }); + + let board = ctx.room_tickers("jazz"); + assert_eq!(board.len(), 1); + assert_eq!(board[0].username, "bob"); +} + +#[test] +fn a_global_message_is_queued_without_touching_room_membership() { + // The global feed carries messages from rooms we have not joined; they + // must not invent a roster for those rooms. + let mut ctx = ClientContext::new(); + ctx.apply_room_event(RoomEvent::GlobalMessage { + room: "elsewhere".into(), + username: "alice".into(), + message: "hi".into(), + }); + assert!(ctx.room_members("elsewhere").is_empty()); + assert_eq!(ctx.take_room_events().len(), 1); +} + +#[test] +fn own_and_global_recommendations_are_kept_apart() { + let mut ctx = ClientContext::new(); + ctx.apply_recommendations( + false, + vec![Recommendation { + item: "jazz".into(), + rating: 2, + }], + Vec::new(), + ); + ctx.apply_recommendations( + true, + vec![Recommendation { + item: "pop".into(), + rating: 9, + }], + Vec::new(), + ); + + assert_eq!(ctx.recommendations(false).unwrap().0[0].item, "jazz"); + assert_eq!(ctx.recommendations(true).unwrap().0[0].item, "pop"); +} + +#[test] +fn asking_again_about_a_users_interests_drops_the_previous_answer() { + let mut ctx = ClientContext::new(); + ctx.apply_user_interests(UserInterests { + username: "alice".into(), + likes: vec!["jazz".into()], + hates: Vec::new(), + }); + assert!(ctx.user_interests("alice").is_some()); + + ctx.invalidate_user_interests("alice"); + assert!( + ctx.user_interests("alice").is_none(), + "a stale answer must not be mistaken for the next one" + ); +} diff --git a/soulseek-rs-lib/src/lib.rs b/soulseek-rs-lib/src/lib.rs index f9474a5..a174fbd 100644 --- a/soulseek-rs-lib/src/lib.rs +++ b/soulseek-rs-lib/src/lib.rs @@ -22,9 +22,10 @@ pub mod utils; pub mod prelude { pub use crate::actor::server_actor::PeerAddress; pub use crate::types::{ - ClientVersion, DownloadStatus, File, RoomEvent, RoomInfo, Search, - SearchResult, Transfer, UploadInfo, UploadStatus, UserInfo, - UserPresence, UserStats, UserStatus, + ClientVersion, DownloadStatus, File, Recommendation, RoomEvent, + RoomInfo, RoomTicker, Search, SearchResult, SimilarUser, Transfer, + UploadInfo, UploadStatus, UserInfo, UserInterests, UserPresence, + UserStats, UserStatus, }; pub use crate::{debug, error, info, trace, warn}; } @@ -36,7 +37,8 @@ pub use error::{Result, SoulseekRs}; pub use message::peer::{SharedDirectory, SharedFileEntry}; pub use peer::{ConnectionType, ParseConnectionTypeError}; pub use types::{ - ClientVersion, DownloadStatus, File, RoomEvent, RoomInfo, Search, - SearchResult, SessionLoss, Transfer, UploadInfo, UploadStatus, UserInfo, - UserPresence, UserStats, UserStatus, + ClientVersion, DownloadStatus, File, Recommendation, RoomEvent, RoomInfo, + RoomTicker, Search, SearchResult, SessionLoss, SimilarUser, Transfer, + UploadInfo, UploadStatus, UserInfo, UserInterests, UserPresence, UserStats, + UserStatus, }; diff --git a/soulseek-rs-lib/src/message/server/cant_connect_to_peer.rs b/soulseek-rs-lib/src/message/server/cant_connect_to_peer.rs new file mode 100644 index 0000000..184e1f2 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/cant_connect_to_peer.rs @@ -0,0 +1,49 @@ +use std::sync::mpsc::Sender; + +use crate::{ + actor::server_actor::ServerMessage, + message::{Message, MessageHandler}, +}; + +/// `CantConnectToPeer` (code 1001): a brokered peer gave up reaching us. +/// +/// The token is the one we quoted in our own `ConnectToPeer` (code 18), and it +/// is the only field the server relays — the username is not repeated, so the +/// token is what a client matches the pending attempt on. +/// +/// Without this the download sits in "connecting" until a timeout; with it the +/// wait ends the moment the peer gives up. +pub struct CantConnectToPeerHandler; + +impl MessageHandler for CantConnectToPeerHandler { + fn get_code(&self) -> u32 { + 1001 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let token = message.read_int32(); + let _ = sender.send(ServerMessage::CantConnectToPeer { token }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::framed; + + #[test] + fn the_token_of_the_failed_attempt_is_reported() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_int32(4242); + }); + + CantConnectToPeerHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::CantConnectToPeer { token }) => { + assert_eq!(token, 4242); + } + other => panic!("unexpected: {other:?}"), + } + } +} diff --git a/soulseek-rs-lib/src/message/server/global_room.rs b/soulseek-rs-lib/src/message/server/global_room.rs new file mode 100644 index 0000000..7e8f704 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/global_room.rs @@ -0,0 +1,58 @@ +use std::sync::mpsc::Sender; + +use crate::{ + actor::server_actor::ServerMessage, + message::{Message, MessageHandler}, +}; + +/// `GlobalRoomMessage` (code 152): a message said in some public room, +/// delivered because we subscribed to the global feed (code 150). +pub struct GlobalRoomMessageHandler; + +impl MessageHandler for GlobalRoomMessageHandler { + fn get_code(&self) -> u32 { + 152 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let room = message.read_string(); + let username = message.read_string(); + let text = message.read_string(); + let _ = sender.send(ServerMessage::GlobalRoomMessageReceived { + room, + username, + message: text, + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::framed; + + #[test] + fn a_global_message_names_the_room_it_was_said_in() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("jazz"); + m.write_string("alice"); + m.write_string("anyone here?"); + }); + + GlobalRoomMessageHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::GlobalRoomMessageReceived { + room, + username, + message, + }) => { + assert_eq!( + (room.as_str(), username.as_str(), message.as_str()), + ("jazz", "alice", "anyone here?") + ); + } + other => panic!("unexpected: {other:?}"), + } + } +} diff --git a/soulseek-rs-lib/src/message/server/interests.rs b/soulseek-rs-lib/src/message/server/interests.rs new file mode 100644 index 0000000..6d623a6 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/interests.rs @@ -0,0 +1,346 @@ +//! Interests and the queries built on them: recommendations (codes 54, 56, +//! 111), similar users (110, 112) and another user's interests (57). +//! +//! Ratings are signed on the wire: the server sends a descending list of +//! things it recommends and an ascending list of things it recommends +//! against, and the latter carry negative ratings. + +use std::sync::mpsc::Sender; + +use crate::{ + actor::server_actor::ServerMessage, + message::{Message, MessageHandler}, + types::{Recommendation, SimilarUser, UserInterests}, +}; + +/// Read a `[count][item][rating]...` vector, stopping if the payload runs out +/// before the count does. +fn read_recommendations(message: &mut Message) -> Vec { + let count = message.read_int32(); + let mut items = Vec::new(); + for _ in 0..count { + let item = message.read_string(); + if item.is_empty() { + break; + } + #[allow(clippy::cast_possible_wrap)] + let rating = message.read_int32() as i32; + items.push(Recommendation { item, rating }); + } + items +} + +/// Read a `[count][string]...` vector, stopping at the end of the payload. +fn read_strings(message: &mut Message) -> Vec { + let count = message.read_int32(); + let mut items = Vec::new(); + for _ in 0..count { + let item = message.read_string(); + if item.is_empty() { + break; + } + items.push(item); + } + items +} + +/// `GetRecommendations` (code 54): recommendations drawn from our own +/// interests. +pub struct RecommendationsHandler; + +impl MessageHandler for RecommendationsHandler { + fn get_code(&self) -> u32 { + 54 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let recommended = read_recommendations(message); + let unrecommended = read_recommendations(message); + let _ = sender.send(ServerMessage::RecommendationsReceived { + recommended, + unrecommended, + }); + } +} + +/// `GlobalRecommendations` (code 56): the same shape, server-wide. +pub struct GlobalRecommendationsHandler; + +impl MessageHandler for GlobalRecommendationsHandler { + fn get_code(&self) -> u32 { + 56 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let recommended = read_recommendations(message); + let unrecommended = read_recommendations(message); + let _ = sender.send(ServerMessage::GlobalRecommendationsReceived { + recommended, + unrecommended, + }); + } +} + +/// `ItemRecommendations` (code 111): what else people who like one item like. +pub struct ItemRecommendationsHandler; + +impl MessageHandler for ItemRecommendationsHandler { + fn get_code(&self) -> u32 { + 111 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let item = message.read_string(); + let recommendations = read_recommendations(message); + let _ = sender.send(ServerMessage::ItemRecommendationsReceived { + item, + recommendations, + }); + } +} + +/// `SimilarUsers` (code 110): users whose interests overlap ours. +pub struct SimilarUsersHandler; + +impl MessageHandler for SimilarUsersHandler { + fn get_code(&self) -> u32 { + 110 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let count = message.read_int32(); + let mut users = Vec::new(); + for _ in 0..count { + let username = message.read_string(); + if username.is_empty() { + break; + } + let weight = message.read_int32(); + users.push(SimilarUser { username, weight }); + } + let _ = sender.send(ServerMessage::SimilarUsersReceived { users }); + } +} + +/// `ItemSimilarUsers` (code 112): who likes one item. The server sends only +/// names here, so every weight reads as zero. +pub struct ItemSimilarUsersHandler; + +impl MessageHandler for ItemSimilarUsersHandler { + fn get_code(&self) -> u32 { + 112 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let item = message.read_string(); + let usernames = read_strings(message); + let _ = sender + .send(ServerMessage::ItemSimilarUsersReceived { item, usernames }); + } +} + +/// `UserInterests` (code 57): what one user likes and hates. +pub struct UserInterestsHandler; + +impl MessageHandler for UserInterestsHandler { + fn get_code(&self) -> u32 { + 57 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let username = message.read_string(); + let likes = read_strings(message); + let hates = read_strings(message); + let _ = + sender.send(ServerMessage::UserInterestsReceived(UserInterests { + username, + likes, + hates, + })); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::framed; + + #[test] + fn recommendations_split_into_recommended_and_not() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_int32(1); + m.write_string("jazz"); + m.write_int32(3); + m.write_int32(1); + m.write_string("polka"); + #[allow(clippy::cast_sign_loss)] + m.write_int32(-2i32 as u32); + }); + + RecommendationsHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::RecommendationsReceived { + recommended, + unrecommended, + }) => { + assert_eq!( + recommended, + vec![Recommendation { + item: "jazz".into(), + rating: 3 + }] + ); + assert_eq!( + unrecommended, + vec![Recommendation { + item: "polka".into(), + rating: -2 + }] + ); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn global_recommendations_parse_the_same_shape() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_int32(1); + m.write_string("dub"); + m.write_int32(9); + m.write_int32(0); + }); + + GlobalRecommendationsHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::GlobalRecommendationsReceived { + recommended, + unrecommended, + }) => { + assert_eq!(recommended.len(), 1); + assert_eq!(recommended[0].item, "dub"); + assert!(unrecommended.is_empty()); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn similar_users_carry_their_weight() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_int32(2); + m.write_string("alice"); + m.write_int32(4); + m.write_string("bob"); + m.write_int32(1); + }); + + SimilarUsersHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::SimilarUsersReceived { users }) => { + assert_eq!( + users, + vec![ + SimilarUser { + username: "alice".into(), + weight: 4 + }, + SimilarUser { + username: "bob".into(), + weight: 1 + }, + ] + ); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn item_similar_users_report_the_item_they_answer_for() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("jazz"); + m.write_int32(1); + m.write_string("alice"); + }); + + ItemSimilarUsersHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::ItemSimilarUsersReceived { item, usernames }) => { + assert_eq!(item, "jazz"); + assert_eq!(usernames, vec!["alice".to_string()]); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn item_recommendations_report_their_item() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("jazz"); + m.write_int32(1); + m.write_string("blues"); + m.write_int32(2); + }); + + ItemRecommendationsHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::ItemRecommendationsReceived { + item, + recommendations, + }) => { + assert_eq!(item, "jazz"); + assert_eq!(recommendations[0].item, "blues"); + assert_eq!(recommendations[0].rating, 2); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn a_users_likes_and_hates_arrive_as_two_lists() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("alice"); + m.write_int32(2); + m.write_string("jazz"); + m.write_string("dub"); + m.write_int32(1); + m.write_string("polka"); + }); + + UserInterestsHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::UserInterestsReceived(interests)) => { + assert_eq!(interests.username, "alice"); + assert_eq!(interests.likes, ["jazz", "dub"]); + assert_eq!(interests.hates, ["polka"]); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn a_user_with_no_interests_parses_to_empty_lists() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("ghost"); + m.write_int32(0); + m.write_int32(0); + }); + + UserInterestsHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::UserInterestsReceived(interests)) => { + assert!(interests.likes.is_empty()); + assert!(interests.hates.is_empty()); + } + other => panic!("unexpected: {other:?}"), + } + } +} diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index 620508a..0f7386b 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -301,6 +301,175 @@ impl MessageFactory { .clone() } + /// A keepalive ping (server code 32, no body). The server sends no reply; + /// its only job is to keep a quiet connection from being reaped by a NAT + /// or the server's own idle timeout, the way other clients ping. + #[must_use] + pub fn build_server_ping() -> Message { + Message::new().write_int32(32).clone() + } + + /// Search one user's shares (server code 42). Replies come back over a + /// peer connection as ordinary `FileSearchResponse` frames, so the token + /// is drawn from the same space as a plain `FileSearch`. + #[must_use] + pub fn build_user_search( + username: &str, + token: u32, + query: &str, + ) -> Message { + Message::new() + .write_int32(42) + .write_string(username) + .write_int32(token) + .write_string(query) + .clone() + } + + /// Search the shares of everyone in `room` (server code 120). + #[must_use] + pub fn build_room_search(room: &str, token: u32, query: &str) -> Message { + Message::new() + .write_int32(120) + .write_string(room) + .write_int32(token) + .write_string(query) + .clone() + } + + /// Tell the server (code 1001) that the peer it asked us to connect to, + /// quoting `token`, could not be reached. The server relays that to the + /// peer so it stops waiting on a connection that is never coming. + #[must_use] + pub fn build_cant_connect_to_peer(token: u32, username: &str) -> Message { + Message::new() + .write_int32(1001) + .write_int32(token) + .write_string(username) + .clone() + } + + /// Set our ticker (the scrolling one-line message) in `room`, server code + /// 116. An empty ticker removes ours. + #[must_use] + pub fn build_set_room_ticker(room: &str, ticker: &str) -> Message { + Message::new() + .write_int32(116) + .write_string(room) + .write_string(ticker) + .clone() + } + + /// Subscribe to the global room feed (server code 150): every public room + /// message on the server, tagged with the room it came from. + #[must_use] + pub fn build_join_global_room() -> Message { + Message::new().write_int32(150).clone() + } + + /// Stop the global room feed (server code 151). + #[must_use] + pub fn build_leave_global_room() -> Message { + Message::new().write_int32(151).clone() + } + + /// Add an interest (server code 51). Interests drive the recommendation + /// and similar-user queries below. + #[must_use] + pub fn build_add_thing_i_like(item: &str) -> Message { + Message::new().write_int32(51).write_string(item).clone() + } + + /// Drop an interest (server code 52). + #[must_use] + pub fn build_remove_thing_i_like(item: &str) -> Message { + Message::new().write_int32(52).write_string(item).clone() + } + + /// Add a dislike (server code 117). + #[must_use] + pub fn build_add_thing_i_hate(item: &str) -> Message { + Message::new().write_int32(117).write_string(item).clone() + } + + /// Drop a dislike (server code 118). + #[must_use] + pub fn build_remove_thing_i_hate(item: &str) -> Message { + Message::new().write_int32(118).write_string(item).clone() + } + + /// Ask for recommendations based on our own interests (server code 54). + #[must_use] + pub fn build_get_recommendations() -> Message { + Message::new().write_int32(54).clone() + } + + /// Ask for the server-wide recommendations (server code 56). + #[must_use] + pub fn build_global_recommendations() -> Message { + Message::new().write_int32(56).clone() + } + + /// Ask what `username` likes and hates (server code 57). + #[must_use] + pub fn build_user_interests(username: &str) -> Message { + Message::new() + .write_int32(57) + .write_string(username) + .clone() + } + + /// Ask for users whose interests overlap ours (server code 110). + #[must_use] + pub fn build_similar_users() -> Message { + Message::new().write_int32(110).clone() + } + + /// Ask what else people who like `item` like (server code 111). + #[must_use] + pub fn build_item_recommendations(item: &str) -> Message { + Message::new().write_int32(111).write_string(item).clone() + } + + /// Ask which users like `item` (server code 112). + #[must_use] + pub fn build_item_similar_users(item: &str) -> Message { + Message::new().write_int32(112).write_string(item).clone() + } + + /// Hand `days` of our own privileges to `username` (server code 123). + #[must_use] + pub fn build_give_privileges(username: &str, days: u32) -> Message { + Message::new() + .write_int32(123) + .write_string(username) + .write_int32(days) + .clone() + } + + /// Change our account password (server code 142). The server echoes the + /// accepted password back under the same code. + #[must_use] + pub fn build_change_password(password: &str) -> Message { + Message::new() + .write_int32(142) + .write_string(password) + .clone() + } + + /// Send one private message to several users at once (server code 149). + #[must_use] + pub fn build_message_users(usernames: &[String], message: &str) -> Message { + let mut msg = Message::new(); + msg.write_int32(149); + msg.write_int32(usernames.len() as u32); + for username in usernames { + msg.write_string(username); + } + msg.write_string(message); + msg + } + #[must_use] pub fn build_peer_init_message( own_username: &str, @@ -517,3 +686,129 @@ fn an_upload_speed_report_carries_the_rate() { decoded.set_pointer(8); assert_eq!(decoded.read_int32(), 900); } + +#[test] +fn a_ping_is_just_its_code() { + let message = MessageFactory::build_server_ping(); + assert_eq!(message.get_data(), [32, 0, 0, 0]); +} + +#[test] +fn a_user_search_names_the_user_before_its_token() { + let message = MessageFactory::build_user_search("bob", 99, "jazz"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 42); + decoded.set_pointer(8); + assert_eq!(decoded.read_string(), "bob"); + assert_eq!(decoded.read_int32(), 99); + assert_eq!(decoded.read_string(), "jazz"); +} + +#[test] +fn a_room_search_names_the_room_before_its_token() { + let message = MessageFactory::build_room_search("nicotine", 7, "dub"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 120); + decoded.set_pointer(8); + assert_eq!(decoded.read_string(), "nicotine"); + assert_eq!(decoded.read_int32(), 7); + assert_eq!(decoded.read_string(), "dub"); +} + +#[test] +fn a_cant_connect_carries_the_token_then_the_user() { + let message = MessageFactory::build_cant_connect_to_peer(4242, "bob"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 1001); + decoded.set_pointer(8); + assert_eq!(decoded.read_int32(), 4242); + assert_eq!(decoded.read_string(), "bob"); +} + +#[test] +fn a_room_ticker_carries_its_room_and_text() { + let message = MessageFactory::build_set_room_ticker("jazz", "hello"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 116); + decoded.set_pointer(8); + assert_eq!(decoded.read_string(), "jazz"); + assert_eq!(decoded.read_string(), "hello"); +} + +#[test] +fn the_global_room_is_joined_and_left_by_code_alone() { + assert_eq!( + MessageFactory::build_join_global_room().get_data(), + [150, 0, 0, 0] + ); + assert_eq!( + MessageFactory::build_leave_global_room().get_data(), + [151, 0, 0, 0] + ); +} + +#[test] +fn each_interest_message_uses_its_own_code() { + for (message, code) in [ + (MessageFactory::build_add_thing_i_like("jazz"), 51), + (MessageFactory::build_remove_thing_i_like("jazz"), 52), + (MessageFactory::build_add_thing_i_hate("polka"), 117), + (MessageFactory::build_remove_thing_i_hate("polka"), 118), + (MessageFactory::build_user_interests("bob"), 57), + (MessageFactory::build_item_recommendations("jazz"), 111), + (MessageFactory::build_item_similar_users("jazz"), 112), + ] { + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), code); + decoded.set_pointer(8); + assert!(!decoded.read_string().is_empty()); + } +} + +#[test] +fn the_bodyless_queries_are_just_their_codes() { + assert_eq!( + MessageFactory::build_get_recommendations().get_data(), + [54, 0, 0, 0] + ); + assert_eq!( + MessageFactory::build_global_recommendations().get_data(), + [56, 0, 0, 0] + ); + assert_eq!( + MessageFactory::build_similar_users().get_data(), + [110, 0, 0, 0] + ); +} + +#[test] +fn a_privilege_gift_carries_the_recipient_and_days() { + let message = MessageFactory::build_give_privileges("bob", 3); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 123); + decoded.set_pointer(8); + assert_eq!(decoded.read_string(), "bob"); + assert_eq!(decoded.read_int32(), 3); +} + +#[test] +fn a_password_change_carries_the_new_password() { + let message = MessageFactory::build_change_password("hunter2"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 142); + decoded.set_pointer(8); + assert_eq!(decoded.read_string(), "hunter2"); +} + +#[test] +fn a_multi_user_message_counts_its_recipients_first() { + let recipients = vec!["alice".to_string(), "bob".to_string()]; + let message = MessageFactory::build_message_users(&recipients, "hi all"); + let mut decoded = Message::new_with_data(message.get_buffer()); + assert_eq!(decoded.get_message_code(), 149); + decoded.set_pointer(8); + assert_eq!(decoded.read_int32(), 2); + assert_eq!(decoded.read_string(), "alice"); + assert_eq!(decoded.read_string(), "bob"); + assert_eq!(decoded.read_string(), "hi all"); +} diff --git a/soulseek-rs-lib/src/message/server/mod.rs b/soulseek-rs-lib/src/message/server/mod.rs index 35fcc3b..76eb9e4 100644 --- a/soulseek-rs-lib/src/message/server/mod.rs +++ b/soulseek-rs-lib/src/message/server/mod.rs @@ -1,9 +1,12 @@ mod admin_message; +mod cant_connect_to_peer; mod connect_to_peer; mod distributed; mod excluded_search_phrases; mod file_search; mod get_peer_address; +mod global_room; +mod interests; mod join_room; mod leave_room; mod login; @@ -14,6 +17,7 @@ mod parent_speed_ratio; mod privileged_users; mod relogged; mod room_list; +mod room_tickers; mod say_chatroom; mod user_info; mod user_joined_room; @@ -22,6 +26,7 @@ mod watch_user; mod wish_list_interval; pub use admin_message::AdminMessageHandler; +pub use cant_connect_to_peer::CantConnectToPeerHandler; pub use connect_to_peer::ConnectToPeerHandler; pub use distributed::{ EmbeddedMessageHandler, PossibleParentsHandler, ResetDistributedHandler, @@ -29,6 +34,12 @@ pub use distributed::{ pub use excluded_search_phrases::ExcludedSearchPhrasesHandler; pub use file_search::FileSearchHandler; pub use get_peer_address::GetPeerAddressHandler; +pub use global_room::GlobalRoomMessageHandler; +pub use interests::{ + GlobalRecommendationsHandler, ItemRecommendationsHandler, + ItemSimilarUsersHandler, RecommendationsHandler, SimilarUsersHandler, + UserInterestsHandler, +}; pub use join_room::JoinRoomHandler; pub use leave_room::LeaveRoomHandler; pub use login::LoginHandler; @@ -39,6 +50,9 @@ pub use parent_speed_ratio::ParentSpeedRatioHandler; pub use privileged_users::{CheckPrivilegesHandler, PrivilegedUsersHandler}; pub use relogged::ReloggedHandler; pub use room_list::RoomListHandler; +pub use room_tickers::{ + RoomTickerAddedHandler, RoomTickerRemovedHandler, RoomTickersHandler, +}; pub use say_chatroom::SayChatroomHandler; pub use user_info::{GetUserStatsHandler, GetUserStatusHandler}; pub use user_joined_room::UserJoinedRoomHandler; diff --git a/soulseek-rs-lib/src/message/server/room_tickers.rs b/soulseek-rs-lib/src/message/server/room_tickers.rs new file mode 100644 index 0000000..d9ab1c9 --- /dev/null +++ b/soulseek-rs-lib/src/message/server/room_tickers.rs @@ -0,0 +1,174 @@ +use std::sync::mpsc::Sender; + +use crate::{ + actor::server_actor::ServerMessage, + message::{Message, MessageHandler}, + types::RoomTicker, +}; + +/// `RoomTickers` (code 113): the whole ticker board of a room, sent when we +/// join it. +pub struct RoomTickersHandler; + +impl MessageHandler for RoomTickersHandler { + fn get_code(&self) -> u32 { + 113 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let room = message.read_string(); + let count = message.read_int32(); + // A truncated or hostile count must not preallocate: grow as entries + // actually parse, and stop at the first one that runs off the end. + let mut tickers = Vec::new(); + for _ in 0..count { + let username = message.read_string(); + let ticker = message.read_string(); + if username.is_empty() { + break; + } + tickers.push(RoomTicker { username, ticker }); + } + let _ = sender.send(ServerMessage::RoomTickers { room, tickers }); + } +} + +/// `RoomTickerAdded` (code 114): one user set or replaced their ticker. +pub struct RoomTickerAddedHandler; + +impl MessageHandler for RoomTickerAddedHandler { + fn get_code(&self) -> u32 { + 114 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let room = message.read_string(); + let username = message.read_string(); + let ticker = message.read_string(); + let _ = sender.send(ServerMessage::RoomTickerAdded { + room, + username, + ticker, + }); + } +} + +/// `RoomTickerRemoved` (code 115): one user cleared their ticker. +pub struct RoomTickerRemovedHandler; + +impl MessageHandler for RoomTickerRemovedHandler { + fn get_code(&self) -> u32 { + 115 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let room = message.read_string(); + let username = message.read_string(); + let _ = + sender.send(ServerMessage::RoomTickerRemoved { room, username }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::framed; + + #[test] + fn a_ticker_board_carries_every_entry_in_order() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("jazz"); + m.write_int32(2); + m.write_string("alice"); + m.write_string("hello there"); + m.write_string("bob"); + m.write_string("back later"); + }); + + RoomTickersHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::RoomTickers { room, tickers }) => { + assert_eq!(room, "jazz"); + assert_eq!( + tickers, + vec![ + RoomTicker { + username: "alice".into(), + ticker: "hello there".into() + }, + RoomTicker { + username: "bob".into(), + ticker: "back later".into() + }, + ] + ); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn a_count_larger_than_the_payload_stops_at_the_data() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("jazz"); + m.write_int32(1_000_000); + m.write_string("alice"); + m.write_string("hi"); + }); + + RoomTickersHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::RoomTickers { tickers, .. }) => { + assert_eq!(tickers.len(), 1); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn an_added_ticker_reports_its_room_user_and_text() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("jazz"); + m.write_string("alice"); + m.write_string("new ticker"); + }); + + RoomTickerAddedHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::RoomTickerAdded { + room, + username, + ticker, + }) => { + assert_eq!( + (room.as_str(), username.as_str(), ticker.as_str()), + ("jazz", "alice", "new ticker") + ); + } + other => panic!("unexpected: {other:?}"), + } + } + + #[test] + fn a_removed_ticker_reports_its_room_and_user() { + let (tx, rx) = std::sync::mpsc::channel(); + let mut message = framed(|m| { + m.write_string("jazz"); + m.write_string("alice"); + }); + + RoomTickerRemovedHandler.handle(&mut message, tx); + match rx.try_recv() { + Ok(ServerMessage::RoomTickerRemoved { room, username }) => { + assert_eq!( + (room.as_str(), username.as_str()), + ("jazz", "alice") + ); + } + other => panic!("unexpected: {other:?}"), + } + } +} diff --git a/soulseek-rs-lib/src/peer/mod.rs b/soulseek-rs-lib/src/peer/mod.rs index 301603c..15494ba 100644 --- a/soulseek-rs-lib/src/peer/mod.rs +++ b/soulseek-rs-lib/src/peer/mod.rs @@ -64,6 +64,10 @@ pub struct Peer { pub privileged: Option, pub unknown: Option, pub obfuscated_port: Option, + /// True when the server asked us to make this connection (`ConnectToPeer`, + /// code 18). Then `token` is the far peer's correlation token, and a dial + /// that fails owes them a `CantConnectToPeer` quoting it. + pub brokered: bool, } impl Peer { #[allow(clippy::too_many_arguments)] @@ -87,6 +91,7 @@ impl Peer { privileged: Some(privileged), unknown: Some(unknown), obfuscated_port: Some(obfuscated_port), + brokered: false, } } pub fn new_from_message(message: &mut Message) -> Option { @@ -120,6 +125,7 @@ impl Peer { privileged: Some(privileged), unknown: Some(unknown), obfuscated_port: Some(u16::from(obfuscated_port)), + brokered: true, }) } } diff --git a/soulseek-rs-lib/src/types.rs b/soulseek-rs-lib/src/types.rs index 1f07ea5..c057acf 100644 --- a/soulseek-rs-lib/src/types.rs +++ b/soulseek-rs-lib/src/types.rs @@ -393,6 +393,58 @@ pub enum RoomEvent { UserJoined { room: String, username: String }, /// `username` left `room`. UserLeft { room: String, username: String }, + /// The full ticker board of `room`, sent when we join it (code 113). + Tickers { + room: String, + tickers: Vec, + }, + /// `username` set their ticker in `room` (code 114). It replaces any + /// previous ticker of theirs. + TickerAdded { + room: String, + username: String, + ticker: String, + }, + /// `username` cleared their ticker in `room` (code 115). + TickerRemoved { room: String, username: String }, + /// A message said in *any* public room, delivered because we joined the + /// global room feed (code 152). `room` names where it was said. + GlobalMessage { + room: String, + username: String, + message: String, + }, +} + +/// One user's ticker (the scrolling one-line message) in a room. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RoomTicker { + pub username: String, + pub ticker: String, +} + +/// One recommended item and how strongly the server recommends it. The rating +/// is signed: a negative rating means the item is recommended *against*. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Recommendation { + pub item: String, + pub rating: i32, +} + +/// A user the server considers similar to us, with the weight (the number of +/// interests shared) it scored them by. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SimilarUser { + pub username: String, + pub weight: u32, +} + +/// What a user likes and hates (`UserInterests`, code 57). +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct UserInterests { + pub username: String, + pub likes: Vec, + pub hates: Vec, } impl Transfer {