Merge pull request #66 from michel/lib/folder-contents
feat(lib): answer FolderContentsRequest with the folder's listing
This commit is contained in:
@@ -2,10 +2,10 @@ use crate::actor::{Actor, ActorHandle, ConnectionState};
|
||||
use crate::client::ClientOperation;
|
||||
use crate::dispatcher::MessageDispatcher;
|
||||
use crate::message::peer::{
|
||||
FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueRequest,
|
||||
PlaceInQueueResponse, QueueUploadHandler, SharedDirectory,
|
||||
SharedFileListResponseHandler, TransferRequest, TransferResponse,
|
||||
UploadDeniedHandler, UploadFailedHandler,
|
||||
FileSearchResponse, FolderContentsRequest, GetShareFileList, PeerInit,
|
||||
PlaceInQueueRequest, PlaceInQueueResponse, QueueUploadHandler,
|
||||
SharedDirectory, SharedFileListResponseHandler, TransferRequest,
|
||||
TransferResponse, UploadDeniedHandler, UploadFailedHandler,
|
||||
};
|
||||
use crate::message::server::MessageFactory;
|
||||
use crate::message::{Handlers, Message, MessageReader, MessageType};
|
||||
@@ -46,6 +46,11 @@ pub enum PeerMessage {
|
||||
ShareListRequested,
|
||||
/// A peer we are browsing sent us their shared-file listing (code 5).
|
||||
ShareListReceived(Vec<SharedDirectory>),
|
||||
/// A peer asked for one folder of our shares (they sent us code 36).
|
||||
FolderContentsRequested {
|
||||
token: u32,
|
||||
folder: String,
|
||||
},
|
||||
/// Offer the queued file to that peer: send an upload TransferRequest.
|
||||
ServeUpload {
|
||||
token: u32,
|
||||
@@ -178,6 +183,7 @@ impl PeerActor {
|
||||
handlers.register_handler(TransferRequest);
|
||||
handlers.register_handler(TransferResponse);
|
||||
handlers.register_handler(GetShareFileList);
|
||||
handlers.register_handler(FolderContentsRequest);
|
||||
handlers.register_handler(UploadDeniedHandler);
|
||||
handlers.register_handler(UploadFailedHandler);
|
||||
handlers.register_handler(PlaceInQueueRequest);
|
||||
@@ -273,6 +279,9 @@ impl PeerActor {
|
||||
PeerMessage::ShareListReceived(directories) => {
|
||||
self.handle_share_list_received(directories);
|
||||
}
|
||||
PeerMessage::FolderContentsRequested { token, folder } => {
|
||||
self.handle_folder_contents_requested(token, folder);
|
||||
}
|
||||
PeerMessage::RequestTransfer(download) => {
|
||||
let message = MessageFactory::build_transfer_request_message(
|
||||
&download.filename,
|
||||
@@ -441,6 +450,20 @@ impl PeerActor {
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_folder_contents_requested(&self, token: u32, folder: String) {
|
||||
let requester_key = self.peer_username();
|
||||
if let Err(e) =
|
||||
self.client_channel
|
||||
.send(ClientOperation::FolderContentsRequested {
|
||||
requester_key,
|
||||
token,
|
||||
folder,
|
||||
})
|
||||
{
|
||||
error!("[peer_actor] forward FolderContentsRequested: {}", e);
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_share_list_received(&self, directories: Vec<SharedDirectory>) {
|
||||
let username = self.peer_username();
|
||||
if let Err(e) =
|
||||
|
||||
@@ -292,6 +292,12 @@ pub enum ClientOperation {
|
||||
ShareListRequested {
|
||||
requester_key: String,
|
||||
},
|
||||
/// A peer asked for one folder of our shares; send a FolderContentsResponse.
|
||||
FolderContentsRequested {
|
||||
requester_key: String,
|
||||
token: u32,
|
||||
folder: String,
|
||||
},
|
||||
/// A peer we are browsing returned their shared-file listing.
|
||||
BrowseResult {
|
||||
username: String,
|
||||
|
||||
@@ -756,35 +756,36 @@ impl Client {
|
||||
}
|
||||
}
|
||||
ClientOperation::ShareListRequested { requester_key } => {
|
||||
// Reply with our full shared-file listing.
|
||||
let (registry, message) = match client_context
|
||||
.read_safe()
|
||||
{
|
||||
Ok(ctx) => {
|
||||
let dirs = ctx
|
||||
.shares
|
||||
.directories()
|
||||
.into_iter()
|
||||
.map(|(name, files)| {
|
||||
crate::message::peer::SharedDirectory {
|
||||
name,
|
||||
files,
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
(
|
||||
ctx.peer_registry.clone(),
|
||||
crate::message::peer::build_shared_file_list(&dirs),
|
||||
Self::reply_to_peer(
|
||||
&client_context,
|
||||
&requester_key,
|
||||
|ctx| {
|
||||
crate::message::peer::build_shared_file_list(
|
||||
&shared_directories(ctx),
|
||||
)
|
||||
}
|
||||
Err(_) => continue,
|
||||
};
|
||||
if let Some(registry) = registry {
|
||||
let _ = registry.send_to_peer(
|
||||
&requester_key,
|
||||
PeerMessage::SendMessage(message),
|
||||
);
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
ClientOperation::FolderContentsRequested {
|
||||
requester_key,
|
||||
token,
|
||||
folder,
|
||||
} => {
|
||||
// ponytail: walks the whole share for one folder, as a
|
||||
// browse does; index by directory if it ever shows up.
|
||||
Self::reply_to_peer(
|
||||
&client_context,
|
||||
&requester_key,
|
||||
|ctx| {
|
||||
let dirs: Vec<_> = shared_directories(ctx)
|
||||
.into_iter()
|
||||
.filter(|dir| dir.name == folder)
|
||||
.collect();
|
||||
crate::message::peer::build_folder_contents(
|
||||
token, &folder, &dirs,
|
||||
)
|
||||
},
|
||||
);
|
||||
}
|
||||
ClientOperation::BrowseResult {
|
||||
username,
|
||||
@@ -845,6 +846,22 @@ impl Client {
|
||||
});
|
||||
}
|
||||
|
||||
/// Answer a peer with the message `build` derives from the client state.
|
||||
fn reply_to_peer(
|
||||
client_context: &Arc<RwLock<ClientContext>>,
|
||||
requester_key: &str,
|
||||
build: impl FnOnce(&ClientContext) -> crate::message::Message,
|
||||
) {
|
||||
let (registry, message) = match client_context.read_safe() {
|
||||
Ok(ctx) => (ctx.peer_registry.clone(), build(&ctx)),
|
||||
Err(_) => return,
|
||||
};
|
||||
if let Some(registry) = registry {
|
||||
let _ = registry
|
||||
.send_to_peer(requester_key, PeerMessage::SendMessage(message));
|
||||
}
|
||||
}
|
||||
|
||||
fn sweep_expired_connects(client_context: &Arc<RwLock<ClientContext>>) {
|
||||
let expired = match client_context.write_safe() {
|
||||
Ok(mut ctx) => ctx.take_expired_connects(Instant::now()),
|
||||
@@ -855,3 +872,17 @@ impl Client {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Our shares in the wire form a browse or folder reply carries.
|
||||
fn shared_directories(
|
||||
ctx: &ClientContext,
|
||||
) -> Vec<crate::message::peer::SharedDirectory> {
|
||||
ctx.shares
|
||||
.directories()
|
||||
.into_iter()
|
||||
.map(|(name, files)| crate::message::peer::SharedDirectory {
|
||||
name,
|
||||
files,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
//! `FolderContentsRequest` (peer code 36) and its reply (code 37): one folder
|
||||
//! of a share in the listing form code 5 uses, which is what "download
|
||||
//! folder" in other clients asks for.
|
||||
|
||||
use crate::message::peer::SharedDirectory;
|
||||
use crate::message::peer::shared_file_list::{
|
||||
decompress_body, read_directories, write_directories,
|
||||
};
|
||||
use crate::message::{Message, MessageHandler};
|
||||
use crate::peer::PeerMessage;
|
||||
use crate::utils::zlib::compress_stored;
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
pub struct FolderContentsRequest;
|
||||
impl MessageHandler<PeerMessage> for FolderContentsRequest {
|
||||
fn get_code(&self) -> u32 {
|
||||
36
|
||||
}
|
||||
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
|
||||
let token = message.read_int32();
|
||||
let folder = message.read_string();
|
||||
let _ =
|
||||
sender.send(PeerMessage::FolderContentsRequested { token, folder });
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a `FolderContentsResponse` (peer code 37).
|
||||
#[must_use]
|
||||
pub fn build_folder_contents(
|
||||
token: u32,
|
||||
folder: &str,
|
||||
dirs: &[SharedDirectory],
|
||||
) -> Message {
|
||||
let mut payload = Message::new();
|
||||
payload.write_int32(token).write_string(folder);
|
||||
write_directories(&mut payload, dirs);
|
||||
Message::new()
|
||||
.write_int32(37)
|
||||
.write_raw_bytes(compress_stored(&payload.get_data()))
|
||||
.clone()
|
||||
}
|
||||
|
||||
/// Parse a `FolderContentsResponse` positioned at its compressed blob, or
|
||||
/// `None` when it does not inflate.
|
||||
#[must_use]
|
||||
pub fn parse_folder_contents(
|
||||
message: &mut Message,
|
||||
) -> Option<(u32, String, Vec<SharedDirectory>)> {
|
||||
let mut body = decompress_body(message)?;
|
||||
let token = body.read_int32();
|
||||
let folder = body.read_string();
|
||||
Some((token, folder, read_directories(&mut body)))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_folder_listing_round_trips() {
|
||||
let dirs = vec![SharedDirectory {
|
||||
name: "music\\album".to_string(),
|
||||
files: vec![("one.flac".to_string(), 40), ("two.flac".to_string(), 50)],
|
||||
}];
|
||||
let built = build_folder_contents(7, "music\\album", &dirs);
|
||||
let mut message = Message::new_with_data(built.get_data());
|
||||
message.set_pointer(4);
|
||||
assert_eq!(
|
||||
parse_folder_contents(&mut message),
|
||||
Some((7, "music\\album".to_string(), dirs))
|
||||
);
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
mod file_search_response;
|
||||
mod folder_contents;
|
||||
mod get_share_file_list;
|
||||
mod peer_init;
|
||||
mod place_in_queue_request;
|
||||
@@ -14,6 +15,9 @@ mod upload_failed;
|
||||
pub use file_search_response::{
|
||||
FileEntry, FileSearchResponse, build_file_search_response,
|
||||
};
|
||||
pub use folder_contents::{
|
||||
FolderContentsRequest, build_folder_contents, parse_folder_contents,
|
||||
};
|
||||
pub use get_share_file_list::GetShareFileList;
|
||||
pub use peer_init::PeerInit;
|
||||
pub use place_in_queue_request::PlaceInQueueRequest;
|
||||
|
||||
@@ -30,20 +30,7 @@ impl MessageHandler<PeerMessage> for SharedFileListResponseHandler {
|
||||
#[must_use]
|
||||
pub fn build_shared_file_list(dirs: &[SharedDirectory]) -> Message {
|
||||
let mut payload = Message::new();
|
||||
payload.write_int32(dirs.len() as u32);
|
||||
for dir in dirs {
|
||||
payload
|
||||
.write_string(&dir.name)
|
||||
.write_int32(dir.files.len() as u32);
|
||||
for (name, size) in &dir.files {
|
||||
payload
|
||||
.write_int8(1)
|
||||
.write_string(name)
|
||||
.write_int64(*size)
|
||||
.write_string("") // extension
|
||||
.write_int32(0); // attribute count
|
||||
}
|
||||
}
|
||||
write_directories(&mut payload, dirs);
|
||||
payload.write_int32(0); // unknown
|
||||
payload.write_int32(0); // number of private directories
|
||||
|
||||
@@ -60,19 +47,35 @@ pub fn build_shared_file_list(dirs: &[SharedDirectory]) -> Message {
|
||||
/// Returns an empty listing if the payload is malformed.
|
||||
#[must_use]
|
||||
pub fn parse_shared_file_list(message: &mut Message) -> Vec<SharedDirectory> {
|
||||
let pointer = message.get_pointer();
|
||||
let size = message.get_size();
|
||||
let compressed = message.get_slice(pointer, size);
|
||||
let Ok(data) = deflate(&compressed) else {
|
||||
return Vec::new();
|
||||
};
|
||||
decompress_body(message)
|
||||
.map_or_else(Vec::new, |mut body| read_directories(&mut body))
|
||||
}
|
||||
|
||||
let mut body = Message::new_with_data(data);
|
||||
/// Write a directory listing in the form both code 5 and code 37 carry.
|
||||
pub fn write_directories(payload: &mut Message, dirs: &[SharedDirectory]) {
|
||||
payload.write_int32(dirs.len() as u32);
|
||||
for dir in dirs {
|
||||
payload
|
||||
.write_string(&dir.name)
|
||||
.write_int32(dir.files.len() as u32);
|
||||
for (name, size) in &dir.files {
|
||||
payload
|
||||
.write_int8(1)
|
||||
.write_string(name)
|
||||
.write_int64(*size)
|
||||
.write_string("") // extension
|
||||
.write_int32(0); // attribute count
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Read back what [`write_directories`] wrote, stopping early when a hostile
|
||||
/// count outruns the payload so a bogus length cannot spin into a huge
|
||||
/// allocation loop.
|
||||
pub fn read_directories(body: &mut Message) -> Vec<SharedDirectory> {
|
||||
let dir_count = body.read_int32();
|
||||
let mut dirs = Vec::new();
|
||||
for _ in 0..dir_count {
|
||||
// Stop if a hostile count outruns the (decompressed) payload, so a
|
||||
// bogus length can't spin us into a huge allocation loop.
|
||||
if body.get_pointer() >= body.get_size() {
|
||||
break;
|
||||
}
|
||||
@@ -104,6 +107,15 @@ pub fn parse_shared_file_list(message: &mut Message) -> Vec<SharedDirectory> {
|
||||
dirs
|
||||
}
|
||||
|
||||
/// Inflate a compressed peer payload positioned at the blob.
|
||||
pub fn decompress_body(message: &mut Message) -> Option<Message> {
|
||||
let pointer = message.get_pointer();
|
||||
let size = message.get_size();
|
||||
deflate(&message.get_slice(pointer, size))
|
||||
.ok()
|
||||
.map(Message::new_with_data)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hostile_dir_count_does_not_hang() {
|
||||
// A compressed body claiming ~4 billion directories with no data must
|
||||
|
||||
@@ -3180,3 +3180,70 @@ fn a_search_reply_reports_no_free_slot_while_the_only_slot_is_taken() {
|
||||
|
||||
let _ = std::fs::remove_dir_all(share);
|
||||
}
|
||||
|
||||
// "Download folder" in SoulseekQt and Nicotine+ asks the sharer for that one
|
||||
// folder (peer code 36) instead of the whole share; a sharer that never
|
||||
// answers leaves the folder download hanging.
|
||||
#[test]
|
||||
fn a_third_party_client_fetches_one_folder_of_our_shares() {
|
||||
let server = server_or_skip!();
|
||||
|
||||
let share_dir = unique_download_dir();
|
||||
std::fs::create_dir_all(share_dir.join("album")).unwrap();
|
||||
std::fs::create_dir_all(share_dir.join("other")).unwrap();
|
||||
std::fs::write(share_dir.join("album").join("one.flac"), b"xxxx").unwrap();
|
||||
std::fs::write(share_dir.join("album").join("two.flac"), b"yy").unwrap();
|
||||
std::fs::write(share_dir.join("other").join("three.flac"), b"z").unwrap();
|
||||
let root = share_dir.file_name().unwrap().to_str().unwrap().to_string();
|
||||
|
||||
let sharer_port = free_port().expect("sharer port");
|
||||
let mut sharer = Client::with_settings(ClientSettings {
|
||||
shared_directories: vec![share_dir.display().to_string()],
|
||||
..server.listening_settings("e2e_folder_sharer", "pw", sharer_port)
|
||||
});
|
||||
sharer.connect().expect("sharer connect");
|
||||
assert!(sharer.login().expect("sharer login"));
|
||||
let server_addr = format!("{}:{}", server.host, server.port);
|
||||
let _qt = login_raw(&server_addr, "e2e_folder_browser", "pw")
|
||||
.expect("third-party client logs in");
|
||||
|
||||
let folder = format!("{root}\\album");
|
||||
let mut p = connect_retry(
|
||||
&format!("127.0.0.1:{sharer_port}"),
|
||||
Duration::from_secs(5),
|
||||
)
|
||||
.expect("dial the sharer");
|
||||
p.set_read_timeout(Some(Duration::from_secs(15))).unwrap();
|
||||
p.write_all(&peer_init_bytes("e2e_folder_browser", "P", 0))
|
||||
.unwrap();
|
||||
let mut request = Message::new();
|
||||
request
|
||||
.write_int32(36)
|
||||
.write_int32(77)
|
||||
.write_string(&folder);
|
||||
p.write_all(&request.get_buffer()).unwrap();
|
||||
p.flush().unwrap();
|
||||
|
||||
let mut response = expect_code(&mut p, 37, Duration::from_secs(15))
|
||||
.expect("the sharer answers with a FolderContentsResponse");
|
||||
response.set_pointer(8);
|
||||
let (token, echoed, directories) =
|
||||
soulseek_rs::message::peer::parse_folder_contents(&mut response)
|
||||
.expect("a well-formed folder listing");
|
||||
assert_eq!(token, 77);
|
||||
assert_eq!(echoed, folder);
|
||||
let mut names: Vec<(String, Vec<String>)> = directories
|
||||
.into_iter()
|
||||
.map(|d| (d.name, d.files.into_iter().map(|(name, _)| name).collect()))
|
||||
.collect();
|
||||
for (_, files) in &mut names {
|
||||
files.sort();
|
||||
}
|
||||
assert_eq!(
|
||||
names,
|
||||
[(folder, vec!["one.flac".to_string(), "two.flac".to_string()])],
|
||||
"only the requested folder, with its files"
|
||||
);
|
||||
|
||||
let _ = std::fs::remove_dir_all(share_dir);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user