Merge branch 'develop'

This commit is contained in:
Michel de Graaf
2026-09-01 15:25:19 +02:00
29 changed files with 1691 additions and 78 deletions
+43 -1
View File
@@ -18,6 +18,7 @@ use crate::message::server::ReloggedHandler;
use crate::message::server::SayChatroomHandler; use crate::message::server::SayChatroomHandler;
use crate::message::server::UserJoinedRoomHandler; use crate::message::server::UserJoinedRoomHandler;
use crate::message::server::UserLeftRoomHandler; use crate::message::server::UserLeftRoomHandler;
use crate::message::server::WatchUserHandler;
use crate::message::server::WishListIntervalHandler; use crate::message::server::WishListIntervalHandler;
use crate::message::server::{ use crate::message::server::{
GetUserStatsHandler, GetUserStatusHandler, RoomListHandler, GetUserStatsHandler, GetUserStatusHandler, RoomListHandler,
@@ -27,7 +28,8 @@ use crate::message::{Message, MessageReader};
use crate::peer::ConnectionType; use crate::peer::ConnectionType;
use crate::peer::Peer; use crate::peer::Peer;
use crate::types::{ use crate::types::{
ClientVersion, RoomEvent, RoomInfo, SessionLoss, SessionWatch, ClientVersion, RoomEvent, RoomInfo, RoomUserStats, SessionLoss,
SessionWatch,
}; };
use crate::utils::lock::RwLockExt; use crate::utils::lock::RwLockExt;
@@ -202,6 +204,17 @@ pub enum ServerMessage {
status: u32, status: u32,
privileged: bool, privileged: bool,
}, },
/// A `WatchUser` (code 5) reply: the initial snapshot for a user we just
/// started watching. The stats are absent when the server does not know
/// the username.
WatchedUserReceived {
username: String,
exists: bool,
status: Option<u32>,
average_speed: Option<u32>,
shared_files: Option<u32>,
shared_folders: Option<u32>,
},
/// A `GetUserStats` (code 36) reply. /// A `GetUserStats` (code 36) reply.
UserStatsReceived { UserStatsReceived {
username: String, username: String,
@@ -213,6 +226,11 @@ pub enum ServerMessage {
room: String, room: String,
users: Vec<String>, users: Vec<String>,
}, },
/// Per-member statistics carried by the same `JoinRoom` (code 14) reply.
RoomMemberStats {
room: String,
stats: Vec<RoomUserStats>,
},
RoomLeft { RoomLeft {
room: String, room: String,
}, },
@@ -359,6 +377,7 @@ impl ServerActor {
handlers.register_handler(ReloggedHandler); handlers.register_handler(ReloggedHandler);
handlers.register_handler(RoomListHandler); handlers.register_handler(RoomListHandler);
handlers.register_handler(GetUserStatusHandler); handlers.register_handler(GetUserStatusHandler);
handlers.register_handler(WatchUserHandler);
handlers.register_handler(GetUserStatsHandler); handlers.register_handler(GetUserStatsHandler);
handlers.register_handler(JoinRoomHandler); handlers.register_handler(JoinRoomHandler);
handlers.register_handler(LeaveRoomHandler); handlers.register_handler(LeaveRoomHandler);
@@ -464,6 +483,23 @@ impl ServerActor {
privileged, privileged,
}); });
} }
ServerMessage::WatchedUserReceived {
username,
exists,
status,
average_speed,
shared_files,
shared_folders,
} => {
self.forward_to_client(ClientOperation::WatchedUserReceived {
username,
exists,
status,
average_speed,
shared_files,
shared_folders,
});
}
ServerMessage::UserStatsReceived { ServerMessage::UserStatsReceived {
username, username,
average_speed, average_speed,
@@ -480,6 +516,12 @@ impl ServerActor {
ServerMessage::RoomJoined { room, users } => { ServerMessage::RoomJoined { room, users } => {
self.forward_room_event(RoomEvent::Joined { room, users }); self.forward_room_event(RoomEvent::Joined { room, users });
} }
ServerMessage::RoomMemberStats { room, stats } => {
self.forward_to_client(ClientOperation::RoomMemberStats {
room,
stats,
});
}
ServerMessage::RoomLeft { room } => { ServerMessage::RoomLeft { room } => {
self.forward_room_event(RoomEvent::Left { room }); self.forward_room_event(RoomEvent::Left { room });
} }
+111 -1
View File
@@ -5,7 +5,8 @@ use crate::actor::server_actor::{
use crate::download_store::{DownloadStore, collect_failed_tokens}; use crate::download_store::{DownloadStore, collect_failed_tokens};
use crate::types::{ use crate::types::{
ClientVersion, DownloadMetadata, DownloadStatus, RoomEvent, RoomInfo, ClientVersion, DownloadMetadata, DownloadStatus, RoomEvent, RoomInfo,
SessionLoss, SessionWatch, UserInfo, UserPresence, UserStats, UserStatus, RoomUserStats, SessionLoss, SessionWatch, UserInfo, UserPresence,
UserStats, UserStatus,
}; };
use crate::utils::logger; use crate::utils::logger;
use crate::{ use crate::{
@@ -245,6 +246,15 @@ pub enum ClientOperation {
shared_files: u32, shared_files: u32,
shared_folders: u32, shared_folders: u32,
}, },
/// The server answered `WatchUser` for a user we started watching.
WatchedUserReceived {
username: String,
exists: bool,
status: Option<u32>,
average_speed: Option<u32>,
shared_files: Option<u32>,
shared_folders: Option<u32>,
},
PeerConnected(String), PeerConnected(String),
/// A search distributed to us by the server; reply if our shares match. /// A search distributed to us by the server; reply if our shares match.
IncomingSearch { IncomingSearch {
@@ -278,6 +288,11 @@ pub enum ClientOperation {
/// Something happened in the chat-room subsystem (list refreshed, a room /// Something happened in the chat-room subsystem (list refreshed, a room
/// joined/left, a message said, a member joined/left). /// joined/left, a message said, a member joined/left).
RoomEvent(RoomEvent), RoomEvent(RoomEvent),
/// Per-member statistics from a `JoinRoom` reply.
RoomMemberStats {
room: String,
stats: Vec<RoomUserStats>,
},
/// The server announced how many seconds must pass between wishlist /// The server announced how many seconds must pass between wishlist
/// searches. /// searches.
WishlistInterval(u32), WishlistInterval(u32),
@@ -325,9 +340,16 @@ pub struct ClientContext {
/// Who is in each room we have joined, kept current from the membership /// Who is in each room we have joined, kept current from the membership
/// the server sends on join plus the later join/leave events. /// the server sends on join plus the later join/leave events.
room_members: HashMap<String, Vec<String>>, room_members: HashMap<String, Vec<String>>,
/// Per-member statistics for each joined room, from the stat vectors the
/// server sends alongside the membership list.
room_member_stats: HashMap<String, Vec<RoomUserStats>>,
/// What the server has told us about other users, merged across the /// What the server has told us about other users, merged across the
/// separate status and statistics replies. /// separate status and statistics replies.
user_info: HashMap<String, UserInfo>, user_info: HashMap<String, UserInfo>,
/// Users we asked the server to watch (code 5), so it keeps pushing their
/// status changes. Kept so a UI can render the watch list and so an
/// unwatch can be rejected for someone we never watched.
watched_users: HashSet<String>,
/// Seconds the server wants between wishlist searches (code 104), once it /// Seconds the server wants between wishlist searches (code 104), once it
/// has told us. /// has told us.
wishlist_interval: Option<u32>, wishlist_interval: Option<u32>,
@@ -415,7 +437,9 @@ impl ClientContext {
room_list: Vec::new(), room_list: Vec::new(),
room_events: Vec::new(), room_events: Vec::new(),
room_members: HashMap::new(), room_members: HashMap::new(),
room_member_stats: HashMap::new(),
user_info: HashMap::new(), user_info: HashMap::new(),
watched_users: HashSet::new(),
wishlist_interval: None, wishlist_interval: None,
privileged_users: HashSet::new(), privileged_users: HashSet::new(),
own_privileges: None, own_privileges: None,
@@ -503,6 +527,69 @@ impl ClientContext {
}); });
} }
/// Record that we are now watching `username`.
pub fn add_watched_user(&mut self, username: &str) {
self.watched_users.insert(username.to_string());
}
/// Forget a watch, dropping any snapshot we held for that user so a later
/// re-watch reports a fresh answer rather than a stale one.
pub fn remove_watched_user(&mut self, username: &str) {
self.watched_users.remove(username);
self.user_info.remove(username);
}
/// Everyone we are currently watching, sorted by name.
#[must_use]
pub fn watched_users(&self) -> Vec<String> {
let mut users: Vec<String> =
self.watched_users.iter().cloned().collect();
users.sort();
users
}
/// Record the initial snapshot from a `WatchUser` reply. A username the
/// server does not know carries no stats and is dropped from the watch
/// list, since the server will never push anything for it.
pub fn apply_watched_user(
&mut self,
username: String,
exists: bool,
status: Option<u32>,
average_speed: Option<u32>,
shared_files: Option<u32>,
shared_folders: Option<u32>,
) {
if !exists {
self.watched_users.remove(&username);
return;
}
if let Some(status) = status {
let entry = self
.user_info
.entry(username.clone())
.or_insert_with(|| UserInfo::pending(username.clone()));
// The watch reply carries no privileged flag, so keep whatever a
// previous GetUserStatus told us rather than asserting `false`.
let privileged =
entry.presence.as_ref().is_some_and(|p| p.privileged);
entry.presence = Some(UserPresence {
status: UserStatus::from_code(status),
privileged,
});
}
if let (Some(average_speed), Some(shared_files), Some(shared_folders)) =
(average_speed, shared_files, shared_folders)
{
self.apply_user_stats(
username,
average_speed,
shared_files,
shared_folders,
);
}
}
/// What the server has said about `username` so far. /// What the server has said about `username` so far.
#[must_use] #[must_use]
pub fn user_info(&self, username: &str) -> Option<UserInfo> { pub fn user_info(&self, username: &str) -> Option<UserInfo> {
@@ -515,6 +602,29 @@ impl ClientContext {
self.room_members.get(room).cloned().unwrap_or_default() self.room_members.get(room).cloned().unwrap_or_default()
} }
/// Record the per-member statistics of a room we just joined, replacing
/// any previous snapshot for it.
pub fn apply_room_member_stats(
&mut self,
room: String,
stats: Vec<RoomUserStats>,
) {
self.room_member_stats.insert(room, stats);
}
/// What the server reported about the members of `room`, sorted by name,
/// or empty for a room we have not joined.
#[must_use]
pub fn room_member_stats(&self, room: &str) -> Vec<RoomUserStats> {
let mut stats = self
.room_member_stats
.get(room)
.cloned()
.unwrap_or_default();
stats.sort_by(|a, b| a.username.cmp(&b.username));
stats
}
/// The latest snapshot of the public chat-room list. /// The latest snapshot of the public chat-room list.
#[must_use] #[must_use]
pub fn room_list(&self) -> Vec<RoomInfo> { pub fn room_list(&self) -> Vec<RoomInfo> {
+30
View File
@@ -473,12 +473,42 @@ impl Client {
error!("[client] UserStatsReceived write: {}", e); error!("[client] UserStatsReceived write: {}", e);
} }
}, },
ClientOperation::WatchedUserReceived {
username,
exists,
status,
average_speed,
shared_files,
shared_folders,
} => match client_context.write_safe() {
Ok(mut ctx) => ctx.apply_watched_user(
username,
exists,
status,
average_speed,
shared_files,
shared_folders,
),
Err(e) => {
error!("[client] WatchedUserReceived write: {}", e);
}
},
ClientOperation::RoomEvent(event) => { ClientOperation::RoomEvent(event) => {
match client_context.write_safe() { match client_context.write_safe() {
Ok(mut ctx) => ctx.apply_room_event(event), Ok(mut ctx) => ctx.apply_room_event(event),
Err(e) => error!("[client] RoomEvent write: {}", e), Err(e) => error!("[client] RoomEvent write: {}", e),
} }
} }
ClientOperation::RoomMemberStats { room, stats } => {
match client_context.write_safe() {
Ok(mut ctx) => {
ctx.apply_room_member_stats(room, stats);
}
Err(e) => {
error!("[client] RoomMemberStats write: {}", e);
}
}
}
ClientOperation::WishlistInterval(seconds) => { ClientOperation::WishlistInterval(seconds) => {
if let Ok(mut ctx) = client_context.write_safe() { if let Ok(mut ctx) = client_context.write_safe() {
ctx.wishlist_interval = Some(seconds); ctx.wishlist_interval = Some(seconds);
+59 -1
View File
@@ -2,7 +2,7 @@ use super::{
Client, PeerMessage, Result, RoomEvent, RoomInfo, RwLockExt, ServerMessage, Client, PeerMessage, Result, RoomEvent, RoomInfo, RwLockExt, ServerMessage,
SharedDirectory, SoulseekRs, UserMessage, error, SharedDirectory, SoulseekRs, UserMessage, error,
}; };
use crate::types::UserInfo; use crate::types::{RoomUserStats, UserInfo};
impl Client { impl Client {
/// Send a private message to another user via the server. /// Send a private message to another user via the server.
@@ -138,6 +138,46 @@ impl Client {
} }
} }
/// Ask the server to watch `username` (code 5).
///
/// The server replies with that user's current status and share
/// statistics — readable with [`Client::user_info`] — and then keeps
/// pushing their status changes until [`Client::unwatch_user`]. Watching
/// a user the server does not know is answered with a "does not exist"
/// flag, and that user is dropped from [`Client::watched_users`].
///
/// # Errors
/// Returns [`SoulseekRs::NotConnected`] if the client is not connected.
pub fn watch_user(&self, username: &str) -> Result<()> {
use crate::message::server::MessageFactory;
self.send_server_message(MessageFactory::build_watch_user(username))?;
self.context.write_safe()?.add_watched_user(username);
Ok(())
}
/// Stop watching `username` (code 6) and drop what we knew about them.
///
/// # Errors
/// Returns [`SoulseekRs::NotConnected`] if the client is not connected.
pub fn unwatch_user(&self, username: &str) -> Result<()> {
use crate::message::server::MessageFactory;
self.send_server_message(MessageFactory::build_unwatch_user(username))?;
self.context.write_safe()?.remove_watched_user(username);
Ok(())
}
/// Everyone this client is currently watching, sorted by name.
#[must_use]
pub fn watched_users(&self) -> Vec<String> {
match self.context.read_safe() {
Ok(ctx) => ctx.watched_users(),
Err(e) => {
error!("[client] watched_users: {}", e);
Vec::new()
}
}
}
/// Who is currently in `room`, sorted by name. /// Who is currently in `room`, sorted by name.
/// ///
/// The roster is built from the membership the server sends when /// The roster is built from the membership the server sends when
@@ -156,6 +196,24 @@ impl Client {
} }
} }
/// What the server reported about the members of `room` when we joined:
/// their status, share statistics, free upload slots and country.
///
/// Populated from the same reply that fills [`Client::room_members`], so
/// it lands shortly after joining. Unlike the roster it is not updated by
/// later join and leave events -- it is the snapshot taken at join time.
/// Returns an empty list for a room this client has not joined.
#[must_use]
pub fn room_member_stats(&self, room: &str) -> Vec<RoomUserStats> {
match self.context.read_safe() {
Ok(ctx) => ctx.room_member_stats(room),
Err(e) => {
error!("[client] room_member_stats: {}", e);
Vec::new()
}
}
}
/// Remove and return all chat-room events received since the last call. /// Remove and return all chat-room events received since the last call.
#[must_use] #[must_use]
pub fn take_room_events(&self) -> Vec<RoomEvent> { pub fn take_room_events(&self) -> Vec<RoomEvent> {
+130
View File
@@ -622,3 +622,133 @@ fn a_replayed_transfer_response_does_not_start_a_second_transfer() {
"a replayed TransferResponse must not dial the peer again" "a replayed TransferResponse must not dial the peer again"
); );
} }
#[test]
fn room_member_stats_are_returned_sorted_and_scoped_to_their_room() {
use crate::types::{RoomUserStats, UserStatus};
let stat = |username: &str| RoomUserStats {
username: username.to_string(),
status: UserStatus::Online,
average_speed: 1,
shared_files: 2,
shared_folders: 3,
slots_full: true,
country: None,
};
let mut context = ClientContext::new();
context.apply_room_member_stats(
"jazz".to_string(),
vec![stat("carol"), stat("alice")],
);
let names: Vec<String> = context
.room_member_stats("jazz")
.into_iter()
.map(|s| s.username)
.collect();
assert_eq!(names, vec!["alice", "carol"]);
assert!(context.room_member_stats("unjoined").is_empty());
}
#[test]
fn rejoining_replaces_the_previous_member_stats() {
use crate::types::{RoomUserStats, UserStatus};
let stat = |username: &str| RoomUserStats {
username: username.to_string(),
status: UserStatus::Online,
average_speed: 1,
shared_files: 2,
shared_folders: 3,
slots_full: false,
country: None,
};
let mut context = ClientContext::new();
context.apply_room_member_stats("jazz".to_string(), vec![stat("alice")]);
context.apply_room_member_stats("jazz".to_string(), vec![stat("bob")]);
let stats = context.room_member_stats("jazz");
assert_eq!(stats.len(), 1, "a rejoin supersedes the old snapshot");
assert_eq!(stats[0].username, "bob");
}
#[test]
fn a_watch_reply_fills_both_halves_of_the_snapshot() {
let mut context = ClientContext::new();
context.add_watched_user("alice");
context.apply_watched_user(
"alice".to_string(),
true,
Some(2),
Some(1024),
Some(20),
Some(3),
);
let info = context.user_info("alice").expect("a snapshot");
assert!(info.is_complete(), "a watch reply carries status and stats");
assert_eq!(info.stats.map(|s| s.average_speed), Some(1024));
assert_eq!(context.watched_users(), vec!["alice".to_string()]);
}
#[test]
fn a_watch_reply_keeps_a_privileged_flag_it_cannot_carry() {
// WatchUser has no privileged field, so it must not overwrite what
// GetUserStatus already told us with a fabricated `false`.
let mut context = ClientContext::new();
context.apply_user_status("bob".to_string(), 1, true);
context.apply_watched_user(
"bob".to_string(),
true,
Some(2),
Some(1),
Some(2),
Some(3),
);
let info = context.user_info("bob").expect("a snapshot");
assert_eq!(info.presence.map(|p| p.privileged), Some(true));
}
#[test]
fn watching_an_unknown_user_drops_them_from_the_list() {
// The server will never push status for a name it does not know, so
// keeping it in the watch list would show a permanently blank row.
let mut context = ClientContext::new();
context.add_watched_user("ghost");
context.apply_watched_user(
"ghost".to_string(),
false,
None,
None,
None,
None,
);
assert!(context.watched_users().is_empty());
assert!(context.user_info("ghost").is_none());
}
#[test]
fn unwatching_forgets_the_users_snapshot() {
let mut context = ClientContext::new();
context.add_watched_user("alice");
context.apply_user_status("alice".to_string(), 2, false);
context.remove_watched_user("alice");
assert!(context.watched_users().is_empty());
assert!(
context.user_info("alice").is_none(),
"a later re-watch must report a fresh answer"
);
}
#[test]
fn watched_users_are_listed_in_a_stable_order() {
let mut context = ClientContext::new();
context.add_watched_user("carol");
context.add_watched_user("alice");
context.add_watched_user("bob");
assert_eq!(context.watched_users(), vec!["alice", "bob", "carol"]);
}
+247 -13
View File
@@ -1,6 +1,7 @@
use crate::{ use crate::{
actor::server_actor::ServerMessage, actor::server_actor::ServerMessage,
message::{Message, MessageHandler}, message::{Message, MessageHandler},
types::{RoomUserStats, UserStatus},
}; };
use std::sync::mpsc::Sender; use std::sync::mpsc::Sender;
@@ -12,24 +13,105 @@ impl MessageHandler<ServerMessage> for JoinRoomHandler {
} }
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) { fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
// JoinRoom (code 14): room name, then a vector of member usernames. // JoinRoom (code 14): room name, the member usernames, then the
// Per-user stat vectors follow but are not needed here, so we stop // parallel per-member stat vectors.
// after reading the names.
let room = message.read_string(); let room = message.read_string();
let user_count = message.read_int32(); let users = read_strings(message);
let mut users = Vec::new(); let stats = parse_member_stats(message, &users);
for _ in 0..user_count { let _ = sender.send(ServerMessage::RoomJoined {
// Guard against a hostile user_count outrunning the payload; each room: room.clone(),
// username is a length-prefixed string (>= 4 bytes). users,
if message.get_pointer() + 4 > message.get_size() { });
break; if !stats.is_empty() {
} let _ = sender.send(ServerMessage::RoomMemberStats { room, stats });
users.push(message.read_string());
} }
let _ = sender.send(ServerMessage::RoomJoined { room, users });
} }
} }
/// Read a length-prefixed vector of strings, stopping early if the count
/// outruns the payload: a bogus (possibly hostile) count must not spin us
/// into an OOM allocation loop. A 4-byte floor is the smallest a
/// length-prefixed string can be.
fn read_strings(message: &mut Message) -> Vec<String> {
let count = message.read_int32();
let mut values = Vec::new();
for _ in 0..count {
if message.get_pointer() + 4 > message.get_size() {
break;
}
values.push(message.read_string());
}
values
}
/// Read a length-prefixed vector of 32-bit values, bounded the same way.
fn read_int32s(message: &mut Message) -> Vec<u32> {
let count = message.read_int32();
let mut values = Vec::new();
for _ in 0..count {
if message.get_pointer() + 4 > message.get_size() {
break;
}
values.push(message.read_int32());
}
values
}
/// Parse the per-member statistics that follow the membership list of a
/// `JoinRoom` (code 14) reply: statuses, then a record of share statistics per
/// member, then whether each member's upload slots are full, then country
/// codes.
///
/// Each vector is independently length-prefixed and a server may send fewer
/// entries than there are members (or, on an older server, omit the country
/// vector entirely), so every field is looked up positionally and defaults
/// when absent rather than shifting the whole roster.
#[must_use]
fn parse_member_stats(
message: &mut Message,
users: &[String],
) -> Vec<RoomUserStats> {
let statuses = read_int32s(message);
let stat_count = message.read_int32();
let mut shares = Vec::new();
for _ in 0..stat_count {
// A share record is 20 bytes: speed, an obsolete 64-bit upload
// number, files and folders.
if message.get_pointer() + 20 > message.get_size() {
break;
}
let average_speed = message.read_int32();
let _upload_number = message.read_int64(); // obsolete
let shared_files = message.read_int32();
let shared_folders = message.read_int32();
shares.push((average_speed, shared_files, shared_folders));
}
let slots_full = read_int32s(message);
let countries = read_strings(message);
users
.iter()
.enumerate()
.map(|(i, username)| {
let (average_speed, shared_files, shared_folders) =
shares.get(i).copied().unwrap_or((0, 0, 0));
RoomUserStats {
username: username.clone(),
status: UserStatus::from_code(
statuses.get(i).copied().unwrap_or(0),
),
average_speed,
shared_files,
shared_folders,
slots_full: slots_full.get(i).copied().unwrap_or(0) != 0,
country: countries.get(i).cloned().filter(|c| !c.is_empty()),
}
})
.collect()
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@@ -53,6 +135,158 @@ mod tests {
} }
} }
/// A full JoinRoom payload: room, members, then the four parallel stat
/// vectors the server sends for them.
fn join_payload() -> Message {
framed(|m| {
m.write_string("nicotine")
.write_int32(2)
.write_string("alice")
.write_string("bob")
// statuses
.write_int32(2)
.write_int32(2)
.write_int32(1)
// share records
.write_int32(2)
.write_int32(1024)
.write_int64(0)
.write_int32(4321)
.write_int32(77)
.write_int32(50)
.write_int64(0)
.write_int32(10)
.write_int32(2)
// slots full
.write_int32(2)
.write_int32(1)
.write_int32(0)
// countries
.write_int32(2)
.write_string("NL")
.write_string("");
})
}
#[test]
fn member_stats_are_parsed_alongside_the_roster() {
let (tx, rx) = std::sync::mpsc::channel();
JoinRoomHandler.handle(&mut join_payload(), tx);
assert!(matches!(
rx.try_recv(),
Ok(ServerMessage::RoomJoined { .. })
));
match rx.try_recv() {
Ok(ServerMessage::RoomMemberStats { room, stats }) => {
assert_eq!(room, "nicotine");
assert_eq!(
stats,
vec![
RoomUserStats {
username: "alice".to_string(),
status: UserStatus::Online,
average_speed: 1024,
shared_files: 4321,
shared_folders: 77,
slots_full: true,
country: Some("NL".to_string()),
},
RoomUserStats {
username: "bob".to_string(),
status: UserStatus::Away,
average_speed: 50,
shared_files: 10,
shared_folders: 2,
slots_full: false,
country: None,
},
]
);
}
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn a_server_that_omits_the_stat_vectors_still_yields_a_roster() {
// Nothing after the membership list: every stat reads as a default
// rather than shifting the roster or dropping members.
let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| {
m.write_string("jazz").write_int32(1).write_string("alice");
});
JoinRoomHandler.handle(&mut message, tx);
assert!(matches!(
rx.try_recv(),
Ok(ServerMessage::RoomJoined { .. })
));
match rx.try_recv() {
Ok(ServerMessage::RoomMemberStats { stats, .. }) => {
assert_eq!(stats.len(), 1);
assert_eq!(stats[0].username, "alice");
assert_eq!(stats[0].average_speed, 0);
assert_eq!(stats[0].country, None);
}
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn short_stat_vectors_do_not_shift_the_roster() {
// Stats for the first member only: the second must still be reported,
// with defaults, rather than inheriting the first member's figures.
let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| {
m.write_string("jazz")
.write_int32(2)
.write_string("alice")
.write_string("bob")
.write_int32(1)
.write_int32(2)
.write_int32(1)
.write_int32(1024)
.write_int64(0)
.write_int32(9)
.write_int32(1)
.write_int32(0)
.write_int32(0);
});
JoinRoomHandler.handle(&mut message, tx);
let _ = rx.try_recv();
match rx.try_recv() {
Ok(ServerMessage::RoomMemberStats { stats, .. }) => {
assert_eq!(stats.len(), 2);
assert_eq!(stats[0].shared_files, 9);
assert_eq!(stats[1].username, "bob");
assert_eq!(stats[1].shared_files, 0);
assert_eq!(stats[1].status, UserStatus::Offline);
}
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn hostile_stat_counts_do_not_hang_or_overallocate() {
// Vectors claiming ~4 billion entries in a tiny frame must return
// promptly, bounded by the payload.
let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| {
m.write_string("jazz")
.write_int32(1)
.write_string("alice")
.write_int32(u32::MAX);
});
JoinRoomHandler.handle(&mut message, tx);
assert!(matches!(
rx.try_recv(),
Ok(ServerMessage::RoomJoined { .. })
));
}
#[test] #[test]
fn forwards_room_and_member_list() { fn forwards_room_and_member_list() {
let (tx, rx) = std::sync::mpsc::channel(); let (tx, rx) = std::sync::mpsc::channel();
@@ -152,6 +152,20 @@ impl MessageFactory {
.clone() .clone()
} }
/// Watch `username` (server code 5): the server replies with their
/// current status and share statistics, then pushes every later status
/// change as a `GetUserStatus` (code 7) until we unwatch them.
#[must_use]
pub fn build_watch_user(username: &str) -> Message {
Message::new().write_int32(5).write_string(username).clone()
}
/// Stop watching `username` (server code 6). The server sends no reply.
#[must_use]
pub fn build_unwatch_user(username: &str) -> Message {
Message::new().write_int32(6).write_string(username).clone()
}
/// Ask the server (code 64) for the list of public chat rooms. /// Ask the server (code 64) for the list of public chat rooms.
#[must_use] #[must_use]
pub fn build_room_list_request() -> Message { pub fn build_room_list_request() -> Message {
@@ -424,3 +438,17 @@ fn test_build_file_search_message() {
.to_vec(); .to_vec();
assert_eq!(expect, message.get_data()); assert_eq!(expect, message.get_data());
} }
#[test]
fn test_build_watch_user() {
let message = MessageFactory::build_watch_user("bob");
let expect: Vec<u8> = [5, 0, 0, 0, 3, 0, 0, 0, b'b', b'o', b'b'].to_vec();
assert_eq!(expect, message.get_data());
}
#[test]
fn test_build_unwatch_user() {
let message = MessageFactory::build_unwatch_user("bob");
let expect: Vec<u8> = [6, 0, 0, 0, 3, 0, 0, 0, b'b', b'o', b'b'].to_vec();
assert_eq!(expect, message.get_data());
}
@@ -16,6 +16,7 @@ mod say_chatroom;
mod user_info; mod user_info;
mod user_joined_room; mod user_joined_room;
mod user_left_room; mod user_left_room;
mod watch_user;
mod wish_list_interval; mod wish_list_interval;
pub use connect_to_peer::ConnectToPeerHandler; pub use connect_to_peer::ConnectToPeerHandler;
@@ -36,4 +37,5 @@ pub use say_chatroom::SayChatroomHandler;
pub use user_info::{GetUserStatsHandler, GetUserStatusHandler}; pub use user_info::{GetUserStatsHandler, GetUserStatusHandler};
pub use user_joined_room::UserJoinedRoomHandler; pub use user_joined_room::UserJoinedRoomHandler;
pub use user_left_room::UserLeftRoomHandler; pub use user_left_room::UserLeftRoomHandler;
pub use watch_user::WatchUserHandler;
pub use wish_list_interval::WishListIntervalHandler; pub use wish_list_interval::WishListIntervalHandler;
@@ -0,0 +1,146 @@
//! `WatchUser` (code 5): subscribe to a user's status and share statistics.
//!
//! Watching is how a client keeps a buddy list current. After a `WatchUser`
//! request the server answers on the same code with the user's current
//! presence and stats, and from then on pushes every later change as an
//! ordinary `GetUserStatus` (code 7). `UnwatchUser` (code 6) cancels the
//! subscription and has no reply.
//!
//! Note on codes: the protocol has no `AddUserList`/`WatchedUserAdded`
//! /`WatchedUserRemoved` messages — 141/142/143 are `EnableRoomInvitations`,
//! `ChangePassword` and `AddRoomOperator`. Watching is 5/6, per the reference
//! protocol documentation.
use crate::actor::server_actor::ServerMessage;
use crate::message::{Message, MessageHandler};
use std::sync::mpsc::Sender;
/// Receives the `WatchUser` reply (code 5): `username`, `exists`, and — when
/// the user exists — status plus the same share statistics `GetUserStats`
/// carries.
pub struct WatchUserHandler;
impl MessageHandler<ServerMessage> for WatchUserHandler {
fn get_code(&self) -> u8 {
5
}
fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
let username = message.read_string();
let exists = message.read_bool();
let watched = if exists {
let status = message.read_int32();
let average_speed = message.read_int32();
// Two obsolete 32-bit fields, exactly as GetUserStats (36)
// carries them -- not one 64-bit value. The byte count is the
// same either way, but reading them as the server writes them
// keeps this handler and its sibling honest.
let _upload_number = message.read_int32(); // obsolete, always 0
let _unknown = message.read_int32(); // obsolete, always 0
let shared_files = message.read_int32();
let shared_folders = message.read_int32();
ServerMessage::WatchedUserReceived {
username,
exists,
status: Some(status),
average_speed: Some(average_speed),
shared_files: Some(shared_files),
shared_folders: Some(shared_folders),
}
} else {
// An unknown username: the server stops after the flag, so there
// is nothing further to read and nothing to record.
ServerMessage::WatchedUserReceived {
username,
exists,
status: None,
average_speed: None,
shared_files: None,
shared_folders: None,
}
};
let _ = sender.send(watched);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn existing_user_carries_status_and_stats() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| {
m.write_string("alice")
.write_int8(1) // exists
.write_int32(2) // status: online
.write_int32(1024) // average speed
.write_int32(0) // obsolete upload number
.write_int32(0) // obsolete unknown
.write_int32(4321) // shared files
.write_int32(77); // shared folders
});
WatchUserHandler.handle(&mut message, tx);
match rx.try_recv() {
Ok(ServerMessage::WatchedUserReceived {
username,
exists,
status,
average_speed,
shared_files,
shared_folders,
}) => {
assert_eq!(username, "alice");
assert!(exists);
assert_eq!(status, Some(2));
assert_eq!(average_speed, Some(1024));
assert_eq!(shared_files, Some(4321));
assert_eq!(shared_folders, Some(77));
}
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn unknown_user_reports_absence_without_stats() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| {
m.write_string("ghost").write_int8(0);
});
WatchUserHandler.handle(&mut message, tx);
match rx.try_recv() {
Ok(ServerMessage::WatchedUserReceived {
username,
exists,
status,
shared_files,
..
}) => {
assert_eq!(username, "ghost");
assert!(!exists);
assert_eq!(status, None);
assert_eq!(shared_files, None);
}
other => panic!("unexpected: {other:?}"),
}
}
#[test]
fn truncated_payload_does_not_panic() {
// exists = 1 but the stats were cut off: the bounds-checked readers
// must yield zeros rather than panicking on a hostile frame.
let (tx, rx) = std::sync::mpsc::channel();
let mut message = framed(|m| {
m.write_string("alice").write_int8(1).write_int32(2);
});
WatchUserHandler.handle(&mut message, tx);
assert!(matches!(
rx.try_recv(),
Ok(ServerMessage::WatchedUserReceived { .. })
));
}
}
+21
View File
@@ -328,6 +328,27 @@ impl UserInfo {
} }
} }
/// What the server reports about one member of a room we joined, from the
/// per-member stat vectors of a `JoinRoom` (code 14) reply.
///
/// A server may send shorter stat vectors than the membership list (older
/// servers omit country codes entirely), so absent fields read as zero and an
/// unreported country is `None` rather than an empty string.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoomUserStats {
pub username: String,
pub status: UserStatus,
pub average_speed: u32,
pub shared_files: u32,
pub shared_folders: u32,
/// True when the member's upload slots are all occupied, so a download
/// from them will queue. The wire field is `slotsfull`, a boolean — not a
/// count, and not a count of *free* slots.
pub slots_full: bool,
/// Two-letter country code, when the server reports one.
pub country: Option<String>,
}
/// A public chat room advertised by the server (`RoomList`, code 64). /// A public chat room advertised by the server (`RoomList`, code 64).
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoomInfo { pub struct RoomInfo {
+179
View File
@@ -2622,3 +2622,182 @@ fn a_peer_that_hangs_up_its_control_connection_still_delivers_the_file() {
let _ = std::fs::remove_dir_all(&download_dir); let _ = std::fs::remove_dir_all(&download_dir);
} }
#[test]
fn a_joined_room_reports_its_members_statistics() {
let server = server_or_skip!();
let room = "e2e_room_stats";
// A member with a known share, so the statistics are checkable rather
// than merely present.
let share_dir = unique_download_dir().join("stats_share");
std::fs::create_dir_all(share_dir.join("album")).unwrap();
for name in ["one.flac", "two.flac", "three.flac"] {
std::fs::write(share_dir.join("album").join(name), b"xxxx").unwrap();
}
let mut sharer = Client::with_settings(ClientSettings {
shared_directories: vec![share_dir.display().to_string()],
..server.settings("e2e_stats_sharer", "pw")
});
sharer.connect().expect("sharer connect");
assert!(sharer.login().expect("sharer login"));
sharer.join_room(room).expect("sharer joins room");
// Let the server register the sharer's membership and share counts, so
// the observer's own join reply describes a room that already has them.
std::thread::sleep(Duration::from_millis(750));
let mut observer =
Client::with_settings(server.settings("e2e_stats_observer", "pw"));
observer.connect().expect("observer connect");
assert!(observer.login().expect("observer login"));
observer.join_room(room).expect("observer joins room");
// The stats ride along with the membership list, so poll for the member
// rather than for a fixed delay.
let deadline = Instant::now() + Duration::from_secs(15);
let mut sharer_stats = None;
while Instant::now() < deadline {
if let Some(found) = observer
.room_member_stats(room)
.into_iter()
.find(|s| s.username == "e2e_stats_sharer")
{
sharer_stats = Some(found);
break;
}
std::thread::sleep(Duration::from_millis(100));
}
let stats =
sharer_stats.expect("the join reply should describe the members");
assert!(
stats.status.is_reachable(),
"a member who is in the room cannot read as offline, got {}",
stats.status
);
assert_eq!(
stats.shared_files, 3,
"the member's file count must survive the parallel stat vectors"
);
assert_eq!(stats.shared_folders, 1);
assert!(
!stats.slots_full,
"a client that never filled its slots must not read as full"
);
// Every member is described, not just the one we looked up: a short or
// misread vector would drop or shift the rest.
let members = observer.room_member_stats(room);
let names: Vec<&str> =
members.iter().map(|s| s.username.as_str()).collect();
assert!(
names.contains(&"e2e_stats_sharer")
&& names.contains(&"e2e_stats_observer"),
"both members should be described, got {names:?}"
);
assert_eq!(
members.len(),
observer.room_members(room).len(),
"the stats must line up one-to-one with the roster"
);
}
#[test]
fn watching_a_user_returns_their_status_and_share_counts() {
let server = server_or_skip!();
// A subject with a known share, so the watch reply's statistics are
// checkable rather than merely present.
let share_dir = unique_download_dir().join("watch_share");
std::fs::create_dir_all(share_dir.join("album")).unwrap();
for name in ["one.flac", "two.flac"] {
std::fs::write(share_dir.join("album").join(name), b"xxxx").unwrap();
}
let mut subject = Client::with_settings(ClientSettings {
shared_directories: vec![share_dir.display().to_string()],
..server.settings("e2e_watch_subject", "pw")
});
subject.connect().expect("subject connect");
assert!(subject.login().expect("subject login"));
let mut watcher =
Client::with_settings(server.settings("e2e_watch_watcher", "pw"));
watcher.connect().expect("watcher connect");
assert!(watcher.login().expect("watcher login"));
// Let the subject's share counts reach the server before watching, so
// the reply describes a user it already knows the statistics for.
std::thread::sleep(Duration::from_millis(750));
watcher
.watch_user("e2e_watch_subject")
.expect("watch the subject");
assert_eq!(
watcher.watched_users(),
vec!["e2e_watch_subject".to_string()]
);
// One reply carries both halves, unlike the two GetUserStatus and
// GetUserStats answers request_user_info waits on.
let deadline = Instant::now() + Duration::from_secs(15);
let mut info = None;
while Instant::now() < deadline {
match watcher.user_info("e2e_watch_subject") {
Some(found) if found.is_complete() => {
info = Some(found);
break;
}
_ => std::thread::sleep(Duration::from_millis(100)),
}
}
let info = info.expect("the watch reply should carry status and stats");
let presence = info.presence.expect("presence should have arrived");
assert!(
presence.status.is_reachable(),
"a logged-in user should not read as offline, got {}",
presence.status
);
let stats = info.stats.expect("stats should have arrived");
assert_eq!(
stats.shared_files, 2,
"the file count must survive the obsolete fields ahead of it"
);
assert_eq!(stats.shared_folders, 1);
// Unwatching drops the user and what we knew about them, so a later
// re-watch reports a fresh answer rather than this stale one.
watcher
.unwatch_user("e2e_watch_subject")
.expect("unwatch the subject");
assert!(watcher.watched_users().is_empty());
assert!(watcher.user_info("e2e_watch_subject").is_none());
}
#[test]
fn watching_a_user_the_server_does_not_know_reports_their_absence() {
let server = server_or_skip!();
let mut watcher =
Client::with_settings(server.settings("e2e_watch_ghost_asker", "pw"));
watcher.connect().expect("watcher connect");
assert!(watcher.login().expect("watcher login"));
watcher
.watch_user("e2e_watch_nobody_here")
.expect("watch an unknown user");
// The server answers that it has no such user; the watch is dropped
// rather than left showing a permanently blank row.
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline && !watcher.watched_users().is_empty() {
std::thread::sleep(Duration::from_millis(100));
}
assert!(
watcher.watched_users().is_empty(),
"a username the server does not know must not stay watched"
);
assert!(watcher.user_info("e2e_watch_nobody_here").is_none());
}
+148
View File
@@ -525,6 +525,21 @@
], ],
"type": "object" "type": "object"
}, },
"MemberStats": {
"description": "What the server said about each member of a room when it was joined.",
"properties": {
"members": {
"items": {
"$ref": "#/components/schemas/RoomUserStatsDto"
},
"type": "array"
}
},
"required": [
"members"
],
"type": "object"
},
"Members": { "Members": {
"properties": { "properties": {
"users": { "users": {
@@ -733,6 +748,52 @@
], ],
"type": "object" "type": "object"
}, },
"RoomUserStatsDto": {
"description": "One member of a room, as the JoinRoom reply described them.",
"properties": {
"average_speed": {
"format": "uint32",
"minimum": 0,
"type": "integer"
},
"country": {
"type": [
"string",
"null"
]
},
"shared_files": {
"format": "uint32",
"minimum": 0,
"type": "integer"
},
"shared_folders": {
"format": "uint32",
"minimum": 0,
"type": "integer"
},
"slots_full": {
"description": "True when the member's upload slots are all occupied.",
"type": "boolean"
},
"status": {
"description": "`online`, `away`, or `offline`.",
"type": "string"
},
"username": {
"type": "string"
}
},
"required": [
"username",
"status",
"average_speed",
"shared_files",
"shared_folders",
"slots_full"
],
"type": "object"
},
"SayParams": { "SayParams": {
"properties": { "properties": {
"message": { "message": {
@@ -1216,6 +1277,21 @@
"shared_folders" "shared_folders"
], ],
"type": "object" "type": "object"
},
"Watched": {
"description": "Everyone this session is watching, sorted by name.",
"properties": {
"users": {
"items": {
"type": "string"
},
"type": "array"
}
},
"required": [
"users"
],
"type": "object"
} }
} }
}, },
@@ -1776,6 +1852,26 @@
}, },
"summary": "Who is currently in a room." "summary": "Who is currently in a room."
}, },
{
"name": "room.member_stats",
"paramStructure": "by-name",
"params": [
{
"name": "room",
"required": true,
"schema": {
"type": "string"
}
}
],
"result": {
"name": "result",
"schema": {
"$ref": "#/components/schemas/MemberStats"
}
},
"summary": "What the server said about each member of a joined room."
},
{ {
"name": "message.send", "name": "message.send",
"paramStructure": "by-name", "paramStructure": "by-name",
@@ -1875,6 +1971,58 @@
}, },
"summary": "What the server has said about a user so far." "summary": "What the server has said about a user so far."
}, },
{
"name": "user.watch",
"paramStructure": "by-name",
"params": [
{
"name": "username",
"required": true,
"schema": {
"type": "string"
}
}
],
"result": {
"name": "result",
"schema": {
"$ref": "#/components/schemas/Ack"
}
},
"summary": "Watch a user, so the server keeps pushing their status."
},
{
"name": "user.unwatch",
"paramStructure": "by-name",
"params": [
{
"name": "username",
"required": true,
"schema": {
"type": "string"
}
}
],
"result": {
"name": "result",
"schema": {
"$ref": "#/components/schemas/Ack"
}
},
"summary": "Stop watching a user."
},
{
"name": "user.watched",
"paramStructure": "by-name",
"params": [],
"result": {
"name": "result",
"schema": {
"$ref": "#/components/schemas/Watched"
}
},
"summary": "Everyone this session is watching."
},
{ {
"name": "shares.status", "name": "shares.status",
"paramStructure": "by-name", "paramStructure": "by-name",
+12 -1
View File
@@ -133,6 +133,13 @@ wishlist later rather than repeating the same search.
be `null` when the server has not answered that half — `null` means unknown, be `null` when the server has not answered that half — `null` means unknown,
not offline. not offline.
**`watch add <user>`** / **`watch remove <user>`** — prints nothing; exit 0 is
the confirmation. **`watch list`** — one object per watched user: `user`,
`status`. A watch lasts only as long as the session holding it, so these are
worth using against a running daemon (`--daemon`), where the connection stays
up and keeps receiving the server's status pushes. A one-shot run's watch ends
with the run.
**`whoami`** — one object: `user`, `server`, `listening`, `listen_port`, **`whoami`** — one object: `user`, `server`, `listening`, `listen_port`,
`shared_folders`, `shared_files`, `download_dir`, `privilege_seconds`. The `shared_folders`, `shared_files`, `download_dir`, `privilege_seconds`. The
cheapest way to prove the credentials and connection work. cheapest way to prove the credentials and connection work.
@@ -140,7 +147,11 @@ cheapest way to prove the credentials and connection work.
server did not answer — those are different facts, so do not treat null as 0. server did not answer — those are different facts, so do not treat null as 0.
**`room list`** — `room`, `users`. **`room list`** — `room`, `users`.
**`room users <room>`** — `room`, `user`. **`room users <room>`** — `room`, `user`, `status`, `average_speed`,
`shared_files`, `shared_folders`, `slots_full`, `country`. Everything but
`room` and `user` is `null` when the server sent no statistics for that
member — `null` means unknown, not zero. `slots_full` is true when the peer's
upload slots are all occupied, so a download from them will queue.
**`room listen <room>`** — streams `type`, `room`, `user`, `message`. **`room listen <room>`** — streams `type`, `room`, `user`, `message`.
**`room say <room> <message>`** — prints nothing; exit 0 is the confirmation. **`room say <room> <message>`** — prints nothing; exit 0 is the confirmation.
+26 -1
View File
@@ -13,7 +13,7 @@
//! remote session has no borrowed client to hand a `&str` out of. //! remote session has no borrowed client to hand a `&str` out of.
use crate::daemon::proto::ChatMessageDto; use crate::daemon::proto::ChatMessageDto;
use soulseek_rs::types::{Download, DownloadMetadata}; use soulseek_rs::types::{Download, DownloadMetadata, RoomUserStats};
use soulseek_rs::{ use soulseek_rs::{
Client, DownloadStatus, Result, RoomEvent, SearchResult, SessionLoss, Client, DownloadStatus, Result, RoomEvent, SearchResult, SessionLoss,
SharedDirectory, UploadInfo, UserInfo, UserMessage, SharedDirectory, UploadInfo, UserInfo, UserMessage,
@@ -126,6 +126,9 @@ pub trait SessionApi: Send + Sync {
fn leave_room(&self, room: &str) -> Result<()>; fn leave_room(&self, room: &str) -> Result<()>;
fn say_in_room(&self, room: &str, message: &str) -> Result<()>; fn say_in_room(&self, room: &str, message: &str) -> Result<()>;
fn room_members(&self, room: &str) -> Vec<String>; fn room_members(&self, room: &str) -> Vec<String>;
/// What the server said about each member when the room was joined:
/// status, share statistics, free upload slots and country.
fn room_member_stats(&self, room: &str) -> Vec<RoomUserStats>;
fn take_room_events(&self) -> Vec<RoomEvent>; fn take_room_events(&self) -> Vec<RoomEvent>;
fn send_private_message(&self, username: &str, message: &str) fn send_private_message(&self, username: &str, message: &str)
@@ -140,6 +143,12 @@ pub trait SessionApi: Send + Sync {
fn request_user_info(&self, username: &str) -> Result<()>; fn request_user_info(&self, username: &str) -> Result<()>;
fn user_info(&self, username: &str) -> Option<UserInfo>; fn user_info(&self, username: &str) -> Option<UserInfo>;
/// Ask the server to keep pushing `username`'s status until unwatched.
fn watch_user(&self, username: &str) -> Result<()>;
fn unwatch_user(&self, username: &str) -> Result<()>;
/// Everyone this session is watching, sorted by name.
fn watched_users(&self) -> Vec<String>;
/// The conversation this session already knows about. /// The conversation this session already knows about.
/// ///
/// Empty locally, where the TUI's own snapshot on disk is the record. A /// Empty locally, where the TUI's own snapshot on disk is the record. A
@@ -329,6 +338,10 @@ impl SessionApi for Client {
Self::room_members(self, room) Self::room_members(self, room)
} }
fn room_member_stats(&self, room: &str) -> Vec<RoomUserStats> {
Self::room_member_stats(self, room)
}
fn take_room_events(&self) -> Vec<RoomEvent> { fn take_room_events(&self) -> Vec<RoomEvent> {
Self::take_room_events(self) Self::take_room_events(self)
} }
@@ -364,6 +377,18 @@ impl SessionApi for Client {
Self::user_info(self, username) Self::user_info(self, username)
} }
fn watch_user(&self, username: &str) -> Result<()> {
Self::watch_user(self, username)
}
fn unwatch_user(&self, username: &str) -> Result<()> {
Self::unwatch_user(self, username)
}
fn watched_users(&self) -> Vec<String> {
Self::watched_users(self)
}
fn shared_counts(&self) -> (u32, u32) { fn shared_counts(&self) -> (u32, u32) {
Self::shared_counts(self) Self::shared_counts(self)
} }
+24
View File
@@ -207,6 +207,10 @@ pub enum Commands {
/// What the server knows about another user /// What the server knows about another user
User(UserArgs), User(UserArgs),
/// Keep a daemon updated about users you care about
#[command(subcommand)]
Watch(WatchCommand),
/// Confirm the credentials and connection work /// Confirm the credentials and connection work
Whoami, Whoami,
@@ -571,6 +575,26 @@ pub struct WishRunArgs {
pub sort: SortKey, pub sort: SortKey,
} }
/// Watching only lasts as long as the session doing it, so these are aimed at
/// a running daemon: it stays connected and keeps receiving the status pushes.
#[derive(Subcommand, Debug)]
pub enum WatchCommand {
/// Start watching a user
Add {
/// The user to watch
user: String,
},
/// Stop watching a user
Remove {
/// The user to stop watching
user: String,
},
/// List the users being watched
List,
}
#[derive(Subcommand, Debug)] #[derive(Subcommand, Debug)]
pub enum RoomCommand { pub enum RoomCommand {
/// List public rooms and their user counts /// List public rooms and their user counts
+1
View File
@@ -309,6 +309,7 @@ fn dispatch(ctx: &Ctx, command: Commands) -> CliResult {
social::message_read(ctx, listen_span(duration, follow)) social::message_read(ctx, listen_span(duration, follow))
} }
Commands::User(args) => peer::user(ctx, &args), Commands::User(args) => peer::user(ctx, &args),
Commands::Watch(command) => peer::watch(ctx, &command),
Commands::Whoami => peer::whoami(ctx), Commands::Whoami => peer::whoami(ctx),
// main.rs runs these itself: some need no account, the rest need the // main.rs runs these itself: some need no account, the rest need the
// config store in scope. The arms only keep the match exhaustive. // config store in scope. The arms only keep the match exhaustive.
+37 -2
View File
@@ -1,9 +1,10 @@
//! Being a peer on the network: `serve`, `whoami`, and `user`. //! Being a peer on the network: `serve`, `whoami`, and `user`.
use super::{Ctx, Session}; use super::{Ctx, Session};
use crate::cli::{ServeArgs, UserArgs}; use crate::cli::{ServeArgs, UserArgs, WatchCommand};
use crate::output::{ use crate::output::{
CliError, CliResult, UploadRecord, UserRecord, WhoamiRecord, CliError, CliResult, UploadRecord, UserRecord, WatchedUserRecord,
WhoamiRecord,
}; };
use soulseek_rs::types::{UploadInfo, UploadStatus, UserInfo}; use soulseek_rs::types::{UploadInfo, UploadStatus, UserInfo};
use std::collections::HashMap; use std::collections::HashMap;
@@ -94,6 +95,40 @@ pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult {
Ok(()) Ok(())
} }
/// Start or stop watching a user, or list who is being watched.
///
/// A watch lives in the session that made it, so `add` against a one-shot CLI
/// run ends with that run. Pointed at a daemon it persists for that daemon's
/// lifetime, which is the point of the command.
pub fn watch(ctx: &Ctx, command: &WatchCommand) -> CliResult {
let session = Session::open(ctx)?;
match command {
WatchCommand::Add { user } => {
session.client.watch_user(user).map_err(|e| {
CliError::connection(format!("cannot watch {user}: {e}"))
})?;
ctx.out.status(&format!("watching {user}"));
}
WatchCommand::Remove { user } => {
session.client.unwatch_user(user).map_err(|e| {
CliError::connection(format!("cannot unwatch {user}: {e}"))
})?;
ctx.out.status(&format!("no longer watching {user}"));
}
WatchCommand::List => {
for user in session.client.watched_users() {
let status = session
.client
.user_info(&user)
.and_then(|info| info.presence)
.map(|p| p.status.to_string());
ctx.out.emit(&WatchedUserRecord { user, status });
}
}
}
Ok(())
}
/// Stay online sharing files, and — with `--wishlist` — re-run the standing /// Stay online sharing files, and — with `--wishlist` — re-run the standing
/// searches on the interval the server sets while doing so. /// searches on the interval the server sets while doing so.
pub fn serve( pub fn serve(
+11
View File
@@ -151,10 +151,21 @@ pub fn room_users(ctx: &Ctx, room: &str, timeout: Duration) -> CliResult {
timeout.as_secs() timeout.as_secs()
))); )));
}; };
// The stats ride along with the membership the server sent, so they are
// already here; a server that sent none leaves every field null rather
// than reporting a zero that reads like an answer.
let stats = session.client.room_member_stats(room);
for user in members { for user in members {
let stat = stats.iter().find(|s| s.username == user);
ctx.out.emit(&RoomMemberRecord { ctx.out.emit(&RoomMemberRecord {
room: room.to_string(), room: room.to_string(),
user, user,
status: stat.map(|s| s.status.to_string()),
average_speed: stat.map(|s| s.average_speed),
shared_files: stat.map(|s| s.shared_files),
shared_folders: stat.map(|s| s.shared_folders),
slots_full: stat.map(|s| s.slots_full),
country: stat.and_then(|s| s.country.clone()),
}); });
} }
Ok(()) Ok(())
+29 -12
View File
@@ -11,10 +11,11 @@
use super::proto::{ use super::proto::{
Ack, AuthParams, AuthResult, BrowseEvent, DaemonStatus, DirectoriesParams, Ack, AuthParams, AuthResult, BrowseEvent, DaemonStatus, DirectoriesParams,
DirectoryParams, DownloadStartParams, DownloadStarted, DownloadStatusEvent, DirectoryParams, DownloadStartParams, DownloadStarted, DownloadStatusEvent,
Downloads, Event, IntervalSeconds, Members, MessageParams, Messages, Downloads, Event, IntervalSeconds, MemberStats, Members, MessageParams,
Method, QueryParams, RoomEventDto, RoomRef, SayParams, SearchResults, Messages, Method, QueryParams, RoomEventDto, RoomRef, SayParams,
Searches, Seconds, SessionLossEvent, SharesStatus, SlotsParams, SearchResults, Searches, Seconds, SessionLossEvent, SharesStatus,
TransferRef, UploadInfoDto, Uploads, UserMessageDto, UserRef, UserResult, SlotsParams, TransferRef, UploadInfoDto, Uploads, UserMessageDto, UserRef,
UserResult, Watched,
}; };
use schemars::{JsonSchema, SchemaGenerator, generate::SchemaSettings}; use schemars::{JsonSchema, SchemaGenerator, generate::SchemaSettings};
use serde_json::{Value, json}; use serde_json::{Value, json};
@@ -46,14 +47,17 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
| Method::UploadCancel => schema_of::<TransferRef>(generator), | Method::UploadCancel => schema_of::<TransferRef>(generator),
Method::DownloadSetDir => schema_of::<DirectoryParams>(generator), Method::DownloadSetDir => schema_of::<DirectoryParams>(generator),
Method::UploadSlots => schema_of::<SlotsParams>(generator), Method::UploadSlots => schema_of::<SlotsParams>(generator),
Method::RoomJoin | Method::RoomLeave | Method::RoomMembers => { Method::RoomJoin
schema_of::<RoomRef>(generator) | Method::RoomLeave
} | Method::RoomMembers
| Method::RoomMemberStats => schema_of::<RoomRef>(generator),
Method::RoomSay => schema_of::<SayParams>(generator), Method::RoomSay => schema_of::<SayParams>(generator),
Method::MessageSend => schema_of::<MessageParams>(generator), Method::MessageSend => schema_of::<MessageParams>(generator),
Method::BrowseUser | Method::UserRequest | Method::UserInfoOf => { Method::BrowseUser
schema_of::<UserRef>(generator) | Method::UserRequest
} | Method::UserInfoOf
| Method::UserWatch
| Method::UserUnwatch => schema_of::<UserRef>(generator),
Method::SharesSet => schema_of::<DirectoriesParams>(generator), Method::SharesSet => schema_of::<DirectoriesParams>(generator),
// Methods that take no parameters. // Methods that take no parameters.
Method::DaemonStatus Method::DaemonStatus
@@ -68,7 +72,8 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
| Method::MessageHistory | Method::MessageHistory
| Method::SearchList | Method::SearchList
| Method::SharesStatusOf | Method::SharesStatusOf
| Method::SharesReindex => json!({ "type": "null" }), | Method::SharesReindex
| Method::UserWatched => json!({ "type": "null" }),
} }
} }
@@ -88,6 +93,8 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
Method::UploadList => schema_of::<Uploads>(generator), Method::UploadList => schema_of::<Uploads>(generator),
Method::PrivilegesOwn => schema_of::<Seconds>(generator), Method::PrivilegesOwn => schema_of::<Seconds>(generator),
Method::RoomMembers => schema_of::<Members>(generator), Method::RoomMembers => schema_of::<Members>(generator),
Method::RoomMemberStats => schema_of::<MemberStats>(generator),
Method::UserWatched => schema_of::<Watched>(generator),
Method::MessageHistory => schema_of::<Messages>(generator), Method::MessageHistory => schema_of::<Messages>(generator),
Method::UserInfoOf => schema_of::<UserResult>(generator), Method::UserInfoOf => schema_of::<UserResult>(generator),
Method::SharesStatusOf | Method::SharesSet | Method::SharesReindex => { Method::SharesStatusOf | Method::SharesSet | Method::SharesReindex => {
@@ -112,7 +119,9 @@ fn result_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
| Method::MessageSend | Method::MessageSend
| Method::BrowseUser | Method::BrowseUser
| Method::SearchForget | Method::SearchForget
| Method::UserRequest => schema_of::<Ack>(generator), | Method::UserRequest
| Method::UserWatch
| Method::UserUnwatch => schema_of::<Ack>(generator),
} }
} }
@@ -172,6 +181,14 @@ fn summary(method: Method) -> &'static str {
Method::RoomLeave => "Leave a chat room.", Method::RoomLeave => "Leave a chat room.",
Method::RoomSay => "Post a message to a room.", Method::RoomSay => "Post a message to a room.",
Method::RoomMembers => "Who is currently in a room.", Method::RoomMembers => "Who is currently in a room.",
Method::RoomMemberStats => {
"What the server said about each member of a joined room."
}
Method::UserWatch => {
"Watch a user, so the server keeps pushing their status."
}
Method::UserUnwatch => "Stop watching a user.",
Method::UserWatched => "Everyone this session is watching.",
Method::MessageSend => "Send a private message.", Method::MessageSend => "Send a private message.",
Method::MessageHistory => { Method::MessageHistory => {
"Private messages the daemon collected, including while nobody was attached." "Private messages the daemon collected, including while nobody was attached."
+56 -6
View File
@@ -7,7 +7,8 @@
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use soulseek_rs::types::{ use soulseek_rs::types::{
Download, DownloadMetadata, UserPresence, UserStats, UserStatus, Download, DownloadMetadata, RoomUserStats, UserPresence, UserStats,
UserStatus,
}; };
use soulseek_rs::{ use soulseek_rs::{
DownloadStatus, File, RoomEvent, RoomInfo, SearchResult, SessionLoss, DownloadStatus, File, RoomEvent, RoomInfo, SearchResult, SessionLoss,
@@ -447,6 +448,59 @@ pub struct UserStatsDto {
pub shared_folders: u32, pub shared_folders: u32,
} }
/// The inverse of [`UserStatus`]'s `Display`. An unrecognised word reads as
/// offline, matching how the wire code is decoded.
fn status_from_str(status: &str) -> UserStatus {
match status {
"online" => UserStatus::Online,
"away" => UserStatus::Away,
_ => UserStatus::Offline,
}
}
/// One member of a room, as the JoinRoom reply described them.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
#[cfg_attr(test, derive(schemars::JsonSchema))]
pub struct RoomUserStatsDto {
pub username: String,
/// `online`, `away`, or `offline`.
pub status: String,
pub average_speed: u32,
pub shared_files: u32,
pub shared_folders: u32,
/// True when the member's upload slots are all occupied.
pub slots_full: bool,
pub country: Option<String>,
}
impl From<&RoomUserStats> for RoomUserStatsDto {
fn from(stats: &RoomUserStats) -> Self {
Self {
username: stats.username.clone(),
status: stats.status.to_string(),
average_speed: stats.average_speed,
shared_files: stats.shared_files,
shared_folders: stats.shared_folders,
slots_full: stats.slots_full,
country: stats.country.clone(),
}
}
}
impl From<RoomUserStatsDto> for RoomUserStats {
fn from(dto: RoomUserStatsDto) -> Self {
Self {
username: dto.username,
status: status_from_str(&dto.status),
average_speed: dto.average_speed,
shared_files: dto.shared_files,
shared_folders: dto.shared_folders,
slots_full: dto.slots_full,
country: dto.country,
}
}
}
/// Each half is `None` until its reply lands, so a consumer can tell "the /// Each half is `None` until its reply lands, so a consumer can tell "the
/// server said offline" from "the server has not answered yet". /// server said offline" from "the server has not answered yet".
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
@@ -480,11 +534,7 @@ impl From<UserInfoDto> for UserInfo {
fn from(dto: UserInfoDto) -> Self { fn from(dto: UserInfoDto) -> Self {
let mut info = Self::pending(dto.username); let mut info = Self::pending(dto.username);
info.presence = dto.presence.map(|presence| UserPresence { info.presence = dto.presence.map(|presence| UserPresence {
status: match presence.status.as_str() { status: status_from_str(&presence.status),
"online" => UserStatus::Online,
"away" => UserStatus::Away,
_ => UserStatus::Offline,
},
privileged: presence.privileged, privileged: presence.privileged,
}); });
info.stats = dto.stats.map(|stats| UserStats { info.stats = dto.stats.map(|stats| UserStats {
+15
View File
@@ -1,5 +1,6 @@
//! What each method takes and returns, and what each pushed event carries. //! What each method takes and returns, and what each pushed event carries.
use super::dto::RoomUserStatsDto;
use super::dto::{ use super::dto::{
DownloadDto, DownloadMetadataDto, DownloadStatusDto, SearchResultDto, DownloadDto, DownloadMetadataDto, DownloadStatusDto, SearchResultDto,
SessionLossDto, SharedDirectoryDto, UploadInfoDto, UserInfoDto, SessionLossDto, SharedDirectoryDto, UploadInfoDto, UserInfoDto,
@@ -139,6 +140,20 @@ pub struct Members {
pub users: Vec<String>, pub users: Vec<String>,
} }
/// What the server said about each member of a room when it was joined.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
#[cfg_attr(test, derive(schemars::JsonSchema))]
pub struct MemberStats {
pub members: Vec<RoomUserStatsDto>,
}
/// Everyone this session is watching, sorted by name.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
#[cfg_attr(test, derive(schemars::JsonSchema))]
pub struct Watched {
pub users: Vec<String>,
}
/// One line of chat history. Unlike a live [`UserMessageDto`] this carries a /// One line of chat history. Unlike a live [`UserMessageDto`] this carries a
/// direction, because a conversation the daemon collected includes what this /// direction, because a conversation the daemon collected includes what this
/// account sent as well as what it received. /// account sent as well as what it received.
+4
View File
@@ -101,11 +101,15 @@ wire_vocabulary! {
RoomLeave => "room.leave", RoomLeave => "room.leave",
RoomSay => "room.say", RoomSay => "room.say",
RoomMembers => "room.members", RoomMembers => "room.members",
RoomMemberStats => "room.member_stats",
MessageSend => "message.send", MessageSend => "message.send",
MessageHistory => "message.history", MessageHistory => "message.history",
BrowseUser => "browse.user", BrowseUser => "browse.user",
UserRequest => "user.request", UserRequest => "user.request",
UserInfoOf => "user.info", UserInfoOf => "user.info",
UserWatch => "user.watch",
UserUnwatch => "user.unwatch",
UserWatched => "user.watched",
SharesStatusOf => "shares.status", SharesStatusOf => "shares.status",
SharesSet => "shares.set", SharesSet => "shares.set",
SharesReindex => "shares.reindex", SharesReindex => "shares.reindex",
+150 -7
View File
@@ -9,11 +9,12 @@ use super::hub::{Hub, PendingBrowses, PendingDownload};
use super::proto::{ use super::proto::{
Ack, AuthResult, CODE_APPLICATION, CODE_INVALID_PARAMS, ChatMessageDto, Ack, AuthResult, CODE_APPLICATION, CODE_INVALID_PARAMS, ChatMessageDto,
DaemonStatus, DirectoriesParams, DirectoryParams, DownloadDto, DaemonStatus, DirectoriesParams, DirectoryParams, DownloadDto,
DownloadStartParams, DownloadStarted, Downloads, IntervalSeconds, Members, DownloadStartParams, DownloadStarted, Downloads, IntervalSeconds,
MessageParams, Messages, Method, OPENRPC, PROTOCOL_VERSION, QueryParams, MemberStats, Members, MessageParams, Messages, Method, OPENRPC,
RoomRef, RpcError, SayParams, SearchResultDto, SearchResults, PROTOCOL_VERSION, QueryParams, RoomRef, RoomUserStatsDto, RpcError,
SearchSummary, Searches, Seconds, SharesStatus, SlotsParams, TransferRef, SayParams, SearchResultDto, SearchResults, SearchSummary, Searches,
UploadInfoDto, Uploads, UserInfoDto, UserRef, UserResult, Seconds, SharesStatus, SlotsParams, TransferRef, UploadInfoDto, Uploads,
UserInfoDto, UserRef, UserResult, Watched,
}; };
use crate::api::SessionApi; use crate::api::SessionApi;
use crate::output::Exit; use crate::output::Exit;
@@ -252,6 +253,18 @@ impl Daemon {
}) })
} }
Method::RoomMemberStats => {
let room: RoomRef = parse(params)?;
ok(MemberStats {
members: self
.session
.room_member_stats(&room.room)
.iter()
.map(RoomUserStatsDto::from)
.collect(),
})
}
Method::MessageSend => { Method::MessageSend => {
let message: MessageParams = parse(params)?; let message: MessageParams = parse(params)?;
self.session self.session
@@ -306,6 +319,24 @@ impl Daemon {
}) })
} }
Method::UserWatch => {
let user: UserRef = parse(params)?;
self.session.watch_user(&user.username).map_err(|e| {
RpcError::application(Exit::Connection, e.to_string())
})?;
ok(Ack::OK)
}
Method::UserUnwatch => {
let user: UserRef = parse(params)?;
self.session.unwatch_user(&user.username).map_err(|e| {
RpcError::application(Exit::Connection, e.to_string())
})?;
ok(Ack::OK)
}
Method::UserWatched => ok(Watched {
users: self.session.watched_users(),
}),
Method::SharesStatusOf => ok(self.shares()), Method::SharesStatusOf => ok(self.shares()),
Method::SharesSet => { Method::SharesSet => {
let directories: DirectoriesParams = parse(params)?; let directories: DirectoriesParams = parse(params)?;
@@ -593,6 +624,72 @@ mod tests {
); );
} }
#[test]
fn member_stats_are_published_for_a_joined_room() {
let daemon = daemon_with_token("t");
let stats = daemon
.call(Method::RoomMemberStats, json!({ "room": STUB_ROOM }))
.expect("member stats");
assert_eq!(stats["members"][0]["username"], "alice");
assert_eq!(stats["members"][0]["status"], "online");
assert_eq!(stats["members"][0]["slots_full"], true);
assert_eq!(stats["members"][0]["country"], "NL");
}
#[test]
fn a_room_that_was_never_joined_has_no_member_stats() {
// Empty, not an error: asking about a room this session is not in is
// an ordinary thing for a client to do.
let daemon = daemon_with_token("t");
let stats = daemon
.call(Method::RoomMemberStats, json!({ "room": "elsewhere" }))
.expect("member stats");
assert_eq!(stats["members"], json!([]));
}
#[test]
fn a_watch_is_readable_back_and_survives_until_unwatched() {
// The three methods are only useful together: what `user.watch`
// records, `user.watched` must report, and `user.unwatch` must undo.
let daemon = daemon_with_token("t");
let user = |name: &str| json!({ "username": name });
daemon
.call(Method::UserWatch, user("alice"))
.expect("watch");
daemon.call(Method::UserWatch, user("bob")).expect("watch");
let watched = daemon
.call(Method::UserWatched, Value::Null)
.expect("watched");
assert_eq!(watched["users"], json!(["alice", "bob"]));
daemon
.call(Method::UserUnwatch, user("alice"))
.expect("unwatch");
let watched = daemon
.call(Method::UserWatched, Value::Null)
.expect("watched");
assert_eq!(
watched["users"],
json!(["bob"]),
"unwatching one must not clear the rest"
);
}
#[test]
fn watching_the_same_user_twice_lists_them_once() {
let daemon = daemon_with_token("t");
let alice = json!({ "username": "alice" });
daemon
.call(Method::UserWatch, alice.clone())
.expect("watch");
daemon.call(Method::UserWatch, alice).expect("watch");
let watched = daemon
.call(Method::UserWatched, Value::Null)
.expect("watched");
assert_eq!(watched["users"], json!(["alice"]));
}
#[test] #[test]
fn stop_is_recorded_so_the_daemon_can_wind_down() { fn stop_is_recorded_so_the_daemon_can_wind_down() {
let daemon = daemon_with_token("t"); let daemon = daemon_with_token("t");
@@ -605,11 +702,22 @@ mod tests {
// A session that answers nothing: enough to exercise dispatch, parameter // A session that answers nothing: enough to exercise dispatch, parameter
// validation, and the auth gate without a Soulseek server. // validation, and the auth gate without a Soulseek server.
struct SilentSession; /// A session that answers nothing, with two exceptions the methods under
/// test need: it remembers watches, so what one call records the next can
/// read back, and it describes the members of one room, so the stats
/// method can return something for a joined room and nothing for any
/// other.
#[derive(Default)]
struct SilentSession {
watched: Mutex<Vec<String>>,
}
/// The one room `SilentSession` claims to have joined.
const STUB_ROOM: &str = "jazz";
fn daemon_with_token(token: &str) -> Daemon { fn daemon_with_token(token: &str) -> Daemon {
Daemon { Daemon {
session: Arc::new(SilentSession), session: Arc::new(SilentSession::default()),
hub: Arc::new(Hub::new()), hub: Arc::new(Hub::new()),
browses: Arc::new(PendingBrowses::default()), browses: Arc::new(PendingBrowses::default()),
downloads: Arc::new(Mutex::new(Vec::new())), downloads: Arc::new(Mutex::new(Vec::new())),
@@ -764,6 +872,23 @@ mod tests {
fn room_members(&self, _room: &str) -> Vec<String> { fn room_members(&self, _room: &str) -> Vec<String> {
Vec::new() Vec::new()
} }
fn room_member_stats(
&self,
room: &str,
) -> Vec<soulseek_rs::types::RoomUserStats> {
if room != STUB_ROOM {
return Vec::new();
}
vec![soulseek_rs::types::RoomUserStats {
username: "alice".to_string(),
status: soulseek_rs::types::UserStatus::Online,
average_speed: 1024,
shared_files: 20,
shared_folders: 3,
slots_full: true,
country: Some("NL".to_string()),
}]
}
fn take_room_events(&self) -> Vec<soulseek_rs::RoomEvent> { fn take_room_events(&self) -> Vec<soulseek_rs::RoomEvent> {
Vec::new() Vec::new()
} }
@@ -792,6 +917,24 @@ mod tests {
) -> soulseek_rs::Result<()> { ) -> soulseek_rs::Result<()> {
Ok(()) Ok(())
} }
fn watch_user(&self, username: &str) -> soulseek_rs::Result<()> {
let mut watched = self.watched.lock().expect("watch list");
if !watched.iter().any(|u| u == username) {
watched.push(username.to_string());
watched.sort();
}
Ok(())
}
fn unwatch_user(&self, username: &str) -> soulseek_rs::Result<()> {
self.watched
.lock()
.expect("watch list")
.retain(|u| u != username);
Ok(())
}
fn watched_users(&self) -> Vec<String> {
self.watched.lock().expect("watch list").clone()
}
fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> { fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> {
None None
} }
+34 -1
View File
@@ -370,19 +370,52 @@ impl Record for UserRecord {
} }
} }
/// One member of a chat room. /// One member of a chat room, with the server's description of them.
///
/// The statistics come from the reply that named the members, and are `None`
/// for a server that sent none — so a consumer never reads "shares nothing"
/// from a missing answer.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RoomMemberRecord { pub struct RoomMemberRecord {
pub room: String, pub room: String,
pub user: String, pub user: String,
/// `online`, `away`, or `offline`.
pub status: Option<String>,
pub average_speed: Option<u32>,
pub shared_files: Option<u32>,
pub shared_folders: Option<u32>,
pub slots_full: Option<bool>,
pub country: Option<String>,
} }
impl Record for RoomMemberRecord { impl Record for RoomMemberRecord {
/// The username alone, unchanged by the statistics added alongside it:
/// the text line is a roster a caller pipes into other commands, and the
/// statistics are there for `--json` to carry.
fn text(&self) -> String { fn text(&self) -> String {
sanitize(&self.user) sanitize(&self.user)
} }
} }
/// One user this session is watching, with whatever the server has already
/// said about them. The status is `None` until the watch reply lands.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct WatchedUserRecord {
pub user: String,
/// `online`, `away`, or `offline`.
pub status: Option<String>,
}
impl Record for WatchedUserRecord {
fn text(&self) -> String {
format!(
"{}\t{}",
sanitize(&self.user),
self.status.clone().unwrap_or_else(|| "-".to_string())
)
}
}
/// One configured share folder. /// One configured share folder.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ShareRecord { pub struct ShareRecord {
+40 -5
View File
@@ -12,14 +12,14 @@ use crate::daemon::proto::{
Ack, AuthParams, AuthResult, CODE_APPLICATION, ChatMessageDto, Ack, AuthParams, AuthResult, CODE_APPLICATION, ChatMessageDto,
DaemonStatus, DirectoriesParams, DirectoryParams, DownloadDto, DaemonStatus, DirectoriesParams, DirectoryParams, DownloadDto,
DownloadStartParams, DownloadStarted, Downloads, Event, IntervalSeconds, DownloadStartParams, DownloadStarted, Downloads, Event, IntervalSeconds,
Members, MessageParams, Messages, Method, PROTOCOL_VERSION, QueryParams, MemberStats, Members, MessageParams, Messages, Method, PROTOCOL_VERSION,
Request, Response, RoomRef, RpcError, SayParams, SearchResults, Searches, QueryParams, Request, Response, RoomRef, RpcError, SayParams,
Seconds, SharesStatus, SlotsParams, TransferRef, Uploads, UserRef, SearchResults, Searches, Seconds, SharesStatus, SlotsParams, TransferRef,
UserResult, Uploads, UserRef, UserResult, Watched,
}; };
use mirror::Mirror; use mirror::Mirror;
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
use soulseek_rs::types::{Download, DownloadMetadata}; use soulseek_rs::types::{Download, DownloadMetadata, RoomUserStats};
use soulseek_rs::{ use soulseek_rs::{
DownloadStatus, Result, RoomEvent, SearchResult, SessionLoss, DownloadStatus, Result, RoomEvent, SearchResult, SessionLoss,
SharedDirectory, SoulseekRs, UploadInfo, UserInfo, UserMessage, SharedDirectory, SoulseekRs, UploadInfo, UserInfo, UserMessage,
@@ -710,6 +710,17 @@ impl SessionApi for RemoteSession {
.unwrap_or_default() .unwrap_or_default()
} }
fn room_member_stats(&self, room: &str) -> Vec<RoomUserStats> {
self.request::<_, MemberStats>(
Method::RoomMemberStats,
RoomRef {
room: room.to_string(),
},
)
.map(|stats| stats.members.into_iter().map(Into::into).collect())
.unwrap_or_default()
}
fn take_room_events(&self) -> Vec<RoomEvent> { fn take_room_events(&self) -> Vec<RoomEvent> {
self.mirror.take_rooms() self.mirror.take_rooms()
} }
@@ -757,6 +768,30 @@ impl SessionApi for RemoteSession {
) )
} }
fn watch_user(&self, username: &str) -> Result<()> {
self.tell(
Method::UserWatch,
UserRef {
username: username.to_string(),
},
)
}
fn unwatch_user(&self, username: &str) -> Result<()> {
self.tell(
Method::UserUnwatch,
UserRef {
username: username.to_string(),
},
)
}
fn watched_users(&self) -> Vec<String> {
self.request::<_, Watched>(Method::UserWatched, ())
.map(|watched| watched.users)
.unwrap_or_default()
}
fn user_info(&self, username: &str) -> Option<UserInfo> { fn user_info(&self, username: &str) -> Option<UserInfo> {
self.request::<_, UserResult>( self.request::<_, UserResult>(
Method::UserInfoOf, Method::UserInfoOf,
+32 -23
View File
@@ -3,6 +3,16 @@ use crate::models::{DownloadEntry, FileDisplayData};
use soulseek_rs::{DownloadStatus, types::Download}; use soulseek_rs::{DownloadStatus, types::Download};
use std::{sync::mpsc, thread}; use std::{sync::mpsc, thread};
/// Done, given up, or ran out of time: the row is all that is left.
const fn is_finished(download: &Download) -> bool {
matches!(
download.status,
DownloadStatus::Completed
| DownloadStatus::Failed(_)
| DownloadStatus::TimedOut
)
}
impl MainTui { impl MainTui {
/// Channel for queued downloads, created on first use. /// Channel for queued downloads, created on first use.
pub(super) fn downloads_sender( pub(super) fn downloads_sender(
@@ -63,26 +73,16 @@ impl MainTui {
// session holds, so dropping them only here would last until the next // session holds, so dropping them only here would last until the next
// poll put them straight back. // poll put them straight back.
for entry in &self.state.downloads { for entry in &self.state.downloads {
if matches!( if is_finished(&entry.download) {
entry.download.status,
DownloadStatus::Completed
| DownloadStatus::Failed(_)
| DownloadStatus::TimedOut
) {
self.client.remove_download( self.client.remove_download(
&entry.download.username, &entry.download.username,
&entry.download.filename, &entry.download.filename,
); );
} }
} }
self.state.downloads.retain(|entry| { self.state
!matches!( .downloads
entry.download.status, .retain(|entry| !is_finished(&entry.download));
DownloadStatus::Completed
| DownloadStatus::Failed(_)
| DownloadStatus::TimedOut
)
});
let len = self.state.downloads.len(); let len = self.state.downloads.len();
if len == 0 { if len == 0 {
self.state.downloads_table_state.select(None); self.state.downloads_table_state.select(None);
@@ -121,7 +121,9 @@ impl MainTui {
} }
} }
pub(super) fn remove_selected_queued_download(&mut self) { /// Drop the selected transfer: a queued one leaves the queue, a finished
/// one leaves the list.
pub(super) fn remove_selected_download(&mut self) {
let Some(index) = self.state.downloads_table_state.selected() else { let Some(index) = self.state.downloads_table_state.selected() else {
return; return;
}; };
@@ -130,14 +132,21 @@ impl MainTui {
}; };
let download = &download_entry.download; let download = &download_entry.download;
if !matches!(download.status, DownloadStatus::Queued) { let dismissed = match download.status {
return; DownloadStatus::Queued => self
} .client
.remove_queued_download(&download.username, &download.filename),
if !self // A finished transfer may exist only in this window (restored from
.client // an earlier run), so the session's "nothing to forget" is not a
.remove_queued_download(&download.username, &download.filename) // reason to keep the row.
{ DownloadStatus::Completed => {
self.client
.remove_download(&download.username, &download.filename);
true
}
_ => false,
};
if !dismissed {
return; return;
} }
+1 -1
View File
@@ -428,7 +428,7 @@ impl MainTui {
self.toggle_selected_download_pause(); self.toggle_selected_download_pause();
} }
KeyCode::Char('d') => { KeyCode::Char('d') => {
self.remove_selected_queued_download(); self.remove_selected_download();
} }
KeyCode::Char('r') => { KeyCode::Char('r') => {
self.retry_selected_download(); self.retry_selected_download();
+74 -2
View File
@@ -265,7 +265,8 @@ pub fn launch_main_tui(
mod tests { mod tests {
use super::*; use super::*;
use crate::daemon::proto::ChatMessageDto; use crate::daemon::proto::ChatMessageDto;
use crate::models::{MessageDirection, SearchStatus}; use crate::models::{FocusedPane, MessageDirection, SearchStatus};
use ratatui::crossterm::event::{KeyCode, KeyEvent, KeyModifiers};
/// A stand-in for a daemon: it already holds a conversation, a transfer /// A stand-in for a daemon: it already holds a conversation, a transfer
/// queue, and a search cache, none of which this window created. /// queue, and a search cache, none of which this window created.
@@ -291,6 +292,21 @@ mod tests {
} }
fn queued(username: &str, filename: &str) -> soulseek_rs::types::Download { fn queued(username: &str, filename: &str) -> soulseek_rs::types::Download {
at_status(username, filename, soulseek_rs::DownloadStatus::Queued)
}
fn completed(
username: &str,
filename: &str,
) -> soulseek_rs::types::Download {
at_status(username, filename, soulseek_rs::DownloadStatus::Completed)
}
fn at_status(
username: &str,
filename: &str,
status: soulseek_rs::DownloadStatus,
) -> soulseek_rs::types::Download {
let (sender, _) = std::sync::mpsc::channel(); let (sender, _) = std::sync::mpsc::channel();
soulseek_rs::types::Download { soulseek_rs::types::Download {
username: username.to_string(), username: username.to_string(),
@@ -298,13 +314,17 @@ mod tests {
token: 1, token: 1,
size: 4096, size: 4096,
download_directory: "/tmp".to_string(), download_directory: "/tmp".to_string(),
status: soulseek_rs::DownloadStatus::Queued, status,
sender, sender,
queue_position: None, queue_position: None,
metadata: soulseek_rs::types::DownloadMetadata::default(), metadata: soulseek_rs::types::DownloadMetadata::default(),
} }
} }
fn key(ch: char) -> KeyEvent {
KeyEvent::new(KeyCode::Char(ch), KeyModifiers::NONE)
}
impl SessionApi for TalkativeSession { impl SessionApi for TalkativeSession {
fn message_history(&self) -> Vec<ChatMessageDto> { fn message_history(&self) -> Vec<ChatMessageDto> {
self.history.clone() self.history.clone()
@@ -441,6 +461,12 @@ mod tests {
fn room_members(&self, _room: &str) -> Vec<String> { fn room_members(&self, _room: &str) -> Vec<String> {
Vec::new() Vec::new()
} }
fn room_member_stats(
&self,
_room: &str,
) -> Vec<soulseek_rs::types::RoomUserStats> {
Vec::new()
}
fn take_room_events(&self) -> Vec<soulseek_rs::RoomEvent> { fn take_room_events(&self) -> Vec<soulseek_rs::RoomEvent> {
Vec::new() Vec::new()
} }
@@ -469,6 +495,15 @@ mod tests {
) -> soulseek_rs::Result<()> { ) -> soulseek_rs::Result<()> {
Ok(()) Ok(())
} }
fn watch_user(&self, _username: &str) -> soulseek_rs::Result<()> {
Ok(())
}
fn unwatch_user(&self, _username: &str) -> soulseek_rs::Result<()> {
Ok(())
}
fn watched_users(&self) -> Vec<String> {
Vec::new()
}
fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> { fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> {
None None
} }
@@ -619,6 +654,43 @@ mod tests {
); );
} }
#[test]
fn d_clears_a_finished_row_but_not_a_running_one() {
// Finished transfers are what a list fills up with after a restart, so
// `d` has to be able to dismiss one — restored or not. A running
// transfer is not ours to discard, and stays put.
let mut tui = with_session(TalkativeSession::default());
tui.state.downloads.push(crate::models::DownloadEntry {
download: completed("bob", "done.mp3"),
receiver: None,
});
tui.state.downloads_table_state.select(Some(0));
tui.state.focused_pane = FocusedPane::Downloads;
tui.handle_key_event(key('d'));
assert!(tui.state.downloads.is_empty(), "the row goes too");
assert_eq!(tui.state.downloads_table_state.selected(), None);
tui.state.downloads.push(crate::models::DownloadEntry {
download: at_status(
"bob",
"still.mp3",
soulseek_rs::DownloadStatus::InProgress {
bytes_downloaded: 1,
total_bytes: 2,
speed_bytes_per_sec: 1.0,
},
),
receiver: None,
});
tui.state.downloads_table_state.select(Some(0));
tui.handle_key_event(key('d'));
assert_eq!(tui.state.downloads.len(), 1, "it is not disposable");
}
#[test] #[test]
fn a_window_shows_searches_run_in_another() { fn a_window_shows_searches_run_in_another() {
let mut tui = let mut tui =
+1 -1
View File
@@ -393,7 +393,7 @@ impl MainTui {
vec![ vec![
("p", "pause/resume"), ("p", "pause/resume"),
("r", "retry failed"), ("r", "retry failed"),
("d", "delete queued"), ("d", "delete queued/done"),
("c", "clear finished"), ("c", "clear finished"),
("b", "browse user"), ("b", "browse user"),
("1-3", "focus pane"), ("1-3", "focus pane"),