Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1f0b541bdb | ||
|
|
25fa5884ed | ||
|
|
727e2bc8db | ||
|
|
ae2cc57c30 | ||
|
|
3205611609 | ||
|
|
68bc3067c3 | ||
|
|
ea884535a8 | ||
|
|
9805b807e5 | ||
|
|
5e979a7425 | ||
|
|
e567fa75ed |
@@ -53,6 +53,9 @@ significant_drop_tightening = "allow"
|
||||
# Subjective nursery lint: the suggested map_or/map_or_else rewrites read worse
|
||||
# than the explicit if let/match they replace.
|
||||
option_if_let_else = "allow"
|
||||
# `assert!(x.is_empty())` in tests is fine; the suggested
|
||||
# `assert_eq!(x, [] as [T; 0])` only trades a clear check for a type-cast chore.
|
||||
assert_is_empty = "allow"
|
||||
# The MD5 round functions use the conventional a/b/c/d register names from the
|
||||
# RFC 1321 reference implementation.
|
||||
many_single_char_names = "allow"
|
||||
|
||||
@@ -606,6 +606,7 @@ file, in that order of precedence. Flags work before or after the subcommand.
|
||||
| `--listener-port` | `SOULSEEK_LISTENER_PORT` | `listener_port` | `2234` |
|
||||
| `--no-listener` / `--listener` | `SOULSEEK_NO_LISTENER` | `disable_listener` | listener on |
|
||||
| — | — | `accept_children` | off |
|
||||
| — | — | `use_username_dirs` | off |
|
||||
| `--max-concurrent-downloads` | `SOULSEEK_MAX_CONCURRENT_DOWNLOADS` | `max_concurrent_downloads` | `20` |
|
||||
| `--search-timeout` | `SOULSEEK_SEARCH_TIMEOUT` | `search_timeout` | `10` |
|
||||
| `--daemon ADDR` | `SOULSEEK_DAEMON` | `daemon` | local socket if one is up |
|
||||
@@ -627,10 +628,10 @@ to them. It is off by default — each child costs a socket and a copy of the
|
||||
network's whole search stream — and it needs the listener, since a child has to
|
||||
be able to dial in.
|
||||
|
||||
`config get` and `config set` cover the twelve settings the file holds
|
||||
`config get` and `config set` cover the thirteen settings the file holds
|
||||
(`username`, `server`, `listener_port`, `disable_listener`, `accept_children`,
|
||||
`download_dir`, `shared_dirs`, `max_concurrent_downloads`, `search_timeout`,
|
||||
`password_cmd`, `daemon`, `daemon_token`), and list them back at you when you
|
||||
`download_dir`, `use_username_dirs`, `shared_dirs`, `max_concurrent_downloads`,
|
||||
`search_timeout`, `password_cmd`, `daemon`, `daemon_token`), and list them back at you when you
|
||||
name something
|
||||
else. An empty string clears a key, and `shared_dirs` takes a comma-separated
|
||||
list. Waits expressed in seconds (`--search-timeout`, `--timeout`,
|
||||
|
||||
@@ -209,33 +209,33 @@ impl Client {
|
||||
/// control.
|
||||
pub fn login(&self) -> Result<bool> {
|
||||
info!("Logging in as {}", self.username);
|
||||
if let Some(handle) = &self.server_handle
|
||||
&& !self.cancel.is_cancelled()
|
||||
{
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
let _ = handle.send(ServerMessage::Login {
|
||||
username: self.username.clone(),
|
||||
password: self.password.clone(),
|
||||
version: self.version,
|
||||
response: tx,
|
||||
});
|
||||
let Some(handle) = &self.server_handle else {
|
||||
return Err(SoulseekRs::NotConnected);
|
||||
};
|
||||
if self.cancel.is_cancelled() {
|
||||
return Err(SoulseekRs::NotConnected);
|
||||
}
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
let _ = handle.send(ServerMessage::Login {
|
||||
username: self.username.clone(),
|
||||
password: self.password.clone(),
|
||||
version: self.version,
|
||||
response: tx,
|
||||
});
|
||||
|
||||
let deadline = Instant::now() + LOGIN_RESPONSE_TIMEOUT;
|
||||
loop {
|
||||
match rx.recv_timeout(LOGIN_POLL_INTERVAL) {
|
||||
Ok(result) => return result,
|
||||
Err(RecvTimeoutError::Timeout)
|
||||
if self.cancel.is_cancelled() =>
|
||||
{
|
||||
return Err(SoulseekRs::NotConnected);
|
||||
}
|
||||
Err(RecvTimeoutError::Timeout)
|
||||
if Instant::now() < deadline => {}
|
||||
Err(_) => return Err(SoulseekRs::Timeout),
|
||||
let deadline = Instant::now() + LOGIN_RESPONSE_TIMEOUT;
|
||||
loop {
|
||||
match rx.recv_timeout(LOGIN_POLL_INTERVAL) {
|
||||
Ok(result) => return result,
|
||||
Err(RecvTimeoutError::Timeout)
|
||||
if self.cancel.is_cancelled() =>
|
||||
{
|
||||
return Err(SoulseekRs::NotConnected);
|
||||
}
|
||||
Err(RecvTimeoutError::Timeout) if Instant::now() < deadline => {
|
||||
}
|
||||
Err(_) => return Err(SoulseekRs::Timeout),
|
||||
}
|
||||
} else {
|
||||
Err(SoulseekRs::NotConnected)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
//! of them having to be widened for the move.
|
||||
|
||||
use super::{
|
||||
ActorSystem, Arc, AtomicUsize, BROKER_CONNECT_TIMEOUT,
|
||||
ActorSystem, Arc, AtomicBool, AtomicUsize, BROKER_CONNECT_TIMEOUT,
|
||||
BROWSE_PROTECT_WINDOW, ClientContext, DEFAULT_MAX_PEERS,
|
||||
DEFAULT_UPLOAD_SLOTS, DownloadStatus, DownloadStore, HashMap, HashSet,
|
||||
Instant, PENDING_PEER_TTL, Recommendation, RoomEvent, RoomInfo, RoomTicker,
|
||||
@@ -50,6 +50,7 @@ impl ClientContext {
|
||||
uploads: HashMap::new(),
|
||||
active_uploads: HashMap::new(),
|
||||
pending_serves: HashMap::new(),
|
||||
brokered_uploads: HashMap::new(),
|
||||
browse_results: HashMap::new(),
|
||||
folder_contents: HashMap::new(),
|
||||
peer_infos: HashMap::new(),
|
||||
@@ -707,6 +708,45 @@ impl ClientContext {
|
||||
.map(|(username, _)| username)
|
||||
}
|
||||
|
||||
/// Returns the claim flag the dialling route races against.
|
||||
pub(crate) fn add_brokered_upload(
|
||||
&mut self,
|
||||
connect_token: u32,
|
||||
token: u32,
|
||||
real_path: std::path::PathBuf,
|
||||
) -> Arc<AtomicBool> {
|
||||
let claimed = Arc::new(AtomicBool::new(false));
|
||||
self.brokered_uploads.insert(
|
||||
connect_token,
|
||||
super::BrokeredUpload {
|
||||
token,
|
||||
real_path,
|
||||
deadline: Instant::now() + BROKER_CONNECT_TIMEOUT,
|
||||
claimed: claimed.clone(),
|
||||
},
|
||||
);
|
||||
claimed
|
||||
}
|
||||
|
||||
pub(crate) fn take_brokered_upload(
|
||||
&mut self,
|
||||
connect_token: u32,
|
||||
) -> Option<super::BrokeredUpload> {
|
||||
self.brokered_uploads.remove(&connect_token)
|
||||
}
|
||||
|
||||
/// Uploads whose downloader never connected back in time.
|
||||
pub(crate) fn take_expired_brokered_uploads(
|
||||
&mut self,
|
||||
now: Instant,
|
||||
) -> Vec<super::BrokeredUpload> {
|
||||
let (expired, live) = std::mem::take(&mut self.brokered_uploads)
|
||||
.into_iter()
|
||||
.partition(|(_, upload)| now >= upload.deadline);
|
||||
self.brokered_uploads = live;
|
||||
expired.into_values().collect()
|
||||
}
|
||||
|
||||
pub fn take_expired_connects(&mut self, now: Instant) -> Vec<String> {
|
||||
let mut expired = Vec::new();
|
||||
self.pending_connect_tokens
|
||||
|
||||
@@ -114,6 +114,24 @@ struct UploadJob {
|
||||
offered: Option<Instant>,
|
||||
}
|
||||
|
||||
/// An upload waiting for its downloader to connect back to our listener while
|
||||
/// we also dial them. Whichever socket arrives first claims the transfer.
|
||||
pub(crate) struct BrokeredUpload {
|
||||
pub(crate) token: u32,
|
||||
pub(crate) real_path: std::path::PathBuf,
|
||||
deadline: Instant,
|
||||
claimed: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
impl BrokeredUpload {
|
||||
/// Take the transfer for this route; false if another route already did.
|
||||
pub(crate) fn claim(&self) -> bool {
|
||||
self.claimed
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.is_ok()
|
||||
}
|
||||
}
|
||||
|
||||
/// Live bookkeeping for an upload being served (or recently finished).
|
||||
struct ActiveUpload {
|
||||
username: String,
|
||||
@@ -504,6 +522,10 @@ pub struct ClientContext {
|
||||
active_uploads: HashMap<u32, ActiveUpload>,
|
||||
/// Upload tokens waiting for the downloader's address to be resolved.
|
||||
pending_serves: HashMap<String, Vec<u32>>,
|
||||
/// Uploads whose downloader we could not dial, keyed by the correlation
|
||||
/// token we sent in `ConnectToPeer(F)`; the peer quotes it back in a
|
||||
/// `PierceFirewall` and we serve over that socket instead.
|
||||
brokered_uploads: HashMap<u32, BrokeredUpload>,
|
||||
/// Shared-file listings received from peers we browsed.
|
||||
browse_results: HashMap<String, Vec<SharedDirectory>>,
|
||||
/// One-folder listings received from peers, keyed by peer and folder.
|
||||
|
||||
@@ -151,26 +151,26 @@ impl Client {
|
||||
// its failure is exactly what makes an
|
||||
// unreachable peer fall back to the
|
||||
// server-brokered path.
|
||||
let waiting_serves = if port == 0 {
|
||||
if port == 0 {
|
||||
warn!(
|
||||
"[client] server reports no listening port for {}",
|
||||
username
|
||||
);
|
||||
Vec::new()
|
||||
} else {
|
||||
match client_context.write_safe() {
|
||||
Ok(mut ctx) => {
|
||||
}
|
||||
let waiting_serves = match client_context.write_safe() {
|
||||
Ok(mut ctx) => {
|
||||
if port != 0 {
|
||||
ctx.cache_peer_address(
|
||||
&username,
|
||||
host.clone(),
|
||||
port,
|
||||
);
|
||||
ctx.pending_serves
|
||||
.remove(&username)
|
||||
.unwrap_or_default()
|
||||
}
|
||||
Err(_) => Vec::new(),
|
||||
ctx.pending_serves
|
||||
.remove(&username)
|
||||
.unwrap_or_default()
|
||||
}
|
||||
Err(_) => Vec::new(),
|
||||
};
|
||||
for token in waiting_serves {
|
||||
Self::spawn_serve(
|
||||
@@ -622,16 +622,29 @@ impl Client {
|
||||
// token is the correlation we sent, so it names the
|
||||
// peer: end the wait now instead of letting every
|
||||
// queued download for them sit until the timeout.
|
||||
let username = match client_context.write_safe() {
|
||||
Ok(mut ctx) => ctx.take_pending_connect(token),
|
||||
Err(e) => {
|
||||
error!(
|
||||
"[client] CantConnectToPeer write: {}",
|
||||
e
|
||||
);
|
||||
None
|
||||
}
|
||||
};
|
||||
let (username, upload) =
|
||||
match client_context.write_safe() {
|
||||
Ok(mut ctx) => (
|
||||
ctx.take_pending_connect(token),
|
||||
ctx.take_brokered_upload(token),
|
||||
),
|
||||
Err(e) => {
|
||||
error!(
|
||||
"[client] CantConnectToPeer write: {}",
|
||||
e
|
||||
);
|
||||
(None, None)
|
||||
}
|
||||
};
|
||||
if let Some(upload) = upload
|
||||
&& upload.claim()
|
||||
{
|
||||
Self::fail_upload(
|
||||
&client_context,
|
||||
upload.token,
|
||||
"peer could not connect back",
|
||||
);
|
||||
}
|
||||
if let Some(username) = username {
|
||||
debug!(
|
||||
"[client] {} cannot connect back (token {})",
|
||||
@@ -788,5 +801,16 @@ impl Client {
|
||||
for username in expired {
|
||||
Self::fail_queued_downloads(client_context, &username);
|
||||
}
|
||||
let expired = match client_context.write_safe() {
|
||||
Ok(mut ctx) => ctx.take_expired_brokered_uploads(Instant::now()),
|
||||
Err(_) => return,
|
||||
};
|
||||
for upload in expired.iter().filter(|u| u.claim()) {
|
||||
Self::fail_upload(
|
||||
client_context,
|
||||
upload.token,
|
||||
"peer never connected back",
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
use super::{
|
||||
ActiveUpload, Arc, Client, ClientContext, DownloadStatus, PeerMessage,
|
||||
RwLock, RwLockExt, ServerMessage, collect_failed_tokens, error,
|
||||
next_upload_token, thread,
|
||||
ActiveUpload, Arc, BrokeredUpload, Client, ClientContext, DownloadStatus,
|
||||
PeerMessage, RwLock, RwLockExt, ServerMessage, collect_failed_tokens,
|
||||
debug, error, next_connect_token, next_upload_token, thread,
|
||||
};
|
||||
use crate::message::server::MessageFactory;
|
||||
use crate::peer::{ConnectionType, upload_peer};
|
||||
use crate::types::UploadStatus;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64};
|
||||
use std::net::TcpStream;
|
||||
use std::path::Path;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
|
||||
impl Client {
|
||||
/// Whether the server listed `username` as privileged. Privileged peers are
|
||||
@@ -136,55 +139,190 @@ impl Client {
|
||||
drop(ctx);
|
||||
let own = own_username.to_string();
|
||||
let real_path = job.real_path;
|
||||
let downloader = job.downloader;
|
||||
let context = client_context.clone();
|
||||
thread::spawn(move || {
|
||||
let result = crate::peer::upload_peer::serve_file(
|
||||
&host,
|
||||
port,
|
||||
// Most downloaders sit behind a NAT, so dialling them is a gamble
|
||||
// that costs a 20s timeout when it fails. Like the reference
|
||||
// clients, ask the server to have them connect back at the same
|
||||
// time; the first socket to arrive claims the transfer, and port 0
|
||||
// (the server not knowing where they listen) skips the dial.
|
||||
let Some((connect_token, claimed)) =
|
||||
Self::broker_upload(&context, token, &real_path, &downloader)
|
||||
else {
|
||||
return;
|
||||
};
|
||||
if port == 0 {
|
||||
return;
|
||||
}
|
||||
let stream = match upload_peer::dial(&host, port) {
|
||||
Ok(stream) => stream,
|
||||
Err(e) => {
|
||||
debug!(
|
||||
"[client] cannot dial {downloader} at {host}:{port} \
|
||||
for upload {token} ({e}); waiting for them to connect"
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
if claimed
|
||||
.compare_exchange(
|
||||
false,
|
||||
true,
|
||||
Ordering::AcqRel,
|
||||
Ordering::Acquire,
|
||||
)
|
||||
.is_err()
|
||||
{
|
||||
return;
|
||||
}
|
||||
if let Ok(mut ctx) = context.write_safe() {
|
||||
ctx.take_brokered_upload(connect_token);
|
||||
}
|
||||
let result = upload_peer::serve_direct(
|
||||
stream,
|
||||
&own,
|
||||
token,
|
||||
&real_path,
|
||||
&bytes_sent,
|
||||
&cancel,
|
||||
);
|
||||
let streamed = result.as_ref().ok().copied();
|
||||
let status = match &result {
|
||||
Ok(_) => UploadStatus::Completed,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
|
||||
UploadStatus::Cancelled
|
||||
}
|
||||
Err(e) => {
|
||||
error!("[client] serve {}: {}", real_path.display(), e);
|
||||
UploadStatus::Failed(e.to_string())
|
||||
}
|
||||
};
|
||||
let mut report = None;
|
||||
if let Ok(mut ctx) = context.write_safe()
|
||||
&& let Some(upload) = ctx.active_uploads.get_mut(&token)
|
||||
{
|
||||
let secs = upload.started.elapsed().as_secs_f64();
|
||||
upload.status = status;
|
||||
if let Some(bytes) = streamed
|
||||
&& secs > 0.0
|
||||
{
|
||||
let speed = (bytes as f64 / secs) as u32;
|
||||
ctx.last_upload_speed = speed;
|
||||
report = ctx.server_sender.clone().map(|s| (s, speed));
|
||||
}
|
||||
}
|
||||
// The server folds each finished upload's rate into the average
|
||||
// speed it shows other users for us.
|
||||
if let Some((server, speed)) = report {
|
||||
let _ = server.send(ServerMessage::SendMessage(
|
||||
MessageFactory::build_send_upload_speed(speed),
|
||||
));
|
||||
}
|
||||
// The slot this transfer held is free now, so whoever is next in
|
||||
// line gets it without waiting for another request to arrive.
|
||||
Self::pump_upload_queue(&context);
|
||||
Self::finish_upload(&context, token, &real_path, result);
|
||||
});
|
||||
}
|
||||
|
||||
/// Park the upload under a correlation token and ask the server to have
|
||||
/// `downloader` connect back to us with it. Returns the token and the claim
|
||||
/// flag the dialling route races against; `None` fails the upload.
|
||||
fn broker_upload(
|
||||
context: &Arc<RwLock<ClientContext>>,
|
||||
token: u32,
|
||||
real_path: &Path,
|
||||
downloader: &str,
|
||||
) -> Option<(u32, Arc<AtomicBool>)> {
|
||||
let connect_token = next_connect_token();
|
||||
let (claimed, sender) = match context.write_safe() {
|
||||
Ok(mut ctx) => (
|
||||
ctx.add_brokered_upload(
|
||||
connect_token,
|
||||
token,
|
||||
real_path.to_path_buf(),
|
||||
),
|
||||
ctx.server_sender.clone(),
|
||||
),
|
||||
Err(_) => return None,
|
||||
};
|
||||
let Some(sender) = sender else {
|
||||
Self::fail_upload(context, token, "not connected to the server");
|
||||
return None;
|
||||
};
|
||||
let _ = sender.send(ServerMessage::SendMessage(
|
||||
MessageFactory::build_connect_to_peer(
|
||||
connect_token,
|
||||
downloader,
|
||||
ConnectionType::F,
|
||||
),
|
||||
));
|
||||
Some((connect_token, claimed))
|
||||
}
|
||||
|
||||
/// The downloader of a brokered upload connected back: stream the file
|
||||
/// over its socket on a background thread, unless the dial beat it.
|
||||
pub(crate) fn serve_pierced(
|
||||
client_context: &Arc<RwLock<ClientContext>>,
|
||||
upload: BrokeredUpload,
|
||||
stream: TcpStream,
|
||||
) {
|
||||
if !upload.claim() {
|
||||
return;
|
||||
}
|
||||
let Some((bytes_sent, cancel)) =
|
||||
client_context.write_safe().ok().and_then(|mut ctx| {
|
||||
let active = ctx.active_uploads.get_mut(&upload.token)?;
|
||||
active.started = std::time::Instant::now();
|
||||
Some((active.bytes_sent.clone(), active.cancel.clone()))
|
||||
})
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let context = client_context.clone();
|
||||
thread::spawn(move || {
|
||||
let result = upload_peer::serve_pierced(
|
||||
stream,
|
||||
upload.token,
|
||||
&upload.real_path,
|
||||
&bytes_sent,
|
||||
&cancel,
|
||||
);
|
||||
Self::finish_upload(
|
||||
&context,
|
||||
upload.token,
|
||||
&upload.real_path,
|
||||
result,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/// Mark upload `token` failed and hand its slot to the next in line.
|
||||
pub(crate) fn fail_upload(
|
||||
client_context: &Arc<RwLock<ClientContext>>,
|
||||
token: u32,
|
||||
reason: &str,
|
||||
) {
|
||||
if let Ok(mut ctx) = client_context.write_safe()
|
||||
&& let Some(upload) = ctx.active_uploads.get_mut(&token)
|
||||
{
|
||||
error!("[client] upload {token} to {}: {reason}", upload.username);
|
||||
upload.status = UploadStatus::Failed(reason.to_string());
|
||||
}
|
||||
Self::pump_upload_queue(client_context);
|
||||
}
|
||||
|
||||
/// Record how a served upload ended, report its rate to the server and free
|
||||
/// its slot.
|
||||
fn finish_upload(
|
||||
context: &Arc<RwLock<ClientContext>>,
|
||||
token: u32,
|
||||
real_path: &Path,
|
||||
result: std::io::Result<u64>,
|
||||
) {
|
||||
let streamed = result.as_ref().ok().copied();
|
||||
let status = match &result {
|
||||
Ok(_) => UploadStatus::Completed,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
|
||||
UploadStatus::Cancelled
|
||||
}
|
||||
Err(e) => {
|
||||
error!("[client] serve {}: {}", real_path.display(), e);
|
||||
UploadStatus::Failed(e.to_string())
|
||||
}
|
||||
};
|
||||
let mut report = None;
|
||||
if let Ok(mut ctx) = context.write_safe()
|
||||
&& let Some(upload) = ctx.active_uploads.get_mut(&token)
|
||||
{
|
||||
let secs = upload.started.elapsed().as_secs_f64();
|
||||
upload.status = status;
|
||||
if let Some(bytes) = streamed
|
||||
&& secs > 0.0
|
||||
{
|
||||
let speed = (bytes as f64 / secs) as u32;
|
||||
ctx.last_upload_speed = speed;
|
||||
report = ctx.server_sender.clone().map(|s| (s, speed));
|
||||
}
|
||||
}
|
||||
// The server folds each finished upload's rate into the average
|
||||
// speed it shows other users for us.
|
||||
if let Some((server, speed)) = report {
|
||||
let _ = server.send(ServerMessage::SendMessage(
|
||||
MessageFactory::build_send_upload_speed(speed),
|
||||
));
|
||||
}
|
||||
// The slot this transfer held is free now, so whoever is next in
|
||||
// line gets it without waiting for another request to arrive.
|
||||
Self::pump_upload_queue(context);
|
||||
}
|
||||
|
||||
pub(crate) fn process_failed_uploads(
|
||||
client_context: Arc<RwLock<ClientContext>>,
|
||||
username: &str,
|
||||
|
||||
@@ -261,7 +261,18 @@ fn handle_pierce_firewall(
|
||||
let token = message.read_int32();
|
||||
|
||||
let username = match context.client_context.write_safe() {
|
||||
Ok(mut ctx) => ctx.take_pending_connect(token),
|
||||
Ok(mut ctx) => {
|
||||
if let Some(upload) = ctx.take_brokered_upload(token) {
|
||||
drop(ctx);
|
||||
crate::client::Client::serve_pierced(
|
||||
&context.client_context,
|
||||
upload,
|
||||
stream,
|
||||
);
|
||||
return;
|
||||
}
|
||||
ctx.take_pending_connect(token)
|
||||
}
|
||||
Err(e) => {
|
||||
error!("[listener] pierce firewall lock: {}", e);
|
||||
return;
|
||||
|
||||
@@ -24,6 +24,7 @@ use crate::trace;
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns any I/O error opening the file or talking to the peer.
|
||||
#[cfg(test)]
|
||||
pub fn serve_file(
|
||||
host: &str,
|
||||
port: u32,
|
||||
@@ -33,16 +34,28 @@ pub fn serve_file(
|
||||
bytes_sent: &AtomicU64,
|
||||
cancel: &AtomicBool,
|
||||
) -> io::Result<u64> {
|
||||
let mut file = File::open(path)?;
|
||||
let stream = dial(host, port)?;
|
||||
serve_direct(stream, own_username, token, path, bytes_sent, cancel)
|
||||
}
|
||||
|
||||
/// Connect to `host:port` with the transfer deadlines applied.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns the connect error, or `InvalidInput` for an unresolvable address.
|
||||
pub fn dial(host: &str, port: u32) -> io::Result<TcpStream> {
|
||||
let socket = format!("{host}:{port}")
|
||||
.to_socket_addrs()?
|
||||
.next()
|
||||
.ok_or_else(|| {
|
||||
io::Error::new(io::ErrorKind::InvalidInput, "no address")
|
||||
})?;
|
||||
let mut stream =
|
||||
TcpStream::connect_timeout(&socket, Duration::from_secs(20))?;
|
||||
with_deadlines(TcpStream::connect_timeout(
|
||||
&socket,
|
||||
Duration::from_secs(20),
|
||||
)?)
|
||||
}
|
||||
|
||||
fn with_deadlines(stream: TcpStream) -> io::Result<TcpStream> {
|
||||
// A downloader that connects and then goes silent must cost seconds, not
|
||||
// the rest of the session: without deadlines a wedged `read_exact` (or a
|
||||
// zero-window `write_all`) holds this thread, the socket, the open file,
|
||||
@@ -55,7 +68,23 @@ pub fn serve_file(
|
||||
stream.set_read_timeout(Some(Duration::from_secs(30)))?;
|
||||
stream.set_write_timeout(Some(Duration::from_secs(30)))?;
|
||||
stream.set_nodelay(true).ok();
|
||||
Ok(stream)
|
||||
}
|
||||
|
||||
/// Stream `path` over a connection we dialled: announce with `PeerInit(F)`
|
||||
/// and the token, then hand over to [`stream_file`].
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns any I/O error opening the file or talking to the peer.
|
||||
pub fn serve_direct(
|
||||
mut stream: TcpStream,
|
||||
own_username: &str,
|
||||
token: u32,
|
||||
path: &Path,
|
||||
bytes_sent: &AtomicU64,
|
||||
cancel: &AtomicBool,
|
||||
) -> io::Result<u64> {
|
||||
let file = File::open(path)?;
|
||||
// PeerInit(F) + the 4-byte token in a single write so they coalesce.
|
||||
let mut init = MessageFactory::build_peer_init_message(
|
||||
own_username,
|
||||
@@ -66,7 +95,36 @@ pub fn serve_file(
|
||||
init.extend_from_slice(&token.to_le_bytes());
|
||||
stream.write_all(&init)?;
|
||||
stream.flush()?;
|
||||
stream_file(stream, file, path, bytes_sent, cancel)
|
||||
}
|
||||
|
||||
/// Stream `path` over a connection the downloader opened to us after the
|
||||
/// server brokered it: it has already sent `PierceFirewall`, so only the
|
||||
/// 4-byte transfer token precedes the offset exchange.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns any I/O error opening the file or talking to the peer.
|
||||
pub fn serve_pierced(
|
||||
stream: TcpStream,
|
||||
token: u32,
|
||||
path: &Path,
|
||||
bytes_sent: &AtomicU64,
|
||||
cancel: &AtomicBool,
|
||||
) -> io::Result<u64> {
|
||||
let file = File::open(path)?;
|
||||
let mut stream = with_deadlines(stream)?;
|
||||
stream.write_all(&token.to_le_bytes())?;
|
||||
stream.flush()?;
|
||||
stream_file(stream, file, path, bytes_sent, cancel)
|
||||
}
|
||||
|
||||
fn stream_file(
|
||||
mut stream: TcpStream,
|
||||
mut file: File,
|
||||
path: &Path,
|
||||
bytes_sent: &AtomicU64,
|
||||
cancel: &AtomicBool,
|
||||
) -> io::Result<u64> {
|
||||
// The downloader replies with an 8-byte START_DOWNLOAD offset first. A
|
||||
// non-zero offset means it is resuming and already holds that prefix, so
|
||||
// seek past it and count it as sent to keep progress absolute.
|
||||
@@ -104,7 +162,11 @@ pub fn serve_file(
|
||||
// and a peer that leaves the socket open, or whose FIN a NAT swallows,
|
||||
// would hold it for the whole wait. Nicotine+ finishes the transfer at
|
||||
// delivery for the same reason; the close below is ours to make.
|
||||
trace!("[upload] served {} to {}:{}", path.display(), host, port);
|
||||
trace!(
|
||||
"[upload] served {} to {:?}",
|
||||
path.display(),
|
||||
stream.peer_addr()
|
||||
);
|
||||
Ok(streamed)
|
||||
}
|
||||
|
||||
|
||||
@@ -482,8 +482,8 @@ fn registered_username_can_relogin_with_same_password() {
|
||||
// ---------------------------------------------------------------------------
|
||||
// Peer-to-peer download coverage.
|
||||
//
|
||||
// The client cannot serve files (the upload side is not implemented), so a real
|
||||
// download is exercised with a minimal in-process "mock uploader" that speaks
|
||||
// To exercise the download path on its own, the peer is a minimal in-process
|
||||
// "mock uploader" rather than a second client. It speaks
|
||||
// the peer protocol using the library's own public `Message` wire format. It
|
||||
// drives the same path a real peer would after a search: a `P` control
|
||||
// connection to our listener, a `QueueUpload` → `TransferRequest` →
|
||||
@@ -5692,3 +5692,65 @@ fn a_cancel_ends_a_login_waiting_for_its_verdict_and_nothing_follows_it() {
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_firewalled_downloader_still_receives_the_upload() {
|
||||
// Issue #102: the leecher cannot be dialled (no listener, as behind a NAT),
|
||||
// so the sharer must ask the server to broker the F connection and serve
|
||||
// the file over the leecher's PierceFirewall connection.
|
||||
let server = server_or_skip!();
|
||||
|
||||
let share_dir = unique_download_dir();
|
||||
let content: Vec<u8> = (0..4096u32).map(|i| (i % 251) as u8).collect();
|
||||
let filename = "e2e_probe_fwleech.bin";
|
||||
std::fs::write(share_dir.join(filename), &content).unwrap();
|
||||
|
||||
let mut sharer = Client::with_settings(ClientSettings {
|
||||
shared_directories: vec![share_dir.display().to_string()],
|
||||
..server.listening_settings(
|
||||
"e2e_fw_sharer",
|
||||
"pw",
|
||||
free_port().expect("sharer port"),
|
||||
)
|
||||
});
|
||||
sharer.connect().expect("sharer connect");
|
||||
assert!(sharer.login().expect("sharer login"));
|
||||
|
||||
let mut leecher =
|
||||
Client::with_settings(server.settings("e2e_fw_leecher", "pw"));
|
||||
leecher.connect().expect("leecher connect");
|
||||
assert!(leecher.login().expect("leecher login"));
|
||||
std::thread::sleep(Duration::from_secs(1));
|
||||
|
||||
let query = "fwleech";
|
||||
let _ = leecher.search(query, Duration::from_secs(3));
|
||||
let (result_path, size) = reply_from(&leecher, query, "e2e_fw_sharer")
|
||||
.and_then(|reply| reply.files.into_iter().next())
|
||||
.map(|file| (file.name, file.size))
|
||||
.expect("leecher should find the sharer's file");
|
||||
|
||||
let download_dir = unique_download_dir();
|
||||
let (_download, status_rx) = leecher
|
||||
.download(
|
||||
result_path.clone(),
|
||||
"e2e_fw_sharer".to_string(),
|
||||
size,
|
||||
download_dir.display().to_string(),
|
||||
)
|
||||
.expect("start download");
|
||||
|
||||
let completed =
|
||||
wait_for_completion(&leecher, &status_rx, Duration::from_secs(40));
|
||||
assert!(
|
||||
completed,
|
||||
"the download to a firewalled leecher should complete"
|
||||
);
|
||||
|
||||
let basename = result_path.rsplit(['\\', '/']).next().unwrap();
|
||||
assert_eq!(
|
||||
std::fs::read(download_dir.join(basename)).expect("downloaded file"),
|
||||
content
|
||||
);
|
||||
let _ = std::fs::remove_dir_all(&download_dir);
|
||||
let _ = std::fs::remove_dir_all(&share_dir);
|
||||
}
|
||||
|
||||
@@ -34,6 +34,8 @@ pub struct Ctx {
|
||||
pub out: Out,
|
||||
pub settings: ClientSettings,
|
||||
pub download_dir: String,
|
||||
/// Save each peer's files under `download_dir/<username>`.
|
||||
pub use_username_dirs: bool,
|
||||
pub max_concurrent_downloads: usize,
|
||||
pub search_timeout: Duration,
|
||||
/// Where to find a daemon, when this run should use one rather than open
|
||||
@@ -47,6 +49,16 @@ pub struct Ctx {
|
||||
}
|
||||
|
||||
impl Ctx {
|
||||
/// Where a download from `username` lands.
|
||||
#[must_use]
|
||||
pub fn dir_for(&self, username: &str) -> String {
|
||||
crate::directories::download_dir_for(
|
||||
&self.download_dir,
|
||||
username,
|
||||
self.use_username_dirs,
|
||||
)
|
||||
}
|
||||
|
||||
/// Re-read a "found nothing", "timed out" or "transfer failed" verdict
|
||||
/// against the session that produced it.
|
||||
///
|
||||
|
||||
@@ -458,7 +458,7 @@ fn download_all(
|
||||
client.as_ref(),
|
||||
&ctx.out,
|
||||
&request,
|
||||
&ctx.download_dir,
|
||||
&ctx.dir_for(&request.user),
|
||||
timeout,
|
||||
);
|
||||
if sender.send((request, outcome)).is_err() {
|
||||
|
||||
@@ -202,6 +202,7 @@ pub fn run(
|
||||
browses: Arc::clone(&browses),
|
||||
downloads: Arc::clone(&downloads),
|
||||
download_dir: Mutex::new(ctx.download_dir.clone()),
|
||||
use_username_dirs: ctx.use_username_dirs,
|
||||
server: ctx.settings.server_address.to_string(),
|
||||
token: token.clone(),
|
||||
started: Instant::now(),
|
||||
|
||||
@@ -38,6 +38,7 @@ pub struct Daemon {
|
||||
/// Where transfers land, on *this* machine. A caller may point it at a
|
||||
/// different folder with `download.set_dir`, but never per download.
|
||||
pub download_dir: Mutex<String>,
|
||||
pub use_username_dirs: bool,
|
||||
pub server: String,
|
||||
pub token: String,
|
||||
pub started: Instant,
|
||||
@@ -428,7 +429,11 @@ impl Daemon {
|
||||
// per download would scatter transfers in ways `daemon.status` cannot
|
||||
// report; changing the folder is a deliberate session-wide act,
|
||||
// `download.set_dir`.
|
||||
let directory = self.download_dir();
|
||||
let directory = crate::directories::download_dir_for(
|
||||
&self.download_dir(),
|
||||
¶ms.username,
|
||||
self.use_username_dirs,
|
||||
);
|
||||
let (download, updates) = self
|
||||
.session
|
||||
.download_with_metadata(
|
||||
@@ -800,6 +805,7 @@ mod tests {
|
||||
browses: Arc::new(PendingBrowses::default()),
|
||||
downloads: Arc::new(Mutex::new(Vec::new())),
|
||||
download_dir: Mutex::new("/tmp".to_string()),
|
||||
use_username_dirs: false,
|
||||
server: "server.invalid:2416".to_string(),
|
||||
token: token.to_string(),
|
||||
started: Instant::now(),
|
||||
|
||||
@@ -50,9 +50,54 @@ pub fn resolve_shared_directories(dirs: &[String]) -> Vec<String> {
|
||||
resolved
|
||||
}
|
||||
|
||||
/// The folder a download from `username` lands in: `download_dir` itself, or
|
||||
/// a per-user subfolder of it when `per_user` is set, as SoulseekQt does.
|
||||
#[must_use]
|
||||
pub fn download_dir_for(
|
||||
download_dir: &str,
|
||||
username: &str,
|
||||
per_user: bool,
|
||||
) -> String {
|
||||
if !per_user {
|
||||
return download_dir.to_string();
|
||||
}
|
||||
std::path::Path::new(download_dir)
|
||||
.join(sanitize_path_component(username))
|
||||
.display()
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// A peer-chosen name made safe as one path component: separators and
|
||||
/// characters Windows refuses become `_`, so `../..` cannot climb out.
|
||||
fn sanitize_path_component(name: &str) -> String {
|
||||
let mut out: String = name
|
||||
.chars()
|
||||
.map(|c| match c {
|
||||
'/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_',
|
||||
c if c.is_control() => '_',
|
||||
c => c,
|
||||
})
|
||||
.collect();
|
||||
if out.is_empty() || out.ends_with(['.', ' ']) {
|
||||
out.push('_');
|
||||
}
|
||||
if is_reserved(&out) {
|
||||
out.insert(0, '_');
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Windows device names (`CON`, `COM1`, …) that cannot be a folder name.
|
||||
fn is_reserved(name: &str) -> bool {
|
||||
let stem = name.split('.').next().unwrap_or(name).to_ascii_uppercase();
|
||||
matches!(stem.as_str(), "CON" | "PRN" | "AUX" | "NUL")
|
||||
|| (matches!(stem.get(..3), Some("COM" | "LPT"))
|
||||
&& matches!(stem.as_bytes().get(3..), Some([b'1'..=b'9'])))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::resolve_shared_directory;
|
||||
use super::{download_dir_for, resolve_shared_directory};
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
|
||||
fn unique_temp_dir() -> std::path::PathBuf {
|
||||
@@ -97,4 +142,23 @@ mod tests {
|
||||
assert!(result.is_err());
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
}
|
||||
|
||||
fn dl(name: &str) -> String {
|
||||
std::path::Path::new("/dl").join(name).display().to_string()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn per_user_folder_is_a_child_of_the_download_dir() {
|
||||
assert_eq!(download_dir_for("/dl", "alice", false), "/dl");
|
||||
assert_eq!(download_dir_for("/dl", "alice", true), dl("alice"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_hostile_username_cannot_leave_the_download_dir() {
|
||||
assert_eq!(download_dir_for("/dl", "../..", true), dl(".._.._"));
|
||||
assert_eq!(download_dir_for("/dl", "/tmp/x", true), dl("_tmp_x"));
|
||||
assert_eq!(download_dir_for("/dl", "..", true), dl(".._"));
|
||||
assert_eq!(download_dir_for("/dl", "", true), dl("_"));
|
||||
assert_eq!(download_dir_for("/dl", "com1", true), dl("_com1"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -38,6 +38,7 @@ impl SettingsMode {
|
||||
pub enum SettingsRow {
|
||||
ChangePassword,
|
||||
Logout,
|
||||
UseUsernameDirs,
|
||||
DownloadDir,
|
||||
/// Share path at this index.
|
||||
Share(usize),
|
||||
@@ -79,6 +80,7 @@ pub struct SettingsState {
|
||||
pub account: AccountInfo,
|
||||
pub download_dir: String,
|
||||
pub share_dirs: Vec<String>,
|
||||
pub use_username_dirs: bool,
|
||||
/// Index into [`Self::rows`].
|
||||
pub selected: usize,
|
||||
pub mode: SettingsMode,
|
||||
@@ -94,11 +96,13 @@ impl SettingsState {
|
||||
account: AccountInfo,
|
||||
download_dir: String,
|
||||
share_dirs: Vec<String>,
|
||||
use_username_dirs: bool,
|
||||
) -> Self {
|
||||
Self {
|
||||
account,
|
||||
download_dir,
|
||||
share_dirs,
|
||||
use_username_dirs,
|
||||
selected: 0,
|
||||
mode: SettingsMode::Navigate,
|
||||
input: String::new(),
|
||||
@@ -119,6 +123,7 @@ impl SettingsState {
|
||||
if self.can_log_out() {
|
||||
rows.push(SettingsRow::Logout);
|
||||
}
|
||||
rows.push(SettingsRow::UseUsernameDirs);
|
||||
rows.push(SettingsRow::DownloadDir);
|
||||
rows.extend((0..self.share_dirs.len()).map(SettingsRow::Share));
|
||||
rows
|
||||
@@ -191,6 +196,10 @@ impl SettingsState {
|
||||
self.input.clear();
|
||||
}
|
||||
SettingsRow::Logout => self.mode = SettingsMode::ConfirmingLogout,
|
||||
SettingsRow::UseUsernameDirs => {
|
||||
self.use_username_dirs = !self.use_username_dirs;
|
||||
return SettingsAction::Apply;
|
||||
}
|
||||
SettingsRow::DownloadDir => {
|
||||
self.mode = SettingsMode::EditingDownloadDir;
|
||||
self.input.clone_from(&self.download_dir);
|
||||
@@ -300,6 +309,7 @@ mod tests {
|
||||
account(),
|
||||
"/dl".into(),
|
||||
vec!["/dl".into(), "/music".into()],
|
||||
false,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -493,11 +503,13 @@ mod tests {
|
||||
},
|
||||
"/dl".into(),
|
||||
vec!["/music".into()],
|
||||
false,
|
||||
);
|
||||
assert_eq!(
|
||||
s.rows(),
|
||||
vec![
|
||||
SettingsRow::ChangePassword,
|
||||
SettingsRow::UseUsernameDirs,
|
||||
SettingsRow::DownloadDir,
|
||||
SettingsRow::Share(0),
|
||||
]
|
||||
@@ -505,6 +517,8 @@ mod tests {
|
||||
// The row below "change password" is the download folder, not a
|
||||
// logout the window cannot perform.
|
||||
s.handle_key(key(KeyCode::Down));
|
||||
assert_eq!(s.selected_row(), SettingsRow::UseUsernameDirs);
|
||||
s.handle_key(key(KeyCode::Down));
|
||||
assert_eq!(s.selected_row(), SettingsRow::DownloadDir);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,6 +40,7 @@ pub struct FileConfig {
|
||||
/// A TOML array rather than a `config set` key, because a query may itself
|
||||
/// contain a comma.
|
||||
pub wishlist: Option<Vec<String>>,
|
||||
pub use_username_dirs: Option<bool>,
|
||||
}
|
||||
|
||||
impl FileConfig {
|
||||
@@ -59,6 +60,7 @@ impl FileConfig {
|
||||
"password_cmd",
|
||||
"daemon",
|
||||
"daemon_token",
|
||||
"use_username_dirs",
|
||||
];
|
||||
|
||||
/// The value of `key`, or `None` when it is unset or unknown. Lists are
|
||||
@@ -83,6 +85,9 @@ impl FileConfig {
|
||||
"password_cmd" => self.password_cmd.clone(),
|
||||
"daemon" => self.daemon.clone(),
|
||||
"daemon_token" => self.daemon_token.clone(),
|
||||
"use_username_dirs" => {
|
||||
self.use_username_dirs.map(|v| v.to_string())
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -162,6 +167,17 @@ impl FileConfig {
|
||||
// The single-folder spelling would otherwise shadow the list.
|
||||
self.shared_dir = None;
|
||||
}
|
||||
"use_username_dirs" => {
|
||||
self.use_username_dirs = Some(match value {
|
||||
"true" | "1" | "yes" => true,
|
||||
"false" | "0" | "no" => false,
|
||||
other => {
|
||||
return Err(format!(
|
||||
"use_username_dirs wants true or false, got '{other}'"
|
||||
));
|
||||
}
|
||||
});
|
||||
}
|
||||
other => {
|
||||
return Err(format!(
|
||||
"unknown setting '{other}' — try one of: {}",
|
||||
@@ -277,6 +293,7 @@ pub struct Resolved {
|
||||
pub accept_children: bool,
|
||||
pub download_dir: String,
|
||||
pub shared_dirs: Vec<String>,
|
||||
pub use_username_dirs: bool,
|
||||
pub max_concurrent_downloads: usize,
|
||||
pub search_timeout: u64,
|
||||
pub password_cmd: Option<String>,
|
||||
@@ -330,6 +347,7 @@ pub fn resolve(cli: &crate::cli::Cli, file: &FileConfig) -> Resolved {
|
||||
&& !disable_listener,
|
||||
download_dir: download_dir.clone(),
|
||||
shared_dirs: resolve_shared_dirs(cli, file, &download_dir),
|
||||
use_username_dirs: file.use_username_dirs.unwrap_or(false),
|
||||
max_concurrent_downloads: cli
|
||||
.max_concurrent_downloads
|
||||
.or(file.max_concurrent_downloads)
|
||||
@@ -573,6 +591,7 @@ mod tests {
|
||||
daemon: Some("nas.local:5030".into()),
|
||||
daemon_token: Some("deadbeef".into()),
|
||||
wishlist: None,
|
||||
use_username_dirs: Some(false),
|
||||
};
|
||||
let resolved = resolve(&bare_cli(), &file);
|
||||
assert_eq!(resolved.username.as_deref(), Some("alice"));
|
||||
|
||||
@@ -276,6 +276,7 @@ fn context(
|
||||
version: ClientVersion::REFERENCE_CLIENT,
|
||||
},
|
||||
download_dir,
|
||||
use_username_dirs: resolved.use_username_dirs,
|
||||
max_concurrent_downloads: resolved.max_concurrent_downloads,
|
||||
search_timeout: Duration::from_secs(resolved.search_timeout),
|
||||
daemon,
|
||||
@@ -522,6 +523,7 @@ fn run_default_tui(
|
||||
terminal,
|
||||
Arc::new(outcome.client),
|
||||
resolved.download_dir.clone(),
|
||||
resolved.use_username_dirs,
|
||||
Duration::from_secs(resolved.search_timeout),
|
||||
store,
|
||||
config_path.clone(),
|
||||
@@ -598,6 +600,7 @@ fn run_attached_tui(
|
||||
terminal,
|
||||
Arc::new(session),
|
||||
download_dir,
|
||||
resolved.use_username_dirs,
|
||||
Duration::from_secs(resolved.search_timeout),
|
||||
None,
|
||||
config_path,
|
||||
|
||||
@@ -2,12 +2,96 @@ use super::MainTui;
|
||||
use super::input::{FilterEdit, edit_filter, jumped, list_jump};
|
||||
use crate::models::BrowseStatus;
|
||||
use ratatui::crossterm::event::{KeyCode, KeyEvent, KeyModifiers};
|
||||
use std::{thread, time::Duration};
|
||||
use std::{
|
||||
collections::HashSet,
|
||||
sync::mpsc::{self, Receiver, TryRecvError},
|
||||
thread,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
/// How long to wait for a browse response before showing a timeout notice.
|
||||
const BROWSE_TIMEOUT: Duration = Duration::from_secs(20);
|
||||
|
||||
/// One full peer listing can satisfy several containing-folder requests.
|
||||
pub(super) struct FolderFetch {
|
||||
username: String,
|
||||
folders: Vec<String>,
|
||||
pub(super) requested_at: Instant,
|
||||
request: Option<Receiver<Result<(), String>>>,
|
||||
}
|
||||
|
||||
/// Soulseek paths use remote separators, independently of the local OS.
|
||||
pub(super) fn containing_folder(path: &str) -> Option<&str> {
|
||||
let (parent, name) = path.rsplit_once('\\')?;
|
||||
if parent.trim_matches('\\').is_empty()
|
||||
|| name.is_empty()
|
||||
|| path.contains('/')
|
||||
|| path.chars().any(char::is_control)
|
||||
|| path
|
||||
.trim_start_matches('\\')
|
||||
.split('\\')
|
||||
.any(|part| matches!(part, "" | "." | ".."))
|
||||
{
|
||||
return None;
|
||||
}
|
||||
Some(parent)
|
||||
}
|
||||
|
||||
impl MainTui {
|
||||
pub(super) fn download_containing_folder(&mut self) {
|
||||
let Some(file) = self.highlighted_result() else {
|
||||
return;
|
||||
};
|
||||
let Some(folder) = containing_folder(&file.filename) else {
|
||||
self.folder_download_status =
|
||||
Some("Cannot download folder: invalid remote file path".into());
|
||||
return;
|
||||
};
|
||||
let username = file.username.clone();
|
||||
let folder = folder.to_string();
|
||||
if username.trim().is_empty() {
|
||||
self.folder_download_status =
|
||||
Some("Cannot download folder: missing peer username".into());
|
||||
return;
|
||||
}
|
||||
self.folder_download_status =
|
||||
Some(format!("Fetching folder {folder} from {username}…"));
|
||||
if let Some(fetch) = self
|
||||
.folder_fetches
|
||||
.iter_mut()
|
||||
.find(|fetch| fetch.username == username)
|
||||
{
|
||||
if !fetch.folders.contains(&folder) {
|
||||
fetch.folders.push(folder);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
let (sender, request) = mpsc::channel();
|
||||
let client = self.client.clone();
|
||||
let peer = username.clone();
|
||||
// SessionApi's browse works for both local and daemon sessions. The
|
||||
// request itself can involve IPC, so even that runs off the UI thread.
|
||||
if self.state.browse.tabs.iter().any(|tab| {
|
||||
!tab.own
|
||||
&& tab.username == username
|
||||
&& tab.status == BrowseStatus::Loading
|
||||
}) {
|
||||
let _ = sender.send(Ok(()));
|
||||
} else {
|
||||
thread::spawn(move || {
|
||||
let _ = sender
|
||||
.send(client.browse_user(&peer).map_err(|e| e.to_string()));
|
||||
});
|
||||
}
|
||||
self.folder_fetches.push(FolderFetch {
|
||||
username,
|
||||
folders: vec![folder],
|
||||
requested_at: Instant::now(),
|
||||
request: Some(request),
|
||||
});
|
||||
}
|
||||
|
||||
pub(super) fn handle_browse_input(&mut self, key: KeyEvent) {
|
||||
// Typing goes to the filter; the other keys still move around.
|
||||
if self.state.browse.active_tab().is_some_and(|b| b.filtering)
|
||||
@@ -218,10 +302,24 @@ impl MainTui {
|
||||
if files.is_empty() {
|
||||
return;
|
||||
}
|
||||
self.queue_peer_files(username, files, self.download_dir.clone());
|
||||
}
|
||||
|
||||
/// Shared browse queue: keep remote paths intact and let the session's
|
||||
/// download machinery choose local names and enforce transfer limits.
|
||||
fn queue_peer_files(
|
||||
&mut self,
|
||||
username: String,
|
||||
files: Vec<(String, u64)>,
|
||||
download_dir: String,
|
||||
) -> Receiver<String> {
|
||||
let sender = self.downloads_sender();
|
||||
let client = self.client.clone();
|
||||
let download_dir = self.download_dir.clone();
|
||||
let (status_sender, status_receiver) = mpsc::channel();
|
||||
thread::spawn(move || {
|
||||
let mut queued = 0;
|
||||
let mut failed = 0;
|
||||
let mut last_error = String::new();
|
||||
for (path, size) in files {
|
||||
match client.download(
|
||||
path.clone(),
|
||||
@@ -231,15 +329,27 @@ impl MainTui {
|
||||
) {
|
||||
Ok((download, rx)) => {
|
||||
let _ = sender.send((download, rx));
|
||||
queued += 1;
|
||||
}
|
||||
Err(e) => {
|
||||
failed += 1;
|
||||
last_error = e.to_string();
|
||||
soulseek_rs::warn!(
|
||||
"Failed to queue download {path}: {e}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
let status = if failed == 0 {
|
||||
format!("Queued {queued} files from {username}")
|
||||
} else {
|
||||
format!(
|
||||
"Queued {queued} files from {username}; {failed} failed: {last_error}"
|
||||
)
|
||||
};
|
||||
let _ = status_sender.send(status);
|
||||
});
|
||||
status_receiver
|
||||
}
|
||||
|
||||
/// Request a user's shared files and open (or focus) their browse tab.
|
||||
@@ -250,7 +360,9 @@ impl MainTui {
|
||||
}
|
||||
// Open/focus the tab; only (re)issue the request when it's new or a
|
||||
// previous attempt timed out.
|
||||
if self.state.browse.open(&username) {
|
||||
if self.state.browse.open(&username)
|
||||
&& !self.folder_fetches.iter().any(|f| f.username == username)
|
||||
{
|
||||
let _ = self.client.browse_user(&username);
|
||||
}
|
||||
self.state.show_browse = true;
|
||||
@@ -289,6 +401,7 @@ impl MainTui {
|
||||
|
||||
/// Drain browse responses into any loading tabs, or time them out.
|
||||
pub(super) fn poll_browse_result(&mut self) {
|
||||
self.poll_folder_requests();
|
||||
// Which loading tabs are waiting, and for whom.
|
||||
let loading: Vec<(usize, String, std::time::Instant)> = self
|
||||
.state
|
||||
@@ -300,20 +413,162 @@ impl MainTui {
|
||||
.map(|(i, b)| (i, b.username.clone(), b.requested_at))
|
||||
.collect();
|
||||
|
||||
for (idx, username, requested_at) in loading {
|
||||
// take_browse_result is destructive. Consume once per peer, then
|
||||
// deliver the listing to both browse tabs and folder downloads.
|
||||
let users: HashSet<String> = loading
|
||||
.iter()
|
||||
.map(|(_, user, _)| user.clone())
|
||||
.chain(self.folder_fetches.iter().map(|f| f.username.clone()))
|
||||
.collect();
|
||||
for username in users {
|
||||
if let Some(directories) = self.client.take_browse_result(&username)
|
||||
{
|
||||
if let Some(browse) = self.state.browse.tabs.get_mut(idx) {
|
||||
for (idx, _, _) in
|
||||
loading.iter().filter(|(_, user, _)| *user == username)
|
||||
{
|
||||
let browse = &mut self.state.browse.tabs[*idx];
|
||||
browse.load(&directories);
|
||||
if *idx == self.state.browse.active {
|
||||
self.state.browse_table_state.select(Some(0));
|
||||
}
|
||||
}
|
||||
if idx == self.state.browse.active {
|
||||
self.state.browse_table_state.select(Some(0));
|
||||
if let Some(index) = self
|
||||
.folder_fetches
|
||||
.iter()
|
||||
.position(|f| f.username == username)
|
||||
{
|
||||
let fetch = self.folder_fetches.remove(index);
|
||||
for folder in fetch.folders {
|
||||
self.queue_containing_folder(
|
||||
&username,
|
||||
&folder,
|
||||
&directories,
|
||||
);
|
||||
}
|
||||
}
|
||||
} else if requested_at.elapsed() > BROWSE_TIMEOUT
|
||||
}
|
||||
}
|
||||
for (idx, _, requested_at) in loading {
|
||||
if requested_at.elapsed() > BROWSE_TIMEOUT
|
||||
&& let Some(browse) = self.state.browse.tabs.get_mut(idx)
|
||||
&& browse.status == BrowseStatus::Loading
|
||||
{
|
||||
browse.status = BrowseStatus::TimedOut;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn queue_containing_folder(
|
||||
&mut self,
|
||||
username: &str,
|
||||
folder: &str,
|
||||
directories: &[soulseek_rs::SharedDirectory],
|
||||
) {
|
||||
let files = match files_in_folder(directories, folder) {
|
||||
Ok(files) if !files.is_empty() => files,
|
||||
Ok(_) => {
|
||||
self.folder_download_status =
|
||||
Some(format!("Folder {folder} from {username} is empty"));
|
||||
return;
|
||||
}
|
||||
Err(error) => {
|
||||
self.folder_download_status = Some(format!(
|
||||
"Cannot download {folder} from {username}: {error}"
|
||||
));
|
||||
return;
|
||||
}
|
||||
};
|
||||
self.folder_download_status = Some(format!(
|
||||
"Queuing {} files from {username}: {folder}…",
|
||||
files.len()
|
||||
));
|
||||
let directory = crate::directories::download_dir_for(
|
||||
&self.download_dir,
|
||||
username,
|
||||
self.use_username_dirs,
|
||||
);
|
||||
let result =
|
||||
self.queue_peer_files(username.to_string(), files, directory);
|
||||
self.folder_queue_results.push(result);
|
||||
}
|
||||
|
||||
fn poll_folder_requests(&mut self) {
|
||||
self.folder_queue_results
|
||||
.retain(|result| match result.try_recv() {
|
||||
Ok(status) => {
|
||||
self.folder_download_status = Some(status);
|
||||
false
|
||||
}
|
||||
Err(TryRecvError::Empty) => true,
|
||||
Err(TryRecvError::Disconnected) => {
|
||||
self.folder_download_status =
|
||||
Some("Folder download worker stopped".into());
|
||||
false
|
||||
}
|
||||
});
|
||||
self.folder_fetches.retain_mut(|fetch| {
|
||||
if let Some(request) = &fetch.request {
|
||||
let error = match request.try_recv() {
|
||||
Ok(Ok(())) => {
|
||||
fetch.request = None;
|
||||
None
|
||||
}
|
||||
Ok(Err(error)) => Some(error),
|
||||
Err(TryRecvError::Disconnected) => {
|
||||
Some("request worker stopped".into())
|
||||
}
|
||||
Err(TryRecvError::Empty) => None,
|
||||
};
|
||||
if let Some(error) = error {
|
||||
self.folder_download_status = Some(format!(
|
||||
"Could not fetch folder from {}: {error}",
|
||||
fetch.username
|
||||
));
|
||||
return false;
|
||||
}
|
||||
}
|
||||
if fetch.requested_at.elapsed() > BROWSE_TIMEOUT {
|
||||
self.folder_download_status = Some(format!(
|
||||
"Folder request to {} timed out; press D to retry",
|
||||
fetch.username
|
||||
));
|
||||
return false;
|
||||
}
|
||||
true
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// Only direct children of an exact directory match, never descendants or
|
||||
/// prefix matches. Reject malformed entries rather than queue a partial folder.
|
||||
fn files_in_folder(
|
||||
directories: &[soulseek_rs::SharedDirectory],
|
||||
folder: &str,
|
||||
) -> Result<Vec<(String, u64)>, &'static str> {
|
||||
let mut found = false;
|
||||
let mut seen = HashSet::new();
|
||||
let mut files = Vec::new();
|
||||
for dir in directories
|
||||
.iter()
|
||||
.filter(|dir| dir.name.trim_end_matches('\\') == folder)
|
||||
{
|
||||
found = true;
|
||||
for file in &dir.files {
|
||||
if file.name.is_empty()
|
||||
|| file.name.contains(['\\', '/'])
|
||||
|| file.name.chars().any(char::is_control)
|
||||
|| matches!(file.name.as_str(), "." | "..")
|
||||
{
|
||||
return Err("invalid file name in peer listing");
|
||||
}
|
||||
let path = format!("{folder}\\{}", file.name);
|
||||
if seen.insert(path.clone()) {
|
||||
files.push((path, file.size));
|
||||
}
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return Err("folder not found in peer listing");
|
||||
}
|
||||
Ok(files)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use super::MainTui;
|
||||
use crate::directories::download_dir_for;
|
||||
use crate::models::{DownloadEntry, FileDisplayData};
|
||||
use crate::ui::panes::{InfoSubject, selected_transfer};
|
||||
use soulseek_rs::types::UploadStatus;
|
||||
@@ -197,6 +198,7 @@ impl MainTui {
|
||||
let sender = self.downloads_sender();
|
||||
let client = self.client.clone();
|
||||
let download_dir = self.download_dir.clone();
|
||||
let use_username_dirs = self.use_username_dirs;
|
||||
|
||||
thread::spawn(move || {
|
||||
for file in selected_files {
|
||||
@@ -210,7 +212,11 @@ impl MainTui {
|
||||
file.filename.clone(),
|
||||
file.username.clone(),
|
||||
file.size,
|
||||
download_dir.clone(),
|
||||
download_dir_for(
|
||||
&download_dir,
|
||||
&file.username,
|
||||
use_username_dirs,
|
||||
),
|
||||
metadata,
|
||||
) {
|
||||
Ok((download, rx)) => {
|
||||
|
||||
@@ -484,6 +484,7 @@ impl MainTui {
|
||||
KeyCode::Enter => {
|
||||
self.queue_selected_downloads();
|
||||
}
|
||||
KeyCode::Char('D') => self.download_containing_folder(),
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,6 +23,7 @@ pub struct MainTui {
|
||||
client: Arc<dyn SessionApi>,
|
||||
state: AppState,
|
||||
download_dir: String,
|
||||
use_username_dirs: bool,
|
||||
search_timeout: Duration,
|
||||
spinner_state: usize,
|
||||
store: Option<StateStore>,
|
||||
@@ -34,12 +35,16 @@ pub struct MainTui {
|
||||
/// Which divider the left mouse button is holding, if any.
|
||||
resize_drag: Option<input::ResizeDrag>,
|
||||
listener_fallback: Option<(u16, u16)>,
|
||||
folder_fetches: Vec<browse::FolderFetch>,
|
||||
folder_queue_results: Vec<std::sync::mpsc::Receiver<String>>,
|
||||
folder_download_status: Option<String>,
|
||||
}
|
||||
|
||||
impl MainTui {
|
||||
pub fn new(
|
||||
client: Arc<dyn SessionApi>,
|
||||
download_dir: String,
|
||||
use_username_dirs: bool,
|
||||
search_timeout: Duration,
|
||||
store: Option<StateStore>,
|
||||
config_path: Option<std::path::PathBuf>,
|
||||
@@ -49,12 +54,16 @@ impl MainTui {
|
||||
state: AppState::new(),
|
||||
download_dir,
|
||||
search_timeout,
|
||||
use_username_dirs,
|
||||
spinner_state: 0,
|
||||
store,
|
||||
config_path,
|
||||
saved_snapshot: Snapshot::default(),
|
||||
resize_drag: None,
|
||||
listener_fallback: None,
|
||||
folder_fetches: Vec::new(),
|
||||
folder_queue_results: Vec::new(),
|
||||
folder_download_status: None,
|
||||
};
|
||||
tui.restore_persisted_state();
|
||||
tui
|
||||
@@ -252,17 +261,25 @@ impl MainTui {
|
||||
}
|
||||
}
|
||||
|
||||
#[expect(clippy::too_many_arguments, reason = "one call site per mode")]
|
||||
pub fn launch_main_tui(
|
||||
terminal: DefaultTerminal,
|
||||
client: Arc<dyn SessionApi>,
|
||||
download_dir: String,
|
||||
use_username_dirs: bool,
|
||||
search_timeout: Duration,
|
||||
store: Option<StateStore>,
|
||||
config_path: Option<std::path::PathBuf>,
|
||||
listener_fallback: Option<(u16, u16)>,
|
||||
) -> Result<TuiExit> {
|
||||
let mut tui =
|
||||
MainTui::new(client, download_dir, search_timeout, store, config_path);
|
||||
let mut tui = MainTui::new(
|
||||
client,
|
||||
download_dir,
|
||||
use_username_dirs,
|
||||
search_timeout,
|
||||
store,
|
||||
config_path,
|
||||
);
|
||||
tui.listener_fallback = listener_fallback;
|
||||
tui.run(terminal)
|
||||
}
|
||||
|
||||
@@ -369,6 +369,16 @@ impl MainTui {
|
||||
lines.push(Line::from(""));
|
||||
|
||||
lines.push(section("Folders"));
|
||||
lines.push(entry(
|
||||
selected == SettingsRow::UseUsernameDirs,
|
||||
"Use username directory ",
|
||||
if settings.use_username_dirs {
|
||||
"[x]"
|
||||
} else {
|
||||
"[ ]"
|
||||
}
|
||||
.to_string(),
|
||||
));
|
||||
lines.push(if settings.mode == SettingsMode::EditingDownloadDir {
|
||||
typing("Download folder: ", settings.input.clone())
|
||||
} else {
|
||||
|
||||
@@ -79,6 +79,7 @@ const HELP_RIGHT: &[(&str, &[(&str, &str)])] = &[
|
||||
("Space", "select / deselect"),
|
||||
("a / A", "select all / none"),
|
||||
("Enter", "download the selection"),
|
||||
("D", "download highlighted file's containing folder"),
|
||||
("/", "filter, Enter keeps it, Esc clears"),
|
||||
("b", "browse the owner"),
|
||||
("c", "chat rooms"),
|
||||
@@ -166,6 +167,7 @@ fn settings_shortcuts(
|
||||
SettingsRow::ChangePassword => ("Enter", "change password"),
|
||||
SettingsRow::Logout => ("Enter", "log out"),
|
||||
SettingsRow::DownloadDir => ("Enter", "edit folder"),
|
||||
SettingsRow::UseUsernameDirs => ("Enter", "toggle option"),
|
||||
SettingsRow::Share(_) => ("d", "remove share"),
|
||||
},
|
||||
("a", "add share"),
|
||||
@@ -396,6 +398,7 @@ impl MainTui {
|
||||
FocusedPane::Results => vec![
|
||||
("Space", "select"),
|
||||
("Enter", "download"),
|
||||
("D", "download containing folder"),
|
||||
("b", "browse owner"),
|
||||
("c", "chat"),
|
||||
("/", "filter"),
|
||||
@@ -425,7 +428,12 @@ impl MainTui {
|
||||
// Borders and padding, and the unread badge's column when it shows.
|
||||
let badge = self.unread_badge().width() as u16;
|
||||
let reserved = 4 + if badge > 0 { badge + 1 } else { 0 };
|
||||
pack_shortcuts(&self.shortcuts(), width.saturating_sub(reserved))
|
||||
let mut rows =
|
||||
pack_shortcuts(&self.shortcuts(), width.saturating_sub(reserved));
|
||||
if let Some(status) = &self.folder_download_status {
|
||||
rows.push(Line::styled(status.clone(), info_style()));
|
||||
}
|
||||
rows
|
||||
}
|
||||
|
||||
/// A prominent unread badge for the shortcuts bar. It rides in the
|
||||
|
||||
@@ -20,6 +20,7 @@ impl MainTui {
|
||||
account,
|
||||
self.download_dir.clone(),
|
||||
self.client.shared_directories(),
|
||||
self.use_username_dirs,
|
||||
));
|
||||
}
|
||||
|
||||
@@ -66,6 +67,7 @@ impl MainTui {
|
||||
};
|
||||
let download_dir = settings.download_dir.clone();
|
||||
let share_dirs = settings.share_dirs.clone();
|
||||
let use_username_dirs = settings.use_username_dirs;
|
||||
|
||||
let shares = if self.client.daemon_endpoint().is_some() {
|
||||
share_dirs.clone()
|
||||
@@ -107,6 +109,7 @@ impl MainTui {
|
||||
if dir_applied {
|
||||
self.download_dir.clone_from(&download_dir);
|
||||
}
|
||||
self.use_username_dirs = use_username_dirs;
|
||||
|
||||
// Persist to config.toml so the change survives a restart. What was
|
||||
// refused live is not written either: the file must not promise a
|
||||
@@ -121,6 +124,7 @@ impl MainTui {
|
||||
// load, so leaving it would resurrect a removed folder.
|
||||
config.shared_dir = None;
|
||||
config.shared_dirs = Some(share_dirs);
|
||||
config.use_username_dirs = Some(use_username_dirs);
|
||||
config.save(&path)
|
||||
});
|
||||
if let Err(e) = result {
|
||||
|
||||
@@ -2,6 +2,366 @@
|
||||
|
||||
use super::*;
|
||||
|
||||
fn directory(
|
||||
name: &str,
|
||||
files: &[(&str, u64)],
|
||||
) -> soulseek_rs::SharedDirectory {
|
||||
soulseek_rs::SharedDirectory {
|
||||
name: name.to_string(),
|
||||
files: files
|
||||
.iter()
|
||||
.map(|(name, size)| soulseek_rs::SharedFileEntry {
|
||||
name: (*name).to_string(),
|
||||
size: *size,
|
||||
attributes: Vec::new(),
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
fn folder_window(session: Arc<TalkativeSession>) -> MainTui {
|
||||
let mut tui = attach(session);
|
||||
tui.state.focused_pane = FocusedPane::Results;
|
||||
tui.state.results_items = results(2);
|
||||
tui.state.results_items[0].filename =
|
||||
"Music\\Artist\\Album\\03 - Track.flac".into();
|
||||
tui.state.results_items[1].filename = "Other\\track.mp3".into();
|
||||
tui.state.results_items[1].username = "alice".into();
|
||||
tui.state.results_table_state.select(Some(0));
|
||||
tui
|
||||
}
|
||||
|
||||
fn wait_for(mut ready: impl FnMut() -> bool) {
|
||||
let deadline = std::time::Instant::now() + Duration::from_secs(5);
|
||||
while !ready() {
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"worker did not finish"
|
||||
);
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
}
|
||||
}
|
||||
|
||||
fn finish_folder(tui: &mut MainTui) {
|
||||
wait_for(|| {
|
||||
tui.poll_session();
|
||||
tui.folder_fetches.is_empty() && tui.folder_queue_results.is_empty()
|
||||
});
|
||||
tui.update_downloads();
|
||||
}
|
||||
|
||||
fn answer(
|
||||
session: &TalkativeSession,
|
||||
directories: Vec<soulseek_rs::SharedDirectory>,
|
||||
) {
|
||||
wait_for(|| !session.browse_requests.lock().unwrap().is_empty());
|
||||
session
|
||||
.browse_results
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert("bob".into(), directories);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn containing_folder_uses_soulseek_paths_and_rejects_malformed_paths() {
|
||||
use super::super::browse::containing_folder;
|
||||
for (path, parent) in [
|
||||
(
|
||||
"Music\\Artist\\Album\\03 - Track.flac",
|
||||
"Music\\Artist\\Album",
|
||||
),
|
||||
("@@share\\song.mp3", "@@share"),
|
||||
("\\Music\\Björk\\曲.flac", "\\Music\\Björk"),
|
||||
("C:\\Music\\song.mp3", "C:\\Music"),
|
||||
] {
|
||||
assert_eq!(containing_folder(path), Some(parent), "{path}");
|
||||
}
|
||||
for path in [
|
||||
"",
|
||||
"track.mp3",
|
||||
"\\track.mp3",
|
||||
"Music\\",
|
||||
"Music\\..\\song.mp3",
|
||||
"Music\\.\\song.mp3",
|
||||
"Music\\\\song.mp3",
|
||||
"Music\\bad\0.mp3",
|
||||
"Music/track.mp3",
|
||||
] {
|
||||
assert_eq!(containing_folder(path), None, "{path:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shift_d_downloads_only_highlighted_folder_from_complete_peer_listing() {
|
||||
for per_user in [false, true] {
|
||||
let (release, gate) = std::sync::mpsc::channel();
|
||||
let session = Arc::new(TalkativeSession {
|
||||
accept_downloads: true,
|
||||
browse_gate: Some(std::sync::Mutex::new(gate)),
|
||||
..Default::default()
|
||||
});
|
||||
let mut tui = folder_window(session.clone());
|
||||
tui.use_username_dirs = per_user;
|
||||
tui.state.results_selected_indices.insert(1);
|
||||
// The kept filter maps the highlighted row to the original result.
|
||||
tui.state.results_items.swap(0, 1);
|
||||
tui.state.results_filter_query = "Track.flac".into();
|
||||
tui.apply_filter();
|
||||
tui.state.results_selected_indices.clear();
|
||||
tui.state.results_selected_indices.insert(0);
|
||||
tui.handle_key_event(KeyEvent::new(
|
||||
KeyCode::Char('D'),
|
||||
KeyModifiers::SHIFT,
|
||||
));
|
||||
// The worker is held at the request: navigation and rendering still run.
|
||||
press(&mut tui, KeyCode::Tab);
|
||||
assert_ne!(tui.state.focused_pane, FocusedPane::Results);
|
||||
assert!(screen_of(&mut tui).contains("Fetching folder"));
|
||||
assert!(!tui.state.show_browse);
|
||||
assert!(tui.state.browse.tabs.is_empty());
|
||||
assert!(session.download_requests.lock().unwrap().is_empty());
|
||||
release.send(()).unwrap();
|
||||
answer(
|
||||
&session,
|
||||
vec![
|
||||
directory("Music\\Artist", &[("parent.mp3", 1)]),
|
||||
directory(
|
||||
"Music\\Artist\\Album\\",
|
||||
&[
|
||||
("03 - Track.flac", 30),
|
||||
("01 - Missing.flac", 10),
|
||||
("cover.jpg", 20),
|
||||
],
|
||||
),
|
||||
directory("Music\\Artist\\Album\\Disc 2", &[("child.flac", 2)]),
|
||||
directory(
|
||||
"Music\\Artist\\Album extras",
|
||||
&[("sibling.flac", 3)],
|
||||
),
|
||||
directory("Other", &[("track.mp3", 4)]),
|
||||
],
|
||||
);
|
||||
finish_folder(&mut tui);
|
||||
let downloads = session.download_requests.lock().unwrap();
|
||||
assert_eq!(downloads.len(), 3);
|
||||
for (download, name, size) in [
|
||||
(&downloads[0], "03 - Track.flac", 30),
|
||||
(&downloads[1], "01 - Missing.flac", 10),
|
||||
(&downloads[2], "cover.jpg", 20),
|
||||
] {
|
||||
assert_eq!(download.username, "bob");
|
||||
assert_eq!(
|
||||
download.filename,
|
||||
format!("Music\\Artist\\Album\\{name}")
|
||||
);
|
||||
assert_eq!(download.size, size);
|
||||
assert_eq!(
|
||||
download.download_directory,
|
||||
if per_user { "/tmp/bob" } else { "/tmp" }
|
||||
);
|
||||
}
|
||||
assert_eq!(tui.state.downloads.len(), 3);
|
||||
assert_eq!(tui.state.results_selected_indices, [0].into());
|
||||
assert!(screen_of(&mut tui).contains("Queued 3 files from bob"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn enter_still_downloads_selected_files_and_b_still_opens_the_owner() {
|
||||
let session = Arc::new(TalkativeSession {
|
||||
accept_downloads: true,
|
||||
..Default::default()
|
||||
});
|
||||
let mut tui = folder_window(session.clone());
|
||||
tui.state.results_selected_indices.insert(1);
|
||||
press(&mut tui, KeyCode::Enter);
|
||||
wait_for(|| !session.download_requests.lock().unwrap().is_empty());
|
||||
let requests = session.download_requests.lock().unwrap();
|
||||
assert_eq!(requests.len(), 1);
|
||||
assert_eq!(requests[0].username, "alice");
|
||||
assert_eq!(requests[0].filename, "Other\\track.mp3");
|
||||
assert!(tui.state.results_selected_indices.is_empty());
|
||||
assert!(session.browse_requests.lock().unwrap().is_empty());
|
||||
drop(requests);
|
||||
|
||||
press(&mut tui, KeyCode::Char('b'));
|
||||
assert!(tui.state.show_browse);
|
||||
assert_eq!(tui.state.browse.active_tab().unwrap().username, "bob");
|
||||
assert_eq!(*session.browse_requests.lock().unwrap(), ["bob"]);
|
||||
assert!(tui.folder_fetches.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn d_without_a_highlight_or_outside_results_does_nothing() {
|
||||
let session = Arc::new(TalkativeSession::default());
|
||||
let mut tui = folder_window(session.clone());
|
||||
for selection in [None, Some(99)] {
|
||||
tui.state.results_table_state.select(selection);
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
assert!(tui.folder_fetches.is_empty());
|
||||
assert!(tui.folder_download_status.is_none());
|
||||
}
|
||||
tui.state.results_items.clear();
|
||||
tui.state.results_table_state.select(Some(0));
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
assert!(tui.folder_fetches.is_empty());
|
||||
tui.state.results_items = results(1);
|
||||
for pane in [FocusedPane::Searches, FocusedPane::Downloads] {
|
||||
tui.state.focused_pane = pane;
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
assert!(tui.folder_fetches.is_empty());
|
||||
}
|
||||
assert!(session.browse_requests.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn d_is_text_while_editing_a_filter_and_ctrl_d_still_pages() {
|
||||
let mut tui = results_tui(40);
|
||||
tui.state.results_table_state.select(Some(0));
|
||||
ctrl(&mut tui, 'd');
|
||||
assert!(selected(&tui).unwrap() > 0);
|
||||
press(&mut tui, KeyCode::Char('/'));
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
assert_eq!(tui.state.results_filter_query, "D");
|
||||
assert!(tui.folder_fetches.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failed_requests_and_timeouts_report_errors_and_allow_retry() {
|
||||
let session = Arc::new(TalkativeSession {
|
||||
browse_fails: true,
|
||||
..Default::default()
|
||||
});
|
||||
let mut tui = folder_window(session.clone());
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
finish_folder(&mut tui);
|
||||
assert!(screen_of(&mut tui).contains("Could not fetch folder from bob"));
|
||||
assert!(session.download_requests.lock().unwrap().is_empty());
|
||||
|
||||
let session = Arc::new(TalkativeSession::default());
|
||||
let mut tui = folder_window(session.clone());
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
wait_for(|| !session.browse_requests.lock().unwrap().is_empty());
|
||||
tui.folder_fetches[0].requested_at = std::time::Instant::now()
|
||||
.checked_sub(Duration::from_secs(21))
|
||||
.expect("a time before the browse timeout");
|
||||
finish_folder(&mut tui);
|
||||
assert!(screen_of(&mut tui).contains("timed out"));
|
||||
// A late response must not queue anything after the timeout.
|
||||
answer(
|
||||
&session,
|
||||
vec![directory("Music\\Artist\\Album", &[("late.flac", 1)])],
|
||||
);
|
||||
tui.poll_session();
|
||||
assert!(session.download_requests.lock().unwrap().is_empty());
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
wait_for(|| session.browse_requests.lock().unwrap().len() == 2);
|
||||
assert_eq!(tui.folder_fetches.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_missing_and_malformed_folders_do_not_queue_files() {
|
||||
for (listing, message) in [
|
||||
(vec![directory("Music\\Artist\\Album", &[])], "is empty"),
|
||||
(vec![], "folder not found"),
|
||||
(
|
||||
vec![directory(
|
||||
"Music\\Artist\\Album\\Child",
|
||||
&[("child.flac", 1)],
|
||||
)],
|
||||
"folder not found",
|
||||
),
|
||||
(
|
||||
vec![directory(
|
||||
"Music\\Artist\\Album",
|
||||
&[("valid.flac", 1), ("..\\escape.flac", 2)],
|
||||
)],
|
||||
"invalid file name",
|
||||
),
|
||||
] {
|
||||
let session = Arc::new(TalkativeSession {
|
||||
accept_downloads: true,
|
||||
..Default::default()
|
||||
});
|
||||
let mut tui = folder_window(session.clone());
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
answer(&session, listing);
|
||||
finish_folder(&mut tui);
|
||||
assert!(
|
||||
tui.folder_download_status
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.contains(message)
|
||||
);
|
||||
assert!(session.download_requests.lock().unwrap().is_empty());
|
||||
}
|
||||
let session = Arc::new(TalkativeSession::default());
|
||||
let mut tui = folder_window(session.clone());
|
||||
tui.state.results_items[0].filename = "track.flac".into();
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
assert!(screen_of(&mut tui).contains("invalid remote file path"));
|
||||
assert!(tui.folder_fetches.is_empty());
|
||||
assert!(session.browse_requests.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn download_queue_failures_are_reported() {
|
||||
let session = Arc::new(TalkativeSession::default());
|
||||
let mut tui = folder_window(session.clone());
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
answer(
|
||||
&session,
|
||||
vec![directory(
|
||||
"Music\\Artist\\Album",
|
||||
&[("a.flac", 1), ("b.flac", 2)],
|
||||
)],
|
||||
);
|
||||
finish_folder(&mut tui);
|
||||
assert!(screen_of(&mut tui).contains("Queued 0 files from bob; 2 failed"));
|
||||
assert!(tui.state.downloads.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_listing_serves_both_browse_and_folder_download_without_duplicate_requests()
|
||||
{
|
||||
for browse_first in [false, true] {
|
||||
let session = Arc::new(TalkativeSession {
|
||||
accept_downloads: true,
|
||||
..Default::default()
|
||||
});
|
||||
let mut tui = folder_window(session.clone());
|
||||
if browse_first {
|
||||
press(&mut tui, KeyCode::Char('b'));
|
||||
press(&mut tui, KeyCode::Esc);
|
||||
}
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
press(&mut tui, KeyCode::Char('D'));
|
||||
assert_eq!(tui.folder_fetches.len(), 1);
|
||||
press(&mut tui, KeyCode::Char('b'));
|
||||
answer(
|
||||
&session,
|
||||
vec![directory("Music\\Artist\\Album", &[("a.flac", 1)])],
|
||||
);
|
||||
finish_folder(&mut tui);
|
||||
assert_eq!(*session.browse_requests.lock().unwrap(), ["bob"]);
|
||||
assert_eq!(
|
||||
tui.state.browse.active_tab().unwrap().status,
|
||||
crate::models::BrowseStatus::Loaded
|
||||
);
|
||||
assert_eq!(session.download_requests.lock().unwrap().len(), 1);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn help_documents_the_containing_folder_key() {
|
||||
let mut tui = results_tui(1);
|
||||
press(&mut tui, KeyCode::Char('?'));
|
||||
let screen = screen_sized(&mut tui, 180, 60);
|
||||
let row =
|
||||
row_with(&screen, "download highlighted file's containing folder");
|
||||
assert!(row.contains('D'));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shift_b_shows_what_this_session_shares_and_r_re_indexes_it() {
|
||||
let session = Arc::new(TalkativeSession::default());
|
||||
|
||||
@@ -51,6 +51,14 @@ struct TalkativeSession {
|
||||
download_cancelled: std::sync::Mutex<Vec<(String, String)>>,
|
||||
upload_cancelled: std::sync::Mutex<Vec<(String, String)>>,
|
||||
download_removed: std::sync::Mutex<Vec<(String, String)>>,
|
||||
browse_requests: std::sync::Mutex<Vec<String>>,
|
||||
browse_results: std::sync::Mutex<
|
||||
std::collections::HashMap<String, Vec<soulseek_rs::SharedDirectory>>,
|
||||
>,
|
||||
browse_fails: bool,
|
||||
browse_gate: Option<std::sync::Mutex<std::sync::mpsc::Receiver<()>>>,
|
||||
accept_downloads: bool,
|
||||
download_requests: std::sync::Mutex<Vec<soulseek_rs::types::Download>>,
|
||||
}
|
||||
|
||||
fn queued(username: &str, filename: &str) -> soulseek_rs::types::Download {
|
||||
@@ -156,28 +164,44 @@ impl SessionApi for TalkativeSession {
|
||||
}
|
||||
fn download(
|
||||
&self,
|
||||
_filename: String,
|
||||
_username: String,
|
||||
_size: u64,
|
||||
_directory: String,
|
||||
filename: String,
|
||||
username: String,
|
||||
size: u64,
|
||||
directory: String,
|
||||
) -> soulseek_rs::Result<(
|
||||
soulseek_rs::types::Download,
|
||||
std::sync::mpsc::Receiver<soulseek_rs::DownloadStatus>,
|
||||
)> {
|
||||
Err(soulseek_rs::SoulseekRs::NotConnected)
|
||||
let mut download = queued(&username, &filename);
|
||||
download.size = size;
|
||||
download.download_directory = directory;
|
||||
self.download_requests
|
||||
.lock()
|
||||
.expect("not poisoned")
|
||||
.push(download.clone());
|
||||
if self.accept_downloads {
|
||||
Ok((download, std::sync::mpsc::channel().1))
|
||||
} else {
|
||||
Err(soulseek_rs::SoulseekRs::NotConnected)
|
||||
}
|
||||
}
|
||||
fn download_with_metadata(
|
||||
&self,
|
||||
_filename: String,
|
||||
_username: String,
|
||||
_size: u64,
|
||||
_directory: String,
|
||||
_metadata: soulseek_rs::types::DownloadMetadata,
|
||||
filename: String,
|
||||
username: String,
|
||||
size: u64,
|
||||
directory: String,
|
||||
metadata: soulseek_rs::types::DownloadMetadata,
|
||||
) -> soulseek_rs::Result<(
|
||||
soulseek_rs::types::Download,
|
||||
std::sync::mpsc::Receiver<soulseek_rs::DownloadStatus>,
|
||||
)> {
|
||||
Err(soulseek_rs::SoulseekRs::NotConnected)
|
||||
self.download(filename, username, size, directory).map(
|
||||
|(mut download, receiver)| {
|
||||
download.metadata = metadata;
|
||||
(download, receiver)
|
||||
},
|
||||
)
|
||||
}
|
||||
fn get_all_downloads(&self) -> Vec<soulseek_rs::types::Download> {
|
||||
self.downloads.clone()
|
||||
@@ -263,14 +287,31 @@ impl SessionApi for TalkativeSession {
|
||||
fn take_private_messages(&self) -> Vec<soulseek_rs::UserMessage> {
|
||||
Vec::new()
|
||||
}
|
||||
fn browse_user(&self, _username: &str) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
fn browse_user(&self, username: &str) -> soulseek_rs::Result<()> {
|
||||
self.browse_requests
|
||||
.lock()
|
||||
.expect("not poisoned")
|
||||
.push(username.to_string());
|
||||
if let Some(gate) = &self.browse_gate {
|
||||
gate.lock()
|
||||
.expect("not poisoned")
|
||||
.recv_timeout(Duration::from_secs(5))
|
||||
.expect("release browse request");
|
||||
}
|
||||
if self.browse_fails {
|
||||
Err(soulseek_rs::SoulseekRs::NotConnected)
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
fn take_browse_result(
|
||||
&self,
|
||||
_username: &str,
|
||||
username: &str,
|
||||
) -> Option<Vec<soulseek_rs::SharedDirectory>> {
|
||||
None
|
||||
self.browse_results
|
||||
.lock()
|
||||
.expect("not poisoned")
|
||||
.remove(username)
|
||||
}
|
||||
fn request_user_info(&self, _username: &str) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
@@ -349,6 +390,7 @@ fn attach(session: Arc<TalkativeSession>) -> MainTui {
|
||||
MainTui::new(
|
||||
session,
|
||||
"/tmp".to_string(),
|
||||
false,
|
||||
Duration::from_secs(1),
|
||||
None,
|
||||
None,
|
||||
|
||||
@@ -17,6 +17,7 @@ fn applying_settings_attached_pushes_both_folders_to_the_daemon() {
|
||||
let mut tui = MainTui::new(
|
||||
session.clone(),
|
||||
"/daemon/old".to_string(),
|
||||
false,
|
||||
Duration::from_secs(1),
|
||||
None,
|
||||
Some(config_path.clone()),
|
||||
@@ -73,6 +74,7 @@ fn applying_settings_locally_stays_on_this_machine() {
|
||||
let mut tui = MainTui::new(
|
||||
session.clone(),
|
||||
"/tmp".to_string(),
|
||||
false,
|
||||
Duration::from_secs(1),
|
||||
None,
|
||||
Some(config_path),
|
||||
@@ -106,6 +108,30 @@ fn applying_settings_locally_stays_on_this_machine() {
|
||||
assert_eq!(tui.download_dir, downloads_dir);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn toggling_username_folders_applies_live_and_persists() {
|
||||
let session = Arc::new(TalkativeSession::default());
|
||||
let scratch = tempfile::tempdir().expect("a temp dir");
|
||||
let config_path = scratch.path().join("config.toml");
|
||||
let mut tui = MainTui::new(
|
||||
session,
|
||||
scratch.path().to_str().expect("utf-8").to_string(),
|
||||
false,
|
||||
Duration::from_secs(1),
|
||||
None,
|
||||
Some(config_path.clone()),
|
||||
);
|
||||
|
||||
tui.open_settings();
|
||||
tui.state.settings.as_mut().expect("open").use_username_dirs = true;
|
||||
tui.apply_settings();
|
||||
|
||||
assert!(tui.use_username_dirs, "the next download uses it");
|
||||
let saved = crate::persist::config::FileConfig::load(&config_path)
|
||||
.expect("config saved");
|
||||
assert_eq!(saved.use_username_dirs, Some(true));
|
||||
}
|
||||
|
||||
/// The account block answers "who is this session, and what is it
|
||||
/// offering" without a trip to another screen.
|
||||
#[test]
|
||||
@@ -190,6 +216,7 @@ fn an_attached_window_offers_no_logout() {
|
||||
let mut tui = attach(session);
|
||||
press(&mut tui, KeyCode::Char('o'));
|
||||
press(&mut tui, KeyCode::Down);
|
||||
press(&mut tui, KeyCode::Down);
|
||||
press(&mut tui, KeyCode::Enter);
|
||||
|
||||
let settings = tui.state.settings.as_ref().expect("still open");
|
||||
|
||||
@@ -56,6 +56,7 @@ impl Window {
|
||||
let tui = MainTui::new(
|
||||
client,
|
||||
downloads.display().to_string(),
|
||||
false,
|
||||
SEARCH_WINDOW,
|
||||
Some(StateStore::new(state.to_path_buf())),
|
||||
None,
|
||||
|
||||
@@ -129,6 +129,11 @@ const SETTINGS: readonly Setting[] = [
|
||||
</>
|
||||
),
|
||||
},
|
||||
{
|
||||
key: 'use_username_dirs',
|
||||
fallback: 'false',
|
||||
body: <>Save each peer's files in a subfolder named after them.</>,
|
||||
},
|
||||
{
|
||||
key: 'max_concurrent_downloads',
|
||||
fallback: '20',
|
||||
|
||||
Reference in New Issue
Block a user