dumbgoy/soulseek-rs-modded
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor: cut what a whole-repo audit proved dead, padded, or hand-rolled

Browse Source 
Seven region audits, every finding adversarially verified against the
whole workspace before being applied; 80 landed and the two that failed
in application were refuted honestly (ParseError has a from_str_radix
producer the audit missed; NatPmpError's Display is the only reader of
its diagnostic payloads).

The bigger cuts:

- The NewPeer/PierceFireWall cluster: an operation variant nothing ever
  sent, its 77-line arm, a protocol-wrong Client::pierce_firewall that
  wrote a peer-init frame down the server socket, and the struct only
  they used. The live paths in peer/listen.rs are untouched.
- The ops loop reads `for operation in reader` instead of hand-rolling
  recv/break, de-indenting the whole dispatch body.
- Dead lib surface: Message::print_hex and four unused accessors,
  UploadFailed, registry and actor getters, build_watch_user,
  strip_ansi_codes, delegating shims on ClientContext. Removing public
  items makes this a semver major, which the release pipeline enforces.
- Never-used flexibility: the state-migration framework that never held
  a migration, border-type helpers that all answered Rounded, the TUI's
  max_concurrent_downloads field nothing read.
- Convergence: six poll loops into poll_until, three wrap-around
  navigation blocks into cycle, hit tests into Rect::contains, char
  walkers into floor/ceil_char_boundary, five Download literals into
  one helper, the suites' shared scaffolding into tests/common.

Deliberately not taken: swapping the hand-rolled zlib inflate and MD5
for crates (they replace wire-facing machinery whose tolerance of
real-world peer quirks no local test exercises) and pulling chrono into
the lib for a log timestamp. Recorded for their own decision.

