Merge branch 'develop': listener transfer-token and download token-race fixes
This commit is contained in:
@@ -57,6 +57,15 @@ implementation for anyone building their own client, and **soulseek-rs**, the
|
|||||||
client built on it. The library stays lean on dependencies and has none today;
|
client built on it. The library stays lean on dependencies and has none today;
|
||||||
the client takes them freely.
|
the client takes them freely.
|
||||||
|
|
||||||
|
### Projects using soulseek-rs-lib
|
||||||
|
|
||||||
|
- [seakarr](https://github.com/binhex/seakarr) — automated Soulseek music
|
||||||
|
downloader with library quality upgrading.
|
||||||
|
|
||||||
|
Built something on the library? Open a PR to add it here, and reserve a client
|
||||||
|
minor version in [#12](https://github.com/michel/soulseek-rs/issues/12) so your
|
||||||
|
project is identifiable on the network.
|
||||||
|
|
||||||
## Installation
|
## Installation
|
||||||
|
|
||||||
Homebrew ships a prebuilt binary, so no Rust toolchain:
|
Homebrew ships a prebuilt binary, so no Rust toolchain:
|
||||||
|
|||||||
@@ -171,8 +171,9 @@ pub struct ClientSettings {
|
|||||||
/// Empty means nothing is shared.
|
/// Empty means nothing is shared.
|
||||||
pub shared_directories: Vec<String>,
|
pub shared_directories: Vec<String>,
|
||||||
/// The version reported to the server on login. Defaults to the
|
/// The version reported to the server on login. Defaults to the
|
||||||
/// soulseek-rs major version with minor version 1; clients and bots built
|
/// soulseek-rs major version with minor version 1 ("unidentified");
|
||||||
/// on this library should pick their own minor version.
|
/// clients and bots built on this library should reserve their own minor
|
||||||
|
/// version (see [`ClientVersion`]).
|
||||||
pub version: ClientVersion,
|
pub version: ClientVersion,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
use crate::message::{Message, handlers::Handlers};
|
use crate::message::{Message, handlers::Handlers};
|
||||||
use std::sync::mpsc::Sender;
|
use std::sync::mpsc::Sender;
|
||||||
|
|
||||||
use crate::warn;
|
use crate::debug;
|
||||||
|
|
||||||
pub struct MessageDispatcher<Op> {
|
pub struct MessageDispatcher<Op> {
|
||||||
owner_name: String,
|
owner_name: String,
|
||||||
@@ -30,7 +30,7 @@ impl<Op> MessageDispatcher<Op> {
|
|||||||
message.set_pointer(8);
|
message.set_pointer(8);
|
||||||
handler.handle(message, self.sender.clone());
|
handler.handle(message, self.sender.clone());
|
||||||
} else {
|
} else {
|
||||||
warn!(
|
debug!(
|
||||||
"[{}:dispatcher] No handler found for message code: {code}",
|
"[{}:dispatcher] No handler found for message code: {code}",
|
||||||
self.owner_name
|
self.owner_name
|
||||||
);
|
);
|
||||||
|
|||||||
@@ -142,8 +142,8 @@ impl DownloadStore {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Returns the tokens of downloads matching `username` (and optionally a
|
/// Returns the tokens of queued downloads matching `username` (and optionally
|
||||||
/// `filename`) after notifying their senders of `Failed`.
|
/// a `filename`) after notifying their senders of `Failed`.
|
||||||
///
|
///
|
||||||
/// Caller is responsible for then calling `update_status` and `remove` for
|
/// Caller is responsible for then calling `update_status` and `remove` for
|
||||||
/// each token, typically under a write lock.
|
/// each token, typically under a write lock.
|
||||||
@@ -159,6 +159,7 @@ pub fn collect_failed_tokens(
|
|||||||
.filter(|d| {
|
.filter(|d| {
|
||||||
d.username == username && filename.is_none_or(|f| d.filename == *f)
|
d.username == username && filename.is_none_or(|f| d.filename == *f)
|
||||||
})
|
})
|
||||||
|
.filter(|d| matches!(d.status, DownloadStatus::Queued))
|
||||||
.map(|d| {
|
.map(|d| {
|
||||||
let _ = d.sender.send(DownloadStatus::Failed(Some(
|
let _ = d.sender.send(DownloadStatus::Failed(Some(
|
||||||
"The upload failed on the other side".to_string(),
|
"The upload failed on the other side".to_string(),
|
||||||
@@ -305,6 +306,61 @@ mod tests {
|
|||||||
assert!(!store.remove_by_file("peer", "song.mp3"), "idempotent");
|
assert!(!store.remove_by_file("peer", "song.mp3"), "idempotent");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn collect_failed_tokens_collects_only_queued_downloads() {
|
||||||
|
let mut store = DownloadStore::new();
|
||||||
|
let (tx_active, rx_active) = mpsc::channel();
|
||||||
|
let (tx_paused, rx_paused) = mpsc::channel();
|
||||||
|
let (tx_completed, rx_completed) = mpsc::channel();
|
||||||
|
let (tx_queued, _rx_queued) = mpsc::channel();
|
||||||
|
|
||||||
|
let mut active = make_download(
|
||||||
|
1,
|
||||||
|
DownloadStatus::InProgress {
|
||||||
|
bytes_downloaded: 25,
|
||||||
|
total_bytes: 100,
|
||||||
|
speed_bytes_per_sec: 10.0,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
active.sender = tx_active;
|
||||||
|
|
||||||
|
let mut paused = make_download(
|
||||||
|
2,
|
||||||
|
DownloadStatus::Paused {
|
||||||
|
bytes_downloaded: 25,
|
||||||
|
total_bytes: 100,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
paused.sender = tx_paused;
|
||||||
|
|
||||||
|
let mut completed = make_download(3, DownloadStatus::Completed);
|
||||||
|
completed.sender = tx_completed;
|
||||||
|
|
||||||
|
let mut queued = make_download(4, DownloadStatus::Queued);
|
||||||
|
queued.sender = tx_queued;
|
||||||
|
|
||||||
|
store.add(active);
|
||||||
|
store.add(paused);
|
||||||
|
store.add(completed);
|
||||||
|
store.add(queued);
|
||||||
|
|
||||||
|
let tokens = collect_failed_tokens(&store, "peer", None);
|
||||||
|
|
||||||
|
assert_eq!(tokens, vec![4], "only the queued download may fail");
|
||||||
|
assert!(
|
||||||
|
rx_active.try_recv().is_err(),
|
||||||
|
"an in-progress transfer must not be notified of failure"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
rx_paused.try_recv().is_err(),
|
||||||
|
"a paused transfer must not be notified of failure"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
rx_completed.try_recv().is_err(),
|
||||||
|
"a completed download must not be notified of failure"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn collect_failed_tokens_notifies_and_lists_matching() {
|
fn collect_failed_tokens_notifies_and_lists_matching() {
|
||||||
let mut store = DownloadStore::new();
|
let mut store = DownloadStore::new();
|
||||||
|
|||||||
@@ -26,6 +26,8 @@ const PEER_INIT_MESSAGE_CODE: u8 = 1;
|
|||||||
/// How long an accepted peer gets to send its peer-init handshake.
|
/// How long an accepted peer gets to send its peer-init handshake.
|
||||||
const PEER_INIT_TIMEOUT: Duration = Duration::from_secs(10);
|
const PEER_INIT_TIMEOUT: Duration = Duration::from_secs(10);
|
||||||
|
|
||||||
|
const TOKEN_REGISTRATION_TIMEOUT: Duration = Duration::from_secs(10);
|
||||||
|
|
||||||
const MAX_HANDSHAKES: usize = 128;
|
const MAX_HANDSHAKES: usize = 128;
|
||||||
|
|
||||||
const MAX_PEER_INIT_BYTES: usize = 4096;
|
const MAX_PEER_INIT_BYTES: usize = 4096;
|
||||||
@@ -96,41 +98,72 @@ fn parse_peer_init_message(mut message: Message) -> Option<PeerInitData> {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
fn extract_download_from_buffer(
|
fn wait_for_transfer_token(
|
||||||
|
stream: &mut TcpStream,
|
||||||
reader: &mut MessageReader,
|
reader: &mut MessageReader,
|
||||||
client_context: &Arc<RwLock<ClientContext>>,
|
deadline: Instant,
|
||||||
username: &str,
|
) -> io::Result<()> {
|
||||||
peer_ip: &str,
|
while reader.buffer_len() < 4 {
|
||||||
peer_port: u16,
|
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||||
) -> Option<Download> {
|
if remaining.is_zero() {
|
||||||
if reader.buffer_len() == 0 {
|
return Err(io::Error::new(
|
||||||
return None;
|
io::ErrorKind::TimedOut,
|
||||||
|
"transfer token deadline passed",
|
||||||
|
));
|
||||||
}
|
}
|
||||||
let buffer = reader.get_buffer();
|
stream.set_read_timeout(Some(remaining))?;
|
||||||
let token = u32::from_le_bytes(buffer.get(0..4)?.try_into().ok()?);
|
reader.read_from_socket(stream)?;
|
||||||
trace!(
|
}
|
||||||
"[listener:{}] got transfer_token: {} from data chunk",
|
stream.set_read_timeout(None)?;
|
||||||
username, token
|
Ok(())
|
||||||
);
|
}
|
||||||
|
|
||||||
let context = match client_context.read_safe() {
|
fn take_transfer_token(reader: &mut MessageReader) -> Option<u32> {
|
||||||
Ok(c) => c,
|
let buffer = reader.get_buffer();
|
||||||
|
Some(u32::from_le_bytes(buffer.get(0..4)?.try_into().ok()?))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn wait_for_download_registration(
|
||||||
|
client_context: &Arc<RwLock<ClientContext>>,
|
||||||
|
token: u32,
|
||||||
|
username: &str,
|
||||||
|
deadline: Instant,
|
||||||
|
) -> Option<Download> {
|
||||||
|
loop {
|
||||||
|
let download = match client_context.read_safe() {
|
||||||
|
Ok(context) => context
|
||||||
|
.get_download_by_token(token)
|
||||||
|
.filter(|d| {
|
||||||
|
d.username == username
|
||||||
|
&& !matches!(
|
||||||
|
d.status,
|
||||||
|
DownloadStatus::InProgress { .. }
|
||||||
|
| DownloadStatus::Paused { .. }
|
||||||
|
| DownloadStatus::Completed
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.cloned(),
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("[listener] client context lock: {}", e);
|
error!("[listener] client context lock: {}", e);
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
let download = context.get_download_by_token(token).cloned();
|
if download.is_some() {
|
||||||
|
return download;
|
||||||
if download.is_none() {
|
}
|
||||||
let download_tokens = context.get_download_tokens();
|
if Instant::now() >= deadline {
|
||||||
|
if let Ok(context) = client_context.read_safe() {
|
||||||
trace!(
|
trace!(
|
||||||
"[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}",
|
"[listener:{}] no claimable download for token {}, known tokens: {:?}",
|
||||||
token, download_tokens
|
username,
|
||||||
|
token,
|
||||||
|
context.get_download_tokens()
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
return None;
|
||||||
download
|
}
|
||||||
|
thread::sleep(Duration::from_millis(50));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn handle_peer_connection(
|
fn handle_peer_connection(
|
||||||
@@ -161,25 +194,35 @@ fn handle_peer_connection(
|
|||||||
|
|
||||||
fn handle_file_connection(
|
fn handle_file_connection(
|
||||||
peer: Peer,
|
peer: Peer,
|
||||||
stream: TcpStream,
|
mut stream: TcpStream,
|
||||||
mut reader: MessageReader,
|
mut reader: MessageReader,
|
||||||
token: u32,
|
token: u32,
|
||||||
context: &ConnectionContext,
|
context: &ConnectionContext,
|
||||||
peer_ip: &str,
|
deadline: Instant,
|
||||||
peer_port: u16,
|
|
||||||
) {
|
) {
|
||||||
trace!(
|
trace!(
|
||||||
"[client] DownloadFromPeer token: {} peer: {:?}",
|
"[client] DownloadFromPeer token: {} peer: {:?}",
|
||||||
token, peer
|
token, peer
|
||||||
);
|
);
|
||||||
|
|
||||||
let download = extract_download_from_buffer(
|
if let Err(e) = wait_for_transfer_token(&mut stream, &mut reader, deadline)
|
||||||
&mut reader,
|
{
|
||||||
&context.client_context,
|
debug!(
|
||||||
&peer.username,
|
"[listener:{}:{}] no transfer token arrived: {e}",
|
||||||
peer_ip,
|
peer.host, peer.port
|
||||||
peer_port,
|
|
||||||
);
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
let download =
|
||||||
|
take_transfer_token(&mut reader).and_then(|transfer_token| {
|
||||||
|
wait_for_download_registration(
|
||||||
|
&context.client_context,
|
||||||
|
transfer_token,
|
||||||
|
&peer.username,
|
||||||
|
Instant::now() + TOKEN_REGISTRATION_TIMEOUT,
|
||||||
|
)
|
||||||
|
});
|
||||||
let failure_token = download.as_ref().map(|d| d.token);
|
let failure_token = download.as_ref().map(|d| d.token);
|
||||||
|
|
||||||
let download_peer = DownloadPeer::new(
|
let download_peer = DownloadPeer::new(
|
||||||
@@ -368,8 +411,7 @@ fn handle_incoming_connection(
|
|||||||
reader,
|
reader,
|
||||||
init_data.token,
|
init_data.token,
|
||||||
&context,
|
&context,
|
||||||
&peer_ip,
|
deadline,
|
||||||
peer_port,
|
|
||||||
),
|
),
|
||||||
ConnectionType::D => {
|
ConnectionType::D => {
|
||||||
debug!(
|
debug!(
|
||||||
@@ -447,8 +489,116 @@ impl Listen {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::types::DownloadMetadata;
|
||||||
use std::io::Write;
|
use std::io::Write;
|
||||||
use std::net::TcpListener;
|
use std::net::TcpListener;
|
||||||
|
use std::sync::mpsc;
|
||||||
|
|
||||||
|
fn registered_download(token: u32) -> Download {
|
||||||
|
Download {
|
||||||
|
username: "peer".to_string(),
|
||||||
|
filename: "song.mp3".to_string(),
|
||||||
|
token,
|
||||||
|
size: 100,
|
||||||
|
download_directory: std::env::temp_dir().display().to_string(),
|
||||||
|
status: DownloadStatus::Queued,
|
||||||
|
sender: mpsc::channel().0,
|
||||||
|
queue_position: None,
|
||||||
|
metadata: DownloadMetadata::default(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_download_registered_during_the_wait_is_found() {
|
||||||
|
let client_context = Arc::new(RwLock::new(ClientContext::new()));
|
||||||
|
let registering = {
|
||||||
|
let client_context = client_context.clone();
|
||||||
|
std::thread::spawn(move || {
|
||||||
|
std::thread::sleep(Duration::from_millis(100));
|
||||||
|
client_context
|
||||||
|
.write()
|
||||||
|
.unwrap()
|
||||||
|
.add_download(registered_download(7));
|
||||||
|
})
|
||||||
|
};
|
||||||
|
|
||||||
|
let download = wait_for_download_registration(
|
||||||
|
&client_context,
|
||||||
|
7,
|
||||||
|
"peer",
|
||||||
|
Instant::now() + Duration::from_secs(5),
|
||||||
|
);
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
download.is_some(),
|
||||||
|
"a token registered during the wait must be found"
|
||||||
|
);
|
||||||
|
let _ = registering.join();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn an_unregistered_token_gives_up_at_the_deadline() {
|
||||||
|
let client_context = Arc::new(RwLock::new(ClientContext::new()));
|
||||||
|
|
||||||
|
let started = Instant::now();
|
||||||
|
let download = wait_for_download_registration(
|
||||||
|
&client_context,
|
||||||
|
7,
|
||||||
|
"peer",
|
||||||
|
started + Duration::from_millis(200),
|
||||||
|
);
|
||||||
|
|
||||||
|
assert!(download.is_none(), "an unknown token must not be invented");
|
||||||
|
assert!(
|
||||||
|
started.elapsed() < Duration::from_secs(2),
|
||||||
|
"the deadline must bound the wait"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_download_for_another_peer_is_not_claimed() {
|
||||||
|
let client_context = Arc::new(RwLock::new(ClientContext::new()));
|
||||||
|
client_context
|
||||||
|
.write()
|
||||||
|
.unwrap()
|
||||||
|
.add_download(registered_download(7));
|
||||||
|
|
||||||
|
let download = wait_for_download_registration(
|
||||||
|
&client_context,
|
||||||
|
7,
|
||||||
|
"impostor",
|
||||||
|
Instant::now() + Duration::from_millis(200),
|
||||||
|
);
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
download.is_none(),
|
||||||
|
"a token must only match the peer it was registered for"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn an_already_claimed_download_is_not_reclaimed() {
|
||||||
|
let client_context = Arc::new(RwLock::new(ClientContext::new()));
|
||||||
|
let mut claimed = registered_download(7);
|
||||||
|
claimed.status = DownloadStatus::InProgress {
|
||||||
|
bytes_downloaded: 25,
|
||||||
|
total_bytes: 100,
|
||||||
|
speed_bytes_per_sec: 10.0,
|
||||||
|
};
|
||||||
|
client_context.write().unwrap().add_download(claimed);
|
||||||
|
|
||||||
|
let download = wait_for_download_registration(
|
||||||
|
&client_context,
|
||||||
|
7,
|
||||||
|
"peer",
|
||||||
|
Instant::now() + Duration::from_millis(200),
|
||||||
|
);
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
download.is_none(),
|
||||||
|
"a transfer already in progress must not be claimed again"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn a_dribbling_handshake_hits_the_total_deadline() {
|
fn a_dribbling_handshake_hits_the_total_deadline() {
|
||||||
@@ -482,6 +632,60 @@ mod tests {
|
|||||||
let _ = writer.join();
|
let _ = writer.join();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn the_transfer_token_is_waited_for_across_segments() {
|
||||||
|
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||||
|
let addr = listener.local_addr().unwrap();
|
||||||
|
let writer = std::thread::spawn(move || {
|
||||||
|
let mut socket = TcpStream::connect(addr).unwrap();
|
||||||
|
std::thread::sleep(Duration::from_millis(100));
|
||||||
|
socket.write_all(&42u32.to_le_bytes()).unwrap();
|
||||||
|
std::thread::sleep(Duration::from_millis(200));
|
||||||
|
});
|
||||||
|
let (mut stream, _) = listener.accept().unwrap();
|
||||||
|
let mut reader = MessageReader::new();
|
||||||
|
|
||||||
|
let result = wait_for_transfer_token(
|
||||||
|
&mut stream,
|
||||||
|
&mut reader,
|
||||||
|
Instant::now() + Duration::from_secs(5),
|
||||||
|
);
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
result.is_ok(),
|
||||||
|
"token in a later segment must be waited for"
|
||||||
|
);
|
||||||
|
assert!(reader.buffer_len() >= 4, "token bytes must be buffered");
|
||||||
|
let _ = writer.join();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn waiting_for_a_transfer_token_stops_at_the_deadline() {
|
||||||
|
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||||
|
let addr = listener.local_addr().unwrap();
|
||||||
|
let writer = std::thread::spawn(move || {
|
||||||
|
let socket = TcpStream::connect(addr).unwrap();
|
||||||
|
std::thread::sleep(Duration::from_millis(500));
|
||||||
|
drop(socket);
|
||||||
|
});
|
||||||
|
let (mut stream, _) = listener.accept().unwrap();
|
||||||
|
let mut reader = MessageReader::new();
|
||||||
|
|
||||||
|
let started = Instant::now();
|
||||||
|
let result = wait_for_transfer_token(
|
||||||
|
&mut stream,
|
||||||
|
&mut reader,
|
||||||
|
started + Duration::from_millis(200),
|
||||||
|
);
|
||||||
|
|
||||||
|
assert!(result.is_err(), "a silent peer must not park the thread");
|
||||||
|
assert!(
|
||||||
|
started.elapsed() < Duration::from_secs(2),
|
||||||
|
"the deadline must bound the wait"
|
||||||
|
);
|
||||||
|
let _ = writer.join();
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn an_oversized_handshake_is_rejected() {
|
fn an_oversized_handshake_is_rejected() {
|
||||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||||
|
|||||||
@@ -15,6 +15,18 @@ pub struct File {
|
|||||||
/// Major versions identify client projects on the Soulseek network and are
|
/// Major versions identify client projects on the Soulseek network and are
|
||||||
/// reserved per project for the project's lifetime:
|
/// reserved per project for the project's lifetime:
|
||||||
/// <https://nicotine-plus.org/doc/SLSKPROTOCOL.html#major-versions>.
|
/// <https://nicotine-plus.org/doc/SLSKPROTOCOL.html#major-versions>.
|
||||||
|
///
|
||||||
|
/// Within the soulseek-rs major, minor versions identify the individual
|
||||||
|
/// clients built on this library. Reserved minors:
|
||||||
|
///
|
||||||
|
/// | Minor | Client |
|
||||||
|
/// |-------|-----------------------------------------------|
|
||||||
|
/// | 1 | unidentified (library default) |
|
||||||
|
/// | 2 | soulseek-rs reference client (CLI/TUI) |
|
||||||
|
///
|
||||||
|
/// Building on this library? Keep [`ClientVersion::MAJOR`] and pick an
|
||||||
|
/// unused minor for your client by opening an issue:
|
||||||
|
/// <https://github.com/michel/soulseek-rs/issues/12>.
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub struct ClientVersion {
|
pub struct ClientVersion {
|
||||||
pub major: u32,
|
pub major: u32,
|
||||||
@@ -23,14 +35,14 @@ pub struct ClientVersion {
|
|||||||
|
|
||||||
impl ClientVersion {
|
impl ClientVersion {
|
||||||
/// The major version reserved for the soulseek-rs project. Clients and
|
/// The major version reserved for the soulseek-rs project. Clients and
|
||||||
/// bots built on this library keep this major and pick their own non-zero
|
/// bots built on this library keep this major and pick their own
|
||||||
/// minor version.
|
/// reserved minor version (see the table on [`ClientVersion`]).
|
||||||
pub const MAJOR: u32 = 176;
|
pub const MAJOR: u32 = 176;
|
||||||
|
|
||||||
/// The version the soulseek-rs reference client (CLI/TUI) logs in with.
|
/// The version the soulseek-rs reference client (CLI/TUI) logs in with.
|
||||||
pub const REFERENCE_CLIENT: Self = Self {
|
pub const REFERENCE_CLIENT: Self = Self {
|
||||||
major: Self::MAJOR,
|
major: Self::MAJOR,
|
||||||
minor: 100,
|
minor: 2,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -478,6 +478,7 @@ struct MockUpload {
|
|||||||
content: Vec<u8>,
|
content: Vec<u8>,
|
||||||
token: u32,
|
token: u32,
|
||||||
ready: Sender<()>,
|
ready: Sender<()>,
|
||||||
|
token_delay: Duration,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Build a `PeerInit` (peer code 1) frame: `[len][1][username][conn_type][token]`.
|
/// Build a `PeerInit` (peer code 1) frame: `[len][1][username][conn_type][token]`.
|
||||||
@@ -519,36 +520,43 @@ fn connect_retry(addr: &str, timeout: Duration) -> std::io::Result<TcpStream> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<u64> {
|
fn open_control_and_await_queue(
|
||||||
// 1. P (control) connection: register ourselves with the downloader.
|
cfg: &MockUpload,
|
||||||
|
) -> std::io::Result<TcpStream> {
|
||||||
let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?;
|
let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?;
|
||||||
p.set_read_timeout(Some(Duration::from_secs(10)))?;
|
p.set_read_timeout(Some(Duration::from_secs(10)))?;
|
||||||
p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?;
|
p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?;
|
||||||
p.flush()?;
|
p.flush()?;
|
||||||
let _ = cfg.ready.send(());
|
let _ = cfg.ready.send(());
|
||||||
|
|
||||||
// 2. Wait for the downloader's QueueUpload (peer code 43).
|
|
||||||
loop {
|
loop {
|
||||||
let mut msg = read_framed(&mut p)?;
|
let msg = read_framed(&mut p)?;
|
||||||
if msg.get_message_code() == 43 {
|
if msg.get_message_code() == 43 {
|
||||||
msg.set_pointer(8);
|
|
||||||
let _requested = msg.read_string();
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Ok(p)
|
||||||
|
}
|
||||||
|
|
||||||
// 3. Offer the transfer with a TransferRequest (peer code 40). The size here
|
/// Build a TransferRequest (peer code 40). The size becomes the download's
|
||||||
// becomes the download's expected size, so it must match the content.
|
/// expected size, so it must match the content.
|
||||||
|
fn transfer_request(cfg: &MockUpload) -> Message {
|
||||||
let mut tr = Message::new();
|
let mut tr = Message::new();
|
||||||
tr.write_int32(40)
|
tr.write_int32(40)
|
||||||
.write_int32(1) // direction: upload
|
.write_int32(1) // direction: upload
|
||||||
.write_int32(cfg.token)
|
.write_int32(cfg.token)
|
||||||
.write_string(&cfg.filename)
|
.write_string(&cfg.filename)
|
||||||
.write_int64(cfg.content.len() as u64);
|
.write_int64(cfg.content.len() as u64);
|
||||||
p.write_all(&tr.get_buffer())?;
|
tr
|
||||||
|
}
|
||||||
|
|
||||||
|
fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<u64> {
|
||||||
|
let mut p = open_control_and_await_queue(cfg)?;
|
||||||
|
|
||||||
|
p.write_all(&transfer_request(cfg).get_buffer())?;
|
||||||
p.flush()?;
|
p.flush()?;
|
||||||
|
|
||||||
// 4. Wait for the downloader to allow it (TransferResponse, peer code 41).
|
// Wait for the downloader to allow it (TransferResponse, peer code 41).
|
||||||
loop {
|
loop {
|
||||||
let msg = read_framed(&mut p)?;
|
let msg = read_framed(&mut p)?;
|
||||||
if msg.get_message_code() == 41 {
|
if msg.get_message_code() == 41 {
|
||||||
@@ -556,31 +564,47 @@ fn run_mock_uploader(cfg: &MockUpload) -> std::io::Result<u64> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 5. Open the F (file) connection to the downloader's listener and stream.
|
|
||||||
serve_file_over_f(
|
serve_file_over_f(
|
||||||
&cfg.listen_addr,
|
&cfg.listen_addr,
|
||||||
&cfg.peer_username,
|
&cfg.peer_username,
|
||||||
cfg.token,
|
cfg.token,
|
||||||
&cfg.content,
|
&cfg.content,
|
||||||
|
cfg.token_delay,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn run_mock_uploader_f_first(cfg: &MockUpload) -> std::io::Result<u64> {
|
||||||
|
let mut p = open_control_and_await_queue(cfg)?;
|
||||||
|
|
||||||
|
let listen_addr = cfg.listen_addr.clone();
|
||||||
|
let peer_username = cfg.peer_username.clone();
|
||||||
|
let token = cfg.token;
|
||||||
|
let content = cfg.content.clone();
|
||||||
|
let token_delay = cfg.token_delay;
|
||||||
|
let f = std::thread::spawn(move || {
|
||||||
|
serve_file_over_f(
|
||||||
|
&listen_addr,
|
||||||
|
&peer_username,
|
||||||
|
token,
|
||||||
|
&content,
|
||||||
|
token_delay,
|
||||||
|
)
|
||||||
|
});
|
||||||
|
|
||||||
|
std::thread::sleep(Duration::from_millis(300));
|
||||||
|
|
||||||
|
p.write_all(&transfer_request(cfg).get_buffer())?;
|
||||||
|
p.flush()?;
|
||||||
|
|
||||||
|
f.join()
|
||||||
|
.map_err(|_| std::io::Error::other("F connection thread panicked"))?
|
||||||
|
}
|
||||||
|
|
||||||
fn run_refusing_uploader(
|
fn run_refusing_uploader(
|
||||||
cfg: &MockUpload,
|
cfg: &MockUpload,
|
||||||
refusal: &Message,
|
refusal: &Message,
|
||||||
) -> std::io::Result<()> {
|
) -> std::io::Result<()> {
|
||||||
let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?;
|
let mut p = open_control_and_await_queue(cfg)?;
|
||||||
p.set_read_timeout(Some(Duration::from_secs(10)))?;
|
|
||||||
p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?;
|
|
||||||
p.flush()?;
|
|
||||||
let _ = cfg.ready.send(());
|
|
||||||
|
|
||||||
loop {
|
|
||||||
let msg = read_framed(&mut p)?;
|
|
||||||
if msg.get_message_code() == 43 {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
p.write_all(&refusal.get_buffer())?;
|
p.write_all(&refusal.get_buffer())?;
|
||||||
p.flush()?;
|
p.flush()?;
|
||||||
@@ -589,10 +613,8 @@ fn run_refusing_uploader(
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Open an F (file transfer) connection to `downloader_addr` and stream
|
/// Open an F (file transfer) connection to `downloader_addr` and stream
|
||||||
/// `content`. PeerInit(F) and the 4-byte transfer token are written together so
|
/// `content`; the downloader sends an 8-byte START_DOWNLOAD offset before we
|
||||||
/// the token lands in the listener's read buffer, where the download is looked
|
/// send the bytes.
|
||||||
/// up by token; the downloader then sends an 8-byte START_DOWNLOAD offset before
|
|
||||||
/// we send the bytes.
|
|
||||||
///
|
///
|
||||||
/// Like a real peer, only the bytes past that offset are sent. Returns the
|
/// Like a real peer, only the bytes past that offset are sent. Returns the
|
||||||
/// offset the downloader asked for.
|
/// offset the downloader asked for.
|
||||||
@@ -601,12 +623,19 @@ fn serve_file_over_f(
|
|||||||
username: &str,
|
username: &str,
|
||||||
token: u32,
|
token: u32,
|
||||||
content: &[u8],
|
content: &[u8],
|
||||||
|
token_delay: Duration,
|
||||||
) -> std::io::Result<u64> {
|
) -> std::io::Result<u64> {
|
||||||
let mut f = connect_retry(downloader_addr, Duration::from_secs(5))?;
|
let mut f = connect_retry(downloader_addr, Duration::from_secs(5))?;
|
||||||
f.set_read_timeout(Some(Duration::from_secs(10)))?;
|
f.set_read_timeout(Some(Duration::from_secs(10)))?;
|
||||||
|
if token_delay.is_zero() {
|
||||||
let mut init = peer_init_bytes(username, "F", token);
|
let mut init = peer_init_bytes(username, "F", token);
|
||||||
init.extend_from_slice(&token.to_le_bytes());
|
init.extend_from_slice(&token.to_le_bytes());
|
||||||
f.write_all(&init)?;
|
f.write_all(&init)?;
|
||||||
|
} else {
|
||||||
|
f.write_all(&peer_init_bytes(username, "F", token))?;
|
||||||
|
std::thread::sleep(token_delay);
|
||||||
|
f.write_all(&token.to_le_bytes())?;
|
||||||
|
}
|
||||||
f.flush()?;
|
f.flush()?;
|
||||||
|
|
||||||
let mut start = [0u8; 8];
|
let mut start = [0u8; 8];
|
||||||
@@ -717,14 +746,17 @@ fn unique_download_dir() -> PathBuf {
|
|||||||
dir
|
dir
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
fn file_downloads_over_p_and_f(
|
||||||
fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
|
peer_username: &str,
|
||||||
|
token_delay: Duration,
|
||||||
|
run_uploader: fn(&MockUpload) -> std::io::Result<u64>,
|
||||||
|
) {
|
||||||
let server = server_or_skip!();
|
let server = server_or_skip!();
|
||||||
|
|
||||||
// Downloader: connected to the server with its peer listener enabled.
|
// Downloader: connected to the server with its peer listener enabled.
|
||||||
let listen_port = free_port().expect("free listen port");
|
let listen_port = free_port().expect("free listen port");
|
||||||
let mut client = Client::with_settings(server.listening_settings(
|
let mut client = Client::with_settings(server.listening_settings(
|
||||||
"e2e_downloader",
|
&format!("{peer_username}_dl"),
|
||||||
"pw",
|
"pw",
|
||||||
listen_port,
|
listen_port,
|
||||||
));
|
));
|
||||||
@@ -741,14 +773,15 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
|
|||||||
let (ready_tx, ready_rx) = std::sync::mpsc::channel();
|
let (ready_tx, ready_rx) = std::sync::mpsc::channel();
|
||||||
let cfg = MockUpload {
|
let cfg = MockUpload {
|
||||||
listen_addr: format!("127.0.0.1:{listen_port}"),
|
listen_addr: format!("127.0.0.1:{listen_port}"),
|
||||||
peer_username: "e2e_mockpeer".to_string(),
|
peer_username: peer_username.to_string(),
|
||||||
filename: filename.to_string(),
|
filename: filename.to_string(),
|
||||||
content: content.clone(),
|
content: content.clone(),
|
||||||
token,
|
token,
|
||||||
ready: ready_tx,
|
ready: ready_tx,
|
||||||
|
token_delay,
|
||||||
};
|
};
|
||||||
let uploader = std::thread::spawn(move || {
|
let uploader = std::thread::spawn(move || {
|
||||||
if let Err(e) = run_mock_uploader(&cfg) {
|
if let Err(e) = run_uploader(&cfg) {
|
||||||
eprintln!("[mock uploader] {e}");
|
eprintln!("[mock uploader] {e}");
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
@@ -764,7 +797,7 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
|
|||||||
let (_download, status_rx) = client
|
let (_download, status_rx) = client
|
||||||
.download(
|
.download(
|
||||||
filename.to_string(),
|
filename.to_string(),
|
||||||
"e2e_mockpeer".to_string(),
|
peer_username.to_string(),
|
||||||
size,
|
size,
|
||||||
download_dir.display().to_string(),
|
download_dir.display().to_string(),
|
||||||
)
|
)
|
||||||
@@ -783,6 +816,33 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
|
|||||||
let _ = std::fs::remove_dir_all(&download_dir);
|
let _ = std::fs::remove_dir_all(&download_dir);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
|
||||||
|
file_downloads_over_p_and_f(
|
||||||
|
"e2e_mockpeer",
|
||||||
|
Duration::ZERO,
|
||||||
|
run_mock_uploader,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_file_downloads_when_the_transfer_token_arrives_late() {
|
||||||
|
file_downloads_over_p_and_f(
|
||||||
|
"e2e_latepeer",
|
||||||
|
Duration::from_millis(300),
|
||||||
|
run_mock_uploader,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_download_completes_when_the_f_connection_beats_token_registration() {
|
||||||
|
file_downloads_over_p_and_f(
|
||||||
|
"e2e_earlypeer",
|
||||||
|
Duration::ZERO,
|
||||||
|
run_mock_uploader_f_first,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
fn refusal_fails_the_download_quickly(
|
fn refusal_fails_the_download_quickly(
|
||||||
peer_username: &str,
|
peer_username: &str,
|
||||||
build_refusal: impl FnOnce(&str) -> Message,
|
build_refusal: impl FnOnce(&str) -> Message,
|
||||||
@@ -808,6 +868,7 @@ fn refusal_fails_the_download_quickly(
|
|||||||
content: Vec::new(),
|
content: Vec::new(),
|
||||||
token: 0,
|
token: 0,
|
||||||
ready: ready_tx,
|
ready: ready_tx,
|
||||||
|
token_delay: Duration::ZERO,
|
||||||
};
|
};
|
||||||
let uploader = std::thread::spawn(move || {
|
let uploader = std::thread::spawn(move || {
|
||||||
if let Err(e) = run_refusing_uploader(&cfg, &refusal) {
|
if let Err(e) = run_refusing_uploader(&cfg, &refusal) {
|
||||||
@@ -905,6 +966,7 @@ fn an_interrupted_download_resumes_from_its_partial_file() {
|
|||||||
content: content.clone(),
|
content: content.clone(),
|
||||||
token: 424_243_u32,
|
token: 424_243_u32,
|
||||||
ready: ready_tx,
|
ready: ready_tx,
|
||||||
|
token_delay: Duration::ZERO,
|
||||||
};
|
};
|
||||||
let (offset_tx, offset_rx) = std::sync::mpsc::channel();
|
let (offset_tx, offset_rx) = std::sync::mpsc::channel();
|
||||||
let uploader = std::thread::spawn(move || match run_mock_uploader(&cfg) {
|
let uploader = std::thread::spawn(move || match run_mock_uploader(&cfg) {
|
||||||
@@ -1122,6 +1184,7 @@ fn run_mock_direct_peer(cfg: &MockDirectUpload) -> std::io::Result<()> {
|
|||||||
&cfg.username,
|
&cfg.username,
|
||||||
cfg.token,
|
cfg.token,
|
||||||
&cfg.content,
|
&cfg.content,
|
||||||
|
Duration::ZERO,
|
||||||
)
|
)
|
||||||
.map(|_| ())
|
.map(|_| ())
|
||||||
}
|
}
|
||||||
@@ -1269,6 +1332,7 @@ fn run_mock_firewalled_peer(cfg: &MockFirewalledUpload) -> std::io::Result<()> {
|
|||||||
&cfg.username,
|
&cfg.username,
|
||||||
cfg.token,
|
cfg.token,
|
||||||
&cfg.content,
|
&cfg.content,
|
||||||
|
Duration::ZERO,
|
||||||
)
|
)
|
||||||
.map(|_| ())
|
.map(|_| ())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,6 +13,8 @@ export const LINKS = {
|
|||||||
slskd: 'https://github.com/slskd/slskd',
|
slskd: 'https://github.com/slskd/slskd',
|
||||||
slsknet: 'https://www.slsknet.org',
|
slsknet: 'https://www.slsknet.org',
|
||||||
seeleseek: 'https://seeleseek.net',
|
seeleseek: 'https://seeleseek.net',
|
||||||
|
seakarr: 'https://github.com/binhex/seakarr',
|
||||||
|
minorVersions: 'https://github.com/michel/soulseek-rs/issues/12',
|
||||||
} as const
|
} as const
|
||||||
|
|
||||||
export const VERSION = __APP_VERSION__
|
export const VERSION = __APP_VERSION__
|
||||||
|
|||||||
@@ -259,6 +259,31 @@ export const Community = () => {
|
|||||||
</Cols>
|
</Cols>
|
||||||
</Section>
|
</Section>
|
||||||
|
|
||||||
|
<Section>
|
||||||
|
<SectionHead
|
||||||
|
eyebrow="built on the library"
|
||||||
|
title="Projects using soulseek-rs-lib."
|
||||||
|
>
|
||||||
|
The protocol crate is there for anyone building their own client, bot, or
|
||||||
|
automation on Soulseek. These projects do.
|
||||||
|
</SectionHead>
|
||||||
|
<Cols start>
|
||||||
|
<Prose>
|
||||||
|
<p>
|
||||||
|
<ExtLink href={LINKS.seakarr}>seakarr</ExtLink> — automated Soulseek music
|
||||||
|
downloader with library quality upgrading: it scans your collection, finds
|
||||||
|
albums below your bitrate bar, and fetches better copies.
|
||||||
|
</p>
|
||||||
|
<p>
|
||||||
|
Built something on <Code>soulseek-rs-lib</Code>? Open a pull request to get
|
||||||
|
listed, and{' '}
|
||||||
|
<ExtLink href={LINKS.minorVersions}>reserve a client minor version</ExtLink>{' '}
|
||||||
|
so your project is identifiable on the network.
|
||||||
|
</p>
|
||||||
|
</Prose>
|
||||||
|
</Cols>
|
||||||
|
</Section>
|
||||||
|
|
||||||
<Section>
|
<Section>
|
||||||
<SectionHead
|
<SectionHead
|
||||||
eyebrow="what this is not"
|
eyebrow="what this is not"
|
||||||
|
|||||||
Reference in New Issue
Block a user