feat(shares): scan shared dir on connect and send a real login handshake
The post-login handshake (SharedFolders / HaveNoParent / SetStatus / SetWaitPort) lived only in ServerActor::login(), which nothing ever called — so a live client never advertised its listen port or share counts. Move it to a testable post_login_messages() fired from the LoginStatus(true) arm (once, on success, on the live path), and delete the dead login(). Client::connect now scans ClientSettings.shared_directory once into ClientContext.shares and passes the real folder/file counts to the server. This is the prerequisite for two live clients to resolve each other's addresses (GetPeerAddress needs the peer to have sent SetWaitPort). Unit-tested post_login_messages; the login/listener e2e still passes, proving soulfind accepts the handshake.
This commit is contained in:
@@ -25,7 +25,7 @@ use std::sync::mpsc::{Receiver, Sender};
|
|||||||
use std::sync::{Arc, RwLock};
|
use std::sync::{Arc, RwLock};
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use crate::{SoulseekRs, debug, error, info, trace, warn};
|
use crate::{SoulseekRs, debug, error, trace, warn};
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct PeerAddress {
|
pub struct PeerAddress {
|
||||||
@@ -177,6 +177,32 @@ pub struct ServerActor {
|
|||||||
dispatcher_receiver: Option<Receiver<ServerMessage>>,
|
dispatcher_receiver: Option<Receiver<ServerMessage>>,
|
||||||
dispatcher_sender: Option<Sender<ServerMessage>>,
|
dispatcher_sender: Option<Sender<ServerMessage>>,
|
||||||
queued_messages: Vec<ServerMessage>,
|
queued_messages: Vec<ServerMessage>,
|
||||||
|
shared_folder_count: u32,
|
||||||
|
shared_file_count: u32,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The messages a client sends right after a successful login: its shared-file
|
||||||
|
/// counts, distributed-network opt-out, online status, and (when listening) the
|
||||||
|
/// port peers should connect to. Kept as a free function so it can be tested
|
||||||
|
/// without a live connection.
|
||||||
|
fn post_login_messages(
|
||||||
|
enable_listen: bool,
|
||||||
|
listen_port: u16,
|
||||||
|
shared_folders: u32,
|
||||||
|
shared_files: u32,
|
||||||
|
) -> Vec<Message> {
|
||||||
|
let mut messages = vec![
|
||||||
|
MessageFactory::build_shared_folders_message(
|
||||||
|
shared_folders,
|
||||||
|
shared_files,
|
||||||
|
),
|
||||||
|
MessageFactory::build_no_parent_message(),
|
||||||
|
MessageFactory::build_set_status_message(2),
|
||||||
|
];
|
||||||
|
if enable_listen {
|
||||||
|
messages.push(MessageFactory::build_set_wait_port_message(listen_port));
|
||||||
|
}
|
||||||
|
messages
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ServerActor {
|
impl ServerActor {
|
||||||
@@ -186,6 +212,8 @@ impl ServerActor {
|
|||||||
client_channel: Sender<ClientOperation>,
|
client_channel: Sender<ClientOperation>,
|
||||||
listen_port: u16,
|
listen_port: u16,
|
||||||
enable_listen: bool,
|
enable_listen: bool,
|
||||||
|
shared_folder_count: u32,
|
||||||
|
shared_file_count: u32,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
address,
|
address,
|
||||||
@@ -201,6 +229,8 @@ impl ServerActor {
|
|||||||
client_channel,
|
client_channel,
|
||||||
self_handle: None,
|
self_handle: None,
|
||||||
queued_messages: Vec::new(),
|
queued_messages: Vec::new(),
|
||||||
|
shared_folder_count,
|
||||||
|
shared_file_count,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -325,54 +355,6 @@ impl ServerActor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn login(
|
|
||||||
&mut self,
|
|
||||||
username: &str,
|
|
||||||
password: &str,
|
|
||||||
) -> Result<bool, SoulseekRs> {
|
|
||||||
self.queue_message(MessageFactory::build_login_message(
|
|
||||||
username, password,
|
|
||||||
));
|
|
||||||
let context = self.context.clone();
|
|
||||||
let mut logged_in;
|
|
||||||
let timeout = Duration::from_secs(5);
|
|
||||||
let start = Instant::now();
|
|
||||||
|
|
||||||
loop {
|
|
||||||
if start.elapsed() > timeout {
|
|
||||||
warn!("Timeout waiting for login response");
|
|
||||||
return Err(SoulseekRs::Timeout);
|
|
||||||
}
|
|
||||||
|
|
||||||
{
|
|
||||||
logged_in = context.read_safe()?.logged_in;
|
|
||||||
}
|
|
||||||
|
|
||||||
if logged_in.is_some() {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
let logged_in = logged_in.ok_or(SoulseekRs::Timeout)?;
|
|
||||||
if logged_in {
|
|
||||||
info!("Logged in as {}", username);
|
|
||||||
self.queue_message(MessageFactory::build_shared_folders_message(
|
|
||||||
1, 499,
|
|
||||||
));
|
|
||||||
self.queue_message(MessageFactory::build_no_parent_message());
|
|
||||||
self.queue_message(MessageFactory::build_set_status_message(2));
|
|
||||||
if self.enable_listen {
|
|
||||||
self.queue_message(
|
|
||||||
MessageFactory::build_set_wait_port_message(
|
|
||||||
self.listen_port,
|
|
||||||
),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok(logged_in)
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn file_search(&mut self, token: u32, query: &str) {
|
pub fn file_search(&mut self, token: u32, query: &str) {
|
||||||
self.queue_message(MessageFactory::build_file_search_message(
|
self.queue_message(MessageFactory::build_file_search_message(
|
||||||
token, query,
|
token, query,
|
||||||
@@ -409,6 +391,20 @@ impl ServerActor {
|
|||||||
error!("[server] LoginStatus write: {}", e);
|
error!("[server] LoginStatus write: {}", e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// Send the post-login handshake exactly once, only on success,
|
||||||
|
// on the live path (the old ServerActor::login did this but was
|
||||||
|
// never called). Advertises real shared counts and, when
|
||||||
|
// listening, the port peers must connect to.
|
||||||
|
if message {
|
||||||
|
for msg in post_login_messages(
|
||||||
|
self.enable_listen,
|
||||||
|
self.listen_port,
|
||||||
|
self.shared_folder_count,
|
||||||
|
self.shared_file_count,
|
||||||
|
) {
|
||||||
|
self.send_message(msg);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
ServerMessage::PierceFirewall(token) => {
|
ServerMessage::PierceFirewall(token) => {
|
||||||
self.send_message(
|
self.send_message(
|
||||||
@@ -725,3 +721,31 @@ impl Actor for ServerActor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::post_login_messages;
|
||||||
|
use crate::message::Message;
|
||||||
|
|
||||||
|
fn code_of(message: &Message) -> u32 {
|
||||||
|
u32::from_le_bytes(message.get_data()[0..4].try_into().unwrap())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn post_login_messages_carry_counts_and_conditional_wait_port() {
|
||||||
|
let messages = post_login_messages(true, 4321, 3, 7);
|
||||||
|
let codes: Vec<u32> = messages.iter().map(code_of).collect();
|
||||||
|
// SharedFolders, HaveNoParent, SetStatus, SetWaitPort.
|
||||||
|
assert_eq!(codes, vec![35, 71, 28, 2]);
|
||||||
|
|
||||||
|
// The SharedFolders message (code 35) carries the real counts.
|
||||||
|
let shared = messages[0].get_data();
|
||||||
|
assert_eq!(u32::from_le_bytes(shared[4..8].try_into().unwrap()), 3);
|
||||||
|
assert_eq!(u32::from_le_bytes(shared[8..12].try_into().unwrap()), 7);
|
||||||
|
|
||||||
|
// Not listening omits SetWaitPort (code 2).
|
||||||
|
let no_listen = post_login_messages(false, 4321, 3, 7);
|
||||||
|
let codes: Vec<u32> = no_listen.iter().map(code_of).collect();
|
||||||
|
assert_eq!(codes, vec![35, 71, 28]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ use crate::{
|
|||||||
actor::{ActorSystem, peer_registry::PeerRegistry},
|
actor::{ActorSystem, peer_registry::PeerRegistry},
|
||||||
error::{Result, SoulseekRs},
|
error::{Result, SoulseekRs},
|
||||||
peer::{ConnectionType, DownloadPeer, NewPeer, Peer, listen::Listen},
|
peer::{ConnectionType, DownloadPeer, NewPeer, Peer, listen::Listen},
|
||||||
|
shares::Shares,
|
||||||
types::{Download, Search, SearchResult},
|
types::{Download, Search, SearchResult},
|
||||||
utils::{lock::RwLockExt, md5, thread_pool::ThreadPool},
|
utils::{lock::RwLockExt, md5, thread_pool::ThreadPool},
|
||||||
};
|
};
|
||||||
@@ -124,6 +125,8 @@ pub struct ClientContext {
|
|||||||
/// Correlation tokens for server-brokered (firewalled) connections, mapping
|
/// Correlation tokens for server-brokered (firewalled) connections, mapping
|
||||||
/// a token we sent in a ConnectToPeer to the peer we expect back.
|
/// a token we sent in a ConnectToPeer to the peer we expect back.
|
||||||
pending_connect_tokens: HashMap<u32, String>,
|
pending_connect_tokens: HashMap<u32, String>,
|
||||||
|
/// Files we share with peers (read-only after connect).
|
||||||
|
pub shares: Arc<Shares>,
|
||||||
actor_system: Arc<ActorSystem>,
|
actor_system: Arc<ActorSystem>,
|
||||||
}
|
}
|
||||||
impl Default for ClientContext {
|
impl Default for ClientContext {
|
||||||
@@ -385,6 +388,7 @@ impl ClientContext {
|
|||||||
searches: HashMap::new(),
|
searches: HashMap::new(),
|
||||||
private_messages: Vec::new(),
|
private_messages: Vec::new(),
|
||||||
pending_connect_tokens: HashMap::new(),
|
pending_connect_tokens: HashMap::new(),
|
||||||
|
shares: Arc::new(Shares::empty()),
|
||||||
downloads: DownloadStore::new(),
|
downloads: DownloadStore::new(),
|
||||||
actor_system,
|
actor_system,
|
||||||
}
|
}
|
||||||
@@ -468,11 +472,39 @@ impl Client {
|
|||||||
|
|
||||||
let listen_sender = sender.clone();
|
let listen_sender = sender.clone();
|
||||||
|
|
||||||
|
// Scan the shared directory once into the read-only index, and report
|
||||||
|
// the real folder/file counts to the server on login.
|
||||||
|
let shares = match self.shared_directory.as_deref() {
|
||||||
|
Some(dir) if !dir.trim().is_empty() => {
|
||||||
|
match Shares::scan(std::path::Path::new(dir)) {
|
||||||
|
Ok(scanned) => {
|
||||||
|
info!(
|
||||||
|
"Sharing {} files in {} folders from {}",
|
||||||
|
scanned.file_count(),
|
||||||
|
scanned.folder_count(),
|
||||||
|
dir
|
||||||
|
);
|
||||||
|
Arc::new(scanned)
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
warn!("Failed to scan shared directory {}: {}", dir, e);
|
||||||
|
Arc::new(Shares::empty())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ => Arc::new(Shares::empty()),
|
||||||
|
};
|
||||||
|
let shared_folder_count = shares.folder_count();
|
||||||
|
let shared_file_count = shares.file_count();
|
||||||
|
ctx.shares = shares;
|
||||||
|
|
||||||
let server_actor = ServerActor::new(
|
let server_actor = ServerActor::new(
|
||||||
self.address.clone(),
|
self.address.clone(),
|
||||||
sender,
|
sender,
|
||||||
self.listen_port,
|
self.listen_port,
|
||||||
self.enable_listen,
|
self.enable_listen,
|
||||||
|
shared_folder_count,
|
||||||
|
shared_file_count,
|
||||||
);
|
);
|
||||||
|
|
||||||
self.server_handle = Some(ctx.actor_system.spawn_with_handle(
|
self.server_handle = Some(ctx.actor_system.spawn_with_handle(
|
||||||
|
|||||||
Reference in New Issue
Block a user