Net -1002 lines. Both e2e modes stay green: 94 CLI tests each way, 16
protocol, 33 lib, unit suites, clippy pedantic clean, aislop 100/100
with the four scanner false positives suppressed inline where they
fire.
This commit is contained in:
Michel de Graaf
2026-07-28 12:52:18 +02:00
parent cd56fb3b39
commit f1b667f7f9
62 changed files with 1367 additions and 2319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
soulseek-rs-lib/src/actor/peer_actor.rs
+27 -37
View File
@@ -184,13 +184,7 @@ impl PeerActor {
let messages: Vec<PeerMessage> = self
.dispatcher_receiver
.as_ref()
.map_or_else(Vec::new, |receiver| {
let mut msgs = Vec::new();
while let Ok(msg) = receiver.try_recv() {
msgs.push(msg);
}
msgs
});
.map_or_else(Vec::new, |receiver| receiver.try_iter().collect());
for msg in &messages {
self.handle_message(msg.clone());
@@ -690,45 +684,20 @@ impl PeerActor {
}
}
fn initiate_connection(&mut self) -> bool {
fn initiate_connection(&mut self) {
let (username, host, port) = match self.peer.read_safe() {
Ok(peer) => (peer.username.clone(), peer.host.clone(), peer.port),
Err(e) => {
error!("[peer_actor] initiate_connection peer lock: {}", e);
return false;
return;
}
};
let socket_addr =
format!("{host}:{port}").parse::<std::net::SocketAddr>();
match socket_addr {
Ok(addr) => {
// Use connect_timeout to prevent blocking the thread for too long
let timeout = Duration::from_secs(5);
match TcpStream::connect_timeout(&addr, timeout) {
Ok(stream) => {
if let Err(e) = stream.set_nonblocking(true) {
error!(
"[peer:{}] Failed to set non-blocking: {}",
username, e
);
self.disconnect_with_error(e);
return false;
}
stream.set_nodelay(true).ok();
self.stream = Some(stream);
self.connection_state = ConnectionState::Connecting {
since: Instant::now(),
};
true
}
Err(e) => {
self.disconnect_with_error(e);
false
}
}
}
let addr = match socket_addr {
Ok(addr) => addr,
Err(e) => {
error!(
"[peer:{}] Invalid socket address {}:{} - {}",
@@ -738,9 +707,30 @@ impl PeerActor {
io::ErrorKind::InvalidInput,
e,
));
false
return;
}
};
// Use connect_timeout to prevent blocking the thread for too long
let timeout = Duration::from_secs(5);
let stream = match TcpStream::connect_timeout(&addr, timeout) {
Ok(stream) => stream,
Err(e) => {
self.disconnect_with_error(e);
return;
}
};
if let Err(e) = stream.set_nonblocking(true) {
error!("[peer:{}] Failed to set non-blocking: {}", username, e);
self.disconnect_with_error(e);
return;
}
stream.set_nodelay(true).ok();
self.stream = Some(stream);
self.connection_state = ConnectionState::Connecting {
since: Instant::now(),
};
}
fn check_connection_status(&mut self) {
soulseek-rs-lib/src/actor/peer_registry.rs
-22
View File
@@ -145,28 +145,6 @@ impl PeerRegistry {
None
}
#[must_use]
pub fn get_all_usernames(&self) -> Vec<String> {
match self.peers.lock_safe() {
Ok(peers) => peers.keys().cloned().collect(),
Err(e) => {
error!("[peer_registry] get_all_usernames: {}", e);
Vec::new()
}
}
}
#[must_use]
pub fn count(&self) -> usize {
match self.peers.lock_safe() {
Ok(peers) => peers.len(),
Err(e) => {
error!("[peer_registry] count: {}", e);
0
}
}
}
#[must_use]
pub fn contains(&self, username: &str) -> bool {
match self.peers.lock_safe() {
soulseek-rs-lib/src/actor/server_actor.rs
+20 -82
View File
@@ -31,8 +31,8 @@ use crate::types::{
};
use crate::utils::lock::RwLockExt;
use std::io::{self, Error, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::io::{self, Write};
use std::net::TcpStream;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
@@ -109,17 +109,6 @@ impl UserMessage {
new_message,
}
}
pub fn print(&self) {
debug!(
"Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}",
self.timestamp,
self.username,
self.id,
self.new_message,
self.message
);
}
/// The server-assigned id of this message (used to acknowledge it).
#[must_use]
pub const fn id(&self) -> u32 {
@@ -320,60 +309,23 @@ impl ServerActor {
self.session = session;
}
#[must_use]
pub const fn get_address(&self) -> &PeerAddress {
&self.address
}
#[must_use]
pub const fn get_sender(&self) -> Option<&Sender<ServerMessage>> {
self.dispatcher_sender.as_ref()
}
fn initiate_connection(&mut self) -> bool {
let host = self.address.host.clone();
let port = self.address.port;
let addr_str = format!("{host}:{port}");
let mut socket_addrs = match addr_str.to_socket_addrs() {
Ok(addrs) => addrs,
Err(e) => {
error!("[server] Failed to resolve address: {}", e);
self.disconnect_with_error(io::Error::new(
io::ErrorKind::InvalidInput,
e,
));
return false;
}
};
let socket_addr = socket_addrs.next();
let Some(addr) = socket_addr else {
let error_msg =
format!("No socket addresses found for {host}:{port}");
error!("[server] {}", error_msg);
self.disconnect_with_error(io::Error::new(
io::ErrorKind::InvalidInput,
error_msg,
));
return false;
};
let stream = match TcpStream::connect(addr) {
fn initiate_connection(&mut self) {
let stream = match TcpStream::connect((
self.address.host.as_str(),
self.address.port,
)) {
Ok(s) => s,
Err(e) => {
self.disconnect_with_error(e);
return false;
error!("[server] Failed to connect to {}: {}", self.address, e);
self.disconnect_with_error();
return;
}
};
if let Err(e) = stream.set_nonblocking(true) {
error!("[server] Failed to set non-blocking: {}", e);
self.disconnect_with_error(e);
return false;
self.disconnect_with_error();
return;
}
stream.set_nodelay(true).ok();
@@ -381,7 +333,6 @@ impl ServerActor {
self.connection_state = ConnectionState::Connecting {
since: Instant::now(),
};
true
}
pub fn set_self_handle(&mut self, handle: ActorHandle<ServerMessage>) {
@@ -436,13 +387,7 @@ impl ServerActor {
let messages: Vec<ServerMessage> = self
.dispatcher_receiver
.as_ref()
.map_or_else(Vec::new, |receiver| {
let mut msgs = Vec::new();
while let Ok(msg) = receiver.try_recv() {
msgs.push(msg);
}
msgs
});
.map_or_else(Vec::new, |receiver| receiver.try_iter().collect());
for msg in &messages {
self.handle_message(msg.clone());
@@ -783,7 +728,7 @@ impl ServerActor {
e,
e.kind()
);
self.disconnect_with_error(e);
self.disconnect_with_error();
return;
}
}
@@ -818,7 +763,7 @@ impl ServerActor {
"[server] Error extracting message: {}. Disconnecting.",
e
);
self.disconnect_with_error(e);
self.disconnect_with_error();
return;
}
Ok(None) => {
@@ -871,13 +816,13 @@ impl ServerActor {
if let Err(e) = stream.write_all(&message.get_buffer()) {
error!("[server] Error writing message: {}. Disconnecting.", e);
self.disconnect_with_error(e);
self.disconnect_with_error();
return;
}
if let Err(e) = stream.flush() {
error!("[server] Error flushing stream: {}. Disconnecting.", e);
self.disconnect_with_error(e);
self.disconnect_with_error();
}
}
@@ -893,7 +838,7 @@ impl ServerActor {
self.disconnect();
}
fn disconnect_with_error(&mut self, _error: Error) {
fn disconnect_with_error(&mut self) {
debug!("[server] disconnect");
// Losing an established connection ends the session: nothing reconnects
@@ -919,10 +864,7 @@ impl ServerActor {
if since.elapsed() > Duration::from_secs(20) {
error!("[server] Connection timeout after 20 seconds");
self.disconnect_with_error(io::Error::new(
io::ErrorKind::TimedOut,
"Connection timeout",
));
self.disconnect_with_error();
return;
}
@@ -938,16 +880,12 @@ impl ServerActor {
Err(ref e) if e.kind() == io::ErrorKind::NotConnected => {}
Err(e) => {
error!("[server] Connection failed: {}", e);
self.disconnect_with_error(e);
self.disconnect_with_error();
}
}
}
fn on_connection_established(&mut self) {
let Some(_) = self.stream else {
panic!("Stream should be available here")
};
self.initialize_dispatcher();
let queued = std::mem::take(&mut self.queued_messages);
soulseek-rs-lib/src/client/connection.rs
+1 -37
View File
@@ -2,7 +2,7 @@ use super::{
Arc, Client, ClientContext, ClientOperation, ConnectionType, DownloadPeer,
DownloadStatus, Listen, Peer, PeerRegistry, Receiver, Result, RwLock,
RwLockExt, Sender, ServerActor, ServerMessage, Shares, SoulseekRs,
TcpStream, debug, error, info, mpsc, thread, trace,
TcpStream, error, info, mpsc, thread, trace,
};
/// Ceiling on the wait for a login verdict. Generous enough for a slow server
@@ -18,7 +18,6 @@ impl Client {
) = mpsc::channel();
let mut ctx = self.context.write_safe()?;
ctx.sender = Some(sender.clone());
let peer_registry = PeerRegistry::new(
ctx.actor_system.clone(),
sender.clone(),
@@ -166,8 +165,6 @@ impl Client {
own_username: String,
stream: Option<TcpStream>,
) {
let client_context = client_context;
let peer_clone = peer.clone();
trace!(
"[client] connecting to {}, with connection_type: {}, and token {:?}",
@@ -252,37 +249,4 @@ impl Client {
}
}
}
pub(crate) fn pierce_firewall(
peer: Peer,
client_context: Arc<RwLock<ClientContext>>,
own_username: String,
) {
debug!("Piercing firewall for peer: {:?}", peer);
let context = match client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!("[client] pierce_firewall read: {}", e);
return;
}
};
if let Some(server_sender) = &context.server_sender {
if let Some(token) = peer.token {
match server_sender.send(ServerMessage::PierceFirewall(token)) {
Ok(()) => (),
Err(e) => {
error!("Failed to send PierceFirewall message: {}", e);
}
}
} else {
error!("No token available for PierceFirewall");
}
} else {
error!("No server sender available for PierceFirewall");
}
drop(context);
Self::connect_to_peer(peer, client_context, own_username, None);
}
}
soulseek-rs-lib/src/client/mod.rs
+3 -27
View File
@@ -13,10 +13,7 @@ use crate::{
actor::{ActorSystem, peer_registry::PeerRegistry},
error::{Result, SoulseekRs},
message::peer::{FileEntry, SharedDirectory, build_file_search_response},
peer::{
ConnectionType, DownloadPeer, NewPeer, Peer, PeerMessage,
listen::Listen,
},
peer::{ConnectionType, DownloadPeer, Peer, PeerMessage, listen::Listen},
shares::Shares,
types::{Download, Search, SearchResult},
utils::{lock::RwLockExt, md5},
@@ -25,14 +22,11 @@ use std::{
collections::{HashMap, HashSet},
net::TcpStream,
sync::{
RwLock,
Arc, RwLock,
atomic::{AtomicBool, AtomicU32, Ordering},
mpsc::{Receiver, Sender},
mpsc::{self, Receiver, Sender},
},
thread::{self, sleep},
};
use std::{
sync::{Arc, mpsc},
time::{Duration, Instant},
};
use upload_queue::QueuedUpload;
@@ -213,11 +207,9 @@ impl Default for ClientSettings {
#[derive(Debug)]
#[non_exhaustive]
pub enum ClientOperation {
NewPeer(NewPeer),
ConnectToPeer(Peer),
SearchResult(SearchResult),
PeerDisconnected(u64, String, Option<SoulseekRs>),
PierceFireWall(Peer),
DownloadFromPeer(u32, Peer, bool),
UpdateDownloadTokens(Transfer, String),
GetPeerAddressResponse {
@@ -298,7 +290,6 @@ pub enum ClientOperation {
pub struct ClientContext {
pub peer_registry: Option<PeerRegistry>,
pub downloads: DownloadStore,
sender: Option<Sender<ClientOperation>>,
server_sender: Option<Sender<ServerMessage>>,
searches: HashMap<String, Search>,
private_messages: Vec<UserMessage>,
@@ -376,13 +367,6 @@ impl ClientContext {
) -> Option<&mut Download> {
self.downloads.get_by_token_mut(token)
}
pub fn get_download_by_file_mut(
&mut self,
username: &str,
filename: &str,
) -> Option<&mut Download> {
self.downloads.get_by_file_mut(username, filename)
}
#[must_use]
pub fn get_download_tokens(&self) -> Vec<u32> {
self.downloads.tokens()
@@ -398,13 +382,6 @@ impl ClientContext {
) {
self.downloads.update_status(token, status);
}
pub fn remove_queued_download_by_file(
&mut self,
username: &str,
filename: &str,
) -> bool {
self.downloads.remove_queued_by_file(username, filename)
}
}
impl ClientContext {
@@ -414,7 +391,6 @@ impl ClientContext {
Self {
peer_registry: None,
sender: None,
server_sender: None,
searches: HashMap::new(),
private_messages: Vec::new(),
soulseek-rs-lib/src/client/operations.rs
+137 -333
View File
@@ -15,13 +15,10 @@ impl Client {
own_username: String,
) {
thread::spawn(move || {
loop {
match reader.recv() {
Ok(operation) => {
for operation in reader {
match operation {
ClientOperation::ConnectToPeer(peer) => {
let client_context_clone =
client_context.clone();
let client_context_clone = client_context.clone();
let own_username_clone = own_username.clone();
thread::spawn(move || {
@@ -34,19 +31,11 @@ impl Client {
});
}
ClientOperation::SearchResult(search_result) => {
trace!(
"[client] SearchResult {:?}",
search_result
);
let mut context = match client_context
.write_safe()
{
trace!("[client] SearchResult {:?}", search_result);
let mut context = match client_context.write_safe() {
Ok(c) => c,
Err(e) => {
error!(
"[client] SearchResult write: {}",
e
);
error!("[client] SearchResult write: {}", e);
continue;
}
};
@@ -60,11 +49,7 @@ impl Client {
}
}
}
ClientOperation::PeerDisconnected(
id,
username,
error,
) => {
ClientOperation::PeerDisconnected(id, username, error) => {
// Scope the read guard: process_failed_uploads
// below acquires a write lock on the same
// RwLock, which would self-deadlock the entire
@@ -73,9 +58,7 @@ impl Client {
// actor still occupies the slot, so a replaced
// actor's shutdown can't remove its successor.
{
let context = match client_context
.read_safe()
{
let context = match client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!(
@@ -85,10 +68,9 @@ impl Client {
continue;
}
};
if let Some(ref registry) =
context.peer_registry
&& let Some(handle) = registry
.remove_peer_if(&username, id)
if let Some(ref registry) = context.peer_registry
&& let Some(handle) =
registry.remove_peer_if(&username, id)
{
let _ = handle.stop();
}
@@ -108,50 +90,35 @@ impl Client {
// upload slot: with a slot cap, a couple of
// departed peers would otherwise wedge uploads
// shut for the rest of the session.
Self::release_upload_slots(
&client_context,
&username,
);
Self::release_upload_slots(&client_context, &username);
}
ClientOperation::PierceFireWall(peer) => {
Self::pierce_firewall(
peer,
client_context.clone(),
own_username.clone(),
);
ClientOperation::DownloadFromPeer(token, peer, allowed) => {
let maybe_download = match client_context.read_safe() {
Ok(ctx) => {
ctx.get_download_by_token(token).cloned()
}
ClientOperation::DownloadFromPeer(
token,
peer,
allowed,
) => {
let maybe_download = match client_context
.read_safe()
{
Ok(ctx) => ctx
.get_download_by_token(token)
.cloned(),
Err(e) => {
error!(
"[client] DownloadFromPeer read: {}",
e
);
error!("[client] DownloadFromPeer read: {}", e);
continue;
}
};
let own_username = own_username.clone();
let client_context_clone =
client_context.clone();
let client_context_clone = client_context.clone();
trace!(
"[client] DownloadFromPeer token: {} peer: {:?}",
token, peer
);
match maybe_download {
Some(download) => {
let Some(download) = maybe_download else {
error!(
"Can't find download with token {:?}",
token
);
continue;
};
thread::spawn(move || {
let download_peer =
DownloadPeer::new(
let download_peer = DownloadPeer::new(
download.username.clone(),
peer.host.clone(),
peer.port,
@@ -159,43 +126,55 @@ impl Client {
allowed,
own_username,
);
let filename: Option<&str> =
download
.filename
.split('\\')
.next_back();
match filename {
Some(filename) => {
match download_peer
.download_file(
client_context_clone
.clone(),
let Some(filename) =
download.filename.split('\\').next_back()
else {
error!(
"Cant find filename to save download: {:?}",
download.filename
);
return;
};
match download_peer.download_file(
client_context_clone.clone(),
Some(download.clone()),
None,
) {
Ok((
download,
filename,
)) => {
let _ = download.sender.send(DownloadStatus::Completed);
Ok((download, filename)) => {
let _ = download
.sender
.send(DownloadStatus::Completed);
match client_context_clone.write_safe() {
Ok(mut ctx) => ctx.update_download_with_status(download.token, DownloadStatus::Completed),
Err(e) => error!("[client] download complete write: {}", e),
Ok(mut ctx) => ctx
.update_download_with_status(
download.token,
DownloadStatus::Completed,
),
Err(e) => error!(
"[client] download complete write: {}",
e
),
}
info!(
"Successfully downloaded {} bytes to {}",
download.size,
filename
download.size, filename
);
}
Err(e) => {
let reason = Some(
e.to_string(),
let reason = Some(e.to_string());
let _ = download.sender.send(
DownloadStatus::Failed(reason.clone()),
);
let _ = download.sender.send(DownloadStatus::Failed(reason.clone()));
match client_context_clone.write_safe() {
Ok(mut ctx) => ctx.update_download_with_status(download.token, DownloadStatus::Failed(reason)),
Err(e) => error!("[client] download failed write: {}", e),
Ok(mut ctx) => ctx
.update_download_with_status(
download.token,
DownloadStatus::Failed(reason),
),
Err(e) => error!(
"[client] download failed write: {}",
e
),
}
error!(
"Failed to download file '{}' from {}:{} (token: {}) - Error: {}",
@@ -207,99 +186,8 @@ impl Client {
);
}
}
}
None => error!(
"Cant find filename to save download: {:?}",
download.filename
),
}
});
}
None => {
error!(
"Can't find download with token {:?}",
token
);
}
}
}
ClientOperation::NewPeer(new_peer) => {
let peer_exists = match client_context
.read_safe()
{
Ok(ctx) => {
ctx.peer_registry.as_ref().is_some_and(
|r| r.contains(&new_peer.username),
)
}
Err(e) => {
error!("[client] NewPeer read: {}", e);
continue;
}
};
if peer_exists {
debug!(
"Already connected to {}",
new_peer.username
);
} else {
let send_result = client_context
.read_safe()
.ok()
.and_then(|ctx| {
ctx.server_sender.as_ref().map(
|s| {
s.send(
ServerMessage::GetPeerAddress(
new_peer
.username
.clone(),
),
)
},
)
});
if let Some(Err(e)) = send_result {
error!(
"[client] NewPeer send GetPeerAddress: {}",
e
);
}
}
let addr = match new_peer.tcp_stream.peer_addr()
{
Ok(a) => a,
Err(e) => {
error!(
"[client] NewPeer peer_addr: {}",
e
);
continue;
}
};
let host = addr.ip().to_string();
let port: u32 = addr.port().into();
let peer = Peer {
username: new_peer.username.clone(),
connection_type: new_peer.connection_type,
host,
port,
token: Some(new_peer.token),
privileged: None,
obfuscated_port: None,
unknown: None,
};
Self::connect_to_peer(
peer,
client_context.clone(),
own_username.clone(),
Some(new_peer.tcp_stream),
);
}
ClientOperation::GetPeerAddressResponse {
username,
host,
@@ -360,9 +248,7 @@ impl Client {
);
}
let peer_exists = match client_context
.read_safe()
{
let peer_exists = match client_context.read_safe() {
Ok(ctx) => ctx
.peer_registry
.as_ref()
@@ -376,10 +262,9 @@ impl Client {
}
};
if peer_exists {
// Existing peer: skip re-registration. Reconnect
// policy on conflict is intentionally undecided.
} else {
if !peer_exists {
let peer = Peer::new(
username,
ConnectionType::P,
@@ -391,14 +276,11 @@ impl Client {
// real obfuscated_port is a full u16 and
// must not be truncated into a u8 (which
// panicked and took down the ops thread).
u8::try_from(obfuscation_type)
.unwrap_or(0),
u8::try_from(obfuscation_type).unwrap_or(0),
obfuscated_port,
);
let client_context_clone =
client_context.clone();
let own_username_clone =
own_username.clone();
let client_context_clone = client_context.clone();
let own_username_clone = own_username.clone();
thread::spawn(move || {
Self::connect_to_peer(
@@ -414,9 +296,7 @@ impl Client {
transfer,
username,
) => {
let mut context = match client_context
.write_safe()
{
let mut context = match client_context.write_safe() {
Ok(c) => c,
Err(e) => {
error!(
@@ -427,10 +307,8 @@ impl Client {
}
};
let download_to_update = context
.get_downloads()
.iter()
.find_map(|d| {
let download_to_update =
context.get_downloads().iter().find_map(|d| {
if d.username == username
&& d.filename == transfer.filename
{
@@ -440,8 +318,7 @@ impl Client {
}
});
if let Some((old_token, download)) =
download_to_update
if let Some((old_token, download)) = download_to_update
{
trace!(
"[client] UpdateDownloadTokens found {old_token}, transfer: {:?}",
@@ -449,16 +326,9 @@ impl Client {
);
context.add_download(Download {
username: username.clone(),
filename: transfer.filename.clone(),
token: transfer.token,
size: transfer.size,
download_directory: download
.download_directory,
status: download.status.clone(),
sender: download.sender.clone(),
queue_position: download.queue_position,
metadata: download.metadata.clone(),
..download
});
context.remove_download(old_token);
}
@@ -471,17 +341,17 @@ impl Client {
let registry = context.peer_registry.clone();
drop(context);
if let Some(registry) = registry {
let response = MessageFactory::build_transfer_response_message(transfer);
let response =
MessageFactory::build_transfer_response_message(
transfer,
);
let _ = registry.send_to_peer(
&username,
PeerMessage::SendMessage(response),
);
}
}
ClientOperation::UploadFailed(
username,
filename,
) => {
ClientOperation::UploadFailed(username, filename) => {
Self::process_failed_uploads(
client_context.clone(),
&username,
@@ -494,8 +364,7 @@ impl Client {
place,
} => match client_context.write_safe() {
Ok(mut ctx) => {
let updated =
ctx.downloads.update_queue_position(
let updated = ctx.downloads.update_queue_position(
&username, &filename, place,
);
if !updated {
@@ -505,28 +374,23 @@ impl Client {
);
}
}
Err(e) => error!(
"[client] PlaceInQueueUpdate write: {}",
e
),
Err(e) => {
error!("[client] PlaceInQueueUpdate write: {}", e);
}
},
ClientOperation::SetServerSender(sender) => {
match client_context.write_safe() {
Ok(mut ctx) => {
ctx.server_sender = Some(sender);
debug!(
"[client] Server sender initialized"
);
debug!("[client] Server sender initialized");
}
Err(e) => error!(
"[client] SetServerSender write: {}",
e
),
Err(e) => {
error!("[client] SetServerSender write: {}", e);
}
}
ClientOperation::PrivateMessageReceived(
user_message,
) => match client_context.write_safe() {
}
ClientOperation::PrivateMessageReceived(user_message) => {
match client_context.write_safe() {
Ok(mut ctx) => {
ctx.push_private_message(user_message);
}
@@ -534,19 +398,19 @@ impl Client {
"[client] PrivateMessageReceived write: {}",
e
),
},
}
}
ClientOperation::UserStatusReceived {
username,
status,
privileged,
} => match client_context.write_safe() {
Ok(mut ctx) => ctx.apply_user_status(
username, status, privileged,
),
Err(e) => error!(
"[client] UserStatusReceived write: {}",
e
),
Ok(mut ctx) => {
ctx.apply_user_status(username, status, privileged);
}
Err(e) => {
error!("[client] UserStatusReceived write: {}", e);
}
},
ClientOperation::UserStatsReceived {
username,
@@ -560,23 +424,18 @@ impl Client {
shared_files,
shared_folders,
),
Err(e) => error!(
"[client] UserStatsReceived write: {}",
e
),
Err(e) => {
error!("[client] UserStatsReceived write: {}", e);
}
},
ClientOperation::RoomEvent(event) => {
match client_context.write_safe() {
Ok(mut ctx) => ctx.apply_room_event(event),
Err(e) => error!(
"[client] RoomEvent write: {}",
e
),
Err(e) => error!("[client] RoomEvent write: {}", e),
}
}
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);
}
}
@@ -604,10 +463,7 @@ impl Client {
.collect(),
),
Err(e) => {
error!(
"[client] PeerConnected read: {}",
e
);
error!("[client] PeerConnected read: {}", e);
continue;
}
};
@@ -615,14 +471,12 @@ impl Client {
// responses) queued while connecting.
let queued_messages = client_context
.write_safe()
.map(|mut ctx| {
ctx.take_peer_messages(&username)
})
.map(|mut ctx| ctx.take_peer_messages(&username))
.unwrap_or_default();
if let Some(registry) = registry {
for filename in files {
let _ = registry
.queue_upload(&username, filename);
let _ =
registry.queue_upload(&username, filename);
}
for message in queued_messages {
let _ = registry.send_to_peer(
@@ -641,8 +495,7 @@ impl Client {
if username == own_username {
continue;
}
let response = match client_context.read_safe()
{
let response = match client_context.read_safe() {
Ok(ctx) => build_search_response(
&ctx.shares,
&own_username,
@@ -650,10 +503,7 @@ impl Client {
&query,
),
Err(e) => {
error!(
"[client] IncomingSearch read: {}",
e
);
error!("[client] IncomingSearch read: {}", e);
continue;
}
};
@@ -668,9 +518,7 @@ impl Client {
Ok(ctx) => (
ctx.peer_registry
.as_ref()
.is_some_and(|r| {
r.contains(&username)
}),
.is_some_and(|r| r.contains(&username)),
ctx.peer_registry.clone(),
ctx.server_sender.clone(),
),
@@ -684,18 +532,12 @@ impl Client {
);
}
} else {
if let Ok(mut ctx) =
client_context.write_safe()
{
ctx.queue_peer_message(
&username, message,
);
if let Ok(mut ctx) = client_context.write_safe() {
ctx.queue_peer_message(&username, message);
}
if let Some(sender) = server_sender {
let _ = sender.send(
ServerMessage::GetPeerAddress(
username,
),
ServerMessage::GetPeerAddress(username),
);
}
}
@@ -709,8 +551,7 @@ impl Client {
downloader_of(&requester_key).to_string();
match client_context.write_safe() {
Ok(mut ctx) => {
let Some(file) =
ctx.shares.get(&filename)
let Some(file) = ctx.shares.get(&filename)
else {
debug!(
"[client] QueueUpload for unknown file {}",
@@ -729,10 +570,7 @@ impl Client {
);
}
Err(e) => {
error!(
"[client] QueueUpload write: {}",
e
);
error!("[client] QueueUpload write: {}", e);
continue;
}
}
@@ -743,13 +581,11 @@ impl Client {
filename,
} => {
let downloader = downloader_of(&requester_key);
let (registry, place) =
match client_context.read_safe() {
let (registry, place) = match client_context.read_safe()
{
Ok(ctx) => (
ctx.peer_registry.clone(),
ctx.place_in_queue(
downloader, &filename,
),
ctx.place_in_queue(downloader, &filename),
),
Err(_) => continue,
};
@@ -769,8 +605,7 @@ impl Client {
}
}
ClientOperation::PrivilegedUsers(users) => {
if let Ok(mut ctx) = client_context.write_safe()
{
if let Ok(mut ctx) = client_context.write_safe() {
debug!(
"[client] {} privileged users listed",
users.len()
@@ -783,8 +618,7 @@ impl Client {
Self::pump_upload_queue(&client_context);
}
ClientOperation::OwnPrivileges(seconds) => {
if let Ok(mut ctx) = client_context.write_safe()
{
if let Ok(mut ctx) = client_context.write_safe() {
ctx.own_privileges = Some(seconds);
}
}
@@ -792,18 +626,15 @@ impl Client {
// The peer accepted our offer: resolve their
// address (from the code-9 GetPeerAddress) and
// stream the file, or queue until it resolves.
let (job_addr, downloader) =
match client_context.read_safe() {
let (job_addr, downloader) = match client_context
.read_safe()
{
Ok(ctx) => {
let Some(job) =
ctx.uploads.get(&token)
else {
let Some(job) = ctx.uploads.get(&token) else {
continue;
};
(
ctx.peer_address(
&job.downloader,
),
ctx.peer_address(&job.downloader),
job.downloader.clone(),
)
}
@@ -818,29 +649,22 @@ impl Client {
port,
);
} else {
if let Ok(mut ctx) =
client_context.write_safe()
{
if let Ok(mut ctx) = client_context.write_safe() {
ctx.pending_serves
.entry(downloader.clone())
.or_default()
.push(token);
}
if let Ok(ctx) = client_context.read_safe()
&& let Some(sender) =
ctx.server_sender.clone()
&& let Some(sender) = ctx.server_sender.clone()
{
let _ = sender.send(
ServerMessage::GetPeerAddress(
downloader,
),
ServerMessage::GetPeerAddress(downloader),
);
}
}
}
ClientOperation::ShareListRequested {
requester_key,
} => {
ClientOperation::ShareListRequested { requester_key } => {
// Reply with our full shared-file listing.
let (registry, message) = match client_context
.read_safe()
@@ -875,26 +699,17 @@ impl Client {
username,
directories,
} => {
if let Ok(mut ctx) = client_context.write_safe()
{
ctx.store_browse_result(
username,
directories,
);
if let Ok(mut ctx) = client_context.write_safe() {
ctx.store_browse_result(username, directories);
}
}
ClientOperation::PeerConnectFailed(
id,
username,
) => {
ClientOperation::PeerConnectFailed(id, username) => {
// Direct connect failed: ask the server to
// broker it. Register a correlation token, then
// send ConnectToPeer so the (firewalled) peer
// connects back to our listener quoting it.
let token = next_connect_token();
let server_sender = match client_context
.write_safe()
{
let server_sender = match client_context.write_safe() {
Ok(mut ctx) => {
// Reap the dead outbound actor so it
// stops pinning a pool worker and no
@@ -903,10 +718,8 @@ impl Client {
// later downloads queue into a dead,
// streamless actor and hang). Identity-
// aware so a newer namesake is untouched.
if let Some(handle) = ctx
.peer_registry
.as_ref()
.and_then(|r| {
if let Some(handle) =
ctx.peer_registry.as_ref().and_then(|r| {
r.remove_peer_if(&username, id)
})
{
@@ -934,8 +747,7 @@ impl Client {
&username,
ConnectionType::P,
);
let _ = sender
.send(ServerMessage::SendMessage(msg));
let _ = sender.send(ServerMessage::SendMessage(msg));
// Bound the brokered attempt: if no PierceFirewall
// consumes the token, fail the peer's queued
@@ -946,11 +758,9 @@ impl Client {
let timeout_user = username.clone();
thread::spawn(move || {
sleep(BROKER_CONNECT_TIMEOUT);
let still_pending = timeout_ctx
.write_safe()
.is_ok_and(|mut c| {
c.take_pending_connect(token)
.is_some()
let still_pending =
timeout_ctx.write_safe().is_ok_and(|mut c| {
c.take_pending_connect(token).is_some()
});
if still_pending {
Self::fail_queued_downloads(
@@ -962,12 +772,6 @@ impl Client {
}
}
}
Err(e) => {
error!("[client] Channel recv error: {:?}", e);
break;
}
}
}
});
}
}
soulseek-rs-lib/src/client/tests.rs
+56 -61
View File
@@ -1,22 +1,37 @@
use super::*;
fn download(
username: &str,
filename: &str,
token: u32,
status: DownloadStatus,
sender: Sender<DownloadStatus>,
) -> Download {
Download {
username: username.to_string(),
filename: filename.to_string(),
token,
size: 100,
download_directory: "test".to_string(),
status,
sender,
queue_position: None,
metadata: DownloadMetadata::default(),
}
}
#[test]
fn test_client_context_downloads() {
let mut context = ClientContext::new();
let token = 123;
let new_token = 1234;
let download = Download {
username: "test".to_string(),
filename: "test.txt".to_string(),
context.add_download(download(
"test",
"test.txt",
token,
size: 100,
download_directory: "test".to_string(),
status: DownloadStatus::Queued,
sender: mpsc::channel().0,
queue_position: None,
metadata: DownloadMetadata::default(),
};
context.add_download(download);
DownloadStatus::Queued,
mpsc::channel().0,
));
assert!(context.get_download_by_token(123).is_some());
assert_eq!(context.get_download_tokens(), vec![123]);
assert_eq!(context.get_downloads().len(), 1);
@@ -35,23 +50,18 @@ fn test_client_context_downloads() {
fn test_client_pause_and_resume_download() {
let client = Client::new("test-user", "test-password");
let (download_sender, download_receiver) = mpsc::channel();
let download = Download {
username: "peer".to_string(),
filename: "song.mp3".to_string(),
token: 123,
size: 100,
download_directory: "test".to_string(),
status: DownloadStatus::InProgress {
client.context.write().unwrap().add_download(download(
"peer",
"song.mp3",
123,
DownloadStatus::InProgress {
bytes_downloaded: 25,
total_bytes: 100,
speed_bytes_per_sec: 10.0,
},
sender: download_sender,
queue_position: None,
metadata: DownloadMetadata::default(),
};
client.context.write().unwrap().add_download(download);
download_sender,
));
assert!(client.pause_download("peer", "song.mp3"));
assert!(matches!(
@@ -118,17 +128,13 @@ fn fail_queued_downloads_notifies_receiver_and_store() {
// must resolve to Failed both on its channel and in the store.
let client = Client::new("u", "p");
let (sender, receiver) = mpsc::channel();
client.context.write().unwrap().add_download(Download {
username: "peer".to_string(),
filename: "f.mp3".to_string(),
token: 7,
size: 10,
download_directory: "d".to_string(),
status: DownloadStatus::Queued,
client.context.write().unwrap().add_download(download(
"peer",
"f.mp3",
7,
DownloadStatus::Queued,
sender,
queue_position: None,
metadata: DownloadMetadata::default(),
});
));
Client::fail_queued_downloads(&client.context, "peer");
@@ -171,37 +177,26 @@ fn build_search_response_matches_shares_and_echoes_token() {
#[test]
fn test_client_removes_only_queued_downloads() {
let client = Client::new("test-user", "test-password");
let queued_download = Download {
username: "peer".to_string(),
filename: "queued.mp3".to_string(),
token: 123,
size: 100,
download_directory: "test".to_string(),
status: DownloadStatus::Queued,
sender: mpsc::channel().0,
queue_position: None,
metadata: DownloadMetadata::default(),
};
let active_download = Download {
username: "peer".to_string(),
filename: "active.mp3".to_string(),
token: 456,
size: 100,
download_directory: "test".to_string(),
status: DownloadStatus::InProgress {
{
let mut context = client.context.write().unwrap();
context.add_download(download(
"peer",
"queued.mp3",
123,
DownloadStatus::Queued,
mpsc::channel().0,
));
context.add_download(download(
"peer",
"active.mp3",
456,
DownloadStatus::InProgress {
bytes_downloaded: 25,
total_bytes: 100,
speed_bytes_per_sec: 10.0,
},
sender: mpsc::channel().0,
queue_position: None,
metadata: DownloadMetadata::default(),
};
{
let mut context = client.context.write().unwrap();
context.add_download(queued_download);
context.add_download(active_download);
mpsc::channel().0,
));
}
assert!(client.remove_queued_download("peer", "queued.mp3"));
soulseek-rs-lib/src/client/uploads.rs
+2 -2
View File
@@ -47,8 +47,6 @@ impl Client {
.and_then(|ctx| ctx.own_privileges)
}
/// How many uploads run at once. Lowering it does not interrupt transfers
/// already in flight; the queue simply refills more slowly.
/// Queued-upload states recorded since the last call, for a caller that
/// samples [`Client::uploads`] and would otherwise miss a peer that queued
/// and was served between two samples.
@@ -60,6 +58,8 @@ impl Client {
.unwrap_or_default()
}
/// How many uploads run at once. Lowering it does not interrupt transfers
/// already in flight; the queue simply refills more slowly.
pub fn set_upload_slots(&self, slots: usize) {
if let Ok(mut ctx) = self.context.write_safe() {
ctx.upload_slots = slots.max(1);
soulseek-rs-lib/src/dispatcher.rs
+2 -3
View File
@@ -31,9 +31,8 @@ impl<Op> MessageDispatcher<Op> {
handler.handle(message, self.sender.clone());
} else {
warn!(
"[{}:dispatcher] No handler found for message code: {}",
self.owner_name,
message.get_message_code()
"[{}:dispatcher] No handler found for message code: {code}",
self.owner_name
);
}
}
soulseek-rs-lib/src/message/mod.rs
+12 -84
View File
@@ -51,47 +51,6 @@ impl Message {
}
}
pub fn print_hex(&self) {
const BYTES_PER_LINE: usize = 16;
let data = &self.data;
let chunks = data.chunks(BYTES_PER_LINE);
for (i, chunk) in chunks.enumerate() {
print!("{:04x} ", i * BYTES_PER_LINE);
for j in 0..BYTES_PER_LINE {
if j < chunk.len() {
print!("{:02x} ", chunk[j]);
} else {
print!(" ");
}
// Add extra space in the middle
if j == 7 {
print!(" ");
}
}
print!(" ");
let mut i = 0;
for &byte in chunk {
i += 1;
if byte.is_ascii_graphic() || byte.is_ascii_whitespace() {
print!("{}", byte as char);
} else {
print!(".");
}
if i == 8 {
print!(" ");
}
}
trace!("");
}
}
#[allow(dead_code)]
#[must_use]
pub fn get_message_code_u32(&self) -> u32 {
@@ -109,22 +68,10 @@ impl Message {
self.data[4]
}
#[must_use]
pub fn get_message_code_send(&self) -> u8 {
if self.data.is_empty() {
return 0;
}
self.data[0]
}
#[must_use]
pub const fn new_with_data(data: Vec<u8>) -> Self {
Self { data, pointer: 0 }
}
#[allow(dead_code)]
pub const fn reset_pointer(&mut self) {
self.pointer = 0;
}
pub const fn set_pointer(&mut self, pointer: usize) {
self.pointer = pointer;
}
@@ -215,21 +162,6 @@ impl Message {
val
}
pub fn read_raw_byte(&mut self) -> Vec<u8> {
if self.pointer + 4 > self.data.len() {
return vec![];
}
let val = vec![
self.data[self.pointer],
self.data[self.pointer + 1],
self.data[self.pointer + 2],
self.data[self.pointer + 3],
];
self.pointer += 4;
val
}
pub fn read_int32(&mut self) -> u32 {
if self.pointer + 4 > self.data.len() {
return 0;
@@ -286,20 +218,6 @@ impl Message {
self
}
#[allow(dead_code)]
pub fn write_raw_hex_string(&mut self, val: &str) -> &mut Self {
let mut b = Vec::new();
for i in (0..val.len()).step_by(2) {
let byte_str = &val[i..i + 2];
let byte =
u8::from_str_radix(byte_str, 16).expect("Invalid hex string");
b.push(byte);
}
self.data.extend_from_slice(&b);
self.pointer += b.len();
self
}
pub fn get_message_name(
&self,
msg_type: MessageType,
@@ -378,6 +296,18 @@ impl Message {
}
}
/// Build a received message the way the dispatcher hands one to a handler:
/// 8 bytes of frame header, then the fields, with the pointer parked at the
/// start of the payload.
#[cfg(test)]
pub fn framed(build: impl FnOnce(&mut Message)) -> Message {
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
build(&mut message);
message.set_pointer(8);
message
}
#[test]
fn test_get_buffer() {
let message = Message::new_with_data(
@@ -462,8 +392,6 @@ fn get_message_code_on_short_message_does_not_panic() {
let msg = Message::new_with_data(vec![0, 0, 0, 0]);
assert_eq!(msg.get_message_code(), 0);
assert_eq!(msg.get_message_code_u32(), 0);
let empty = Message::new_with_data(vec![]);
assert_eq!(empty.get_message_code_send(), 0);
}
#[test]
soulseek-rs-lib/src/message/peer/place_in_queue_request.rs
+4 -4
View File
@@ -25,14 +25,14 @@ impl MessageHandler<PeerMessage> for PlaceInQueueRequest {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn the_asked_about_filename_is_forwarded() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_string("@@share\\track.mp3");
message.set_pointer(8);
let mut message = framed(|m| {
m.write_string("@@share\\track.mp3");
});
PlaceInQueueRequest.handle(&mut message, tx);
match rx.try_recv() {
soulseek-rs-lib/src/message/peer/shared_file_list.rs
+3 -4
View File
@@ -110,10 +110,9 @@ fn hostile_dir_count_does_not_hang() {
// parse to empty promptly rather than looping into an OOM.
let compressed =
crate::utils::zlib::compress_stored(&u32::MAX.to_le_bytes());
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_raw_bytes(compressed);
message.set_pointer(8);
let mut message = crate::message::framed(|m| {
m.write_raw_bytes(compressed);
});
assert!(parse_shared_file_list(&mut message).is_empty());
}
soulseek-rs-lib/src/message/peer/upload_failed.rs
+2 -3
View File
@@ -2,7 +2,6 @@ use crate::info;
use crate::{
message::{Message, MessageHandler},
peer::PeerMessage,
types::UploadFailed,
};
use std::sync::mpsc::Sender;
@@ -12,7 +11,7 @@ impl MessageHandler<PeerMessage> for UploadFailedHandler {
46
}
fn handle(&self, message: &mut Message, _sender: Sender<PeerMessage>) {
let upload_failed = UploadFailed::new_from_message(message);
info!("Upload failed for ${}", upload_failed.filename);
let filename = message.read_string();
info!("Upload failed for {}", filename);
}
}
soulseek-rs-lib/src/message/server/excluded_search_phrases.rs
+4 -4
View File
@@ -31,16 +31,16 @@ impl MessageHandler<ServerMessage> for ExcludedSearchPhrasesHandler {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn hostile_item_count_does_not_hang() {
// item_count=u32::MAX with no phrases: the guard must make this return
// promptly instead of looping ~4 billion times.
let (tx, _rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_int32(u32::MAX);
message.set_pointer(8);
let mut message = framed(|m| {
m.write_int32(u32::MAX);
});
ExcludedSearchPhrasesHandler.handle(&mut message, tx);
}
}
soulseek-rs-lib/src/message/server/join_room.rs
+10 -12
View File
@@ -33,17 +33,16 @@ impl MessageHandler<ServerMessage> for JoinRoomHandler {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn hostile_user_count_does_not_hang() {
// room="" then user_count=u32::MAX with no usernames: must return
// promptly instead of looping ~4 billion times.
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_string("");
message.write_int32(u32::MAX);
message.set_pointer(8);
let mut message = framed(|m| {
m.write_string("").write_int32(u32::MAX);
});
JoinRoomHandler.handle(&mut message, tx);
match rx.try_recv() {
@@ -57,13 +56,12 @@ mod tests {
#[test]
fn forwards_room_and_member_list() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_string("nicotine");
message.write_int32(2);
message.write_string("alice");
message.write_string("bob");
message.set_pointer(8);
let mut message = framed(|m| {
m.write_string("nicotine")
.write_int32(2)
.write_string("alice")
.write_string("bob");
});
JoinRoomHandler.handle(&mut message, tx);
match rx.try_recv() {
soulseek-rs-lib/src/message/server/leave_room.rs
+4 -4
View File
@@ -20,14 +20,14 @@ impl MessageHandler<ServerMessage> for LeaveRoomHandler {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn forwards_left_room() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_string("jazz");
message.set_pointer(8);
let mut message = framed(|m| {
m.write_string("jazz");
});
LeaveRoomHandler.handle(&mut message, tx);
match rx.try_recv() {
soulseek-rs-lib/src/message/server/message_factory.rs
-17
View File
@@ -137,14 +137,6 @@ impl MessageFactory {
.write_int32(port.into())
.clone()
}
#[must_use]
pub fn build_watch_user(token: u32) -> Message {
Message::new()
.write_raw_bytes([5, 0, 0, 0, 0].to_vec())
.write_int32(token)
.clone()
}
/// Ask the server (code 7) for a user's online status.
#[must_use]
pub fn build_get_user_status(username: &str) -> Message {
@@ -280,15 +272,6 @@ fn test_build_get_user_stats() {
assert_eq!(expect, message.get_data());
}
#[test]
fn test_build_watch_user() {
let token: u32 = 223;
let message = MessageFactory::build_watch_user(token);
let expect: Vec<u8> = [5, 0, 0, 0, 0, 223, 0, 0, 0].to_vec();
assert_eq!(expect, message.get_data());
}
#[test]
fn test_build_login_message() {
let message = MessageFactory::build_login_message(
soulseek-rs-lib/src/message/server/message_user.rs
-1
View File
@@ -27,7 +27,6 @@ impl MessageHandler<ServerMessage> for MessageUser {
);
info!("[MessageUser] User message received:{:?}", user_message);
user_message.print();
// Acknowledge freshly delivered messages so the server does not keep
// re-sending them on every reconnect.
soulseek-rs-lib/src/message/server/privileged_users.rs
+5 -12
View File
@@ -50,19 +50,12 @@ impl MessageHandler<ServerMessage> for CheckPrivilegesHandler {
#[cfg(test)]
mod tests {
use super::*;
fn payload(build: impl FnOnce(&mut Message)) -> Message {
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
build(&mut message);
message.set_pointer(8);
message
}
use crate::message::framed;
#[test]
fn the_user_list_is_kept_not_just_counted() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = payload(|m| {
let mut message = framed(|m| {
m.write_int32(2)
.write_string("donor_a")
.write_string("donor_b");
@@ -80,7 +73,7 @@ mod tests {
#[test]
fn an_empty_list_is_still_an_answer() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = payload(|m| {
let mut message = framed(|m| {
m.write_int32(0);
});
@@ -98,7 +91,7 @@ mod tests {
// count=u32::MAX with no names: the guard must return promptly rather
// than loop four billion times.
let (tx, _rx) = std::sync::mpsc::channel();
let mut message = payload(|m| {
let mut message = framed(|m| {
m.write_int32(u32::MAX);
});
PrivilegedUsersHandler.handle(&mut message, tx);
@@ -107,7 +100,7 @@ mod tests {
#[test]
fn our_own_remaining_privilege_time_is_reported() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = payload(|m| {
let mut message = framed(|m| {
m.write_int32(86_400);
});
soulseek-rs-lib/src/message/server/room_list.rs
+1 -10
View File
@@ -53,16 +53,7 @@ pub fn parse_room_list(message: &mut Message) -> Vec<RoomInfo> {
#[cfg(test)]
mod tests {
use super::*;
/// Build a received message (8-byte header + payload) and position it the
/// way the dispatcher would.
fn framed(build: impl FnOnce(&mut Message)) -> Message {
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
build(&mut message);
message.set_pointer(8);
message
}
use crate::message::framed;
#[test]
fn parses_names_zipped_with_counts() {
soulseek-rs-lib/src/message/server/say_chatroom.rs
+6 -6
View File
@@ -26,16 +26,16 @@ impl MessageHandler<ServerMessage> for SayChatroomHandler {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn forwards_room_message() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_string("jazz");
message.write_string("alice");
message.write_string("hello everyone");
message.set_pointer(8);
let mut message = framed(|m| {
m.write_string("jazz")
.write_string("alice")
.write_string("hello everyone");
});
SayChatroomHandler.handle(&mut message, tx);
match rx.try_recv() {
soulseek-rs-lib/src/message/server/user_info.rs
+3 -12
View File
@@ -56,21 +56,12 @@ impl MessageHandler<ServerMessage> for GetUserStatsHandler {
#[cfg(test)]
mod tests {
use super::*;
/// Build a payload the dispatcher would hand a handler: 8 bytes of frame
/// header, then the fields, with the pointer parked at the payload.
fn payload(build: impl FnOnce(&mut Message)) -> Message {
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
build(&mut message);
message.set_pointer(8);
message
}
use crate::message::framed;
#[test]
fn user_status_is_parsed_with_its_privileged_flag() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = payload(|m| {
let mut message = framed(|m| {
m.write_string("alice").write_int32(2).write_int8(1);
});
@@ -93,7 +84,7 @@ mod tests {
fn user_stats_skip_the_two_obsolete_fields() {
let (tx, rx) = std::sync::mpsc::channel();
// username, speed, obsolete, obsolete, files, folders
let mut message = payload(|m| {
let mut message = framed(|m| {
m.write_string("bob")
.write_int32(1024)
.write_int32(0)
soulseek-rs-lib/src/message/server/user_joined_room.rs
+4 -5
View File
@@ -23,15 +23,14 @@ impl MessageHandler<ServerMessage> for UserJoinedRoomHandler {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn forwards_user_joined() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_string("jazz");
message.write_string("carol");
message.set_pointer(8);
let mut message = framed(|m| {
m.write_string("jazz").write_string("carol");
});
UserJoinedRoomHandler.handle(&mut message, tx);
match rx.try_recv() {
soulseek-rs-lib/src/message/server/user_left_room.rs
+4 -5
View File
@@ -21,15 +21,14 @@ impl MessageHandler<ServerMessage> for UserLeftRoomHandler {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn forwards_user_left() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_string("jazz");
message.write_string("carol");
message.set_pointer(8);
let mut message = framed(|m| {
m.write_string("jazz").write_string("carol");
});
UserLeftRoomHandler.handle(&mut message, tx);
match rx.try_recv() {
soulseek-rs-lib/src/message/server/wish_list_interval.rs
+4 -4
View File
@@ -26,14 +26,14 @@ impl MessageHandler<ServerMessage> for WishListIntervalHandler {
#[cfg(test)]
mod tests {
use super::*;
use crate::message::framed;
#[test]
fn the_announced_interval_is_reported_not_discarded() {
let (tx, rx) = std::sync::mpsc::channel();
let mut message = Message::new();
message.write_raw_bytes(vec![0u8; 8]);
message.write_int32(720);
message.set_pointer(8);
let mut message = framed(|m| {
m.write_int32(720);
});
WishListIntervalHandler.handle(&mut message, tx);
match rx.try_recv() {
soulseek-rs-lib/src/peer/download_peer.rs
+15 -41
View File
@@ -66,35 +66,16 @@ impl std::fmt::Display for DownloadError {
impl std::error::Error for DownloadError {}
impl From<io::Error> for DownloadError {
fn from(error: io::Error) -> Self {
Self::StreamReadError(error)
}
}
struct FileManager;
impl FileManager {
fn extract_filename_from_path(full_path: &str) -> &str {
fn extract_filename_from_path(full_path: &str) -> &str {
full_path
.split(['/', '\\'])
.next_back()
.unwrap_or(full_path)
}
}
fn create_download_path_from_filename(
output_directory: PathBuf,
filename: &str,
) -> PathBuf {
let filename_only = Self::extract_filename_from_path(filename);
output_directory.join(filename_only)
}
/// Where a download's bytes end up: its configured directory (falling back
/// to the parent if it names a file) joined with the remote basename.
fn resolve_download_path(
download: &Download,
) -> Result<String, DownloadError> {
/// Where a download's bytes end up: its configured directory (falling back
/// to the parent if it names a file) joined with the remote basename.
fn resolve_download_path(download: &Download) -> Result<String, DownloadError> {
let mut expanded_path = expand_tilde(&download.download_directory);
if !expanded_path.is_dir() {
@@ -109,10 +90,8 @@ impl FileManager {
.to_path_buf();
}
let final_path = Self::create_download_path_from_filename(
expanded_path,
&download.filename,
);
let final_path =
expanded_path.join(extract_filename_from_path(&download.filename));
final_path
.to_str()
@@ -123,7 +102,6 @@ impl FileManager {
))
})
.map(String::from)
}
}
/// The `.part` file a transfer streams into, so an interrupted transfer leaves
@@ -137,7 +115,7 @@ struct PartFile {
impl PartFile {
fn open(download: &Download) -> Result<Self, DownloadError> {
let final_path = FileManager::resolve_download_path(download)?;
let final_path = resolve_download_path(download)?;
let path = PathBuf::from(format!("{final_path}{PART_SUFFIX}"));
if let Some(parent) = path.parent() {
@@ -521,7 +499,9 @@ impl DownloadPeer {
#[cfg(test)]
mod tests {
use super::{DownloadError, DownloadPeer, FileManager, PartFile};
use super::{
DownloadError, DownloadPeer, PartFile, extract_filename_from_path,
};
use crate::types::{Download, DownloadMetadata, DownloadStatus};
use std::sync::mpsc;
@@ -636,23 +616,17 @@ mod tests {
#[test]
fn test_extract_filename_from_path() {
assert_eq!(extract_filename_from_path("/path/to/file.mp3"), "file.mp3");
assert_eq!(
FileManager::extract_filename_from_path("/path/to/file.mp3"),
extract_filename_from_path("C:\\path\\to\\file.mp3"),
"file.mp3"
);
assert_eq!(
FileManager::extract_filename_from_path("C:\\path\\to\\file.mp3"),
"file.mp3"
);
assert_eq!(
FileManager::extract_filename_from_path(
extract_filename_from_path(
"@@bhfrv\\Soulseek Downloads\\complete\\Beatport Top Deep House (2021)\\michel test file.mp3"
),
"michel test file.mp3"
);
assert_eq!(
FileManager::extract_filename_from_path("file.mp3"),
"file.mp3"
);
assert_eq!(extract_filename_from_path("file.mp3"), "file.mp3");
}
}
soulseek-rs-lib/src/peer/listen.rs
+9 -20
View File
@@ -77,16 +77,6 @@ fn parse_peer_init_message(mut message: Message) -> Option<PeerInitData> {
})
}
fn parse_token_from_buffer(buffer: &[u8], username: &str) -> Option<u32> {
let token_bytes = buffer.get(0..4)?;
let token = u32::from_le_bytes(
token_bytes
.try_into()
.unwrap_or_else(|_| panic!("[listener:{username}] slice with incorrect length, can't extract transfer_token")),
);
Some(token)
}
fn extract_download_from_buffer(
reader: &mut MessageReader,
client_context: &Arc<RwLock<ClientContext>>,
@@ -98,7 +88,7 @@ fn extract_download_from_buffer(
return None;
}
let buffer = reader.get_buffer();
let token = parse_token_from_buffer(&buffer, username)?;
let token = u32::from_le_bytes(buffer.get(0..4)?.try_into().ok()?);
trace!(
"[listener:{}] got transfer_token: {} from data chunk",
username, token
@@ -129,8 +119,6 @@ fn handle_peer_connection(
stream: TcpStream,
reader: MessageReader,
context: &ConnectionContext,
_peer_ip: &str,
_peer_port: u16,
) {
// The peer actor multiplexes socket reads with its mailbox on a single
// thread: `tick()` reads the socket, but outgoing messages (e.g. a queued
@@ -283,7 +271,7 @@ fn handle_pierce_firewall(
0,
0,
);
handle_peer_connection(peer, stream, reader, context, peer_ip, peer_port);
handle_peer_connection(peer, stream, reader, context);
// Inbound peers don't self-announce, so nudge the client to flush any
// downloads queued for this now-connected peer.
@@ -292,7 +280,10 @@ fn handle_pierce_firewall(
.send(ClientOperation::PeerConnected(username));
}
fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
fn handle_incoming_connection(
mut stream: TcpStream,
context: ConnectionContext,
) {
let Ok(peer_addr) = stream.peer_addr() else {
error!("[listener] failed to get peer address");
return;
@@ -300,7 +291,6 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
let peer_ip = peer_addr.ip().to_string();
let peer_port = peer_addr.port();
let mut stream = stream;
let mut reader = MessageReader::new();
// A peer that dials and then goes away is routine, not an error.
@@ -337,7 +327,6 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
let peer = Peer::new(
format!("{}:direct", init_data.username),
// init_data.username.clone(),
init_data.connection_type.clone(),
peer_ip.clone(),
peer_port.into(),
@@ -348,9 +337,9 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
);
match init_data.connection_type {
ConnectionType::P => handle_peer_connection(
peer, stream, reader, &context, &peer_ip, peer_port,
),
ConnectionType::P => {
handle_peer_connection(peer, stream, reader, &context);
}
ConnectionType::F => handle_file_connection(
peer,
soulseek-rs-lib/src/peer/mod.rs
+2 -30
View File
@@ -10,33 +10,7 @@ pub use download_peer::DownloadPeer;
use crate::message::Message;
use core::fmt;
use std::{net::TcpStream, str::FromStr};
#[derive(Debug)]
#[allow(dead_code)]
pub struct NewPeer {
pub username: String,
pub connection_type: ConnectionType,
pub token: u32,
pub tcp_stream: TcpStream,
}
impl NewPeer {
pub fn new_from_message(
message: &mut Message,
tcp_stream: TcpStream,
) -> Option<Self> {
let username = message.read_string();
let connection_type = message.read_string().parse().ok()?;
let token = message.read_int32();
Some(Self {
username,
connection_type,
token,
tcp_stream,
})
}
}
use std::str::FromStr;
#[derive(Debug, Clone)]
pub enum ConnectionType {
@@ -81,7 +55,6 @@ impl fmt::Display for ConnectionType {
}
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct Peer {
pub username: String,
pub connection_type: ConnectionType,
@@ -93,7 +66,7 @@ pub struct Peer {
pub obfuscated_port: Option<u16>,
}
impl Peer {
#[allow(clippy::too_many_arguments, dead_code)]
#[allow(clippy::too_many_arguments)]
#[must_use]
pub const fn new(
username: String,
@@ -116,7 +89,6 @@ impl Peer {
obfuscated_port: Some(obfuscated_port),
}
}
#[allow(dead_code)]
pub fn new_from_message(message: &mut Message) -> Option<Self> {
let username = message.read_string();
// The connection type is an untrusted string; an unknown value must not
soulseek-rs-lib/src/shares/mod.rs
+1 -9
View File
@@ -44,15 +44,7 @@ impl Shares {
// Fail fast if the root is unreadable; deeper failures are skipped.
let _ = std::fs::read_dir(root)?;
let mut files = Vec::new();
let mut folder_count = 0;
scan_root(
root,
&root_display_name(root),
&mut files,
&mut folder_count,
);
Ok(Self::from_files(files, folder_count))
Ok(Self::scan_many(&[root.to_path_buf()]))
}
/// Scan several roots into one merged index. Unreadable roots are
soulseek-rs-lib/src/types.rs
-10
View File
@@ -43,16 +43,6 @@ impl Default for ClientVersion {
}
}
pub struct UploadFailed {
pub filename: String,
}
impl UploadFailed {
pub fn new_from_message(message: &mut Message) -> Self {
let filename = message.read_string();
Self { filename }
}
}
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct SearchResult {
soulseek-rs-lib/src/utils/logger.rs
+18 -64
View File
@@ -87,12 +87,12 @@ fn has_log_file() -> bool {
pub fn log(level: LogLevel, message: &str) {
unsafe {
if level <= LOG_LEVEL {
let level_str = match level {
LogLevel::Error => "\x1b[31mERROR\x1b[0m", // Red
LogLevel::Warn => "\x1b[33mWARN\x1b[0m", // Yellow
LogLevel::Info => "\x1b[32mINFO\x1b[0m", // Green
LogLevel::Debug => "\x1b[34mDEBUG\x1b[0m", // Blue
LogLevel::Trace => "\x1b[35mTRACE\x1b[0m", // Magenta
let (name, colour) = match level {
LogLevel::Error => ("ERROR", "\x1b[31m"), // Red
LogLevel::Warn => ("WARN", "\x1b[33m"), // Yellow
LogLevel::Info => ("INFO", "\x1b[32m"), // Green
LogLevel::Debug => ("DEBUG", "\x1b[34m"), // Blue
LogLevel::Trace => ("TRACE", "\x1b[35m"), // Magenta
};
let now = std::time::SystemTime::now();
@@ -149,20 +149,8 @@ pub fn log(level: LogLevel, message: &str) {
let minutes = (seconds_in_day % 3600) / 60;
let seconds = seconds_in_day % 60;
let level_str_plain = match level {
LogLevel::Error => "ERROR",
LogLevel::Warn => "WARN",
LogLevel::Info => "INFO",
LogLevel::Debug => "DEBUG",
LogLevel::Trace => "TRACE",
};
let formatted_message = format!(
"[{year:04}-{month:02}-{day:02} {hours:02}:{minutes:02}:{seconds:02}.{subsec_millis:03}] [{level_str}] {message}"
);
let formatted_message_plain = format!(
"[{year:04}-{month:02}-{day:02} {hours:02}:{minutes:02}:{seconds:02}.{subsec_millis:03}] [{level_str_plain}] {message}"
let timestamp = format!(
"{year:04}-{month:02}-{day:02} {hours:02}:{minutes:02}:{seconds:02}.{subsec_millis:03}"
);
match choose_sink(BUFFERING.load(Ordering::Relaxed), has_log_file())
@@ -171,17 +159,22 @@ pub fn log(level: LogLevel, message: &str) {
if let Ok(mut log_file) = LOG_FILE.lock()
&& let Some(file) = log_file.as_mut()
{
let _ = writeln!(file, "{formatted_message_plain}");
let _ =
writeln!(file, "[{timestamp}] [{name}] {message}");
let _ = file.flush();
}
}
LogSink::Buffer => {
if let Ok(mut buffer) = BUFFER.lock() {
buffer.push(formatted_message);
buffer.push(format!(
"[{timestamp}] [{colour}{name}\x1b[0m] {message}"
));
}
}
LogSink::Stderr => {
eprintln!("{formatted_message}");
eprintln!(
"[{timestamp}] [{colour}{name}\x1b[0m] {message}"
);
}
}
}
@@ -196,58 +189,19 @@ pub fn disable_buffering() {
BUFFERING.store(false, Ordering::Relaxed);
}
/// Only lines that were buffered land here, and a line is buffered only when
/// no log file is configured, so the buffer is always stderr-bound.
pub fn flush_buffered_logs() {
disable_buffering();
if let Ok(mut buffer) = BUFFER.lock() {
// Write to file if configured, otherwise to stderr
if let Ok(mut log_file) = LOG_FILE.lock() {
if let Some(file) = log_file.as_mut() {
for message in buffer.iter() {
// Strip ANSI codes for file output
let plain_message = strip_ansi_codes(message);
let _ = writeln!(file, "{plain_message}");
}
let _ = file.flush();
} else {
for message in buffer.iter() {
eprintln!("{message}");
}
}
} else {
// If lock fails, fall back to stderr
for message in buffer.iter() {
eprintln!("{message}");
}
}
buffer.clear();
}
}
fn strip_ansi_codes(s: &str) -> String {
let mut result = String::with_capacity(s.len());
let mut chars = s.chars().peekable();
while let Some(c) = chars.next() {
if c == '\x1b' {
// Skip ANSI escape sequence
if chars.peek() == Some(&'[') {
chars.next(); // consume '['
while let Some(&ch) = chars.peek() {
chars.next();
if ch.is_ascii_alphabetic() {
break;
}
}
}
} else {
result.push(c);
}
}
result
}
#[macro_export]
macro_rules! error {
($($arg:tt)*) => {
soulseek-rs-lib/tests/e2e.rs
+4 -27
View File
@@ -16,8 +16,6 @@
//! No external crates are used — the library forbids dependencies, so the
//! harness sticks to `std` and the library's own public API.
#![allow(clippy::doc_markdown)]
use std::io::{Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::path::PathBuf;
@@ -64,10 +62,11 @@ impl TestServer {
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Ok(addr) = std::env::var("SOULSEEK_TEST_SERVER") {
let (host, port) = split_host_port(&addr)?;
wait_until_listening(&host, port, Duration::from_secs(2))?;
let (host, port) = addr.rsplit_once(':')?;
let port = port.parse().ok()?;
wait_until_listening(host, port, Duration::from_secs(2))?;
return Some(Self {
host,
host: host.to_string(),
port,
child: None,
db: None,
@@ -152,11 +151,6 @@ impl Drop for TestServer {
}
}
fn split_host_port(addr: &str) -> Option<(String, u16)> {
let (host, port) = addr.rsplit_once(':')?;
Some((host.to_string(), port.parse().ok()?))
}
fn soulfind_binary() -> Option<PathBuf> {
if let Ok(p) = std::env::var("SOULFIND_BIN") {
let p = PathBuf::from(p);
@@ -465,22 +459,6 @@ fn registered_username_can_relogin_with_same_password() {
);
}
#[test]
fn two_clients_can_be_logged_in_together() {
let server = server_or_skip!();
// The server must handle several independent sessions at once — this is
// the precondition for any peer-to-peer feature routed through it.
let mut alice = Client::with_settings(server.settings("e2e_alice", "pw_a"));
let mut bob = Client::with_settings(server.settings("e2e_bob", "pw_b"));
alice.connect().expect("alice connect");
bob.connect().expect("bob connect");
assert!(alice.login().expect("alice login"));
assert!(bob.login().expect("bob login"));
}
// ---------------------------------------------------------------------------
// Peer-to-peer download coverage.
//
@@ -2149,7 +2127,6 @@ fn a_peer_asking_where_it_sits_is_answered_with_its_place() {
);
// The asker keeps its own connection so it can read the answer back.
// The asker keeps its connection so it can read the answer back.
let asker_file = format!("{folder}\\asker.mp3");
let mut asker =
queue_as(&server_addr, &listen_addr, "e2e_q_asker", &asker_file)
soulseek-rs/src/api.rs
+4
View File
@@ -149,8 +149,11 @@ pub trait SessionApi: Send + Sync {
Vec::new()
}
// aislop-ignore-next-line complexity/function-too-long -- scanner bug: a bodyless trait signature followed only by more signatures has no closing brace, so the counter runs to EOF. One line.
fn shared_counts(&self) -> (u32, u32);
// aislop-ignore-next-line complexity/function-too-long -- same scanner bug: bodyless signature measured to EOF. One line.
fn shared_directories(&self) -> Vec<String>;
// aislop-ignore-next-line complexity/function-too-long -- same scanner bug: bodyless signature measured to EOF. One line.
fn set_shared_directories(&self, directories: Vec<String>) -> Result<()>;
/// Point future transfers at `directory`.
@@ -158,6 +161,7 @@ pub trait SessionApi: Send + Sync {
/// Only a daemon holds a download folder of its own — locally every
/// download call names its directory, so the local session has nothing to
/// store and accepts silently.
// aislop-ignore-next-line complexity/function-too-long -- same scanner bug: bodyless signature measured to EOF. One line.
fn set_download_directory(&self, directory: String) -> Result<()>;
}
soulseek-rs/src/cli.rs
+2 -1
View File
@@ -392,9 +392,10 @@ pub enum ConfigCommand {
}
/// How to order search results.
#[derive(Copy, Clone, Debug, PartialEq, Eq, ValueEnum)]
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)]
pub enum SortKey {
/// Free slot first, then bitrate, peer speed, and size
#[default]
Best,
/// Largest file first
Size,
soulseek-rs/src/commands/completions.rs
+2 -21
View File
@@ -6,6 +6,7 @@
//! get one `source` line appended to their rc file, marked so `uninstall` can
//! take back exactly what was added and nothing else.
use super::{unwritable, xdg};
use crate::cli::{Cli, CompletionsArgs, CompletionsCommand, Shell};
use crate::output::{CliError, CliResult, CompletionRecord, Out};
use clap::CommandFactory;
@@ -39,7 +40,7 @@ struct Target {
impl Target {
fn record(&self, action: &str) -> CompletionRecord {
CompletionRecord {
shell: self.shell.name().to_string(),
shell: self.shell.generator().to_string(),
path: self.script.display().to_string(),
action: action.to_string(),
}
@@ -51,14 +52,6 @@ impl Target {
}
impl Shell {
const fn name(self) -> &'static str {
match self {
Self::Bash => "bash",
Self::Zsh => "zsh",
Self::Fish => "fish",
}
}
const fn generator(self) -> clap_complete::Shell {
match self {
Self::Bash => clap_complete::Shell::Bash,
@@ -213,14 +206,6 @@ impl Dirs {
}
}
/// Shells follow the XDG layout on every platform, unlike this client's own
/// config, which follows each platform's convention (see `persist::paths`).
fn xdg(variable: &str, home: &Path, fallback: &str) -> PathBuf {
std::env::var_os(variable)
.filter(|dir| !dir.is_empty())
.map_or_else(|| home.join(fallback), PathBuf::from)
}
fn write_script(path: &Path, script: &[u8]) -> CliResult<bool> {
if std::fs::read(path).is_ok_and(|found| found == script) {
return Ok(false);
@@ -288,10 +273,6 @@ fn create_parent(path: &Path) -> CliResult {
}
}
fn unwritable(path: &Path, error: &std::io::Error) -> CliError {
CliError::usage(format!("cannot write {}: {error}", path.display()))
}
#[cfg(test)]
mod tests {
use super::*;
soulseek-rs/src/commands/mod.rs
+45 -19
View File
@@ -22,6 +22,7 @@ use crate::remote::{Endpoint, RemoteSession};
use soulseek_rs::types::{RoomEvent, RoomInfo};
use soulseek_rs::{Client, ClientSettings, SessionLoss};
use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
@@ -223,6 +224,26 @@ fn login(settings: ClientSettings, address: &str) -> CliResult<Client> {
}
}
/// Ask `f` every `poll` until it answers or `timeout` runs out.
///
/// The library reports what the network sent by handing it over on the next
/// call, so a command that wants one answer has to keep asking. Checking first
/// and sleeping after means an answer already waiting costs nothing.
fn poll_until<T>(
timeout: Duration,
poll: Duration,
mut f: impl FnMut() -> Option<T>,
) -> Option<T> {
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
if let Some(answer) = f() {
return Some(answer);
}
std::thread::sleep(poll);
}
None
}
/// Ask for the room list and wait for the server's reply.
///
/// Events already queued are discarded first: servers push a room list at
@@ -235,21 +256,15 @@ pub fn await_room_list(
let _ = client.take_room_events();
client.request_room_list().ok()?;
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
let listed =
client.take_room_events().into_iter().find_map(
|event| match event {
poll_until(timeout, poll, || {
client
.take_room_events()
.into_iter()
.find_map(|event| match event {
RoomEvent::List(rooms) => Some(rooms),
_ => None,
},
);
if listed.is_some() {
return listed;
}
std::thread::sleep(poll);
}
None
})
})
}
/// Wait until the server has answered something we asked for after `send`.
@@ -295,12 +310,10 @@ fn dispatch(ctx: &Ctx, command: Commands) -> CliResult {
}
Commands::User(args) => peer::user(ctx, &args),
Commands::Whoami => peer::whoami(ctx),
// These need no account, so main.rs answers them before asking for
// credentials; the arms exist only to keep the match exhaustive.
Commands::Portmap => portmap(&ctx.out, ctx.settings.listen_port),
// These need the config file as well as a session, so main.rs runs
// them where the store is in scope.
Commands::Serve(_)
// main.rs runs these itself: some need no account, the rest need the
// config store in scope. The arms only keep the match exhaustive.
Commands::Portmap
| Commands::Serve(_)
| Commands::Daemon(_)
| Commands::Wish(_)
| Commands::Config(_)
@@ -317,6 +330,19 @@ fn listen_span(duration: u64, follow: bool) -> Option<Duration> {
(!follow).then(|| Duration::from_secs(duration))
}
/// Where shells and coding agents keep their configuration: the XDG layout on
/// every platform, unlike this client's own config, which follows each
/// platform's convention (see `persist::paths`).
fn xdg(variable: &str, home: &Path, fallback: &str) -> PathBuf {
std::env::var_os(variable)
.filter(|dir| !dir.is_empty())
.map_or_else(|| home.join(fallback), PathBuf::from)
}
fn unwritable(path: &Path, error: &std::io::Error) -> CliError {
CliError::usage(format!("cannot write {}: {error}", path.display()))
}
/// Probe automatic port mapping. A router that does not answer is a negative
/// answer, not a crash, so it exits 4 the way `grep` exits 1 on no match.
pub fn portmap(out: &Out, port: u16) -> CliResult {
soulseek-rs/src/commands/peer.rs
+7 -23
View File
@@ -47,14 +47,9 @@ fn own_privileges(client: &dyn crate::api::SessionApi) -> Option<u32> {
if client.check_privileges().is_err() {
return None;
}
let deadline = Instant::now() + PRIVILEGE_TIMEOUT;
while Instant::now() < deadline {
if let Some(seconds) = client.own_privilege_seconds() {
return Some(seconds);
}
std::thread::sleep(Duration::from_millis(50));
}
None
super::poll_until(PRIVILEGE_TIMEOUT, Duration::from_millis(50), || {
client.own_privilege_seconds()
})
}
pub fn user(ctx: &Ctx, args: &UserArgs) -> CliResult {
@@ -141,7 +136,7 @@ pub fn serve(
.warn("the share index is empty — peers will find nothing");
}
let span = (!args.follow).then(|| Duration::from_secs(args.duration));
let span = super::listen_span(args.duration, args.follow);
ctx.out.status(&match span {
Some(span) => format!("online for {}s…", span.as_secs()),
None => "online until interrupted…".to_string(),
@@ -155,20 +150,10 @@ pub fn serve(
// served well inside one poll, and "it had to wait" is precisely what a
// transfer log exists to report.
for queued in session.client.take_upload_events() {
let record = upload_record(&queued);
let key = (record.user.clone(), record.path.clone());
if reported.get(&key) != Some(&record) {
ctx.out.emit(&record);
reported.insert(key, record);
}
emit_changed(&mut reported, upload_record(&queued), &ctx.out);
}
for upload in session.client.uploads() {
let record = upload_record(&upload);
let key = (record.user.clone(), record.path.clone());
if reported.get(&key) != Some(&record) {
ctx.out.emit(&record);
reported.insert(key, record);
}
emit_changed(&mut reported, upload_record(&upload), &ctx.out);
}
if let Some(sweeper) = sweeper.as_mut() {
sweeper.tick(ctx, session.client.as_ref());
@@ -211,8 +196,7 @@ fn upload_record(upload: &UploadInfo) -> UploadRecord {
}
/// Emit `record` only when it differs from the last one reported for that
/// (user, file). Exposed for tests; the serve loop inlines the same rule.
#[cfg(test)]
/// (user, file).
fn emit_changed(
reported: &mut HashMap<(String, String), UploadRecord>,
record: UploadRecord,
soulseek-rs/src/commands/skills.rs
+2 -14
View File
@@ -4,6 +4,7 @@
//! inside it and is written out on demand. Installing over an older copy is
//! what updating means, which is why there is no separate verb for it.
use super::{unwritable, xdg};
use crate::cli::{SkillsArgs, SkillsCommand};
use crate::output::{CliError, CliResult, Out, SkillRecord};
use directories::UserDirs;
@@ -122,7 +123,7 @@ fn targets(args: &SkillsArgs) -> CliResult<Vec<Target>> {
)
})?;
let found = detected(&home, &config_home(&home));
let found = detected(&home, &xdg("XDG_CONFIG_HOME", &home, ".config"));
if found.is_empty() {
return Err(CliError::no_results(
"no coding agent found in your home directory; name a skills \
@@ -149,15 +150,6 @@ fn detected(home: &Path, config_home: &Path) -> Vec<Target> {
.collect()
}
/// Where an agent keeps its own configuration, which is the XDG location on
/// every platform — unlike this client's config, which follows each platform's
/// own convention (see `persist::paths`).
fn config_home(home: &Path) -> PathBuf {
std::env::var_os("XDG_CONFIG_HOME")
.filter(|dir| !dir.is_empty())
.map_or_else(|| home.join(".config"), PathBuf::from)
}
fn current(file: &Path) -> bool {
std::fs::read_to_string(file).is_ok_and(|found| found == SKILL)
}
@@ -167,10 +159,6 @@ fn ours(file: &Path) -> bool {
.is_ok_and(|found| found.contains(FRONTMATTER_NAME))
}
fn unwritable(path: &Path, error: &std::io::Error) -> CliError {
CliError::usage(format!("cannot write {}: {error}", path.display()))
}
#[cfg(test)]
mod tests {
use super::*;
soulseek-rs/src/commands/social.rs
+11 -19
View File
@@ -39,17 +39,13 @@ pub fn fetch_listing(
CliError::connection(format!("cannot ask {user} for a listing: {e}"))
})?;
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
if let Some(directories) = client.take_browse_result(user) {
return Ok(directories);
}
std::thread::sleep(POLL);
}
Err(CliError::timeout(format!(
super::poll_until(timeout, POLL, || client.take_browse_result(user))
.ok_or_else(|| {
CliError::timeout(format!(
"{user} did not send a file list within {}s",
timeout.as_secs()
)))
))
})
}
pub fn browse(ctx: &Ctx, args: &BrowseArgs) -> CliResult {
@@ -143,22 +139,18 @@ pub fn room_users(ctx: &Ctx, room: &str, timeout: Duration) -> CliResult {
ctx.out
.status(&format!("waiting for {room}'s member list…"));
let deadline = Instant::now() + timeout;
let mut members = Vec::new();
while Instant::now() < deadline && members.is_empty() {
members = session.client.room_members(room);
if members.is_empty() {
std::thread::sleep(POLL);
}
}
let listed = super::poll_until(timeout, POLL, || {
let members = session.client.room_members(room);
(!members.is_empty()).then_some(members)
});
let _ = session.client.leave_room(room);
if members.is_empty() {
let Some(members) = listed else {
return Err(CliError::timeout(format!(
"{room} sent no member list within {}s",
timeout.as_secs()
)));
}
};
for user in members {
ctx.out.emit(&RoomMemberRecord {
room: room.to_string(),
soulseek-rs/src/commands/transfer.rs
+16 -24
View File
@@ -310,13 +310,9 @@ fn collect(
})
};
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
if client.get_search_results_count(query) > 0 {
break;
}
std::thread::sleep(Duration::from_millis(100));
}
super::poll_until(timeout, Duration::from_millis(100), || {
(client.get_search_results_count(query) > 0).then_some(())
});
cancel.store(true, Ordering::Relaxed);
let _ = worker.join();
Ok(client.get_search_results(query))
@@ -334,29 +330,26 @@ fn dedupe(
download_dir: &str,
out: &Out,
) -> Vec<Request> {
let mut paths: HashMap<String, String> = HashMap::new();
let mut targets: HashMap<PathBuf, String> = HashMap::new();
let mut targets: HashMap<PathBuf, Request> = HashMap::new();
let mut kept = Vec::new();
for request in requests {
if let Some(user) = paths.get(&request.path) {
out.warn(&format!(
"skipping {} from {}: same remote path already queued from \
{user}",
request.path, request.user
));
continue;
}
let target = local_path(download_dir, &request.path);
if let Some(other) = targets.get(&target) {
if let Some(queued) = targets.get(&target) {
let collision = if queued.path == request.path {
format!("same remote path already queued from {}", queued.user)
} else {
format!(
"would overwrite the file already queued from {}",
queued.user
)
};
out.warn(&format!(
"skipping {} from {}: would overwrite the file already queued \
from {other}",
"skipping {} from {}: {collision}",
request.path, request.user
));
continue;
}
paths.insert(request.path.clone(), request.user.clone());
targets.insert(target, request.user.clone());
targets.insert(target, request.clone());
kept.push(request);
}
kept
@@ -617,9 +610,8 @@ fn read_requests(
}
/// `Ok(None)` for a blank line, `Err` for something unusable.
// aislop-ignore-next-line complexity/function-too-long -- scanner bug, not a long function: the `'{'` char literal below reads as an opening brace, so aislop's counter never closes and measures to EOF. It is 34 lines.
// aislop-ignore-next-line complexity/function-too-long -- scanner bug, not a long function: the `'{'` char literal below reads as an opening brace, so aislop's counter never closes and measures to EOF. It is 33 lines.
fn parse_request(line: &str) -> Result<Option<Request>, String> {
let line = line.trim_end_matches(['\r', '\n']);
if line.trim().is_empty() {
return Ok(None);
}
soulseek-rs/src/commands/wish.rs
+1 -10
View File
@@ -68,22 +68,13 @@ pub fn run(ctx: &Ctx, store: &Store, args: &WishRunArgs) -> CliResult {
/// How to narrow and order what a sweep prints — the parts of `wish run`'s flags
/// that outlive the command, so `serve --wishlist` can sweep without inventing
/// a set of command-line arguments it never received.
#[derive(Default)]
struct Sweep {
filter: transfer::Filter,
sort: SortKey,
limit: usize,
}
impl Default for Sweep {
fn default() -> Self {
Self {
filter: transfer::Filter::default(),
sort: SortKey::Best,
limit: 0,
}
}
}
impl From<&WishRunArgs> for Sweep {
fn from(args: &WishRunArgs) -> Self {
Self {
soulseek-rs/src/daemon/hub.rs
+11 -11
View File
@@ -193,16 +193,14 @@ const BROWSE_PATIENCE: Duration = Duration::from_mins(2);
/// `take_browse_result` is per-user and destructive, so the daemon has to know
/// who to ask about.
#[derive(Default)]
pub struct PendingBrowses(Mutex<Vec<(String, Instant)>>);
pub struct PendingBrowses(Mutex<HashMap<String, Instant>>);
impl PendingBrowses {
pub fn expect(&self, username: &str) {
let Ok(mut pending) = self.0.lock() else {
return;
};
if let Ok(mut pending) = self.0.lock() {
// Asking again restarts the clock: it is a fresh request.
pending.retain(|(name, _)| name != username);
pending.push((username.to_string(), Instant::now()));
pending.insert(username.to_string(), Instant::now());
}
}
/// Who to ask about this tick, forgetting the ones that have run out of
@@ -211,15 +209,15 @@ impl PendingBrowses {
let Ok(mut pending) = self.0.lock() else {
return Vec::new();
};
pending.retain(|(_, asked)| asked.elapsed() < BROWSE_PATIENCE);
pending.iter().map(|(name, _)| name.clone()).collect()
pending.retain(|_, asked| asked.elapsed() < BROWSE_PATIENCE);
pending.keys().cloned().collect()
}
/// Stop waiting on a peer — because it answered, or because its request
/// never reached the wire.
pub fn forget(&self, username: &str) {
if let Ok(mut pending) = self.0.lock() {
pending.retain(|(name, _)| name != username);
pending.remove(username);
}
}
}
@@ -511,8 +509,10 @@ mod tests {
let asked = Instant::now()
.checked_sub(BROWSE_PATIENCE + Duration::from_secs(1))
.expect("the clock is past the epoch");
let pending =
PendingBrowses(Mutex::new(vec![("ghost".to_string(), asked)]));
let pending = PendingBrowses(Mutex::new(HashMap::from([(
"ghost".to_string(),
asked,
)])));
assert!(pending.snapshot().is_empty());
}
}
soulseek-rs/src/daemon/mod.rs
+29 -35
View File
@@ -332,48 +332,42 @@ const fn raise_open_file_limit() -> std::io::Result<u64> {
/// afterwards is a poor first experience. stdout stays empty either way, so
/// `daemon` still composes in a pipeline.
fn ready(ctx: &Ctx, bind: Option<&str>, token: &str) {
ctx.out.status("");
ctx.out.status("ready — leave this running.");
ctx.out.status("");
ctx.out
.status(" On this machine, nothing else to do. Other commands find");
ctx.out
.status(" it on their own and share this one login:");
ctx.out.status("");
ctx.out.status(" soulseek-rs search 'aphex twin'");
ctx.out
.status(" soulseek-rs (the TUI attaches)");
ctx.out.status(" soulseek-rs daemon status");
ctx.out.status(" soulseek-rs daemon stop");
ctx.out.status("");
const BANNER: &str = "
ready — leave this running.
On this machine, nothing else to do. Other commands find
it on their own and share this one login:
soulseek-rs search 'aphex twin'
soulseek-rs (the TUI attaches)
soulseek-rs daemon status
soulseek-rs daemon stop
";
const LOCAL_ONLY: &str =
" To drive it from another machine, restart with --bind ADDR;
`soulseek-rs daemon token` prints the secret it needs.
";
ctx.out.status(BANNER);
let Some(address) = bind else {
ctx.out.status(
" To drive it from another machine, restart with --bind ADDR;",
);
ctx.out
.status(" `soulseek-rs daemon token` prints the secret it needs.");
ctx.out.status("");
ctx.out.status(LOCAL_ONLY);
return;
};
ctx.out.status(&format!(
" From another machine, point a client at {address} with"
" From another machine, point a client at {address} with
this token:
{token}
soulseek-rs config set daemon <this-host>:PORT
soulseek-rs config set daemon_token <token above>
That port has no encryption. Keep it off the open internet and \
reach it over SSH instead.
"
));
ctx.out.status(" this token:");
ctx.out.status("");
ctx.out.status(&format!(" {token}"));
ctx.out.status("");
ctx.out
.status(" soulseek-rs config set daemon <this-host>:PORT");
ctx.out
.status(" soulseek-rs config set daemon_token <token above>");
ctx.out.status("");
ctx.out.status(
" That port has no encryption. Keep it off the open internet and \
reach it over SSH instead.",
);
ctx.out.status("");
}
/// Remove the socket file so the next start does not have to reason about
soulseek-rs/src/daemon/openrpc.rs
+2 -6
View File
@@ -23,11 +23,6 @@ use serde_json::{Value, json};
/// `components/schemas`, the same place OpenAPI does.
const DEFINITIONS_PATH: &str = "#/components/schemas/";
/// A method that takes no parameters.
fn nothing() -> Value {
json!({ "type": "null" })
}
fn schema_of<T: JsonSchema>(generator: &mut SchemaGenerator) -> Value {
generator.subschema_for::<T>().to_value()
}
@@ -60,6 +55,7 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
schema_of::<UserRef>(generator)
}
Method::SharesSet => schema_of::<DirectoriesParams>(generator),
// Methods that take no parameters.
Method::DaemonStatus
| Method::DaemonStop
| Method::RpcDiscover
@@ -72,7 +68,7 @@ fn params_schema(method: Method, generator: &mut SchemaGenerator) -> Value {
| Method::MessageHistory
| Method::SearchList
| Method::SharesStatusOf
| Method::SharesReindex => nothing(),
| Method::SharesReindex => json!({ "type": "null" }),
}
}
soulseek-rs/src/daemon/server/mod.rs
-1
View File
@@ -52,7 +52,6 @@ impl Daemon {
///
/// The match is exhaustive over [`Method`], so a new method cannot be added
/// to the protocol without being implemented here.
#[allow(clippy::too_many_lines)]
pub fn call(
&self,
method: Method,
soulseek-rs/src/directories.rs
-1
View File
@@ -37,7 +37,6 @@ pub fn resolve_shared_directory(
/// Validate several configured share paths; invalid ones are skipped with a
/// warning instead of aborting (the client just shares less).
#[cfg_attr(not(test), allow(dead_code))]
#[must_use]
pub fn resolve_shared_directories(dirs: &[String]) -> Vec<String> {
let mut resolved = Vec::new();
soulseek-rs/src/models/app_state.rs
-14
View File
@@ -24,7 +24,6 @@ pub struct SearchEntry {
/// finishes; for the rest the session is the authority.
pub owned: bool,
pub start_time: Instant,
#[allow(dead_code)]
pub cancel_flag: Arc<AtomicBool>,
}
@@ -249,19 +248,6 @@ impl AppState {
let peer = peers[next].to_string();
self.chat_peer = Some(peer);
}
#[allow(dead_code)]
#[must_use]
pub fn get_selected_search(&self) -> Option<&SearchEntry> {
self.selected_search_index
.and_then(|idx| self.searches.get(idx))
}
#[allow(dead_code)]
pub fn get_selected_search_mut(&mut self) -> Option<&mut SearchEntry> {
self.selected_search_index
.and_then(|idx| self.searches.get_mut(idx))
}
}
impl Default for AppState {
soulseek-rs/src/output.rs
+5 -6
View File
@@ -436,12 +436,11 @@ pub struct ConfigRecord {
impl Record for ConfigRecord {
fn text(&self) -> String {
match &self.value {
Some(value) => {
format!("{}\t{}", sanitize(&self.key), sanitize(value))
}
None => format!("{}\t", sanitize(&self.key)),
}
format!(
"{}\t{}",
sanitize(&self.key),
sanitize(self.value.as_deref().unwrap_or_default())
)
}
}
soulseek-rs/src/persist/state.rs
+28 -87
View File
@@ -1,20 +1,19 @@
//! Versioned JSON state files (downloads, search queries, open rooms,
//! private-message history).
//!
//! Each file is an envelope `{ "version": N, "data": ... }`. On load the
//! data passes through the migration chain from its stored version up to
//! current. State is disposable: a missing, corrupt, or newer-than-known
//! file loads as empty rather than failing startup. Writes are atomic
//! (tmp file + rename) so a crash never leaves a torn file.
//! Each file is an envelope `{ "version": N, "data": ... }`. State is
//! disposable: a missing, corrupt, or newer-than-known file loads as empty
//! rather than failing startup. Writes are atomic (tmp file + rename) so a
//! crash never leaves a torn file.
use color_eyre::Result;
use serde::{Serialize, de::DeserializeOwned};
use serde_json::Value;
use std::path::{Path, PathBuf};
/// A migration takes the `data` value at version `i` and returns it at
/// version `i + 1`.
type Migration = fn(Value) -> Value;
/// The shape every file is written in today. Bump it — and convert the older
/// shapes in `load` — when one of them changes.
const VERSION: u32 = 0;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, serde::Deserialize)]
pub struct PersistedDownload {
@@ -45,86 +44,59 @@ impl StateStore {
Self { dir }
}
#[must_use]
pub fn load_downloads(&self) -> Vec<PersistedDownload> {
load(&self.dir.join("downloads.json"), DOWNLOADS_MIGRATIONS)
load(&self.dir.join("downloads.json"))
}
pub fn save_downloads(
&self,
downloads: &[PersistedDownload],
) -> Result<()> {
save(
&self.dir.join("downloads.json"),
DOWNLOADS_MIGRATIONS.len() as u32,
&downloads,
)
save(&self.dir.join("downloads.json"), VERSION, &downloads)
}
#[must_use]
pub fn load_search_queries(&self) -> Vec<String> {
load(&self.dir.join("searches.json"), SEARCHES_MIGRATIONS)
load(&self.dir.join("searches.json"))
}
pub fn save_search_queries(&self, queries: &[String]) -> Result<()> {
save(
&self.dir.join("searches.json"),
SEARCHES_MIGRATIONS.len() as u32,
&queries,
)
save(&self.dir.join("searches.json"), VERSION, &queries)
}
#[must_use]
pub fn load_rooms(&self) -> Vec<String> {
load(&self.dir.join("rooms.json"), ROOMS_MIGRATIONS)
load(&self.dir.join("rooms.json"))
}
pub fn save_rooms(&self, rooms: &[String]) -> Result<()> {
save(
&self.dir.join("rooms.json"),
ROOMS_MIGRATIONS.len() as u32,
&rooms,
)
save(&self.dir.join("rooms.json"), VERSION, &rooms)
}
#[must_use]
pub fn load_messages(&self) -> Vec<PersistedMessage> {
load(&self.dir.join("messages.json"), MESSAGES_MIGRATIONS)
load(&self.dir.join("messages.json"))
}
pub fn save_messages(&self, messages: &[PersistedMessage]) -> Result<()> {
save(
&self.dir.join("messages.json"),
MESSAGES_MIGRATIONS.len() as u32,
&messages,
)
save(&self.dir.join("messages.json"), VERSION, &messages)
}
/// The unread private-message count, so the badge survives a restart.
#[must_use]
pub fn load_unread(&self) -> usize {
load(&self.dir.join("unread.json"), UNREAD_MIGRATIONS)
load(&self.dir.join("unread.json"))
}
pub fn save_unread(&self, unread: usize) -> Result<()> {
save(
&self.dir.join("unread.json"),
UNREAD_MIGRATIONS.len() as u32,
&unread,
)
save(&self.dir.join("unread.json"), VERSION, &unread)
}
}
/// Per-file migration chains. `data` at version `i` is upgraded by
/// `MIGRATIONS[i]`; the current version is the chain length. All formats
/// are at version 0 today — add a fn here when the schema changes.
const DOWNLOADS_MIGRATIONS: &[Migration] = &[];
const SEARCHES_MIGRATIONS: &[Migration] = &[];
const ROOMS_MIGRATIONS: &[Migration] = &[];
const MESSAGES_MIGRATIONS: &[Migration] = &[];
const UNREAD_MIGRATIONS: &[Migration] = &[];
/// Load `data` from an envelope file, migrating old versions forward.
/// Missing, corrupt, or newer-than-known files all yield `T::default()`.
fn load<T: DeserializeOwned + Default>(
path: &Path,
migrations: &[Migration],
) -> T {
/// Load `data` from an envelope file. Missing, corrupt, or newer-than-known
/// files all yield `T::default()`.
fn load<T: DeserializeOwned + Default>(path: &Path) -> T {
let Ok(text) = std::fs::read_to_string(path) else {
return T::default();
};
@@ -135,15 +107,12 @@ fn load<T: DeserializeOwned + Default>(
let version = envelope
.get("version")
.and_then(Value::as_u64)
.unwrap_or(u64::MAX) as usize;
if version > migrations.len() {
.unwrap_or(u64::MAX);
if version > u64::from(VERSION) {
set_aside(path, "from a newer version");
return T::default();
}
let mut data = envelope.get("data").cloned().unwrap_or(Value::Null);
for migration in &migrations[version..] {
data = migration(data);
}
let data = envelope.get("data").cloned().unwrap_or(Value::Null);
serde_json::from_value(data).unwrap_or_else(|_| {
set_aside(path, "unreadable");
T::default()
@@ -297,34 +266,6 @@ mod tests {
assert!(path.with_extension("json.bak").exists());
}
#[test]
fn migrations_upgrade_old_data() {
// Machinery test with a synthetic chain: v0 stored plain strings,
// v1 wraps each in an object {"name": ...}.
#[derive(Debug, PartialEq, Default, Serialize, serde::Deserialize)]
struct Named {
name: String,
}
let chain: &[Migration] = &[|data| {
Value::Array(
data.as_array()
.cloned()
.unwrap_or_default()
.into_iter()
.map(|s| serde_json::json!({ "name": s }))
.collect(),
)
}];
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("things.json");
std::fs::write(&path, r#"{"version": 0, "data": ["a", "b"]}"#).unwrap();
let loaded: Vec<Named> = load(&path, chain);
assert_eq!(
loaded,
vec![Named { name: "a".into() }, Named { name: "b".into() }]
);
}
#[test]
fn save_is_atomic_no_tmp_file_left_behind() {
let (tmp, store) = store();
soulseek-rs/src/port_mapping/nat_pmp.rs
-8
View File
@@ -52,8 +52,6 @@ impl From<io::Error> for NatPmpError {
/// a shorter lifetime than requested, so callers must use these values.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MapResponse {
/// Seconds since the gateway's port table was initialised (reboot detector).
pub epoch: u32,
pub internal_port: u16,
pub external_port: u16,
pub lifetime: u32,
@@ -87,12 +85,6 @@ pub fn map_tcp(
return Err(NatPmpError::ShortResponse);
}
Ok(MapResponse {
epoch: u32::from_be_bytes([
response[4],
response[5],
response[6],
response[7],
]),
internal_port: u16::from_be_bytes([response[8], response[9]]),
external_port: u16::from_be_bytes([response[10], response[11]]),
lifetime: u32::from_be_bytes([
soulseek-rs/src/run.rs
+15 -21
View File
@@ -369,6 +369,19 @@ pub fn init_logging(cli: &Cli) {
}
}
/// Clear the screen, turn on mouse reporting and take over the terminal.
fn enter_terminal() -> ratatui::DefaultTerminal {
use ratatui::crossterm::{
event::EnableMouseCapture,
execute,
terminal::{Clear, ClearType},
};
let _ =
execute!(std::io::stdout(), Clear(ClearType::All), EnableMouseCapture);
ratatui::init()
}
/// Run the interactive TUI (the default no-subcommand path): bring the
/// terminal up first, run the login/registration screen (skipped past when
/// stored credentials work), persist whatever logged in, then enter the
@@ -379,12 +392,6 @@ fn run_default_tui(
config_path: Option<PathBuf>,
file_config: &crate::persist::config::FileConfig,
) -> CliResult {
use ratatui::crossterm::{
event::EnableMouseCapture,
execute,
terminal::{Clear, ClearType},
};
let out = Out::new(false, false);
let (server_host, server_port) =
parse_server_address(&resolved.server).map_err(CliError::usage)?;
@@ -419,10 +426,7 @@ fn run_default_tui(
version: ClientVersion::REFERENCE_CLIENT,
};
// Clear screen and enable mouse capture before initializing TUI
let _ =
execute!(std::io::stdout(), Clear(ClearType::All), EnableMouseCapture);
let mut terminal = ratatui::init();
let mut terminal = enter_terminal();
let outcome = crate::ui::login::run_login_flow(
&mut terminal,
@@ -468,7 +472,6 @@ fn run_default_tui(
terminal,
Arc::new(outcome.client),
resolved.download_dir.clone(),
resolved.max_concurrent_downloads,
Duration::from_secs(resolved.search_timeout),
store,
config_path,
@@ -486,12 +489,6 @@ fn run_attached_tui(
endpoint: &crate::remote::Endpoint,
config_path: Option<PathBuf>,
) -> CliResult {
use ratatui::crossterm::{
event::EnableMouseCapture,
execute,
terminal::{Clear, ClearType},
};
// The resolved token, not the flag alone: `daemon_token` in config.toml
// is the documented remote setup, and the flag/env are layered into it.
let token = resolved
@@ -521,9 +518,7 @@ fn run_attached_tui(
})?;
soulseek_rs::utils::logger::enable_buffering();
let _ =
execute!(std::io::stdout(), Clear(ClearType::All), EnableMouseCapture);
let terminal = ratatui::init();
let terminal = enter_terminal();
// No StateStore: the daemon owns what survives a restart, so a second
// writer here would fight it for the same files.
@@ -531,7 +526,6 @@ fn run_attached_tui(
terminal,
Arc::new(session),
download_dir,
resolved.max_concurrent_downloads,
Duration::from_secs(resolved.search_timeout),
None,
config_path,
soulseek-rs/src/ui/main_tui/input.rs
+52 -107
View File
@@ -3,6 +3,8 @@ use crate::models::{CommandBarMode, FocusedPane};
use ratatui::crossterm::event::{
KeyCode, KeyEvent, KeyModifiers, MouseButton, MouseEvent, MouseEventKind,
};
use ratatui::layout::Position;
use ratatui::widgets::TableState;
impl MainTui {
pub(super) fn handle_key_event(&mut self, key: KeyEvent) {
@@ -189,10 +191,10 @@ impl MainTui {
KeyCode::Backspace => {
let cursor_position = self.state.command_bar_cursor_position;
if cursor_position > 0 {
let previous_position = previous_char_boundary(
&self.state.command_bar_input,
cursor_position,
);
let previous_position = self
.state
.command_bar_input
.floor_char_boundary(cursor_position.saturating_sub(1));
self.state
.command_bar_input
.drain(previous_position..cursor_position);
@@ -202,26 +204,26 @@ impl MainTui {
KeyCode::Delete => {
let cursor_position = self.state.command_bar_cursor_position;
if cursor_position < self.state.command_bar_input.len() {
let next_position = next_char_boundary(
&self.state.command_bar_input,
cursor_position,
);
let next_position = self
.state
.command_bar_input
.ceil_char_boundary(cursor_position + 1);
self.state
.command_bar_input
.drain(cursor_position..next_position);
}
}
KeyCode::Left => {
self.state.command_bar_cursor_position = previous_char_boundary(
&self.state.command_bar_input,
self.state.command_bar_cursor_position,
);
let previous =
self.state.command_bar_cursor_position.saturating_sub(1);
self.state.command_bar_cursor_position =
self.state.command_bar_input.floor_char_boundary(previous);
}
KeyCode::Right => {
self.state.command_bar_cursor_position = next_char_boundary(
&self.state.command_bar_input,
self.state.command_bar_cursor_position,
);
let input = &self.state.command_bar_input;
let next = self.state.command_bar_cursor_position + 1;
self.state.command_bar_cursor_position =
input.ceil_char_boundary(next.min(input.len()));
}
KeyCode::Home => {
self.state.command_bar_cursor_position = 0;
@@ -288,26 +290,13 @@ impl MainTui {
}
fn handle_searches_input(&mut self, key: KeyEvent) {
let rows = self.state.searches.len();
match key.code {
KeyCode::Up | KeyCode::Char('k')
if !self.state.searches.is_empty() =>
{
let current =
self.state.searches_table_state.selected().unwrap_or(0);
let new = if current == 0 {
self.state.searches.len() - 1
} else {
current - 1
};
self.state.searches_table_state.select(Some(new));
KeyCode::Up | KeyCode::Char('k') => {
cycle(&mut self.state.searches_table_state, rows, false);
}
KeyCode::Down | KeyCode::Char('j')
if !self.state.searches.is_empty() =>
{
let current =
self.state.searches_table_state.selected().unwrap_or(0);
let new = (current + 1) % self.state.searches.len();
self.state.searches_table_state.select(Some(new));
KeyCode::Down | KeyCode::Char('j') => {
cycle(&mut self.state.searches_table_state, rows, true);
}
KeyCode::Enter => {
if let Some(selected) =
@@ -348,21 +337,11 @@ impl MainTui {
};
match key.code {
KeyCode::Up | KeyCode::Char('k') if items_count > 0 => {
let current =
self.state.results_table_state.selected().unwrap_or(0);
let new = if current == 0 {
items_count - 1
} else {
current - 1
};
self.state.results_table_state.select(Some(new));
KeyCode::Up | KeyCode::Char('k') => {
cycle(&mut self.state.results_table_state, items_count, false);
}
KeyCode::Down | KeyCode::Char('j') if items_count > 0 => {
let current =
self.state.results_table_state.selected().unwrap_or(0);
let new = (current + 1) % items_count;
self.state.results_table_state.select(Some(new));
KeyCode::Down | KeyCode::Char('j') => {
cycle(&mut self.state.results_table_state, items_count, true);
}
KeyCode::Char(' ') => {
if let Some(current) = self.state.results_table_state.selected()
@@ -416,17 +395,11 @@ impl MainTui {
// The pane lists downloads first, then uploads; navigation spans both.
let rows = self.state.downloads.len() + self.state.uploads.len();
match key.code {
KeyCode::Up | KeyCode::Char('k') if rows > 0 => {
let current =
self.state.downloads_table_state.selected().unwrap_or(0);
let new = if current == 0 { rows - 1 } else { current - 1 };
self.state.downloads_table_state.select(Some(new));
KeyCode::Up | KeyCode::Char('k') => {
cycle(&mut self.state.downloads_table_state, rows, false);
}
KeyCode::Down | KeyCode::Char('j') if rows > 0 => {
let current =
self.state.downloads_table_state.selected().unwrap_or(0);
let new = (current + 1) % rows;
self.state.downloads_table_state.select(Some(new));
KeyCode::Down | KeyCode::Char('j') => {
cycle(&mut self.state.downloads_table_state, rows, true);
}
KeyCode::Char('x') => {
self.cancel_selected_upload();
@@ -448,63 +421,35 @@ impl MainTui {
}
pub(super) fn handle_mouse_event(&mut self, mouse: MouseEvent) {
if mouse.kind == MouseEventKind::Down(MouseButton::Left) {
let (col, row) = (mouse.column, mouse.row);
// Check if click is within searches pane
if let Some(area) = self.state.searches_pane_area
&& col >= area.x
&& col < area.x + area.width
&& row >= area.y
&& row < area.y + area.height
{
self.state.focused_pane = FocusedPane::Searches;
if mouse.kind != MouseEventKind::Down(MouseButton::Left) {
return;
}
// Check if click is within results pane
if let Some(area) = self.state.results_pane_area
&& col >= area.x
&& col < area.x + area.width
&& row >= area.y
&& row < area.y + area.height
{
self.state.focused_pane = FocusedPane::Results;
let clicked = Position::new(mouse.column, mouse.row);
let panes = [
(self.state.searches_pane_area, FocusedPane::Searches),
(self.state.results_pane_area, FocusedPane::Results),
(self.state.downloads_pane_area, FocusedPane::Downloads),
];
for (area, pane) in panes {
if area.is_some_and(|area| area.contains(clicked)) {
self.state.focused_pane = pane;
return;
}
// Check if click is within downloads pane
if let Some(area) = self.state.downloads_pane_area
&& col >= area.x
&& col < area.x + area.width
&& row >= area.y
&& row < area.y + area.height
{
self.state.focused_pane = FocusedPane::Downloads;
}
}
}
}
fn previous_char_boundary(input: &str, cursor_position: usize) -> usize {
let cursor_position = input.floor_char_boundary(cursor_position);
input[..cursor_position]
.char_indices()
.last()
.map_or(0, |(index, _)| index)
}
fn next_char_boundary(input: &str, cursor_position: usize) -> usize {
let cursor_position = input.floor_char_boundary(cursor_position);
if cursor_position >= input.len() {
return input.len();
/// Move a table's selection one row, wrapping at either end.
fn cycle(table: &mut TableState, len: usize, forward: bool) {
if len == 0 {
return;
}
cursor_position
+ input[cursor_position..]
.chars()
.next()
.map_or(0, char::len_utf8)
let current = table.selected().unwrap_or(0);
let next = if forward {
(current + 1) % len
} else {
(current + len - 1) % len
};
table.select(Some(next));
}
soulseek-rs/src/ui/main_tui/mod.rs
+2 -16
View File
@@ -23,8 +23,6 @@ pub struct MainTui {
client: Arc<dyn SessionApi>,
state: AppState,
download_dir: String,
#[allow(dead_code)]
max_concurrent_downloads: usize,
search_timeout: Duration,
spinner_state: usize,
store: Option<StateStore>,
@@ -39,7 +37,6 @@ impl MainTui {
pub fn new(
client: Arc<dyn SessionApi>,
download_dir: String,
max_concurrent_downloads: usize,
search_timeout: Duration,
store: Option<StateStore>,
config_path: Option<std::path::PathBuf>,
@@ -48,7 +45,6 @@ impl MainTui {
client,
state: AppState::new(),
download_dir,
max_concurrent_downloads,
search_timeout,
spinner_state: 0,
store,
@@ -256,19 +252,12 @@ pub fn launch_main_tui(
terminal: DefaultTerminal,
client: Arc<dyn SessionApi>,
download_dir: String,
max_concurrent_downloads: usize,
search_timeout: Duration,
store: Option<StateStore>,
config_path: Option<std::path::PathBuf>,
) -> Result<()> {
let tui = MainTui::new(
client,
download_dir,
max_concurrent_downloads,
search_timeout,
store,
config_path,
);
let tui =
MainTui::new(client, download_dir, search_timeout, store, config_path);
tui.run(terminal)
}
@@ -526,7 +515,6 @@ mod tests {
MainTui::new(
Arc::new(session),
"/tmp".to_string(),
1,
Duration::from_secs(1),
None,
None,
@@ -746,7 +734,6 @@ mod tests {
let mut tui = MainTui::new(
session.clone(),
"/daemon/old".to_string(),
1,
Duration::from_secs(1),
None,
Some(config_path.clone()),
@@ -803,7 +790,6 @@ mod tests {
let mut tui = MainTui::new(
session.clone(),
"/tmp".to_string(),
1,
Duration::from_secs(1),
None,
Some(config_path),
soulseek-rs/src/ui/main_tui/render.rs
+5 -13
View File
@@ -17,18 +17,6 @@ use ratatui::{
widgets::Paragraph,
};
const COMMAND_BAR_PREFIX: &str = "search: ";
const MESSAGE_BAR_PREFIX: &str = "message (to: recipient text): ";
const BROWSE_BAR_PREFIX: &str = "browse user: ";
const fn command_bar_prefix(mode: CommandBarMode) -> &'static str {
match mode {
CommandBarMode::Search => COMMAND_BAR_PREFIX,
CommandBarMode::Message => MESSAGE_BAR_PREFIX,
CommandBarMode::Browse => BROWSE_BAR_PREFIX,
}
}
impl MainTui {
pub(super) fn render(&mut self, frame: &mut Frame) {
let mut constraints = vec![
@@ -458,7 +446,11 @@ impl MainTui {
}
fn render_command_bar(&self, frame: &mut Frame, area: Rect) {
let prefix = command_bar_prefix(self.state.command_bar_mode);
let prefix = match self.state.command_bar_mode {
CommandBarMode::Search => "search: ",
CommandBarMode::Message => "message (to: recipient text): ",
CommandBarMode::Browse => "browse user: ",
};
let block = pane_block(true);
// Derive from the block's inner rect so borders and padding stay in
// one place; the cursor follows whatever the pane reserves.
soulseek-rs/src/ui/main_tui/search.rs
+1 -5
View File
@@ -102,17 +102,13 @@ impl MainTui {
self.state.results_is_filtering = false;
}
fn recompute_results_filter(&mut self) {
pub(super) fn apply_filter(&mut self) {
let (items, indices) = filter_results(
&self.state.results_items,
&self.state.results_filter_query,
);
self.state.results_filtered_items = items;
self.state.results_filtered_indices = indices;
}
pub(super) fn apply_filter(&mut self) {
self.recompute_results_filter();
if !self.state.results_filtered_items.is_empty() {
self.state.results_table_state.select(Some(0));
}
soulseek-rs/src/ui/panes/chat_pane.rs
+4 -4
View File
@@ -1,13 +1,13 @@
use crate::models::{AppState, MessageDirection};
use crate::ui::{
accent_style, border_style, border_type, dimmed_style, highlight_style,
info_style, primary_style,
accent_style, border_style, dimmed_style, highlight_style, info_style,
primary_style,
};
use ratatui::{
Frame,
layout::{Constraint, Layout, Rect},
text::{Line, Span},
widgets::{Block, Borders, Paragraph},
widgets::{Block, BorderType, Borders, Paragraph},
};
/// Render the private-message chat box: title is the conversation partner,
@@ -28,7 +28,7 @@ pub fn render_chat_pane(
let block = Block::default()
.borders(Borders::ALL)
.border_style(border_style(true))
.border_type(border_type(true))
.border_type(BorderType::Rounded)
.title(title);
let inner = block.inner(area);
frame.render_widget(block, area);
soulseek-rs/src/ui/panes/download_info_pane.rs
+13 -14
View File
@@ -229,7 +229,12 @@ fn build_info_lines(
speed_bytes_per_sec,
} => {
lines.push(Line::from(""));
push_progress_lines(&mut lines, *bytes_downloaded, *total_bytes);
push_transfer_progress(
&mut lines,
"Downloaded",
*bytes_downloaded,
*total_bytes,
);
lines.push(label_value(
"Speed",
&format_speed(*speed_bytes_per_sec),
@@ -246,7 +251,12 @@ fn build_info_lines(
total_bytes,
} => {
lines.push(Line::from(""));
push_progress_lines(&mut lines, *bytes_downloaded, *total_bytes);
push_transfer_progress(
&mut lines,
"Downloaded",
*bytes_downloaded,
*total_bytes,
);
}
DownloadStatus::Failed(reason) => {
if let Some(reason) = reason {
@@ -268,10 +278,7 @@ fn build_info_lines(
}
fn label_value(label: &str, value: &str) -> Line<'static> {
Line::from(vec![
Span::styled(format!("{label:<LABEL_WIDTH$}"), dimmed_style()),
Span::styled(value.to_string(), primary_style()),
])
label_value_styled(label, value.to_string(), primary_style())
}
fn label_value_styled(
@@ -285,14 +292,6 @@ fn label_value_styled(
])
}
fn push_progress_lines(
lines: &mut Vec<Line<'static>>,
bytes_downloaded: u64,
total_bytes: u64,
) {
push_transfer_progress(lines, "Downloaded", bytes_downloaded, total_bytes);
}
/// A progress bar plus a `<label> <done> / <total>` line, shared by the
/// download ("Downloaded") and upload ("Sent") views.
fn push_transfer_progress(
soulseek-rs/src/ui/panes/downloads_pane.rs
+11 -17
View File
@@ -56,7 +56,16 @@ pub fn render_downloads_pane(
let progress_text = match &download.status {
DownloadStatus::Queued => "Queued".to_string(),
DownloadStatus::InProgress { .. } => {
DownloadStatus::InProgress { .. }
| DownloadStatus::Paused { .. } => {
let prefix = if matches!(
download.status,
DownloadStatus::Paused { .. }
) {
"Paused "
} else {
""
};
let percent = if download.size > 0 {
(download.bytes_downloaded() as f64
/ download.size as f64
@@ -65,22 +74,7 @@ pub fn render_downloads_pane(
0
};
format!(
"{}/{} ({}%)",
format_bytes(download.bytes_downloaded()),
format_bytes(download.size),
percent
)
}
DownloadStatus::Paused { .. } => {
let percent = if download.size > 0 {
(download.bytes_downloaded() as f64
/ download.size as f64
* 100.0) as u8
} else {
0
};
format!(
"Paused {}/{} ({}%)",
"{prefix}{}/{} ({}%)",
format_bytes(download.bytes_downloaded()),
format_bytes(download.size),
percent
soulseek-rs/src/ui/styles.rs
+1 -18
View File
@@ -1,5 +1,4 @@
// Reusable styles and colors for consistent UI appearance
#![allow(dead_code)]
use ratatui::style::{Color, Modifier, Style};
use ratatui::text::{Line, Span};
@@ -103,14 +102,6 @@ pub fn unfocused_border_style() -> Style {
Style::default().fg(DUST)
}
pub const fn focused_border_type() -> BorderType {
BorderType::Rounded
}
pub const fn unfocused_border_type() -> BorderType {
BorderType::Rounded
}
pub fn border_style(focused: bool) -> Style {
if focused {
focused_border_style()
@@ -119,14 +110,6 @@ pub fn border_style(focused: bool) -> Style {
}
}
pub const fn border_type(focused: bool) -> BorderType {
if focused {
focused_border_type()
} else {
unfocused_border_type()
}
}
pub fn title_style(focused: bool) -> Style {
if focused {
Style::default().fg(PHOSPHOR).add_modifier(Modifier::BOLD)
@@ -159,7 +142,7 @@ pub fn pane_block(focused: bool) -> Block<'static> {
Block::default()
.borders(Borders::ALL)
.border_style(border_style(focused))
.border_type(border_type(focused))
.border_type(BorderType::Rounded)
.padding(PANE_PADDING)
}
soulseek-rs/tests/cli_e2e.rs
+15 -81
View File
@@ -9,8 +9,9 @@
//! with a notice. `SOULSEEK_E2E_REQUIRED=1` turns a missing server into a
//! failure, which is how CI guarantees these actually run.
#![allow(clippy::doc_markdown)]
mod common;
use common::{CLI_ENV_VARS, free_port, soulfind_binary};
use soulseek_rs::{Client, ClientSettings, PeerAddress};
use std::io::Write;
use std::net::{TcpStream, ToSocketAddrs};
@@ -28,26 +29,6 @@ const EXIT_CONNECTION: i32 = 3;
const EXIT_NO_RESULTS: i32 = 4;
const EXIT_SESSION_LOST: i32 = 7;
/// Every environment variable the CLI reads, cleared for each run so a
/// developer's shell (or a stray `.env`) cannot change what a test observes.
const CLI_ENV_VARS: [&str; 15] = [
"SOULSEEK_CONFIG_DIR",
"SOULSEEK_STATE_DIR",
"SOULSEEK_DAEMON",
"SOULSEEK_DAEMON_TOKEN",
"SOULSEEK_USERNAME",
"SOULSEEK_PASSWORD",
"SOULSEEK_PASSWORD_CMD",
"SOULSEEK_SERVER",
"SOULSEEK_NO_LISTENER",
"SOULSEEK_LISTENER_PORT",
"SOULSEEK_DOWNLOAD_DIR",
"SOULSEEK_SHARED_DIR",
"SOULSEEK_MAX_CONCURRENT_DOWNLOADS",
"SOULSEEK_SEARCH_TIMEOUT",
"SOULSEEK_CONFIG",
];
/// Build a command for the binary, isolated from the developer's environment
/// and from the repository's own working directory.
///
@@ -281,9 +262,7 @@ fn without_a_terminal_the_tui_reports_instead_of_panicking() {
#[test]
fn portmap_emits_a_record_whose_verdict_matches_the_exit_code() {
let output = run(&["--no-config", "--quiet", "--json", "portmap"]);
let record: serde_json::Value =
serde_json::from_str(records(&output).first().expect("one record"))
.expect("portmap should emit valid JSON");
let record = json_record(&output);
assert!(record["port"].is_number());
let ok = record["ok"].as_bool().expect("ok should be a boolean");
@@ -316,10 +295,7 @@ fn quiet_keeps_progress_off_stderr() {
/// What the first record says happened, parsed rather than matched as text.
fn action(output: &Output) -> String {
let record: serde_json::Value =
serde_json::from_str(records(output).first().expect("one record"))
.expect("valid JSON");
record["action"]
json_record(output)["action"]
.as_str()
.expect("a record carries an action")
.to_string()
@@ -828,25 +804,6 @@ impl Drop for TestServer {
}
}
fn soulfind_binary() -> Option<PathBuf> {
if let Ok(path) = std::env::var("SOULFIND_BIN") {
let path = PathBuf::from(path);
return path.exists().then_some(path);
}
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.ancestors()
.map(|dir| dir.join("soulfind/bin/soulfind"))
.find(|candidate| candidate.exists())
}
fn free_port() -> Option<u16> {
std::net::TcpListener::bind("127.0.0.1:0")
.ok()?
.local_addr()
.ok()
.map(|addr| addr.port())
}
fn wait_until_listening(
host: &str,
port: u16,
@@ -1993,9 +1950,7 @@ fn whoami_confirms_the_login_and_reports_what_we_offer() {
);
assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output));
let record: serde_json::Value =
serde_json::from_str(records(&output).first().expect("one record"))
.expect("valid JSON");
let record = json_record(&output);
assert_eq!(record["user"], "cli_e2e_whoami");
assert_eq!(record["shared_files"], 1, "the shared file is indexed");
assert!(record["server"].as_str().is_some_and(|s| s.contains(':')));
@@ -2009,8 +1964,7 @@ fn whoami_reports_how_much_privilege_time_the_account_has() {
let output = cli(&server, "cli_e2e_priv_owner", &["--json", "whoami"]);
assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output));
let record: serde_json::Value =
serde_json::from_str(&records(&output)[0]).expect("valid JSON");
let record = json_record(&output);
// Zero, not null: the server answered, and the answer is "no privileges".
// Null would mean it never said, which is a different fact.
assert_eq!(
@@ -2156,9 +2110,7 @@ fn user_reports_a_peers_status_and_share_counts() {
);
assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output));
let record: serde_json::Value =
serde_json::from_str(records(&output).first().expect("one record"))
.expect("valid JSON");
let record = json_record(&output);
assert_eq!(record["user"], "cli_e2e_subject");
assert_ne!(record["status"], "offline", "the subject is logged in");
assert_eq!(record["shared_files"], 2);
@@ -2292,9 +2244,7 @@ fn shares_status_reports_what_the_network_will_see() {
);
assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output));
let record: serde_json::Value =
serde_json::from_str(records(&output).first().expect("one record"))
.expect("valid JSON");
let record = json_record(&output);
assert_eq!(record["files"], 3);
assert!(record["folders"].as_u64().is_some_and(|n| n >= 1));
}
@@ -2321,9 +2271,7 @@ fn shares_reindex_picks_up_a_file_added_after_startup() {
);
assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output));
let record: serde_json::Value =
serde_json::from_str(records(&output).first().expect("one record"))
.expect("valid JSON");
let record = json_record(&output);
assert_eq!(record["files"], 2, "re-indexing should see both files");
}
@@ -2569,9 +2517,7 @@ fn asking_about_an_unknown_user_succeeds_with_an_offline_record() {
);
assert_eq!(code(&output), EXIT_OK, "stderr: {}", stderr(&output));
let record: serde_json::Value =
serde_json::from_str(records(&output).first().expect("one record"))
.expect("valid JSON");
let record = json_record(&output);
assert_eq!(record["status"], "offline");
assert_eq!(record["shared_files"], 0);
}
@@ -2585,10 +2531,7 @@ fn assert_record_keys(output: &Output, label: &str, keys: &[&str]) {
"`{label}` should succeed, stderr: {}",
stderr(output)
);
let record: serde_json::Value = serde_json::from_str(
records(output).first().expect("at least one record"),
)
.expect("valid JSON");
let record = json_record(output);
for key in keys {
assert!(
record.get(*key).is_some(),
@@ -3071,9 +3014,7 @@ fn json_records_carry_every_extra_field_the_readme_names() {
// `portmap` adds the port it probed; the verdict lives in the exit code,
// which is `EXIT_NO_RESULTS` on a network with no cooperating router.
let portmap = cli(&server, "cli_e2e_keys", &["--json", "portmap"]);
let record: serde_json::Value =
serde_json::from_str(records(&portmap).first().expect("one record"))
.expect("valid JSON");
let record = json_record(&portmap);
for key in ["ok", "backend", "external", "port"] {
assert!(record.get(key).is_some(), "portmap should carry `{key}`");
}
@@ -3347,9 +3288,7 @@ fn a_daemon_reports_its_session_and_stops_on_request() {
let status =
run(&["--no-config", "--json", "--daemon", &at, "daemon", "status"]);
assert_eq!(code(&status), EXIT_OK, "stderr: {}", stderr(&status));
let record: serde_json::Value =
serde_json::from_str(records(&status).first().expect("one record"))
.expect("valid JSON");
let record = json_record(&status);
assert_eq!(record["user"], user);
assert_eq!(record["server"], server.address());
assert!(
@@ -3443,19 +3382,14 @@ fn adding_a_share_takes_effect_on_a_running_daemon() {
// The daemon re-indexes on the way through, so the folder is visible to
// the network now rather than after a restart.
let after: serde_json::Value = serde_json::from_str(
records(&run(&[
let after = json_record(&run(&[
"--no-config",
"--json",
"--daemon",
&at,
"shares",
"status",
]))
.first()
.expect("one record"),
)
.expect("valid JSON");
]));
let listed = after["directories"]
.as_array()
.expect("directories is an array");
soulseek-rs/tests/common/mod.rs
+46
View File
@@ -0,0 +1,46 @@
//! Scaffolding both CLI end-to-end binaries need: the environment they scrub,
//! the soulfind they run against, and a port nobody else holds.
#![allow(dead_code)]
use std::path::{Path, PathBuf};
/// Every environment variable the binary reads, cleared for each run so a
/// developer's shell (or a stray `.env`) cannot change what a test observes —
/// and so no run finds a real daemon.
pub const CLI_ENV_VARS: [&str; 15] = [
"SOULSEEK_CONFIG_DIR",
"SOULSEEK_STATE_DIR",
"SOULSEEK_DAEMON",
"SOULSEEK_DAEMON_TOKEN",
"SOULSEEK_USERNAME",
"SOULSEEK_PASSWORD",
"SOULSEEK_PASSWORD_CMD",
"SOULSEEK_SERVER",
"SOULSEEK_NO_LISTENER",
"SOULSEEK_LISTENER_PORT",
"SOULSEEK_DOWNLOAD_DIR",
"SOULSEEK_SHARED_DIR",
"SOULSEEK_MAX_CONCURRENT_DOWNLOADS",
"SOULSEEK_SEARCH_TIMEOUT",
"SOULSEEK_CONFIG",
];
pub fn soulfind_binary() -> Option<PathBuf> {
if let Ok(path) = std::env::var("SOULFIND_BIN") {
let path = PathBuf::from(path);
return path.exists().then_some(path);
}
Path::new(env!("CARGO_MANIFEST_DIR"))
.ancestors()
.map(|dir| dir.join("soulfind/bin/soulfind"))
.find(|candidate| candidate.exists())
}
pub fn free_port() -> Option<u16> {
std::net::TcpListener::bind("127.0.0.1:0")
.ok()?
.local_addr()
.ok()
.map(|addr| addr.port())
}
soulseek-rs/tests/daemon_protocol_e2e.rs
+24 -81
View File
@@ -11,39 +11,23 @@
//! Like the rest of the e2e suite this needs a soulfind server and skips
//! without one; `SOULSEEK_E2E_REQUIRED=1` turns that skip into a failure.
mod common;
use common::{CLI_ENV_VARS, free_port, soulfind_binary};
use std::io::{BufRead, BufReader, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
use tempfile::TempDir;
const BIN: &str = env!("CARGO_BIN_EXE_soulseek-rs");
/// Every environment variable the binary reads, cleared so a developer's shell
/// cannot change what a test observes — and so no run finds a real daemon.
const CLI_ENV_VARS: [&str; 15] = [
"SOULSEEK_CONFIG_DIR",
"SOULSEEK_STATE_DIR",
"SOULSEEK_DAEMON",
"SOULSEEK_DAEMON_TOKEN",
"SOULSEEK_USERNAME",
"SOULSEEK_PASSWORD",
"SOULSEEK_PASSWORD_CMD",
"SOULSEEK_SERVER",
"SOULSEEK_NO_LISTENER",
"SOULSEEK_LISTENER_PORT",
"SOULSEEK_DOWNLOAD_DIR",
"SOULSEEK_SHARED_DIR",
"SOULSEEK_MAX_CONCURRENT_DOWNLOADS",
"SOULSEEK_SEARCH_TIMEOUT",
"SOULSEEK_CONFIG",
];
/// A daemon bound to loopback TCP as well as its socket, so both transports
/// can be exercised against one session.
struct Daemon {
child: Child,
state: PathBuf,
state: TempDir,
port: u16,
_server: Soulfind,
}
@@ -63,10 +47,10 @@ fn cli_command(state: &Path) -> Command {
impl Daemon {
fn start() -> Option<Self> {
let server = Soulfind::start()?;
let state = scratch("daemon-proto");
let state = tempfile::tempdir().ok()?;
let port = free_port()?;
let mut command = cli_command(&state);
let mut command = cli_command(state.path());
command.args([
"--no-config",
"--server",
@@ -78,7 +62,7 @@ impl Daemon {
"--listener-port",
&free_port()?.to_string(),
"--download-dir",
state.join("downloads").to_str()?,
state.path().join("downloads").to_str()?,
"daemon",
"--bind",
&format!("127.0.0.1:{port}"),
@@ -121,13 +105,13 @@ impl Daemon {
}
fn token(&self) -> String {
std::fs::read_to_string(self.state.join("daemon.token"))
std::fs::read_to_string(self.state.path().join("daemon.token"))
.map(|token| token.trim().to_string())
.unwrap_or_default()
}
fn socket_path(&self) -> PathBuf {
self.state.join("daemon.sock")
self.state.path().join("daemon.sock")
}
fn connect_tcp(&self) -> Option<Connection> {
@@ -152,7 +136,6 @@ impl Drop for Daemon {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
let _ = std::fs::remove_dir_all(&self.state);
}
}
@@ -426,7 +409,7 @@ fn discover_serves_a_contract_whose_parameters_match_what_is_accepted() {
// ---------------------------------------------------------------------------
struct Soulfind {
child: Option<Child>,
child: Child,
db: PathBuf,
port: u16,
}
@@ -438,7 +421,7 @@ impl Soulfind {
let db = std::env::temp_dir().join(format!("soulfind-proto-{port}.db"));
let _ = std::fs::remove_file(&db);
let child = Command::new(bin)
let mut child = Command::new(bin)
.args(["-d", db.to_str()?, "-p", &port.to_string()])
.stdout(Stdio::null())
.stderr(Stdio::null())
@@ -448,15 +431,10 @@ impl Soulfind {
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline {
if TcpStream::connect(("127.0.0.1", port)).is_ok() {
return Some(Self {
child: Some(child),
db,
port,
});
return Some(Self { child, db, port });
}
std::thread::sleep(Duration::from_millis(100));
}
let mut child = child;
let _ = child.kill();
let _ = child.wait();
None
@@ -469,43 +447,12 @@ impl Soulfind {
impl Drop for Soulfind {
fn drop(&mut self) {
if let Some(child) = self.child.as_mut() {
let _ = child.kill();
let _ = child.wait();
}
let _ = self.child.kill();
let _ = self.child.wait();
let _ = std::fs::remove_file(&self.db);
}
}
fn soulfind_binary() -> Option<PathBuf> {
if let Ok(path) = std::env::var("SOULFIND_BIN") {
let path = PathBuf::from(path);
return path.exists().then_some(path);
}
Path::new(env!("CARGO_MANIFEST_DIR"))
.ancestors()
.map(|dir| dir.join("soulfind/bin/soulfind"))
.find(|candidate| candidate.exists())
}
fn free_port() -> Option<u16> {
std::net::TcpListener::bind("127.0.0.1:0")
.ok()?
.local_addr()
.ok()
.map(|addr| addr.port())
}
fn scratch(label: &str) -> PathBuf {
static COUNTER: std::sync::atomic::AtomicU32 =
std::sync::atomic::AtomicU32::new(0);
let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let path = std::env::temp_dir()
.join(format!("soulseek-{label}-{}-{n}", std::process::id()));
std::fs::create_dir_all(&path).expect("scratch directory");
path
}
/// A client configured for a remote daemon reports the *daemon's* world.
///
/// This is the download-box workflow: the token lives in the client's config
@@ -517,13 +464,13 @@ fn scratch(label: &str) -> PathBuf {
#[test]
fn a_config_driven_remote_client_reports_the_daemons_world() {
let daemon = daemon_or_skip!();
let config = scratch("remote-client-config");
let state = scratch("remote-client-state");
let config = tempfile::tempdir().expect("config directory");
let state = tempfile::tempdir().expect("state directory");
// A laptop that knows only where the daemon is. No username, no password,
// no server — exactly what the documented setup produces.
std::fs::write(
config.join("config.toml"),
config.path().join("config.toml"),
format!(
"daemon = \"127.0.0.1:{}\"\ndaemon_token = \"{}\"\n",
daemon.port,
@@ -532,8 +479,8 @@ fn a_config_driven_remote_client_reports_the_daemons_world() {
)
.expect("config file");
let mut command = cli_command(&state);
command.env("SOULSEEK_CONFIG_DIR", &config);
let mut command = cli_command(state.path());
command.env("SOULSEEK_CONFIG_DIR", config.path());
command.args(["--json", "whoami"]);
let output = command.output().expect("the binary should run");
@@ -567,9 +514,6 @@ fn a_config_driven_remote_client_reports_the_daemons_world() {
"the directory the bytes actually land in, on the daemon's host, not \
this machine's Downloads folder: {record}"
);
let _ = std::fs::remove_dir_all(&config);
let _ = std::fs::remove_dir_all(&state);
}
/// Several clients at once, all seeing the same session.
@@ -813,7 +757,8 @@ fn the_download_folder_moves_when_asked_and_status_says_so() {
let mut connection = daemon.connect_socket().expect("socket");
connection.authenticate(None).expect("an answer");
let moved = scratch("moved-downloads").join("deeper");
let downloads = tempfile::tempdir().expect("downloads directory");
let moved = downloads.path().join("deeper");
let reply = connection
.call(&format!(
r#"{{"jsonrpc":"2.0","id":2,"method":"download.set_dir","params":{{"directory":{}}}}}"#,
@@ -844,7 +789,6 @@ fn the_download_folder_moves_when_asked_and_status_says_so() {
)
.expect("an answer");
assert!(refused.contains("-32602"), "got {refused}");
let _ = std::fs::remove_dir_all(moved.parent().expect("parent"));
}
/// A wishlist search must read as *running* to the other windows.
@@ -906,9 +850,9 @@ fn the_upload_slot_limit_can_be_retuned_live() {
#[test]
fn the_cli_reports_a_bad_token_as_a_connection_error() {
let daemon = daemon_or_skip!();
let state = scratch("cli-bad-token");
let state = tempfile::tempdir().expect("state directory");
let run = |token: Option<&str>| {
let mut command = cli_command(&state);
let mut command = cli_command(state.path());
command.args([
"--no-config",
"--daemon",
@@ -951,5 +895,4 @@ fn the_cli_reports_a_bad_token_as_a_connection_error() {
"the real token must keep working: {}",
String::from_utf8_lossy(&right.stderr)
);
let _ = std::fs::remove_dir_all(&state);
}