From 7bae6b36da7f1e89ade5942bfe000d4bd7fe35e0 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Sun, 9 Nov 2025 21:19:52 +0100 Subject: [PATCH 1/9] chore: cargo.lock --- Cargo.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 4bc772f..53fe6dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,4 +4,4 @@ version = 4 [[package]] name = "soulseek-rs" -version = "0.1.0" +version = "1.0.0" From 3dc667fe3049ae042a68b48d8ba5c1b85433e63c Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 13:54:44 +0100 Subject: [PATCH 2/9] feat: actor refactoring default peer --- src/actor/mod.rs | 211 +++++++++ src/actor/peer_actor.rs | 405 ++++++++++++++++++ src/actor/peer_registry.rs | 119 ++++++ src/client.rs | 165 +++++-- src/lib.rs | 1 + src/main.rs | 2 +- src/message/handlers.rs | 6 +- src/message/peer/file_search_response.rs | 8 +- src/message/peer/get_share_file_list.rs | 8 +- src/message/peer/peer_init.rs | 8 +- src/message/peer/place_in_queue_response.rs | 8 +- src/message/peer/transfer_request.rs | 8 +- src/message/peer/transfer_response.rs | 8 +- src/message/peer/upload_failed.rs | 6 +- src/peer/default_peer.rs | 450 -------------------- src/peer/listen.rs | 24 +- src/peer/mod.rs | 6 +- 17 files changed, 908 insertions(+), 535 deletions(-) create mode 100644 src/actor/mod.rs create mode 100644 src/actor/peer_actor.rs create mode 100644 src/actor/peer_registry.rs delete mode 100644 src/peer/default_peer.rs diff --git a/src/actor/mod.rs b/src/actor/mod.rs new file mode 100644 index 0000000..7587f7e --- /dev/null +++ b/src/actor/mod.rs @@ -0,0 +1,211 @@ +// src/actor/mod.rs +// Lightweight actor system using the existing thread pool + +use std::sync::mpsc::{channel, Receiver, Sender}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use crate::utils::thread_pool::ThreadPool; +use crate::trace; + +pub mod peer_actor; +pub mod peer_registry; + +/// Core actor trait - each actor processes messages +pub trait Actor: Send + 'static { + type Message: Send + Clone + 'static; + + /// Handle a single message + fn handle(&mut self, msg: Self::Message); + + /// Called when actor starts (optional hook) + fn on_start(&mut self) {} + + /// Called when actor stops (optional hook) + fn on_stop(&mut self) {} + + /// Optional periodic tick for background work + fn tick(&mut self) {} +} + +/// Handle to send messages to an actor +#[derive(Clone)] +pub struct ActorHandle { + pub(crate) sender: Sender>, +} + +impl ActorHandle { + /// Send a message to the actor (non-blocking) + pub fn send(&self, msg: M) -> Result<(), String> { + self.sender + .send(ActorMessage::UserMessage(msg)) + .map_err(|e| format!("Failed to send message: {}", e)) + } + + /// Request actor to stop gracefully + pub fn stop(&self) -> Result<(), String> { + self.sender + .send(ActorMessage::Stop) + .map_err(|e| format!("Failed to send stop signal: {}", e)) + } +} + +/// Internal actor message wrapper +pub(crate) enum ActorMessage { + UserMessage(M), + Stop, + Tick, +} + +/// Actor system that manages actor lifecycle +pub struct ActorSystem { + thread_pool: Arc, +} + +impl ActorSystem { + pub fn new(thread_pool: Arc) -> Self { + ActorSystem { thread_pool } + } + + /// Spawn a new actor and return its handle + pub fn spawn(&self, mut actor: A) -> ActorHandle { + let (sender, receiver) = channel::>(); + let handle = ActorHandle { sender: sender.clone() }; + + let thread_pool = self.thread_pool.clone(); + + // Start the actor event loop on the thread pool + self.thread_pool.execute(move || { + actor.on_start(); + Self::run_actor_loop(&mut actor, receiver); + actor.on_stop(); + }); + + handle + } + + /// Spawn a new actor with initialization callback and return its handle + /// The callback receives the actor handle before on_start is called + pub fn spawn_with_handle(&self, mut actor: A, init: F) -> ActorHandle + where + F: FnOnce(&mut A, ActorHandle) + Send + 'static, + { + let (sender, receiver) = channel::>(); + let handle = ActorHandle { sender: sender.clone() }; + let handle_for_init = handle.clone(); + + let thread_pool = self.thread_pool.clone(); + + // Start the actor event loop on the thread pool + self.thread_pool.execute(move || { + trace!("[actor_system] Actor task starting on thread pool"); + init(&mut actor, handle_for_init); + trace!("[actor_system] Actor init callback completed"); + actor.on_start(); + trace!("[actor_system] Actor on_start completed, starting event loop"); + Self::run_actor_loop(&mut actor, receiver); + trace!("[actor_system] Actor event loop ended"); + actor.on_stop(); + trace!("[actor_system] Actor on_stop completed"); + }); + + handle + } + + /// Run the actor's message processing loop + fn run_actor_loop(actor: &mut A, receiver: Receiver>) { + trace!("[actor_system] run_actor_loop STARTED"); + let tick_interval = Duration::from_millis(100); + let mut last_tick = Instant::now(); + let mut message_count = 0; + let mut tick_count = 0; + + loop { + // Try to receive messages with timeout for periodic ticks + match receiver.recv_timeout(tick_interval) { + Ok(ActorMessage::UserMessage(msg)) => { + message_count += 1; + trace!("[actor_system] Processing message #{}", message_count); + actor.handle(msg); + } + Ok(ActorMessage::Stop) => { + trace!("[actor_system] Received Stop message, breaking loop"); + break; + } + Ok(ActorMessage::Tick) => { + tick_count += 1; + trace!("[actor_system] Received explicit Tick message #{}", tick_count); + actor.tick(); + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + // Periodic tick + if last_tick.elapsed() >= tick_interval { + tick_count += 1; + if tick_count % 10 == 0 { + trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count); + } + actor.tick(); + last_tick = Instant::now(); + } + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + trace!("[actor_system] Channel disconnected, breaking loop"); + // Channel closed, actor should stop + break; + } + } + } + trace!("[actor_system] run_actor_loop ENDED - processed {} messages, {} ticks", message_count, tick_count); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct CounterActor { + count: Arc, + } + + impl Actor for CounterActor { + type Message = usize; + + fn handle(&mut self, msg: Self::Message) { + self.count.fetch_add(msg, Ordering::SeqCst); + } + + fn on_start(&mut self) { + println!("Counter actor started"); + } + + fn on_stop(&mut self) { + println!("Counter actor stopped"); + } + } + + #[test] + fn test_actor_system() { + let thread_pool = Arc::new(ThreadPool::new(4)); + let system = ActorSystem::new(thread_pool); + + let count = Arc::new(AtomicUsize::new(0)); + let actor = CounterActor { + count: count.clone(), + }; + + let handle = system.spawn(actor); + + // Send some messages + handle.send(1).unwrap(); + handle.send(2).unwrap(); + handle.send(3).unwrap(); + + // Give actor time to process + std::thread::sleep(Duration::from_millis(100)); + + assert_eq!(count.load(Ordering::SeqCst), 6); + + handle.stop().unwrap(); + } +} diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs new file mode 100644 index 0000000..7f5ac89 --- /dev/null +++ b/src/actor/peer_actor.rs @@ -0,0 +1,405 @@ +use crate::actor::{Actor, ActorHandle}; +use crate::client::ClientOperation; +use crate::dispatcher::MessageDispatcher; +use crate::message::peer::{ + FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueResponse, TransferRequest, + TransferResponse, UploadFailedHandler, +}; +use crate::message::server::MessageFactory; +use crate::message::{Handlers, Message, MessageReader, MessageType}; +use crate::peer::Peer; +use crate::types::{Download, FileSearchResult, Transfer}; +use crate::{debug, error, trace, warn}; + +use std::io::{self, Write}; +use std::net::TcpStream; +use std::sync::mpsc::{Receiver, Sender}; +use std::sync::{Arc, RwLock}; + +/// Messages that can be sent to a PeerActor +#[derive(Debug, Clone)] +pub enum PeerMessage { + /// Send a raw message to the peer + SendMessage(Message), + /// Handle a file search result + FileSearchResult(FileSearchResult), + /// Handle a transfer request + TransferRequest(Transfer), + /// Handle a transfer response + TransferResponse { + token: u32, + allowed: bool, + reason: Option, + }, + /// Handle a place in queue response + PlaceInQueueResponse { filename: String, place: u32 }, + /// Set the peer's username + SetUsername(String), + /// Queue an upload for a file + QueueUpload(String), + /// Request a transfer for a download + RequestTransfer(Download), + /// Internal: process read operations + ProcessRead, +} + +/// Actor that manages a single peer connection +pub struct PeerActor { + peer: Arc>, + stream: Option, + reader: MessageReader, + client_channel: Sender, + self_handle: Option>, + dispatcher: Option>, + dispatcher_receiver: Option>, +} + +impl PeerActor { + /// Create a new PeerActor with an existing socket connection + pub fn new( + peer: Peer, + stream: TcpStream, + reader: Option, + client_channel: Sender, + ) -> Self { + Self { + peer: Arc::new(RwLock::new(peer)), + stream: Some(stream), + reader: reader.unwrap_or_default(), + client_channel, + self_handle: None, + dispatcher: None, + dispatcher_receiver: None, + } + } + + /// Set the actor's own handle for self-messaging + pub fn set_self_handle(&mut self, handle: ActorHandle) { + self.self_handle = Some(handle); + } + + /// Initialize the message dispatcher with handlers + fn initialize_dispatcher(&mut self) { + // Create a channel for the dispatcher to send messages to + let (dispatcher_sender, dispatcher_receiver) = std::sync::mpsc::channel::(); + + self.dispatcher_receiver = Some(dispatcher_receiver); + + let mut handlers = Handlers::new(); + handlers.register_handler(FileSearchResponse); + handlers.register_handler(TransferRequest); + handlers.register_handler(TransferResponse); + handlers.register_handler(GetShareFileList); + handlers.register_handler(UploadFailedHandler); + handlers.register_handler(PlaceInQueueResponse); + handlers.register_handler(PeerInit); + + self.dispatcher = Some(MessageDispatcher::new( + "peer_actor".to_string(), + dispatcher_sender, + handlers, + )); + } + + /// Process any messages from the dispatcher + fn process_dispatcher_messages(&mut self) { + // Collect all messages first to avoid borrow issues + let messages: Vec = if let Some(ref receiver) = self.dispatcher_receiver { + let mut msgs = Vec::new(); + while let Ok(msg) = receiver.try_recv() { + msgs.push(msg); + } + msgs + } else { + Vec::new() + }; + + // Process all collected messages + for msg in messages { + self.handle_message(msg); + } + } + + /// Internal handler that doesn't call Actor::handle + fn handle_message(&mut self, msg: PeerMessage) { + match msg { + PeerMessage::SendMessage(message) => { + self.send_message(message); + } + PeerMessage::FileSearchResult(file_search) => { + self.client_channel + .send(ClientOperation::SearchResult(file_search)) + .unwrap(); + } + PeerMessage::TransferRequest(transfer) => { + let username = self.peer.read().unwrap().username.clone(); + debug!( + "[peer_actor:{}] TransferRequest for {}", + username, transfer.token + ); + + self.client_channel + .send(ClientOperation::UpdateDownloadTokens( + transfer.clone(), + username.clone(), + )) + .unwrap(); + + let transfer_response = + MessageFactory::build_transfer_response_message(transfer); + + if let Some(ref handle) = self.self_handle { + if let Err(e) = handle.send(PeerMessage::SendMessage(transfer_response)) { + error!("[peer_actor:{}] Failed to send TransferResponse message: {}", username, e); + } + } + } + PeerMessage::TransferResponse { + token, + allowed, + reason, + } => { + let username = self.peer.read().unwrap().username.clone(); + debug!( + "[peer_actor:{}] transfer response token: {} allowed: {}", + username, token, allowed + ); + + if !allowed { + if let Some(reason_text) = reason { + debug!( + "[peer_actor:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...", + username, reason_text, token + ); + } + } else { + debug!( + "[peer_actor:{}] Transfer allowed, ready to connect with token {:}", + username, token + ); + self.client_channel + .send(ClientOperation::DownloadFromPeer( + token, + self.peer.read().unwrap().clone(), + allowed, + )) + .unwrap(); + } + } + PeerMessage::PlaceInQueueResponse { filename, place } => { + let username = self.peer.read().unwrap().username.clone(); + debug!( + "[peer_actor:{}] Place in queue response - file: {}, place: {}", + username, filename, place + ); + } + PeerMessage::SetUsername(username) => { + trace!( + "[peer_actor:{}] SetUsername: {}", + self.peer.read().unwrap().username, + username + ); + self.peer.write().unwrap().username = username; + } + PeerMessage::QueueUpload(filename) => { + let message = MessageFactory::build_queue_upload_message(&filename); + self.send_message(message); + } + PeerMessage::RequestTransfer(download) => { + let message = MessageFactory::build_transfer_request_message( + &download.filename, + download.token, + ); + self.send_message(message); + } + PeerMessage::ProcessRead => { + let username = self.peer.read().unwrap().username.clone(); + trace!("[peer_actor:{}] Handling ProcessRead message", username); + self.process_read(); + } + } + } + + /// Handle reading from the socket and dispatching messages + fn process_read(&mut self) { + let username = self.peer.read().unwrap().username.clone(); + trace!("[peer_actor:{}] process_read() called", username); + + // Try to read more data from the socket first + { + let stream = match self.stream.as_mut() { + Some(s) => s, + None => { + trace!("[peer_actor:{}] process_read: stream is None, returning", username); + return; + } + }; + + match self.reader.read_from_socket(stream) { + Ok(()) => {} + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {} + Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { + let peer_lock = self.peer.read().unwrap(); + debug!( + "Read operation timed out for peer actor {:}:{:}", + peer_lock.host, peer_lock.port + ); + } + Err(e) => { + let username = self.peer.read().unwrap().username.clone(); + error!( + "[peer_actor:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.", + username, e, e.kind() + ); + self.disconnect(); + return; + } + } + } + + // Extract all available messages from the buffer + let username = self.peer.read().unwrap().username.clone(); + let mut extracted_count = 0; + loop { + match self.reader.extract_message() { + Ok(Some(mut message)) => { + extracted_count += 1; + trace!( + "[peer_actor:{}] ← Message #{}: {:?}", + username, + extracted_count, + message + .get_message_name(MessageType::Peer, message.get_message_code() as u32) + .map_err(|e| e.to_string()) + ); + if let Some(ref dispatcher) = self.dispatcher { + dispatcher.dispatch(&mut message); + } else { + warn!("[peer_actor:{}] No dispatcher available!", username); + } + } + Err(e) => { + warn!( + "[peer_actor:{}] Error extracting message: {}. Disconnecting peer.", + username, e + ); + self.disconnect(); + return; + } + Ok(None) => { + trace!("[peer_actor:{}] No more messages in buffer (extracted {} total)", username, extracted_count); + break; + } + } + } + + // Process any messages that were dispatched + trace!("[peer_actor:{}] Processing dispatcher messages", username); + self.process_dispatcher_messages(); + trace!("[peer_actor:{}] process_read() completed", username); + } + + /// Handle writing a message to the socket + fn send_message(&mut self, message: Message) { + let stream = match self.stream.as_mut() { + Some(s) => s, + None => { + error!("Cannot send message: stream is None"); + return; + } + }; + + let username = self.peer.read().unwrap().username.clone(); + trace!( + "[peer_actor:{}] ➡ {:?}", + username, + message + .get_message_name( + MessageType::Peer, + u32::from_le_bytes(message.get_slice(0, 4).try_into().unwrap()) + ) + .map_err(|e| e.to_string()) + ); + + if let Err(e) = stream.write_all(&message.get_buffer()) { + error!( + "[peer_actor:{}] Error writing message: {}. Disconnecting.", + username, e + ); + self.disconnect(); + return; + } + + if let Err(e) = stream.flush() { + error!( + "[peer_actor:{}] Error flushing stream: {}. Disconnecting.", + username, e + ); + self.disconnect(); + } + } + + /// Disconnect the peer and notify the client + fn disconnect(&mut self) { + let username = self.peer.read().unwrap().username.clone(); + debug!("[peer_actor:{}] disconnect", username); + + self.stream.take(); + + if let Err(e) = self + .client_channel + .send(ClientOperation::PeerDisconnected(username)) + { + error!("Failed to send disconnect notification: {}", e); + } + } +} + +impl Actor for PeerActor { + type Message = PeerMessage; + + fn handle(&mut self, msg: Self::Message) { + self.handle_message(msg); + } + + fn on_start(&mut self) { + let username = self.peer.read().unwrap().username.clone(); + trace!("[peer_actor:{}] actor starting", username); + + // Initialize the dispatcher + self.initialize_dispatcher(); + trace!("[peer_actor:{}] dispatcher initialized", username); + + // Trigger initial read + if let Some(ref handle) = self.self_handle { + trace!("[peer_actor:{}] Sending initial ProcessRead message", username); + match handle.send(PeerMessage::ProcessRead) { + Ok(_) => trace!("[peer_actor:{}] ProcessRead message sent successfully", username), + Err(e) => error!("[peer_actor:{}] FAILED to send ProcessRead: {}", username, e), + } + } else { + error!("[peer_actor:{}] self_handle is None! Cannot send ProcessRead", username); + } + + // Also do an immediate read as fallback + trace!("[peer_actor:{}] Performing immediate process_read as fallback", username); + self.process_read(); + } + + fn on_stop(&mut self) { + let username = self.peer.read().unwrap().username.clone(); + trace!("[peer_actor:{}] actor stopping", username); + self.disconnect(); + } + + fn tick(&mut self) { + // Periodically check for new data on the socket + if self.stream.is_some() { + self.process_read(); + } else { + let username = self.peer.read().unwrap().username.clone(); + trace!("[peer_actor:{}] tick() called but stream is None", username); + } + } +} + diff --git a/src/actor/peer_registry.rs b/src/actor/peer_registry.rs new file mode 100644 index 0000000..02de074 --- /dev/null +++ b/src/actor/peer_registry.rs @@ -0,0 +1,119 @@ +use crate::actor::{ActorHandle, ActorSystem}; +use crate::actor::peer_actor::{PeerActor, PeerMessage}; +use crate::client::ClientOperation; +use crate::peer::Peer; +use crate::message::MessageReader; +use crate::{debug, trace}; + +use std::collections::HashMap; +use std::net::TcpStream; +use std::sync::mpsc::Sender; +use std::sync::{Arc, Mutex}; + +/// Registry for managing peer actors +pub struct PeerRegistry { + peers: Arc>>>, + actor_system: Arc, + client_channel: Sender, +} + +impl PeerRegistry { + pub fn new(actor_system: Arc, client_channel: Sender) -> Self { + Self { + peers: Arc::new(Mutex::new(HashMap::new())), + actor_system, + client_channel, + } + } + + /// Spawn a new peer actor and register it + pub fn register_peer( + &self, + peer: Peer, + stream: TcpStream, + reader: Option, + ) -> Result, String> { + let username = peer.username.clone(); + + debug!("[peer_registry] Registering peer actor for {}", username); + + let actor = PeerActor::new( + peer, + stream, + reader, + self.client_channel.clone(), + ); + + // Spawn the actor with initialization callback to set self_handle + let handle = self.actor_system.spawn_with_handle(actor, |actor, handle| { + actor.set_self_handle(handle); + }); + + // Store in registry + let mut peers = self.peers.lock().unwrap(); + peers.insert(username.clone(), handle.clone()); + + trace!("[peer_registry] Peer actor {} registered successfully", username); + + Ok(handle) + } + + /// Get a peer actor handle by username + pub fn get_peer(&self, username: &str) -> Option> { + let peers = self.peers.lock().unwrap(); + peers.get(username).cloned() + } + + /// Remove a peer actor from the registry + pub fn remove_peer(&self, username: &str) -> Option> { + let mut peers = self.peers.lock().unwrap(); + let handle = peers.remove(username); + + if handle.is_some() { + debug!("[peer_registry] Removed peer actor for {}", username); + } + + handle + } + + /// Get all registered peer usernames + pub fn get_all_usernames(&self) -> Vec { + let peers = self.peers.lock().unwrap(); + peers.keys().cloned().collect() + } + + /// Get count of registered peers + pub fn count(&self) -> usize { + let peers = self.peers.lock().unwrap(); + peers.len() + } + + /// Check if a peer is registered + pub fn contains(&self, username: &str) -> bool { + let peers = self.peers.lock().unwrap(); + peers.contains_key(username) + } + + /// Send a message to a peer by username + pub fn send_to_peer(&self, username: &str, message: PeerMessage) -> Result<(), String> { + let handle = self.get_peer(username) + .ok_or_else(|| format!("Peer {} not found in registry", username))?; + + handle.send(message) + } + + /// Queue an upload for a peer + pub fn queue_upload(&self, username: &str, filename: String) -> Result<(), String> { + self.send_to_peer(username, PeerMessage::QueueUpload(filename)) + } +} + +impl Clone for PeerRegistry { + fn clone(&self) -> Self { + Self { + peers: self.peers.clone(), + actor_system: self.actor_system.clone(), + client_channel: self.client_channel.clone(), + } + } +} diff --git a/src/client.rs b/src/client.rs index 1dda7ed..c87cc64 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,8 +1,9 @@ use crate::types::{DownloadResult, DownloadStatus}; use crate::{ + actor::{ActorSystem, peer_registry::PeerRegistry}, error::{Result, SoulseekRs}, peer::{ - listen::Listen, ConnectionType, DefaultPeer, DownloadPeer, NewPeer, + listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer, }, server::{PeerAddress, Server, ServerOperation}, @@ -25,7 +26,6 @@ use std::{ }; use crate::{debug, error, info, trace}; -const MAX_THREADS: usize = 50; const DEFALT_LISTEN_PORT: u32 = 2234; pub enum ClientOperation { @@ -45,12 +45,13 @@ pub enum ClientOperation { }, } pub struct ClientContext { - peers: HashMap, + pub peer_registry: Option, sender: Option>, server_sender: Option>, search_results: Vec, pub download_tokens: HashMap, - thread_pool: ThreadPool, + thread_pool: Arc, + actor_system: Arc, } impl Default for ClientContext { fn default() -> Self { @@ -61,13 +62,21 @@ impl Default for ClientContext { impl ClientContext { #[must_use] pub fn new() -> Self { + let max_threads = thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(8); + + let thread_pool = Arc::new(ThreadPool::new(max_threads)); + let actor_system = Arc::new(ActorSystem::new(thread_pool.clone())); + Self { - peers: HashMap::new(), + peer_registry: None, sender: None, server_sender: None, search_results: Vec::new(), download_tokens: HashMap::new(), - thread_pool: ThreadPool::new(MAX_THREADS), + thread_pool, + actor_system, } } } @@ -120,10 +129,20 @@ impl Client { ) = mpsc::channel(); let listen_port = self.listen_port; - self.context.write().unwrap().sender = Some(sender.clone()); let context = self.context.clone(); let own_username = self.username.clone(); + // Initialize PeerRegistry with the actor system + { + let mut ctx = self.context.write().unwrap(); + ctx.sender = Some(sender.clone()); + let peer_registry = PeerRegistry::new( + ctx.actor_system.clone(), + sender.clone(), + ); + ctx.peer_registry = Some(peer_registry); + } + let client_sender = sender.clone(); self.server = match Server::new( @@ -181,9 +200,11 @@ impl Client { #[allow(dead_code)] pub fn remove_peer(&self, username: &str) { - let mut context = self.context.write().unwrap(); - if let Some(peer) = context.peers.remove(username) { - drop(peer); + let context = self.context.read().unwrap(); + if let Some(ref registry) = context.peer_registry { + if let Some(handle) = registry.remove_peer(username) { + let _ = handle.stop(); + } } } @@ -239,11 +260,12 @@ impl Client { let mut context = self.context.write().unwrap(); context.download_tokens.insert(token, download.clone()); - let download_initiated = context - .peers - .get(&username) - .map(|p| p.queue_upload(download.filename)) - .is_some(); + + let download_initiated = if let Some(ref registry) = context.peer_registry { + registry.queue_upload(&username, download.filename.clone()).is_ok() + } else { + false + }; drop(context); @@ -291,9 +313,11 @@ impl Client { .push(file_search.clone()); } ClientOperation::PeerDisconnected(username) => { - let mut context = client_context.write().unwrap(); - if let Some(peer) = context.peers.remove(&username) { - drop(peer); // Explicitly drop to trigger cleanup + let context = client_context.read().unwrap(); + if let Some(ref registry) = context.peer_registry { + if let Some(handle) = registry.remove_peer(&username) { + let _ = handle.stop(); + } } } ClientOperation::PierceFireWall(peer) => { @@ -364,12 +388,15 @@ impl Client { }; } ClientOperation::NewPeer(new_peer) => { - if client_context + let peer_exists = client_context .read() .unwrap() - .peers - .contains_key(&new_peer.username) - { + .peer_registry + .as_ref() + .map(|r| r.contains(&new_peer.username)) + .unwrap_or(false); + + if peer_exists { debug!( "Already connected to {}", new_peer.username @@ -418,9 +445,13 @@ impl Client { username, host, port, obfuscation_type, obfuscated_port ); - if let Some(_peer) = - client_context.read().unwrap().peers.get(&username) - { + let peer_exists = client_context.read().unwrap() + .peer_registry + .as_ref() + .map(|r| r.contains(&username)) + .unwrap_or(false); + + if peer_exists { // don't know if i should update? and or reconnect the peer // debug!( // "existing peer: {:?}, new peer details: @@ -517,29 +548,75 @@ impl Client { trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); match peer.connection_type { ConnectionType::P => { - let default_peer = - DefaultPeer::new(peer_clone, sender_clone); + let username = peer.username.clone(); - let connect_result = match stream { - Some(s) => default_peer.connect_with_socket(s, None), - _ => default_peer.connect() - }; + // Establish TCP connection or use provided stream + let mut tcp_stream = match stream { + Some(s) => s, + None => { + let socket_addr = format!("{}:{}", peer_clone.host, peer_clone.port) + .parse::(); - match connect_result { - Ok(p) => { - trace!("[client] connected to: {}", peer.username); - client_context2.write().unwrap().peers.insert(peer.username, p); + match socket_addr { + Ok(addr) => { + match TcpStream::connect_timeout(&addr, Duration::from_secs(20)) { + Ok(s) => s, + Err(e) => { + trace!( + "Can't connect to {:?} {:?}:{:?} - {:?}", + username, peer_clone.host, peer_clone.port, e + ); + return; + } + } + } + Err(e) => { + trace!("Invalid socket address for {:?}: {:?}", username, e); + return; + } + } } - Err(e) => { - trace!( - "Can't connect to {:?} {:?}:{:?} {:?} - {:?}", - peer.username, - peer.host, - peer.port, - peer.connection_type, - e - ); + }; + + // Send connection handshake message + use std::io::Write; + if let Some(token) = peer_clone.token { + // For indirect connections, send watch user message + let handshake_msg = crate::message::server::MessageFactory::build_watch_user(token); + if let Err(e) = tcp_stream.write_all(&handshake_msg.get_data()) { + trace!("Failed to send watch user handshake to {:?}: {:?}", username, e); + return; } + trace!("[client] Sent watch user handshake for token: {}", token); + } + + // Configure socket + if let Err(e) = tcp_stream.set_read_timeout(Some(Duration::from_secs(5))) { + trace!("Failed to set read timeout for {:?}: {:?}", username, e); + return; + } + if let Err(e) = tcp_stream.set_write_timeout(Some(Duration::from_secs(5))) { + trace!("Failed to set write timeout for {:?}: {:?}", username, e); + return; + } + if let Err(e) = tcp_stream.set_nodelay(true) { + trace!("Failed to set nodelay for {:?}: {:?}", username, e); + return; + } + + // Spawn peer actor + let context = client_context2.read().unwrap(); + if let Some(ref registry) = context.peer_registry { + match registry.register_peer(peer_clone, tcp_stream, None) { + Ok(_) => { + trace!("[client] peer actor spawned for: {}", username); + } + Err(e) => { + trace!("Failed to spawn peer actor for {:?}: {:?}", username, e); + } + } + } else { + trace!("PeerRegistry not initialized"); } } diff --git a/src/lib.rs b/src/lib.rs index 558d920..7bedec5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,5 @@ // Core modules +pub mod actor; pub mod client; pub mod dispatcher; pub mod error; diff --git a/src/main.rs b/src/main.rs index 64c7710..b9d8880 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,7 +6,7 @@ fn main() { PeerAddress::new(String::from("server.slsknet.org"), 2242), String::from("xxxxxx"), String::from("xxxxxx"), - true, + false, None, ); diff --git a/src/message/handlers.rs b/src/message/handlers.rs index c8f4432..57f1e99 100644 --- a/src/message/handlers.rs +++ b/src/message/handlers.rs @@ -3,12 +3,12 @@ use std::collections::HashMap; use crate::message::Message; use std::sync::mpsc::Sender; -pub trait MessageHandler { +pub trait MessageHandler: Send { fn get_code(&self) -> u8; fn handle(&self, message: &mut Message, sender: Sender); } pub struct Handlers { - handlers: HashMap>>, + handlers: HashMap + Send>>, } impl Default for Handlers { @@ -31,7 +31,7 @@ impl Handlers { self.handlers.insert(handler.get_code(), Box::new(handler)); self } - pub fn get_handler(&self, code: u8) -> Option<&dyn MessageHandler> { + pub fn get_handler(&self, code: u8) -> Option<&(dyn MessageHandler + Send)> { self.handlers.get(&code).map(|v| &**v) } } diff --git a/src/message/peer/file_search_response.rs b/src/message/peer/file_search_response.rs index 539c171..33f09f1 100644 --- a/src/message/peer/file_search_response.rs +++ b/src/message/peer/file_search_response.rs @@ -1,21 +1,21 @@ use crate::message::{Message, MessageHandler}; -use crate::peer::PeerOperation; +use crate::peer::PeerMessage; use crate::types::FileSearchResult; use std::sync::mpsc::Sender; pub struct FileSearchResponse; -impl MessageHandler for FileSearchResponse { +impl MessageHandler for FileSearchResponse { fn get_code(&self) -> u8 { 9 } - fn handle(&self, message: &mut Message, sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { let file_search = match FileSearchResult::new_from_message(message) { Ok(result) => result, Err(_) => return, // Skip malformed search results }; sender - .send(PeerOperation::FileSearchResult(file_search)) + .send(PeerMessage::FileSearchResult(file_search)) .unwrap(); } } diff --git a/src/message/peer/get_share_file_list.rs b/src/message/peer/get_share_file_list.rs index 60f7d11..683b94f 100644 --- a/src/message/peer/get_share_file_list.rs +++ b/src/message/peer/get_share_file_list.rs @@ -1,17 +1,17 @@ use crate::{ message::{server::MessageFactory, Message, MessageHandler}, - peer::PeerOperation, + peer::PeerMessage, }; use std::sync::mpsc::Sender; pub struct GetShareFileList; -impl MessageHandler for GetShareFileList { +impl MessageHandler for GetShareFileList { fn get_code(&self) -> u8 { 4 } - fn handle(&self, _message: &mut Message, sender: Sender) { + fn handle(&self, _message: &mut Message, sender: Sender) { let message = MessageFactory::build_shared_folders_message(100, 800); - sender.send(PeerOperation::SendMessage(message)).unwrap(); + sender.send(PeerMessage::SendMessage(message)).unwrap(); } } diff --git a/src/message/peer/peer_init.rs b/src/message/peer/peer_init.rs index 5d37f77..ea1ff0a 100644 --- a/src/message/peer/peer_init.rs +++ b/src/message/peer/peer_init.rs @@ -2,17 +2,17 @@ use std::sync::mpsc::Sender; use crate::{ message::{Message, MessageHandler}, - peer::PeerOperation, + peer::PeerMessage, trace, }; pub struct PeerInit; -impl MessageHandler for PeerInit { +impl MessageHandler for PeerInit { fn get_code(&self) -> u8 { 1 } - fn handle(&self, message: &mut Message, sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { message.set_pointer(4); let _message_code = message.read_int8(); let username = message.read_string(); @@ -25,6 +25,6 @@ impl MessageHandler for PeerInit { token ); - sender.send(PeerOperation::SetUsername(username)).unwrap(); + sender.send(PeerMessage::SetUsername(username)).unwrap(); } } diff --git a/src/message/peer/place_in_queue_response.rs b/src/message/peer/place_in_queue_response.rs index 08ac7cf..0d4a38e 100644 --- a/src/message/peer/place_in_queue_response.rs +++ b/src/message/peer/place_in_queue_response.rs @@ -1,22 +1,22 @@ use crate::{ message::{Message, MessageHandler}, - peer::PeerOperation, + peer::PeerMessage, }; use std::sync::mpsc::Sender; pub struct PlaceInQueueResponse; -impl MessageHandler for PlaceInQueueResponse { +impl MessageHandler for PlaceInQueueResponse { fn get_code(&self) -> u8 { 43 } - fn handle(&self, message: &mut Message, sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { let filename = message.read_string(); let place = message.read_int32(); sender - .send(PeerOperation::PlaceInQueueResponse { filename, place }) + .send(PeerMessage::PlaceInQueueResponse { filename, place }) .unwrap(); } } diff --git a/src/message/peer/transfer_request.rs b/src/message/peer/transfer_request.rs index 828b6b0..914bc0e 100644 --- a/src/message/peer/transfer_request.rs +++ b/src/message/peer/transfer_request.rs @@ -1,20 +1,20 @@ use crate::{ message::{Message, MessageHandler}, - peer::PeerOperation, + peer::PeerMessage, types::Transfer, }; use std::sync::mpsc::Sender; pub struct TransferRequest; -impl MessageHandler for TransferRequest { +impl MessageHandler for TransferRequest { fn get_code(&self) -> u8 { 40 } - fn handle(&self, message: &mut Message, sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { let transfer = Transfer::new_from_message(message); sender - .send(PeerOperation::TransferRequest(transfer)) + .send(PeerMessage::TransferRequest(transfer)) .unwrap(); } } diff --git a/src/message/peer/transfer_response.rs b/src/message/peer/transfer_response.rs index baa19d2..7191b09 100644 --- a/src/message/peer/transfer_response.rs +++ b/src/message/peer/transfer_response.rs @@ -1,23 +1,23 @@ use crate::{ message::{Message, MessageHandler}, - peer::PeerOperation, + peer::PeerMessage, }; use std::sync::mpsc::Sender; pub struct TransferResponse; -impl MessageHandler for TransferResponse { +impl MessageHandler for TransferResponse { fn get_code(&self) -> u8 { 41 } - fn handle(&self, message: &mut Message, sender: Sender) { + fn handle(&self, message: &mut Message, sender: Sender) { let token = message.read_int32(); let allowed = message.read_int8(); let reason = (allowed == 0).then(|| message.read_string()); sender - .send(PeerOperation::TransferResponse { + .send(PeerMessage::TransferResponse { token, allowed: allowed == 1, reason, diff --git a/src/message/peer/upload_failed.rs b/src/message/peer/upload_failed.rs index d03ea53..27f1ec9 100644 --- a/src/message/peer/upload_failed.rs +++ b/src/message/peer/upload_failed.rs @@ -1,17 +1,17 @@ use crate::info; use crate::{ message::{Message, MessageHandler}, - peer::PeerOperation, + peer::PeerMessage, types::UploadFailed, }; use std::sync::mpsc::Sender; pub struct UploadFailedHandler; -impl MessageHandler for UploadFailedHandler { +impl MessageHandler for UploadFailedHandler { fn get_code(&self) -> u8 { 46 } - fn handle(&self, message: &mut Message, _sender: Sender) { + fn handle(&self, message: &mut Message, _sender: Sender) { let upload_failed = UploadFailed::new_from_message(message); info!("Upload failed for ${}", upload_failed.filename); } diff --git a/src/peer/default_peer.rs b/src/peer/default_peer.rs deleted file mode 100644 index 85faac3..0000000 --- a/src/peer/default_peer.rs +++ /dev/null @@ -1,450 +0,0 @@ -use crate::dispatcher::MessageDispatcher; -use crate::message::peer::{FileSearchResponse, PeerInit}; -use crate::message::peer::{ - GetShareFileList, PlaceInQueueResponse, TransferRequest, TransferResponse, - UploadFailedHandler, -}; -use crate::message::server::MessageFactory; -use crate::message::{Handlers, Message, MessageReader, MessageType}; -use crate::types::{Download, FileSearchResult, Transfer}; - -use core::result::Result::Ok; -use std::sync::mpsc::{Receiver, Sender}; -use std::sync::{mpsc, Arc, RwLock}; -use std::thread::{self, JoinHandle}; - -use crate::client::ClientOperation; -use crate::peer::Peer; -use crate::{debug, error, trace, warn}; -use std::io::{self, Write}; -use std::net::TcpStream; -use std::net::ToSocketAddrs; -use std::time::Duration; - -#[allow(dead_code)] -#[derive(Debug)] -pub struct DefaultPeer { - peer: Arc>, - peer_channel: Option>, - client_channel: Sender, - read_thread: Option>, - write_thread: Option>, -} - -#[allow(dead_code)] -pub enum PeerOperation { - SendMessage(Message), - FileSearchResult(FileSearchResult), - TransferRequest(Transfer), - TransferResponse { - token: u32, - allowed: bool, - reason: Option, - }, - PlaceInQueueResponse { - filename: String, - place: u32, - }, - SetUsername(String), -} - -impl DefaultPeer { - pub fn new(peer: Peer, client_channel: Sender) -> Self { - Self { - peer: Arc::new(RwLock::new(peer)), - peer_channel: None, - client_channel, - read_thread: None, - write_thread: None, - } - } - pub fn disconnect(&mut self) { - let username = self.peer.read().unwrap().username.clone(); - debug!("[default_peer:{}] disconnect", username); - if let Err(e) = self - .client_channel - .send(ClientOperation::PeerDisconnected(username)) - { - error!("Failed to send disconnect notification: {}", e); - } - - self.peer_channel.take(); - } - - pub fn connect_with_socket( - mut self, - mut stream: TcpStream, - reader: Option, - ) -> Result { - if let Some(token) = self.peer.read().unwrap().token { - let mut message: Vec = [0, 5, 0, 0, 0, 0].to_vec(); - message.extend_from_slice(&token.to_le_bytes()); - stream.write_all(&message).unwrap(); - } - - stream.set_read_timeout(Some(Duration::from_secs(5)))?; - stream.set_write_timeout(Some(Duration::from_secs(5)))?; - stream.set_nodelay(true)?; - - trace!( - "[default_peer:{}] connect_with_socket: direct", - self.peer.read().unwrap().username - ); - self.start_read_write_loops(stream, reader)?; - - Ok(self) - } - - pub fn connect(mut self) -> Result { - { - let peer = self.peer.read().unwrap(); - trace!( - "[default_peer] Connecting to {} on port {}", - peer.host, - peer.port - ); - } - let socket_address = { - let peer = self.peer.read().unwrap(); - format!("{}:{}", peer.host, peer.port) - } - .to_socket_addrs()? - .next() - .ok_or_else(|| { - io::Error::new(io::ErrorKind::InvalidInput, "Invalid address") - })?; - - trace!( - "[default_peer:{}] connect: indirect", - self.peer.read().unwrap().username - ); - let mut stream = TcpStream::connect_timeout( - &socket_address, - Duration::from_secs(20), - )?; - - stream.set_read_timeout(Some(Duration::from_secs(5)))?; - stream.set_write_timeout(Some(Duration::from_secs(5)))?; - stream.set_nodelay(true)?; - - if let Some(token) = self.peer.read().unwrap().token { - stream - .write_all(&MessageFactory::build_watch_user(token).get_data()) - .unwrap(); - } - self.start_read_write_loops(stream, None)?; - - Ok(self) - } - - fn start_read_write_loops( - &mut self, - stream: TcpStream, - reader: Option, - ) -> Result<(), io::Error> { - let (peer_sender, peer_reader): ( - Sender, - Receiver, - ) = mpsc::channel(); - - self.peer_channel = Some(peer_sender.clone()); - - let mut write_stream = match stream.try_clone() { - Ok(s) => s, - Err(e) => { - error!( - "[default_peer:{}] Failed to clone stream: {}", - self.peer.read().unwrap().username, - e - ); - return Err(e); - } - }; - let mut read_stream = stream; - - let peer = self.peer.clone(); - let peer_clone = self.peer.clone(); - let client_channel_for_read = self.client_channel.clone(); - - let mut buffered_reader = reader.unwrap_or_default(); - let buffer_size = buffered_reader.buffer_len(); - - self.read_thread = Some(thread::spawn(move || { - let mut handlers = Handlers::new(); - handlers.register_handler(FileSearchResponse); - handlers.register_handler(TransferRequest); - handlers.register_handler(TransferResponse); - handlers.register_handler(GetShareFileList); - handlers.register_handler(UploadFailedHandler); - handlers.register_handler(PlaceInQueueResponse); - handlers.register_handler(PeerInit); - - let dispatcher = MessageDispatcher::new( - "default_peer".to_string(), - peer_sender, - handlers, - ); - trace!( - "[default_peer:{}] read thread starting", - peer.read().unwrap().username - ); - - if buffer_size > 0 { - trace!( - "[default_peer:{}] starting with {} buffered bytes", - peer.read().unwrap().username, - buffer_size - ); - } - - loop { - let mut should_terminate = false; - loop { - match buffered_reader.extract_message() { - Ok(Some(mut message)) => { - trace!( - "[default_peer:{}] ← {:?}", - peer.read().unwrap().username, - message - .get_message_name( - MessageType::Peer, - message.get_message_code() as u32 - ) - .map_err(|e| e.to_string()) - ); - dispatcher.dispatch(&mut message) - } - Err(e) => { - warn!( - "Error extracting message in default peer: {}. Terminating read loop.", - e - ); - let username = - peer.read().unwrap().username.clone(); - let _ = client_channel_for_read.send( - ClientOperation::PeerDisconnected(username), - ); - should_terminate = true; - break; - } - Ok(None) => { - break; - } - } - } - - if should_terminate { - break; - } - - match buffered_reader.read_from_socket(&mut read_stream) { - Ok(()) => continue, - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { - continue - } - - Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { - { - let peer_lock = peer.read().unwrap(); - debug!( - "Read operation timed out in default peer {:}:{:}", - peer_lock.host, peer_lock.port - ); - } - continue; - } - Err(e) => { - let username = peer.read().unwrap().username.clone(); - error!("[default_peer:{}] Error reading from peer: {} (kind: {:?}). Terminating read loop.", username, e, e.kind()); - let _ = client_channel_for_read - .send(ClientOperation::PeerDisconnected(username)); - break; - } - } - } - trace!( - "[default_peer:{}] read thread exiting", - peer.read().unwrap().username - ); - })); - - let client_channel = self.client_channel.clone(); - let peer_channel = self.peer_channel.clone(); - let peer_clone_for_write = self.peer.clone(); - - self.write_thread = Some(thread::spawn(move || loop { - match peer_reader.recv() { - Ok(operation) => match operation { - PeerOperation::SendMessage(message) => { - trace!( - "[default_peer:{}] ➡ {:?} - {:?}", - peer_clone_for_write.read().unwrap().username, - message - .get_message_name( - MessageType::Peer, - u32::from_le_bytes( - message - .get_slice(0, 4) - .try_into() - .unwrap() - ) - ) - .map_err(|e| e.to_string()), - u32::from_le_bytes( - message.get_slice(0, 4).try_into().unwrap() - ) - ); - - if let Err(e) = - write_stream.write_all(&message.get_buffer()) - { - error!("Error writing message to stream: {} - {}. Terminating write loop.", peer_clone_for_write.read().unwrap().username, e); - let _ = client_channel.send( - ClientOperation::PeerDisconnected( - peer_clone_for_write - .read() - .unwrap() - .username - .clone(), - ), - ); - break; - } - write_stream.flush().unwrap(); - } - PeerOperation::FileSearchResult(file_search) => { - client_channel - .send(ClientOperation::SearchResult(file_search)) - .unwrap(); - } - PeerOperation::TransferRequest(transfer) => { - debug!( - "[default_peer:{}] TransferRequest for {}", - peer_clone_for_write.read().unwrap().username, - transfer.token - ); - client_channel - .send(ClientOperation::UpdateDownloadTokens( - transfer.clone(), - peer_clone.read().unwrap().username.clone(), - )) - .unwrap(); - - let transfer_response = - MessageFactory::build_transfer_response_message( - transfer.clone(), - ); - - trace!( - "[default_peer:{}] TransferResponse for {:?}", - peer_clone_for_write.read().unwrap().username, - transfer_response.get_buffer() - ); - if let Some(sender) = peer_channel.clone() { - sender - .send(PeerOperation::SendMessage( - transfer_response, - )) - .unwrap(); - } - } - PeerOperation::TransferResponse { - token, - allowed, - reason, - } => { - debug!( - "[default_peer:{}] transfer response token: {} allowed: {}", - peer_clone_for_write.read().unwrap().username, token, allowed - ); - - if !allowed { - if let Some(reason_text) = reason { - debug!( - "[default_peer:{}] Transfer rejected: {} - token {}, I will receive TransferRequest soon...", - peer_clone_for_write.read().unwrap().username.clone(), - reason_text, - token - ); - } - } else { - debug!("[default_peer:{}] Transfer allowed, ready to connect with token {:}",peer_clone_for_write.read().unwrap().username, token); - client_channel - .send(ClientOperation::DownloadFromPeer( - token, - peer_clone.read().unwrap().clone(), - allowed, - )) - .unwrap(); - } - } - PeerOperation::PlaceInQueueResponse { filename, place } => { - debug!( - "[default_peer:{}] Place in queue response - file: {}, place: {}", - peer_clone_for_write.read().unwrap().username, filename, place - ); - } - PeerOperation::SetUsername(username) => { - trace!( - "[default_peer:{}] SetUsername: {}", - peer_clone_for_write.read().unwrap().username, - username - ); - - peer_clone.write().unwrap().username = username; - } - }, - Err(_) => { - debug!("[default_peer:{}] Peer channel closed. Terminating write loop.", peer_clone_for_write.read().unwrap().username); - break; - } - } - })); - - Ok(()) - } - - pub fn queue_upload(&self, filename: String) -> Result<(), io::Error> { - let message = MessageFactory::build_queue_upload_message(&filename); - if let Some(sender) = &self.peer_channel { - sender.send(PeerOperation::SendMessage(message)).unwrap(); - } - Ok(()) - } - pub fn transfer_request( - &self, - download: Download, - ) -> Result<(), io::Error> { - let message = MessageFactory::build_transfer_request_message( - &download.filename, - download.token, - ); - if let Some(sender) = &self.peer_channel { - sender.send(PeerOperation::SendMessage(message)).unwrap(); - } - Ok(()) - } -} - -impl Drop for DefaultPeer { - fn drop(&mut self) { - self.peer_channel = None; - trace!("[default_peer:{}] drop", self.peer.read().unwrap().username); - - if let Some(handle) = self.read_thread.take() { - debug!("Joining read thread..."); - match handle.join() { - Ok(_) => debug!("Read thread joined successfully."), - Err(e) => error!("Read thread panicked: {:?}", e), - } - } - - if let Some(handle) = self.write_thread.take() { - debug!("Joining write thread..."); - match handle.join() { - Ok(_) => debug!("Write thread joined successfully."), - Err(e) => error!("Write thread panicked: {:?}", e), - } - } - } -} diff --git a/src/peer/listen.rs b/src/peer/listen.rs index 82a2918..dd58b10 100644 --- a/src/peer/listen.rs +++ b/src/peer/listen.rs @@ -7,7 +7,7 @@ use std::thread; use crate::client::{ClientContext, ClientOperation}; use crate::message::{Message, MessageReader}; -use crate::peer::{ConnectionType, DefaultPeer, DownloadPeer, Peer}; +use crate::peer::{ConnectionType, DownloadPeer, Peer}; use crate::types::Download; use crate::{debug, error, info, trace, DownloadStatus}; @@ -105,17 +105,25 @@ fn handle_peer_connection( peer: Peer, stream: TcpStream, reader: MessageReader, - client_sender: Sender, + context: &ConnectionContext, peer_ip: &str, peer_port: u16, ) { debug!("[listener:{peer_ip}:{peer_port}] connection type is P, reader buffer has {} bytes", reader.buffer_len()); - let default_peer = DefaultPeer::new(peer, client_sender); - if let Ok(default_peer) = - default_peer.connect_with_socket(stream, Some(reader)) - { - drop(default_peer); + // Spawn peer actor directly using the peer registry + let client_context = context.client_context.read().unwrap(); + if let Some(ref registry) = client_context.peer_registry { + match registry.register_peer(peer.clone(), stream, Some(reader)) { + Ok(_) => { + trace!("[listener] peer actor spawned for: {}", peer.username); + } + Err(e) => { + error!("Failed to spawn peer actor for {:?}: {:?}", peer.username, e); + } + } + } else { + error!("PeerRegistry not initialized"); } } @@ -216,7 +224,7 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { peer, stream, reader, - context.client_sender, + &context, &peer_ip, peer_port, ), diff --git a/src/peer/mod.rs b/src/peer/mod.rs index 7032047..9386c18 100644 --- a/src/peer/mod.rs +++ b/src/peer/mod.rs @@ -1,8 +1,10 @@ -mod default_peer; mod download_peer; pub mod listen; -pub use default_peer::{DefaultPeer, PeerOperation}; +// Export actor types +pub use crate::actor::peer_actor::{PeerActor, PeerMessage}; +pub use crate::actor::peer_registry::PeerRegistry; + pub use download_peer::DownloadPeer; use crate::message::Message; From 2584428170e1ad7bb97f0c2a3ccc656b85529dac Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 13:59:35 +0100 Subject: [PATCH 3/9] chore: linting --- src/actor/mod.rs | 48 +++++++++++----- src/actor/peer_actor.rs | 83 +++++++++++++++++++--------- src/actor/peer_registry.rs | 51 +++++++++++------ src/client.rs | 37 +++++++------ src/main.rs | 2 +- src/message/handlers.rs | 5 +- src/message/peer/transfer_request.rs | 4 +- src/peer/listen.rs | 13 ++--- 8 files changed, 156 insertions(+), 87 deletions(-) diff --git a/src/actor/mod.rs b/src/actor/mod.rs index 7587f7e..f56c2d8 100644 --- a/src/actor/mod.rs +++ b/src/actor/mod.rs @@ -5,8 +5,8 @@ use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; use std::time::{Duration, Instant}; -use crate::utils::thread_pool::ThreadPool; use crate::trace; +use crate::utils::thread_pool::ThreadPool; pub mod peer_actor; pub mod peer_registry; @@ -54,6 +54,7 @@ impl ActorHandle { pub(crate) enum ActorMessage { UserMessage(M), Stop, + #[allow(dead_code)] Tick, } @@ -70,9 +71,9 @@ impl ActorSystem { /// Spawn a new actor and return its handle pub fn spawn(&self, mut actor: A) -> ActorHandle { let (sender, receiver) = channel::>(); - let handle = ActorHandle { sender: sender.clone() }; - - let thread_pool = self.thread_pool.clone(); + let handle = ActorHandle { + sender: sender.clone(), + }; // Start the actor event loop on the thread pool self.thread_pool.execute(move || { @@ -86,23 +87,29 @@ impl ActorSystem { /// Spawn a new actor with initialization callback and return its handle /// The callback receives the actor handle before on_start is called - pub fn spawn_with_handle(&self, mut actor: A, init: F) -> ActorHandle + pub fn spawn_with_handle( + &self, + mut actor: A, + init: F, + ) -> ActorHandle where F: FnOnce(&mut A, ActorHandle) + Send + 'static, { let (sender, receiver) = channel::>(); - let handle = ActorHandle { sender: sender.clone() }; + let handle = ActorHandle { + sender: sender.clone(), + }; let handle_for_init = handle.clone(); - let thread_pool = self.thread_pool.clone(); - // Start the actor event loop on the thread pool self.thread_pool.execute(move || { trace!("[actor_system] Actor task starting on thread pool"); init(&mut actor, handle_for_init); trace!("[actor_system] Actor init callback completed"); actor.on_start(); - trace!("[actor_system] Actor on_start completed, starting event loop"); + trace!( + "[actor_system] Actor on_start completed, starting event loop" + ); Self::run_actor_loop(&mut actor, receiver); trace!("[actor_system] Actor event loop ended"); actor.on_stop(); @@ -113,7 +120,10 @@ impl ActorSystem { } /// Run the actor's message processing loop - fn run_actor_loop(actor: &mut A, receiver: Receiver>) { + fn run_actor_loop( + actor: &mut A, + receiver: Receiver>, + ) { trace!("[actor_system] run_actor_loop STARTED"); let tick_interval = Duration::from_millis(100); let mut last_tick = Instant::now(); @@ -125,16 +135,24 @@ impl ActorSystem { match receiver.recv_timeout(tick_interval) { Ok(ActorMessage::UserMessage(msg)) => { message_count += 1; - trace!("[actor_system] Processing message #{}", message_count); + trace!( + "[actor_system] Processing message #{}", + message_count + ); actor.handle(msg); } Ok(ActorMessage::Stop) => { - trace!("[actor_system] Received Stop message, breaking loop"); + trace!( + "[actor_system] Received Stop message, breaking loop" + ); break; } Ok(ActorMessage::Tick) => { tick_count += 1; - trace!("[actor_system] Received explicit Tick message #{}", tick_count); + trace!( + "[actor_system] Received explicit Tick message #{}", + tick_count + ); actor.tick(); } Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { @@ -149,7 +167,9 @@ impl ActorSystem { } } Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - trace!("[actor_system] Channel disconnected, breaking loop"); + trace!( + "[actor_system] Channel disconnected, breaking loop" + ); // Channel closed, actor should stop break; } diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index 7f5ac89..b0efbe5 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -2,8 +2,8 @@ use crate::actor::{Actor, ActorHandle}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::peer::{ - FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueResponse, TransferRequest, - TransferResponse, UploadFailedHandler, + FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueResponse, + TransferRequest, TransferResponse, UploadFailedHandler, }; use crate::message::server::MessageFactory; use crate::message::{Handlers, Message, MessageReader, MessageType}; @@ -81,7 +81,8 @@ impl PeerActor { /// Initialize the message dispatcher with handlers fn initialize_dispatcher(&mut self) { // Create a channel for the dispatcher to send messages to - let (dispatcher_sender, dispatcher_receiver) = std::sync::mpsc::channel::(); + let (dispatcher_sender, dispatcher_receiver) = + std::sync::mpsc::channel::(); self.dispatcher_receiver = Some(dispatcher_receiver); @@ -104,15 +105,16 @@ impl PeerActor { /// Process any messages from the dispatcher fn process_dispatcher_messages(&mut self) { // Collect all messages first to avoid borrow issues - let messages: Vec = if let Some(ref receiver) = self.dispatcher_receiver { - let mut msgs = Vec::new(); - while let Ok(msg) = receiver.try_recv() { - msgs.push(msg); - } - msgs - } else { - Vec::new() - }; + let messages: Vec = + if let Some(ref receiver) = self.dispatcher_receiver { + let mut msgs = Vec::new(); + while let Ok(msg) = receiver.try_recv() { + msgs.push(msg); + } + msgs + } else { + Vec::new() + }; // Process all collected messages for msg in messages { @@ -149,7 +151,9 @@ impl PeerActor { MessageFactory::build_transfer_response_message(transfer); if let Some(ref handle) = self.self_handle { - if let Err(e) = handle.send(PeerMessage::SendMessage(transfer_response)) { + if let Err(e) = + handle.send(PeerMessage::SendMessage(transfer_response)) + { error!("[peer_actor:{}] Failed to send TransferResponse message: {}", username, e); } } @@ -202,7 +206,8 @@ impl PeerActor { self.peer.write().unwrap().username = username; } PeerMessage::QueueUpload(filename) => { - let message = MessageFactory::build_queue_upload_message(&filename); + let message = + MessageFactory::build_queue_upload_message(&filename); self.send_message(message); } PeerMessage::RequestTransfer(download) => { @@ -214,7 +219,10 @@ impl PeerActor { } PeerMessage::ProcessRead => { let username = self.peer.read().unwrap().username.clone(); - trace!("[peer_actor:{}] Handling ProcessRead message", username); + trace!( + "[peer_actor:{}] Handling ProcessRead message", + username + ); self.process_read(); } } @@ -269,13 +277,19 @@ impl PeerActor { username, extracted_count, message - .get_message_name(MessageType::Peer, message.get_message_code() as u32) + .get_message_name( + MessageType::Peer, + message.get_message_code() as u32 + ) .map_err(|e| e.to_string()) ); if let Some(ref dispatcher) = self.dispatcher { dispatcher.dispatch(&mut message); } else { - warn!("[peer_actor:{}] No dispatcher available!", username); + warn!( + "[peer_actor:{}] No dispatcher available!", + username + ); } } Err(e) => { @@ -316,7 +330,9 @@ impl PeerActor { message .get_message_name( MessageType::Peer, - u32::from_le_bytes(message.get_slice(0, 4).try_into().unwrap()) + u32::from_le_bytes( + message.get_slice(0, 4).try_into().unwrap() + ) ) .map_err(|e| e.to_string()) ); @@ -372,17 +388,32 @@ impl Actor for PeerActor { // Trigger initial read if let Some(ref handle) = self.self_handle { - trace!("[peer_actor:{}] Sending initial ProcessRead message", username); + trace!( + "[peer_actor:{}] Sending initial ProcessRead message", + username + ); match handle.send(PeerMessage::ProcessRead) { - Ok(_) => trace!("[peer_actor:{}] ProcessRead message sent successfully", username), - Err(e) => error!("[peer_actor:{}] FAILED to send ProcessRead: {}", username, e), + Ok(_) => trace!( + "[peer_actor:{}] ProcessRead message sent successfully", + username + ), + Err(e) => error!( + "[peer_actor:{}] FAILED to send ProcessRead: {}", + username, e + ), } } else { - error!("[peer_actor:{}] self_handle is None! Cannot send ProcessRead", username); + error!( + "[peer_actor:{}] self_handle is None! Cannot send ProcessRead", + username + ); } // Also do an immediate read as fallback - trace!("[peer_actor:{}] Performing immediate process_read as fallback", username); + trace!( + "[peer_actor:{}] Performing immediate process_read as fallback", + username + ); self.process_read(); } @@ -398,8 +429,10 @@ impl Actor for PeerActor { self.process_read(); } else { let username = self.peer.read().unwrap().username.clone(); - trace!("[peer_actor:{}] tick() called but stream is None", username); + trace!( + "[peer_actor:{}] tick() called but stream is None", + username + ); } } } - diff --git a/src/actor/peer_registry.rs b/src/actor/peer_registry.rs index 02de074..5d99014 100644 --- a/src/actor/peer_registry.rs +++ b/src/actor/peer_registry.rs @@ -1,8 +1,8 @@ -use crate::actor::{ActorHandle, ActorSystem}; use crate::actor::peer_actor::{PeerActor, PeerMessage}; +use crate::actor::{ActorHandle, ActorSystem}; use crate::client::ClientOperation; -use crate::peer::Peer; use crate::message::MessageReader; +use crate::peer::Peer; use crate::{debug, trace}; use std::collections::HashMap; @@ -18,7 +18,10 @@ pub struct PeerRegistry { } impl PeerRegistry { - pub fn new(actor_system: Arc, client_channel: Sender) -> Self { + pub fn new( + actor_system: Arc, + client_channel: Sender, + ) -> Self { Self { peers: Arc::new(Mutex::new(HashMap::new())), actor_system, @@ -37,23 +40,23 @@ impl PeerRegistry { debug!("[peer_registry] Registering peer actor for {}", username); - let actor = PeerActor::new( - peer, - stream, - reader, - self.client_channel.clone(), - ); + let actor = + PeerActor::new(peer, stream, reader, self.client_channel.clone()); // Spawn the actor with initialization callback to set self_handle - let handle = self.actor_system.spawn_with_handle(actor, |actor, handle| { - actor.set_self_handle(handle); - }); + let handle = + self.actor_system.spawn_with_handle(actor, |actor, handle| { + actor.set_self_handle(handle); + }); // Store in registry let mut peers = self.peers.lock().unwrap(); peers.insert(username.clone(), handle.clone()); - trace!("[peer_registry] Peer actor {} registered successfully", username); + trace!( + "[peer_registry] Peer actor {} registered successfully", + username + ); Ok(handle) } @@ -65,7 +68,10 @@ impl PeerRegistry { } /// Remove a peer actor from the registry - pub fn remove_peer(&self, username: &str) -> Option> { + pub fn remove_peer( + &self, + username: &str, + ) -> Option> { let mut peers = self.peers.lock().unwrap(); let handle = peers.remove(username); @@ -95,15 +101,24 @@ impl PeerRegistry { } /// Send a message to a peer by username - pub fn send_to_peer(&self, username: &str, message: PeerMessage) -> Result<(), String> { - let handle = self.get_peer(username) - .ok_or_else(|| format!("Peer {} not found in registry", username))?; + pub fn send_to_peer( + &self, + username: &str, + message: PeerMessage, + ) -> Result<(), String> { + let handle = self.get_peer(username).ok_or_else(|| { + format!("Peer {} not found in registry", username) + })?; handle.send(message) } /// Queue an upload for a peer - pub fn queue_upload(&self, username: &str, filename: String) -> Result<(), String> { + pub fn queue_upload( + &self, + username: &str, + filename: String, + ) -> Result<(), String> { self.send_to_peer(username, PeerMessage::QueueUpload(filename)) } } diff --git a/src/client.rs b/src/client.rs index c87cc64..862269a 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,11 +1,8 @@ use crate::types::{DownloadResult, DownloadStatus}; use crate::{ - actor::{ActorSystem, peer_registry::PeerRegistry}, + actor::{peer_registry::PeerRegistry, ActorSystem}, error::{Result, SoulseekRs}, - peer::{ - listen::Listen, ConnectionType, DownloadPeer, NewPeer, - Peer, - }, + peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer}, server::{PeerAddress, Server, ServerOperation}, types::{Download, FileSearchResult}, utils::{md5, thread_pool::ThreadPool}, @@ -136,10 +133,8 @@ impl Client { { let mut ctx = self.context.write().unwrap(); ctx.sender = Some(sender.clone()); - let peer_registry = PeerRegistry::new( - ctx.actor_system.clone(), - sender.clone(), - ); + let peer_registry = + PeerRegistry::new(ctx.actor_system.clone(), sender.clone()); ctx.peer_registry = Some(peer_registry); } @@ -261,11 +256,14 @@ impl Client { let mut context = self.context.write().unwrap(); context.download_tokens.insert(token, download.clone()); - let download_initiated = if let Some(ref registry) = context.peer_registry { - registry.queue_upload(&username, download.filename.clone()).is_ok() - } else { - false - }; + let download_initiated = + if let Some(ref registry) = context.peer_registry { + registry + .queue_upload(&username, download.filename.clone()) + .is_ok() + } else { + false + }; drop(context); @@ -315,7 +313,9 @@ impl Client { ClientOperation::PeerDisconnected(username) => { let context = client_context.read().unwrap(); if let Some(ref registry) = context.peer_registry { - if let Some(handle) = registry.remove_peer(&username) { + if let Some(handle) = + registry.remove_peer(&username) + { let _ = handle.stop(); } } @@ -445,7 +445,9 @@ impl Client { username, host, port, obfuscation_type, obfuscated_port ); - let peer_exists = client_context.read().unwrap() + let peer_exists = client_context + .read() + .unwrap() .peer_registry .as_ref() .map(|r| r.contains(&username)) @@ -541,9 +543,8 @@ impl Client { let unlocked_context = client_context.read().unwrap(); trace!("[client] connect_to_peer: {}", peer.username); - if let Some(sender) = unlocked_context.sender.clone() { + if let Some(_sender) = unlocked_context.sender.clone() { let peer_clone = peer.clone(); - let sender_clone = sender; unlocked_context.thread_pool.execute(move || { trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); match peer.connection_type { diff --git a/src/main.rs b/src/main.rs index b9d8880..64c7710 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,7 +6,7 @@ fn main() { PeerAddress::new(String::from("server.slsknet.org"), 2242), String::from("xxxxxx"), String::from("xxxxxx"), - false, + true, None, ); diff --git a/src/message/handlers.rs b/src/message/handlers.rs index 57f1e99..0f343c8 100644 --- a/src/message/handlers.rs +++ b/src/message/handlers.rs @@ -31,7 +31,10 @@ impl Handlers { self.handlers.insert(handler.get_code(), Box::new(handler)); self } - pub fn get_handler(&self, code: u8) -> Option<&(dyn MessageHandler + Send)> { + pub fn get_handler( + &self, + code: u8, + ) -> Option<&(dyn MessageHandler + Send)> { self.handlers.get(&code).map(|v| &**v) } } diff --git a/src/message/peer/transfer_request.rs b/src/message/peer/transfer_request.rs index 914bc0e..a4d7499 100644 --- a/src/message/peer/transfer_request.rs +++ b/src/message/peer/transfer_request.rs @@ -13,8 +13,6 @@ impl MessageHandler for TransferRequest { fn handle(&self, message: &mut Message, sender: Sender) { let transfer = Transfer::new_from_message(message); - sender - .send(PeerMessage::TransferRequest(transfer)) - .unwrap(); + sender.send(PeerMessage::TransferRequest(transfer)).unwrap(); } } diff --git a/src/peer/listen.rs b/src/peer/listen.rs index dd58b10..789a82f 100644 --- a/src/peer/listen.rs +++ b/src/peer/listen.rs @@ -15,6 +15,7 @@ const PEER_INIT_MESSAGE_CODE: u8 = 1; #[derive(Clone)] struct ConnectionContext { + #[allow(dead_code)] client_sender: Sender, client_context: Arc>, own_username: String, @@ -119,7 +120,10 @@ fn handle_peer_connection( trace!("[listener] peer actor spawned for: {}", peer.username); } Err(e) => { - error!("Failed to spawn peer actor for {:?}: {:?}", peer.username, e); + error!( + "Failed to spawn peer actor for {:?}: {:?}", + peer.username, e + ); } } } else { @@ -221,12 +225,7 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { match init_data.connection_type { ConnectionType::P => handle_peer_connection( - peer, - stream, - reader, - &context, - &peer_ip, - peer_port, + peer, stream, reader, &context, &peer_ip, peer_port, ), ConnectionType::F => handle_file_connection( peer, From f6921224c6792ea090d1f93220b5077aa24dacfb Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 15:48:02 +0100 Subject: [PATCH 4/9] fix: listener --- src/actor/peer_actor.rs | 31 ++++++++++++++++++++++++++++++- src/main.rs | 2 +- src/peer/listen.rs | 5 +++-- 3 files changed, 34 insertions(+), 4 deletions(-) diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index b0efbe5..45cede8 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -232,7 +232,26 @@ impl PeerActor { fn process_read(&mut self) { let username = self.peer.read().unwrap().username.clone(); trace!("[peer_actor:{}] process_read() called", username); + if self.reader.buffer_len() > 0 { + trace!( + "[peer_actor:{}] reader buffer has {} bytes", + username, + self.reader.buffer_len() + ); + self.extract_and_process_messages(); // This will process all messages in the buffer + } + trace!("[peer_actor:{}] Trying to read from socket", username); + trace!( + "[peer_actor:{}] reader buffer has {} bytes", + username, + self.reader.buffer_len() + ); + trace!( + "[peer_actor:{}] reader buffer has {:?}", + username, + self.reader.get_buffer() + ); // Try to read more data from the socket first { let stream = match self.stream.as_mut() { @@ -264,6 +283,17 @@ impl PeerActor { } } } + trace!("[peer_actor:{}] After reading from socket", username); + self.extract_and_process_messages(); // This will process all messages in the buffer + trace!("[peer_actor:{}] process_read() completed", username); + } + + fn extract_and_process_messages(&mut self) { + let username = self.peer.read().unwrap().username.clone(); + trace!( + "[peer_actor:{}] Extracting and processing messages", + username + ); // Extract all available messages from the buffer let username = self.peer.read().unwrap().username.clone(); @@ -310,7 +340,6 @@ impl PeerActor { // Process any messages that were dispatched trace!("[peer_actor:{}] Processing dispatcher messages", username); self.process_dispatcher_messages(); - trace!("[peer_actor:{}] process_read() completed", username); } /// Handle writing a message to the socket diff --git a/src/main.rs b/src/main.rs index 64c7710..5e68ab7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,7 +13,7 @@ fn main() { client.connect(); match client.login() { Ok(_) => { - match client.search("michel test file", Duration::from_secs(4)) { + match client.search("michel test file", Duration::from_secs(10)) { Ok(results) => { if let Some(result) = results.iter().find(|r| !r.files.is_empty()) diff --git a/src/peer/listen.rs b/src/peer/listen.rs index 789a82f..1b922be 100644 --- a/src/peer/listen.rs +++ b/src/peer/listen.rs @@ -213,8 +213,9 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { ); let peer = Peer::new( - init_data.username.clone(), - ConnectionType::P, + format!("{}:direct", init_data.username), // Add direct suffix to distinguish from regular peers + // init_data.username.clone(), + init_data.connection_type.clone(), peer_ip.clone(), peer_port.into(), None, From 71547aeb8578f1aba14d7fe79780d0e5603216fe Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 17:03:12 +0100 Subject: [PATCH 5/9] chore: remove threadpool from client --- src/actor/listener_actor.rs | 344 ++++++++++++++++++++++++++++++++++++ src/client.rs | 234 ++++++++++++++---------- 2 files changed, 482 insertions(+), 96 deletions(-) create mode 100644 src/actor/listener_actor.rs diff --git a/src/actor/listener_actor.rs b/src/actor/listener_actor.rs new file mode 100644 index 0000000..19d4313 --- /dev/null +++ b/src/actor/listener_actor.rs @@ -0,0 +1,344 @@ +use crate::actor::{Actor, ActorSystem}; +use crate::client::ClientContext; +use crate::message::{Message, MessageReader}; +use crate::peer::{ConnectionType, DownloadPeer, Peer}; +use crate::types::Download; +use crate::{debug, error, info, trace, DownloadStatus}; + +use std::io; +use std::net::{TcpListener, TcpStream}; +use std::sync::{Arc, RwLock}; + +const PEER_INIT_MESSAGE_CODE: u8 = 1; + +struct PeerInitData { + username: String, + connection_type: ConnectionType, + token: u32, +} + +fn read_peer_init_message( + stream: &mut TcpStream, + reader: &mut MessageReader, +) -> io::Result { + loop { + reader.read_from_socket(stream)?; + + if let Ok(Some(msg)) = reader.extract_message() { + return Ok(msg); + } + } +} + +fn parse_peer_init_message(mut message: Message) -> Option { + message.set_pointer(4); + let message_code = message.read_int8(); + + if message_code != PEER_INIT_MESSAGE_CODE { + return None; + } + + Some(PeerInitData { + username: message.read_string(), + connection_type: message.read_string().parse().unwrap(), + token: message.read_int32(), + }) +} + +fn parse_token_from_buffer(buffer: &[u8], username: &str) -> Option { + let token_bytes = buffer.get(0..4)?; + let token = u32::from_le_bytes( + token_bytes + .try_into() + .unwrap_or_else(|_| panic!("[listener:{}] slice with incorrect length, can't extract transfer_token", username)), + ); + Some(token) +} + +fn extract_download_from_buffer( + reader: &mut MessageReader, + client_context: &Arc>, + username: &str, + peer_ip: &str, + peer_port: u16, +) -> Option { + if reader.buffer_len() == 0 { + return None; + } + + let buffer = reader.get_buffer(); + trace!( + "[listener:{peer_ip}:{peer_port}] reader buffer has {} bytes. {:?}", + buffer.len(), + buffer + ); + + let token = parse_token_from_buffer(&buffer, username)?; + trace!( + "[listener:{}] got transfer_token: {} from data chunk", + username, + token + ); + + let context = client_context.read().unwrap(); + let download = context.download_tokens.get(&token).cloned(); + + if download.is_none() { + let download_tokens: Vec = context.download_tokens.keys().cloned().collect(); + trace!("[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}", token, download_tokens); + } + + download +} + +/// Message types for ConnectionHandlerActor +#[derive(Debug, Clone)] +pub enum ConnectionHandlerMessage { + /// Process the incoming connection + Process, +} + +/// Actor that handles a single incoming connection +pub struct ConnectionHandlerActor { + stream: Option, + client_context: Arc>, + own_username: String, + peer_ip: String, + peer_port: u16, +} + +impl ConnectionHandlerActor { + pub fn new( + stream: TcpStream, + client_context: Arc>, + own_username: String, + ) -> Self { + let peer_addr = stream.peer_addr().unwrap(); + let peer_ip = peer_addr.ip().to_string(); + let peer_port = peer_addr.port(); + + Self { + stream: Some(stream), + client_context, + own_username, + peer_ip, + peer_port, + } + } + + fn handle_peer_connection( + &self, + peer: Peer, + stream: TcpStream, + reader: MessageReader, + ) { + debug!("[listener:{}:{}] connection type is P, reader buffer has {} bytes", + self.peer_ip, self.peer_port, reader.buffer_len()); + + let client_context = self.client_context.read().unwrap(); + if let Some(ref registry) = client_context.peer_registry { + match registry.register_peer(peer.clone(), stream, Some(reader)) { + Ok(_) => trace!("[listener] peer actor spawned for: {}", peer.username), + Err(e) => error!("Failed to spawn peer actor for {:?}: {:?}", peer.username, e), + } + } else { + error!("PeerRegistry not initialized"); + } + } + + fn handle_file_connection( + &self, + peer: Peer, + stream: TcpStream, + mut reader: MessageReader, + token: u32, + ) { + trace!( + "[client] DownloadFromPeer token: {} peer: {:?}", + token, + peer + ); + + // Set stream to blocking mode - DownloadPeer expects blocking I/O + if let Err(e) = stream.set_nonblocking(false) { + error!( + "Failed to set stream to blocking mode for {}:{} (token: {}) - Error: {}", + peer.host, peer.port, token, e + ); + return; + } + + let download = extract_download_from_buffer( + &mut reader, + &self.client_context, + &peer.username, + &self.peer_ip, + self.peer_port, + ); + + let download_peer = DownloadPeer::new( + format!("{}:direct", peer.username), + peer.host.clone(), + peer.port, + token, + true, + self.own_username.clone(), + ); + + match download_peer.download_file( + self.client_context.clone(), + download, + Some(stream), + ) { + Ok((download, filename)) => { + download.sender.send(DownloadStatus::Completed).unwrap(); + info!( + "Successfully downloaded {} bytes to {}", + download.size, filename + ); + } + Err(e) => { + error!( + "Failed to download file from {}:{} (token: {}) - Error: {}", + peer.host, peer.port, token, e + ); + } + } + } + + fn process_connection(&mut self) { + let Some(mut stream) = self.stream.take() else { + error!("[listener] stream already taken"); + return; + }; + + let mut reader = MessageReader::new(); + + let Ok(message) = read_peer_init_message(&mut stream, &mut reader) else { + error!( + "[listener:{}:{}] Failed to read peer init message", + self.peer_ip, self.peer_port + ); + return; + }; + + let Some(init_data) = parse_peer_init_message(message) else { + error!("[listener:{}:{}] Invalid or unknown peer init message", + self.peer_ip, self.peer_port); + return; + }; + + debug!( + "[listener:{}:{}] peerInit (0) username: {} connection_type: {} token: {}", + self.peer_ip, self.peer_port, init_data.username, init_data.connection_type, init_data.token + ); + + let peer = Peer::new( + format!("{}:direct", init_data.username), + init_data.connection_type.clone(), + self.peer_ip.clone(), + self.peer_port.into(), + None, + 0, + 0, + 0, + ); + + match init_data.connection_type { + ConnectionType::P => self.handle_peer_connection(peer, stream, reader), + ConnectionType::F => self.handle_file_connection(peer, stream, reader, init_data.token), + ConnectionType::D => { + debug!("[listener:{}:{}] connection type is D, not supported yet, closing connection.", + self.peer_ip, self.peer_port); + } + } + } +} + +impl Actor for ConnectionHandlerActor { + type Message = ConnectionHandlerMessage; + + fn handle(&mut self, msg: Self::Message) { + match msg { + ConnectionHandlerMessage::Process => self.process_connection(), + } + } + + fn on_start(&mut self) { + trace!("[connection_handler:{}:{}] starting", self.peer_ip, self.peer_port); + self.process_connection(); + } +} + +/// Message types for ListenerActor +#[derive(Debug, Clone)] +pub enum ListenerMessage { + /// Stop the listener + Stop, +} + +/// Actor that accepts incoming TCP connections +pub struct ListenerActor { + listener: Option, + client_context: Arc>, + own_username: String, + actor_system: Arc, +} + +impl ListenerActor { + pub fn new( + port: u32, + client_context: Arc>, + own_username: String, + actor_system: Arc, + ) -> Result { + let listener = TcpListener::bind(format!("0.0.0.0:{port}"))?; + listener.set_nonblocking(true)?; + + info!("[listener] starting listener on port {port}"); + + Ok(Self { + listener: Some(listener), + client_context, + own_username, + actor_system, + }) + } +} + +impl Actor for ListenerActor { + type Message = ListenerMessage; + + fn handle(&mut self, msg: Self::Message) { + match msg { + ListenerMessage::Stop => { + info!("[listener] stopping listener"); + self.listener = None; + } + } + } + + fn tick(&mut self) { + let Some(ref listener) = self.listener else { + return; + }; + + match listener.accept() { + Ok((stream, _addr)) => { + trace!("[listener] accepted new connection"); + let actor = ConnectionHandlerActor::new( + stream, + self.client_context.clone(), + self.own_username.clone(), + ); + self.actor_system.spawn(actor); + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + // No incoming connections, this is normal for non-blocking + } + Err(e) => { + error!("[listener] Failed to accept connection: {}", e); + } + } + } +} diff --git a/src/client.rs b/src/client.rs index 862269a..794ed17 100644 --- a/src/client.rs +++ b/src/client.rs @@ -47,7 +47,6 @@ pub struct ClientContext { server_sender: Option>, search_results: Vec, pub download_tokens: HashMap, - thread_pool: Arc, actor_system: Arc, } impl Default for ClientContext { @@ -72,7 +71,6 @@ impl ClientContext { server_sender: None, search_results: Vec::new(), download_tokens: HashMap::new(), - thread_pool, actor_system, } } @@ -97,10 +95,6 @@ impl Client { ) -> Self { crate::utils::logger::init(); let context = Arc::new(RwLock::new(ClientContext::new())); - debug!( - "ThreadPool initialized with {} threads", - context.read().unwrap().thread_pool.thread_count() - ); Self { enable_listen, listen_port: listen_port.unwrap_or(DEFALT_LISTEN_PORT), @@ -545,113 +539,161 @@ impl Client { trace!("[client] connect_to_peer: {}", peer.username); if let Some(_sender) = unlocked_context.sender.clone() { let peer_clone = peer.clone(); - unlocked_context.thread_pool.execute(move || { - trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); - match peer.connection_type { - ConnectionType::P => { - let username = peer.username.clone(); + trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); + match peer.connection_type { + ConnectionType::P => { + let username = peer.username.clone(); - // Establish TCP connection or use provided stream - let mut tcp_stream = match stream { - Some(s) => s, - None => { - let socket_addr = format!("{}:{}", peer_clone.host, peer_clone.port) - .parse::(); + // Establish TCP connection or use provided stream + let mut tcp_stream = match stream { + Some(s) => s, + None => { + let socket_addr = format!( + "{}:{}", + peer_clone.host, peer_clone.port + ) + .parse::(); - match socket_addr { - Ok(addr) => { - match TcpStream::connect_timeout(&addr, Duration::from_secs(20)) { - Ok(s) => s, - Err(e) => { - trace!( + match socket_addr { + Ok(addr) => { + match TcpStream::connect_timeout( + &addr, + Duration::from_secs(20), + ) { + Ok(s) => s, + Err(e) => { + trace!( "Can't connect to {:?} {:?}:{:?} - {:?}", username, peer_clone.host, peer_clone.port, e ); - return; - } - } - } - Err(e) => { - trace!("Invalid socket address for {:?}: {:?}", username, e); return; } } } - }; - - // Send connection handshake message - use std::io::Write; - if let Some(token) = peer_clone.token { - // For indirect connections, send watch user message - let handshake_msg = crate::message::server::MessageFactory::build_watch_user(token); - if let Err(e) = tcp_stream.write_all(&handshake_msg.get_data()) { - trace!("Failed to send watch user handshake to {:?}: {:?}", username, e); + Err(e) => { + trace!( + "Invalid socket address for {:?}: {:?}", + username, + e + ); return; } - trace!("[client] Sent watch user handshake for token: {}", token); - } - - // Configure socket - if let Err(e) = tcp_stream.set_read_timeout(Some(Duration::from_secs(5))) { - trace!("Failed to set read timeout for {:?}: {:?}", username, e); - return; - } - if let Err(e) = tcp_stream.set_write_timeout(Some(Duration::from_secs(5))) { - trace!("Failed to set write timeout for {:?}: {:?}", username, e); - return; - } - if let Err(e) = tcp_stream.set_nodelay(true) { - trace!("Failed to set nodelay for {:?}: {:?}", username, e); - return; - } - - // Spawn peer actor - let context = client_context2.read().unwrap(); - if let Some(ref registry) = context.peer_registry { - match registry.register_peer(peer_clone, tcp_stream, None) { - Ok(_) => { - trace!("[client] peer actor spawned for: {}", username); - } - Err(e) => { - trace!("Failed to spawn peer actor for {:?}: {:?}", username, e); - } - } - } else { - trace!("PeerRegistry not initialized"); } } + }; - ConnectionType::F => { - trace!("[client] downloading from: {}, {:?}", peer.username, peer.token); - let download_peer = DownloadPeer::new( - peer.username, - peer.host, - peer.port, - peer.token.unwrap(), - false, - own_username.clone(), - ); + // Send connection handshake message + use std::io::Write; + if let Some(token) = peer_clone.token { + // For indirect connections, send watch user message + let handshake_msg = crate::message::server::MessageFactory::build_watch_user(token); + if let Err(e) = + tcp_stream.write_all(&handshake_msg.get_data()) + { + trace!("Failed to send watch user handshake to {:?}: {:?}", username, e); + return; + } + trace!( + "[client] Sent watch user handshake for token: {}", + token + ); + } - match download_peer - .download_file( - client_context2.clone(), - None, - None - ) { - Ok((download, filename)) => { - trace!("[client] downloaded {} bytes {:?} ", filename, download.size); - download.sender.send(DownloadStatus::Completed).unwrap(); - } - Err(e) => { - trace!("[client] failed to download: {}", e); - } - } - } - ConnectionType::D => { - error!("ConnectionType::D not implemented") + // Configure socket + if let Err(e) = tcp_stream + .set_read_timeout(Some(Duration::from_secs(5))) + { + trace!( + "Failed to set read timeout for {:?}: {:?}", + username, + e + ); + return; + } + if let Err(e) = tcp_stream + .set_write_timeout(Some(Duration::from_secs(5))) + { + trace!( + "Failed to set write timeout for {:?}: {:?}", + username, + e + ); + return; + } + if let Err(e) = tcp_stream.set_nodelay(true) { + trace!( + "Failed to set nodelay for {:?}: {:?}", + username, + e + ); + return; + } + + // Spawn peer actor + let context = client_context2.read().unwrap(); + if let Some(ref registry) = context.peer_registry { + match registry + .register_peer(peer_clone, tcp_stream, None) + { + Ok(_) => { + trace!( + "[client] peer actor spawned for: {}", + username + ); + } + Err(e) => { + trace!( + "Failed to spawn peer actor for {:?}: {:?}", + username, + e + ); + } + } + } else { + trace!("PeerRegistry not initialized"); + } + } + + ConnectionType::F => { + trace!( + "[client] downloading from: {}, {:?}", + peer.username, + peer.token + ); + let download_peer = DownloadPeer::new( + peer.username, + peer.host, + peer.port, + peer.token.unwrap(), + false, + own_username.clone(), + ); + + match download_peer.download_file( + client_context2.clone(), + None, + None, + ) { + Ok((download, filename)) => { + trace!( + "[client] downloaded {} bytes {:?} ", + filename, + download.size + ); + download + .sender + .send(DownloadStatus::Completed) + .unwrap(); + } + Err(e) => { + trace!("[client] failed to download: {}", e); } } - }); + } + ConnectionType::D => { + error!("ConnectionType::D not implemented") + } + } } else { debug!("[client] Peer already connected: {}", peer.username); } From 1ee5755ed1e38a5edbce17aa01862424936faa6a Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 17:36:44 +0100 Subject: [PATCH 6/9] feat: only for download peer spawn a thread --- src/peer/listen.rs | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/src/peer/listen.rs b/src/peer/listen.rs index 1b922be..5f90be5 100644 --- a/src/peer/listen.rs +++ b/src/peer/listen.rs @@ -228,15 +228,23 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { ConnectionType::P => handle_peer_connection( peer, stream, reader, &context, &peer_ip, peer_port, ), - ConnectionType::F => handle_file_connection( - peer, - stream, - reader, - init_data.token, - &context, - &peer_ip, - peer_port, - ), + + ConnectionType::F => { + thread::spawn(move || { + trace!( + "[listener:{peer_ip}:{peer_port}] handling file connection in thread" + ); + handle_file_connection( + peer, + stream, + reader, + init_data.token, + &context, + &peer_ip, + peer_port, + ) + }); + } ConnectionType::D => { debug!("[listener:{peer_ip}:{peer_port}] connection type is D, not supported yet, closing connection. "); } @@ -273,7 +281,7 @@ impl Listen { }; let context = context.clone(); - thread::spawn(move || handle_incoming_connection(stream, context)); + handle_incoming_connection(stream, context); } } } From a4aee59e1ef66f3fb83071d9cc3a3d43ec883653 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 17:36:55 +0100 Subject: [PATCH 7/9] chore: remove logging --- src/actor/peer_actor.rs | 50 ++--------------------------------------- 1 file changed, 2 insertions(+), 48 deletions(-) diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index 45cede8..f555695 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -218,11 +218,6 @@ impl PeerActor { self.send_message(message); } PeerMessage::ProcessRead => { - let username = self.peer.read().unwrap().username.clone(); - trace!( - "[peer_actor:{}] Handling ProcessRead message", - username - ); self.process_read(); } } @@ -230,28 +225,11 @@ impl PeerActor { /// Handle reading from the socket and dispatching messages fn process_read(&mut self) { - let username = self.peer.read().unwrap().username.clone(); - trace!("[peer_actor:{}] process_read() called", username); if self.reader.buffer_len() > 0 { - trace!( - "[peer_actor:{}] reader buffer has {} bytes", - username, - self.reader.buffer_len() - ); self.extract_and_process_messages(); // This will process all messages in the buffer } + let username = self.peer.read().unwrap().username.clone(); - trace!("[peer_actor:{}] Trying to read from socket", username); - trace!( - "[peer_actor:{}] reader buffer has {} bytes", - username, - self.reader.buffer_len() - ); - trace!( - "[peer_actor:{}] reader buffer has {:?}", - username, - self.reader.get_buffer() - ); // Try to read more data from the socket first { let stream = match self.stream.as_mut() { @@ -283,19 +261,10 @@ impl PeerActor { } } } - trace!("[peer_actor:{}] After reading from socket", username); self.extract_and_process_messages(); // This will process all messages in the buffer - trace!("[peer_actor:{}] process_read() completed", username); } fn extract_and_process_messages(&mut self) { - let username = self.peer.read().unwrap().username.clone(); - trace!( - "[peer_actor:{}] Extracting and processing messages", - username - ); - - // Extract all available messages from the buffer let username = self.peer.read().unwrap().username.clone(); let mut extracted_count = 0; loop { @@ -331,14 +300,12 @@ impl PeerActor { return; } Ok(None) => { - trace!("[peer_actor:{}] No more messages in buffer (extracted {} total)", username, extracted_count); break; } } } // Process any messages that were dispatched - trace!("[peer_actor:{}] Processing dispatcher messages", username); self.process_dispatcher_messages(); } @@ -409,23 +376,14 @@ impl Actor for PeerActor { fn on_start(&mut self) { let username = self.peer.read().unwrap().username.clone(); - trace!("[peer_actor:{}] actor starting", username); // Initialize the dispatcher self.initialize_dispatcher(); - trace!("[peer_actor:{}] dispatcher initialized", username); // Trigger initial read if let Some(ref handle) = self.self_handle { - trace!( - "[peer_actor:{}] Sending initial ProcessRead message", - username - ); match handle.send(PeerMessage::ProcessRead) { - Ok(_) => trace!( - "[peer_actor:{}] ProcessRead message sent successfully", - username - ), + Ok(_) => {}, Err(e) => error!( "[peer_actor:{}] FAILED to send ProcessRead: {}", username, e @@ -439,10 +397,6 @@ impl Actor for PeerActor { } // Also do an immediate read as fallback - trace!( - "[peer_actor:{}] Performing immediate process_read as fallback", - username - ); self.process_read(); } From a1b88f58bd3040c5c8685d0839eb0a0648e10d60 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 20:04:06 +0100 Subject: [PATCH 8/9] chore: debug logging --- src/actor/mod.rs | 24 ++------ src/actor/peer_actor.rs | 68 ++++++++++++++--------- src/actor/peer_registry.rs | 1 - src/client.rs | 110 +++++++++++++++++++++++++++---------- src/error.rs | 1 - src/main.rs | 6 +- src/server.rs | 29 ++++++---- 7 files changed, 148 insertions(+), 91 deletions(-) diff --git a/src/actor/mod.rs b/src/actor/mod.rs index f56c2d8..0b619ad 100644 --- a/src/actor/mod.rs +++ b/src/actor/mod.rs @@ -101,44 +101,29 @@ impl ActorSystem { }; let handle_for_init = handle.clone(); - // Start the actor event loop on the thread pool self.thread_pool.execute(move || { - trace!("[actor_system] Actor task starting on thread pool"); init(&mut actor, handle_for_init); - trace!("[actor_system] Actor init callback completed"); actor.on_start(); - trace!( - "[actor_system] Actor on_start completed, starting event loop" - ); Self::run_actor_loop(&mut actor, receiver); - trace!("[actor_system] Actor event loop ended"); actor.on_stop(); - trace!("[actor_system] Actor on_stop completed"); }); handle } - /// Run the actor's message processing loop fn run_actor_loop( actor: &mut A, receiver: Receiver>, ) { - trace!("[actor_system] run_actor_loop STARTED"); let tick_interval = Duration::from_millis(100); let mut last_tick = Instant::now(); let mut message_count = 0; let mut tick_count = 0; loop { - // Try to receive messages with timeout for periodic ticks match receiver.recv_timeout(tick_interval) { Ok(ActorMessage::UserMessage(msg)) => { message_count += 1; - trace!( - "[actor_system] Processing message #{}", - message_count - ); actor.handle(msg); } Ok(ActorMessage::Stop) => { @@ -153,15 +138,15 @@ impl ActorSystem { "[actor_system] Received explicit Tick message #{}", tick_count ); + actor.tick(); } Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { - // Periodic tick if last_tick.elapsed() >= tick_interval { tick_count += 1; - if tick_count % 10 == 0 { - trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count); - } + // if tick_count % 10 == 0 { + // trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count); + // } actor.tick(); last_tick = Instant::now(); } @@ -170,7 +155,6 @@ impl ActorSystem { trace!( "[actor_system] Channel disconnected, breaking loop" ); - // Channel closed, actor should stop break; } } diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index f555695..4be1294 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -11,35 +11,29 @@ use crate::peer::Peer; use crate::types::{Download, FileSearchResult, Transfer}; use crate::{debug, error, trace, warn}; -use std::io::{self, Write}; +use std::io::{self, Error, Write}; use std::net::TcpStream; use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, RwLock}; -/// Messages that can be sent to a PeerActor #[derive(Debug, Clone)] pub enum PeerMessage { - /// Send a raw message to the peer SendMessage(Message), - /// Handle a file search result FileSearchResult(FileSearchResult), - /// Handle a transfer request TransferRequest(Transfer), - /// Handle a transfer response + UploadFailed(String, String), TransferResponse { token: u32, allowed: bool, reason: Option, }, - /// Handle a place in queue response - PlaceInQueueResponse { filename: String, place: u32 }, - /// Set the peer's username + PlaceInQueueResponse { + filename: String, + place: u32, + }, SetUsername(String), - /// Queue an upload for a file QueueUpload(String), - /// Request a transfer for a download RequestTransfer(Download), - /// Internal: process read operations ProcessRead, } @@ -52,6 +46,7 @@ pub struct PeerActor { self_handle: Option>, dispatcher: Option>, dispatcher_receiver: Option>, + queued_filenames: Vec, } impl PeerActor { @@ -70,6 +65,7 @@ impl PeerActor { self_handle: None, dispatcher: None, dispatcher_receiver: None, + queued_filenames: Vec::new(), } } @@ -172,15 +168,15 @@ impl PeerActor { if !allowed { if let Some(reason_text) = reason { debug!( - "[peer_actor:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...", - username, reason_text, token - ); + "[peer_actor:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...", + username, reason_text, token + ); } } else { debug!( - "[peer_actor:{}] Transfer allowed, ready to connect with token {:}", - username, token - ); + "[peer_actor:{}] Transfer allowed, ready to connect with token {:}", + username, token + ); self.client_channel .send(ClientOperation::DownloadFromPeer( token, @@ -193,9 +189,9 @@ impl PeerActor { PeerMessage::PlaceInQueueResponse { filename, place } => { let username = self.peer.read().unwrap().username.clone(); debug!( - "[peer_actor:{}] Place in queue response - file: {}, place: {}", - username, filename, place - ); + "[peer_actor:{}] Place in queue response - file: {}, place: {}", + username, filename, place + ); } PeerMessage::SetUsername(username) => { trace!( @@ -206,6 +202,7 @@ impl PeerActor { self.peer.write().unwrap().username = username; } PeerMessage::QueueUpload(filename) => { + self.queued_filenames.push(filename.clone()); let message = MessageFactory::build_queue_upload_message(&filename); self.send_message(message); @@ -220,6 +217,11 @@ impl PeerActor { PeerMessage::ProcessRead => { self.process_read(); } + PeerMessage::UploadFailed(username, filename) => { + self.client_channel + .send(ClientOperation::UploadFailed(username, filename)) + .unwrap(); + } } } @@ -256,7 +258,7 @@ impl PeerActor { "[peer_actor:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.", username, e, e.kind() ); - self.disconnect(); + self.disconnect_with_error(e); return; } } @@ -296,7 +298,7 @@ impl PeerActor { "[peer_actor:{}] Error extracting message: {}. Disconnecting peer.", username, e ); - self.disconnect(); + self.disconnect_with_error(e); return; } Ok(None) => { @@ -338,7 +340,7 @@ impl PeerActor { "[peer_actor:{}] Error writing message: {}. Disconnecting.", username, e ); - self.disconnect(); + self.disconnect_with_error(e); return; } @@ -347,10 +349,22 @@ impl PeerActor { "[peer_actor:{}] Error flushing stream: {}. Disconnecting.", username, e ); - self.disconnect(); + self.disconnect_with_error(e); } } + fn disconnect_with_error(&mut self, error: Error) { + let username = self.peer.read().unwrap().username.clone(); + debug!("[peer_actor:{}] disconnect", username); + + self.stream.take(); + + if let Err(e) = self.client_channel.send( + ClientOperation::PeerDisconnected(username, Some(error.into())), + ) { + error!("Failed to send disconnect notification: {}", e); + } + } /// Disconnect the peer and notify the client fn disconnect(&mut self) { let username = self.peer.read().unwrap().username.clone(); @@ -360,7 +374,7 @@ impl PeerActor { if let Err(e) = self .client_channel - .send(ClientOperation::PeerDisconnected(username)) + .send(ClientOperation::PeerDisconnected(username, None)) { error!("Failed to send disconnect notification: {}", e); } @@ -383,7 +397,7 @@ impl Actor for PeerActor { // Trigger initial read if let Some(ref handle) = self.self_handle { match handle.send(PeerMessage::ProcessRead) { - Ok(_) => {}, + Ok(_) => {} Err(e) => error!( "[peer_actor:{}] FAILED to send ProcessRead: {}", username, e diff --git a/src/actor/peer_registry.rs b/src/actor/peer_registry.rs index 5d99014..992396e 100644 --- a/src/actor/peer_registry.rs +++ b/src/actor/peer_registry.rs @@ -113,7 +113,6 @@ impl PeerRegistry { handle.send(message) } - /// Queue an upload for a peer pub fn queue_upload( &self, username: &str, diff --git a/src/client.rs b/src/client.rs index 794ed17..e5adc61 100644 --- a/src/client.rs +++ b/src/client.rs @@ -25,11 +25,13 @@ use std::{ use crate::{debug, error, info, trace}; const DEFALT_LISTEN_PORT: u32 = 2234; +#[derive(Debug)] + pub enum ClientOperation { NewPeer(NewPeer), ConnectToPeer(Peer), SearchResult(FileSearchResult), - PeerDisconnected(String), + PeerDisconnected(String, Option), PierceFireWall(Peer), DownloadFromPeer(u32, Peer, bool), UpdateDownloadTokens(Transfer, String), @@ -40,6 +42,7 @@ pub enum ClientOperation { obfuscation_type: u32, obfuscated_port: u16, }, + UploadFailed(String, String), } pub struct ClientContext { pub peer_registry: Option, @@ -228,6 +231,7 @@ impl Client { size: u64, download_directory: String, ) -> Result { + info!("[client] Downloading {} from {}", filename, username); let start_time = Instant::now(); let hash = md5::md5(&filename); @@ -280,6 +284,34 @@ impl Client { }) } + fn process_failed_uploads( + client_context: Arc>, + username: &str, + filename: Option<&str>, + ) { + let context = client_context.read().unwrap(); + let download_tokens = &context.download_tokens; + let failed_downloads: Vec<_> = download_tokens + .iter() + .filter(|(_, download)| { + download.username == username + && (filename.is_none_or(|f| download.filename == *f)) + }) + .map(|(token, download)| { + download.sender.send(DownloadStatus::Failed).unwrap(); + *token + }) + .collect(); + + failed_downloads.iter().for_each(|token| { + client_context + .write() + .unwrap() + .download_tokens + .remove(token); + }); + } + fn listen_to_client_operations( reader: Receiver, client_context: Arc>, @@ -289,6 +321,7 @@ impl Client { if let Ok(operation) = reader.recv() { match operation { ClientOperation::ConnectToPeer(peer) => { + trace!("[client] ConnectToPeer {:?}", peer); Self::connect_to_peer( peer, client_context.clone(), @@ -304,7 +337,7 @@ impl Client { .search_results .push(file_search.clone()); } - ClientOperation::PeerDisconnected(username) => { + ClientOperation::PeerDisconnected(username, error) => { let context = client_context.read().unwrap(); if let Some(ref registry) = context.peer_registry { if let Some(handle) = @@ -313,6 +346,17 @@ impl Client { let _ = handle.stop(); } } + if let Some(error) = error { + error!( + "[client] Peer {} disconnected with error: {:?}", + username, error + ); + Self::process_failed_uploads( + client_context.clone(), + &username, + None, + ); + } } ClientOperation::PierceFireWall(peer) => { Self::pierce_firewall( @@ -350,27 +394,27 @@ impl Client { .split('\\') .next_back(); match filename { - Some(filename) => { - match download_peer.download_file( - client_context_clone, - Some(download.clone()), - None - ) { - Ok((download, filename)) => { - download.sender.send(DownloadStatus::Completed).unwrap(); - info!("Successfully downloaded {} bytes to {}", download.size, filename); - } - Err(e) => { - download.sender.send(DownloadStatus::Failed).unwrap(); - error!( - "Failed to download file '{}' from {}:{} (token: {}) - Error: {}", - filename, peer.host, peer.port, download.token, e - ); - } - } - } - None => error!("Cant find filename to save download: {:?}", download.filename), - } + Some(filename) => { + match download_peer.download_file( + client_context_clone, + Some(download.clone()), + None + ) { + Ok((download, filename)) => { + download.sender.send(DownloadStatus::Completed).unwrap(); + info!("Successfully downloaded {} bytes to {}", download.size, filename); + } + Err(e) => { + download.sender.send(DownloadStatus::Failed).unwrap(); + error!( + "Failed to download file '{}' from {}:{} (token: {}) - Error: {}", + filename, peer.host, peer.port, download.token, e + ); + } + } + } + None => error!("Cant find filename to save download: {:?}", download.filename), + } }); } None => { @@ -435,9 +479,9 @@ impl Client { obfuscated_port, } => { debug!( - "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", - username, host, port, obfuscation_type, obfuscated_port - ); + "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", + username, host, port, obfuscation_type, obfuscated_port + ); let peer_exists = client_context .read() @@ -503,9 +547,9 @@ impl Client { if let Some((key, download)) = key_to_remove { trace!( - "[client] UpdateDownloadTokens found {key}, transfer: {:?}", - transfer - ); + "[client] UpdateDownloadTokens found {key}, transfer: {:?}", + transfer + ); context.download_tokens.insert( transfer.token, @@ -522,6 +566,13 @@ impl Client { context.download_tokens.remove(&key); } } + ClientOperation::UploadFailed(username, filename) => { + Self::process_failed_uploads( + client_context.clone(), + &username, + Some(&filename), + ); + } } } }); @@ -544,7 +595,6 @@ impl Client { ConnectionType::P => { let username = peer.username.clone(); - // Establish TCP connection or use provided stream let mut tcp_stream = match stream { Some(s) => s, None => { diff --git a/src/error.rs b/src/error.rs index 74fddab..81adadd 100644 --- a/src/error.rs +++ b/src/error.rs @@ -3,7 +3,6 @@ use std::{error::Error, fmt}; /// Custom error type for the Soulseek download library #[derive(Debug)] pub enum SoulseekRs { - /// Network-related errors (I/O, connection failures) NetworkError(std::io::Error), /// Authentication failed during login AuthenticationFailed, diff --git a/src/main.rs b/src/main.rs index 5e68ab7..55d15bd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,14 +6,16 @@ fn main() { PeerAddress::new(String::from("server.slsknet.org"), 2242), String::from("xxxxxx"), String::from("xxxxxx"), - true, + false, None, ); client.connect(); match client.login() { Ok(_) => { - match client.search("michel test file", Duration::from_secs(10)) { + match client + .search("Eternity Alex Kassian", Duration::from_secs(10)) + { Ok(results) => { if let Some(result) = results.iter().find(|r| !r.files.is_empty()) diff --git a/src/server.rs b/src/server.rs index 4ad5082..d8cc7d9 100644 --- a/src/server.rs +++ b/src/server.rs @@ -341,21 +341,30 @@ impl Server { if let Ok(operation) = server_channel.recv() { match operation { ServerOperation::ConnectToPeer(peer) => { - debug!( - "[server] ConnectToPeer {} - ConnectionType {}", - peer.username, peer.connection_type + trace!( + "[server] → ConnectToPeer {}", + peer.username ); - if let Some(op) = match peer.connection_type { - ConnectionType::P => { - Some(ClientOperation::ConnectToPeer(peer)) - } - ConnectionType::F => { - Some(ClientOperation::ConnectToPeer(peer)) + ConnectionType::P | ConnectionType::F => { + Some(ClientOperation::ConnectToPeer( + peer.clone(), + )) } ConnectionType::D => None, } { - client_channel.send(op).unwrap(); + trace!( + "[server] → ConnectToPeer {} - op {:?}", + peer.username, + op + ); + + match client_channel.send(op) { + Ok(_) => {} + Err(e) => { + error!("[server] Failed to send: {}", e) + } + } } } ServerOperation::LoginStatus(message) => { From 13f4783f65c35ce58f3f291e9a65a526b06b829b Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 21:12:39 +0100 Subject: [PATCH 9/9] chore: refactor peer connection to actor and non blocking --- src/actor/listener_actor.rs | 344 ------------------- src/actor/peer_actor.rs | 262 +++++++++++---- src/actor/peer_registry.rs | 21 +- src/client.rs | 638 ++++++++++++++++-------------------- src/main.rs | 2 +- src/peer/listen.rs | 22 +- src/server.rs | 17 +- 7 files changed, 490 insertions(+), 816 deletions(-) delete mode 100644 src/actor/listener_actor.rs diff --git a/src/actor/listener_actor.rs b/src/actor/listener_actor.rs deleted file mode 100644 index 19d4313..0000000 --- a/src/actor/listener_actor.rs +++ /dev/null @@ -1,344 +0,0 @@ -use crate::actor::{Actor, ActorSystem}; -use crate::client::ClientContext; -use crate::message::{Message, MessageReader}; -use crate::peer::{ConnectionType, DownloadPeer, Peer}; -use crate::types::Download; -use crate::{debug, error, info, trace, DownloadStatus}; - -use std::io; -use std::net::{TcpListener, TcpStream}; -use std::sync::{Arc, RwLock}; - -const PEER_INIT_MESSAGE_CODE: u8 = 1; - -struct PeerInitData { - username: String, - connection_type: ConnectionType, - token: u32, -} - -fn read_peer_init_message( - stream: &mut TcpStream, - reader: &mut MessageReader, -) -> io::Result { - loop { - reader.read_from_socket(stream)?; - - if let Ok(Some(msg)) = reader.extract_message() { - return Ok(msg); - } - } -} - -fn parse_peer_init_message(mut message: Message) -> Option { - message.set_pointer(4); - let message_code = message.read_int8(); - - if message_code != PEER_INIT_MESSAGE_CODE { - return None; - } - - Some(PeerInitData { - username: message.read_string(), - connection_type: message.read_string().parse().unwrap(), - token: message.read_int32(), - }) -} - -fn parse_token_from_buffer(buffer: &[u8], username: &str) -> Option { - let token_bytes = buffer.get(0..4)?; - let token = u32::from_le_bytes( - token_bytes - .try_into() - .unwrap_or_else(|_| panic!("[listener:{}] slice with incorrect length, can't extract transfer_token", username)), - ); - Some(token) -} - -fn extract_download_from_buffer( - reader: &mut MessageReader, - client_context: &Arc>, - username: &str, - peer_ip: &str, - peer_port: u16, -) -> Option { - if reader.buffer_len() == 0 { - return None; - } - - let buffer = reader.get_buffer(); - trace!( - "[listener:{peer_ip}:{peer_port}] reader buffer has {} bytes. {:?}", - buffer.len(), - buffer - ); - - let token = parse_token_from_buffer(&buffer, username)?; - trace!( - "[listener:{}] got transfer_token: {} from data chunk", - username, - token - ); - - let context = client_context.read().unwrap(); - let download = context.download_tokens.get(&token).cloned(); - - if download.is_none() { - let download_tokens: Vec = context.download_tokens.keys().cloned().collect(); - trace!("[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}", token, download_tokens); - } - - download -} - -/// Message types for ConnectionHandlerActor -#[derive(Debug, Clone)] -pub enum ConnectionHandlerMessage { - /// Process the incoming connection - Process, -} - -/// Actor that handles a single incoming connection -pub struct ConnectionHandlerActor { - stream: Option, - client_context: Arc>, - own_username: String, - peer_ip: String, - peer_port: u16, -} - -impl ConnectionHandlerActor { - pub fn new( - stream: TcpStream, - client_context: Arc>, - own_username: String, - ) -> Self { - let peer_addr = stream.peer_addr().unwrap(); - let peer_ip = peer_addr.ip().to_string(); - let peer_port = peer_addr.port(); - - Self { - stream: Some(stream), - client_context, - own_username, - peer_ip, - peer_port, - } - } - - fn handle_peer_connection( - &self, - peer: Peer, - stream: TcpStream, - reader: MessageReader, - ) { - debug!("[listener:{}:{}] connection type is P, reader buffer has {} bytes", - self.peer_ip, self.peer_port, reader.buffer_len()); - - let client_context = self.client_context.read().unwrap(); - if let Some(ref registry) = client_context.peer_registry { - match registry.register_peer(peer.clone(), stream, Some(reader)) { - Ok(_) => trace!("[listener] peer actor spawned for: {}", peer.username), - Err(e) => error!("Failed to spawn peer actor for {:?}: {:?}", peer.username, e), - } - } else { - error!("PeerRegistry not initialized"); - } - } - - fn handle_file_connection( - &self, - peer: Peer, - stream: TcpStream, - mut reader: MessageReader, - token: u32, - ) { - trace!( - "[client] DownloadFromPeer token: {} peer: {:?}", - token, - peer - ); - - // Set stream to blocking mode - DownloadPeer expects blocking I/O - if let Err(e) = stream.set_nonblocking(false) { - error!( - "Failed to set stream to blocking mode for {}:{} (token: {}) - Error: {}", - peer.host, peer.port, token, e - ); - return; - } - - let download = extract_download_from_buffer( - &mut reader, - &self.client_context, - &peer.username, - &self.peer_ip, - self.peer_port, - ); - - let download_peer = DownloadPeer::new( - format!("{}:direct", peer.username), - peer.host.clone(), - peer.port, - token, - true, - self.own_username.clone(), - ); - - match download_peer.download_file( - self.client_context.clone(), - download, - Some(stream), - ) { - Ok((download, filename)) => { - download.sender.send(DownloadStatus::Completed).unwrap(); - info!( - "Successfully downloaded {} bytes to {}", - download.size, filename - ); - } - Err(e) => { - error!( - "Failed to download file from {}:{} (token: {}) - Error: {}", - peer.host, peer.port, token, e - ); - } - } - } - - fn process_connection(&mut self) { - let Some(mut stream) = self.stream.take() else { - error!("[listener] stream already taken"); - return; - }; - - let mut reader = MessageReader::new(); - - let Ok(message) = read_peer_init_message(&mut stream, &mut reader) else { - error!( - "[listener:{}:{}] Failed to read peer init message", - self.peer_ip, self.peer_port - ); - return; - }; - - let Some(init_data) = parse_peer_init_message(message) else { - error!("[listener:{}:{}] Invalid or unknown peer init message", - self.peer_ip, self.peer_port); - return; - }; - - debug!( - "[listener:{}:{}] peerInit (0) username: {} connection_type: {} token: {}", - self.peer_ip, self.peer_port, init_data.username, init_data.connection_type, init_data.token - ); - - let peer = Peer::new( - format!("{}:direct", init_data.username), - init_data.connection_type.clone(), - self.peer_ip.clone(), - self.peer_port.into(), - None, - 0, - 0, - 0, - ); - - match init_data.connection_type { - ConnectionType::P => self.handle_peer_connection(peer, stream, reader), - ConnectionType::F => self.handle_file_connection(peer, stream, reader, init_data.token), - ConnectionType::D => { - debug!("[listener:{}:{}] connection type is D, not supported yet, closing connection.", - self.peer_ip, self.peer_port); - } - } - } -} - -impl Actor for ConnectionHandlerActor { - type Message = ConnectionHandlerMessage; - - fn handle(&mut self, msg: Self::Message) { - match msg { - ConnectionHandlerMessage::Process => self.process_connection(), - } - } - - fn on_start(&mut self) { - trace!("[connection_handler:{}:{}] starting", self.peer_ip, self.peer_port); - self.process_connection(); - } -} - -/// Message types for ListenerActor -#[derive(Debug, Clone)] -pub enum ListenerMessage { - /// Stop the listener - Stop, -} - -/// Actor that accepts incoming TCP connections -pub struct ListenerActor { - listener: Option, - client_context: Arc>, - own_username: String, - actor_system: Arc, -} - -impl ListenerActor { - pub fn new( - port: u32, - client_context: Arc>, - own_username: String, - actor_system: Arc, - ) -> Result { - let listener = TcpListener::bind(format!("0.0.0.0:{port}"))?; - listener.set_nonblocking(true)?; - - info!("[listener] starting listener on port {port}"); - - Ok(Self { - listener: Some(listener), - client_context, - own_username, - actor_system, - }) - } -} - -impl Actor for ListenerActor { - type Message = ListenerMessage; - - fn handle(&mut self, msg: Self::Message) { - match msg { - ListenerMessage::Stop => { - info!("[listener] stopping listener"); - self.listener = None; - } - } - } - - fn tick(&mut self) { - let Some(ref listener) = self.listener else { - return; - }; - - match listener.accept() { - Ok((stream, _addr)) => { - trace!("[listener] accepted new connection"); - let actor = ConnectionHandlerActor::new( - stream, - self.client_context.clone(), - self.own_username.clone(), - ); - self.actor_system.spawn(actor); - } - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { - // No incoming connections, this is normal for non-blocking - } - Err(e) => { - error!("[listener] Failed to accept connection: {}", e); - } - } - } -} diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index 4be1294..36cf6e0 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -15,6 +15,14 @@ use std::io::{self, Error, Write}; use std::net::TcpStream; use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + +#[derive(Debug, Clone)] +enum ConnectionState { + Disconnected, + Connecting { since: Instant }, + Connected, +} #[derive(Debug, Clone)] pub enum PeerMessage { @@ -37,46 +45,49 @@ pub enum PeerMessage { ProcessRead, } -/// Actor that manages a single peer connection pub struct PeerActor { peer: Arc>, stream: Option, + connection_state: ConnectionState, reader: MessageReader, client_channel: Sender, self_handle: Option>, dispatcher: Option>, dispatcher_receiver: Option>, - queued_filenames: Vec, + queued_messages: Vec, } impl PeerActor { - /// Create a new PeerActor with an existing socket connection pub fn new( peer: Peer, - stream: TcpStream, + stream: Option, reader: Option, client_channel: Sender, ) -> Self { + let connection_state = if stream.is_some() { + ConnectionState::Connected + } else { + ConnectionState::Disconnected + }; + Self { peer: Arc::new(RwLock::new(peer)), - stream: Some(stream), + stream, + connection_state, reader: reader.unwrap_or_default(), client_channel, self_handle: None, dispatcher: None, dispatcher_receiver: None, - queued_filenames: Vec::new(), + queued_messages: Vec::new(), } } - /// Set the actor's own handle for self-messaging pub fn set_self_handle(&mut self, handle: ActorHandle) { self.self_handle = Some(handle); } - /// Initialize the message dispatcher with handlers fn initialize_dispatcher(&mut self) { - // Create a channel for the dispatcher to send messages to let (dispatcher_sender, dispatcher_receiver) = std::sync::mpsc::channel::(); @@ -98,28 +109,34 @@ impl PeerActor { )); } - /// Process any messages from the dispatcher fn process_dispatcher_messages(&mut self) { - // Collect all messages first to avoid borrow issues - let messages: Vec = - if let Some(ref receiver) = self.dispatcher_receiver { + let messages: Vec = self + .dispatcher_receiver + .as_ref() + .map_or_else(Vec::new, |receiver| { let mut msgs = Vec::new(); while let Ok(msg) = receiver.try_recv() { msgs.push(msg); } msgs - } else { - Vec::new() - }; + }); - // Process all collected messages - for msg in messages { - self.handle_message(msg); - } + messages + .iter() + .for_each(|msg| self.handle_message(msg.clone())); } - /// Internal handler that doesn't call Actor::handle fn handle_message(&mut self, msg: PeerMessage) { + if matches!(self.connection_state, ConnectionState::Connecting { .. }) { + match &msg { + PeerMessage::SetUsername(_) | PeerMessage::ProcessRead => {} + _ => { + self.queued_messages.push(msg); + return; + } + } + } + match msg { PeerMessage::SendMessage(message) => { self.send_message(message); @@ -202,7 +219,6 @@ impl PeerActor { self.peer.write().unwrap().username = username; } PeerMessage::QueueUpload(filename) => { - self.queued_filenames.push(filename.clone()); let message = MessageFactory::build_queue_upload_message(&filename); self.send_message(message); @@ -225,21 +241,15 @@ impl PeerActor { } } - /// Handle reading from the socket and dispatching messages fn process_read(&mut self) { if self.reader.buffer_len() > 0 { - self.extract_and_process_messages(); // This will process all messages in the buffer + self.extract_and_process_messages(); } - let username = self.peer.read().unwrap().username.clone(); - // Try to read more data from the socket first { let stream = match self.stream.as_mut() { Some(s) => s, - None => { - trace!("[peer_actor:{}] process_read: stream is None, returning", username); - return; - } + None => return, }; match self.reader.read_from_socket(stream) { @@ -263,7 +273,7 @@ impl PeerActor { } } } - self.extract_and_process_messages(); // This will process all messages in the buffer + self.extract_and_process_messages(); } fn extract_and_process_messages(&mut self) { @@ -307,11 +317,9 @@ impl PeerActor { } } - // Process any messages that were dispatched self.process_dispatcher_messages(); } - /// Handle writing a message to the socket fn send_message(&mut self, message: Message) { let stream = match self.stream.as_mut() { Some(s) => s, @@ -365,7 +373,6 @@ impl PeerActor { error!("Failed to send disconnect notification: {}", e); } } - /// Disconnect the peer and notify the client fn disconnect(&mut self) { let username = self.peer.read().unwrap().username.clone(); debug!("[peer_actor:{}] disconnect", username); @@ -379,6 +386,145 @@ impl PeerActor { error!("Failed to send disconnect notification: {}", e); } } + + fn initiate_connection(&mut self) -> bool { + let peer = self.peer.read().unwrap(); + let username = peer.username.clone(); + let host = peer.host.clone(); + let port = peer.port; + drop(peer); + + let socket_addr = + format!("{}:{}", host, port).parse::(); + + match socket_addr { + Ok(addr) => { + if let Ok(stream) = TcpStream::connect(addr) { + if let Err(e) = stream.set_nonblocking(true) { + error!( + "[peer_actor:{}] Failed to set non-blocking: {}", + username, e + ); + self.disconnect_with_error(e); + return false; + } + + stream.set_nodelay(true).ok(); + + self.stream = Some(stream); + self.connection_state = ConnectionState::Connecting { + since: Instant::now(), + }; + true + } else { + match TcpStream::connect(addr) { + Ok(stream) => { + if let Err(e) = stream.set_nonblocking(true) { + error!("[peer_actor:{}] Failed to set non-blocking: {}", username, e); + self.disconnect_with_error(e); + return false; + } + stream.set_nodelay(true).ok(); + self.stream = Some(stream); + self.connection_state = + ConnectionState::Connecting { + since: Instant::now(), + }; + true + } + Err(e) => { + self.disconnect_with_error(e); + false + } + } + } + } + Err(e) => { + error!( + "[peer_actor:{}] Invalid socket address {}:{} - {}", + username, host, port, e + ); + self.disconnect_with_error(io::Error::new( + io::ErrorKind::InvalidInput, + e, + )); + false + } + } + } + + fn check_connection_status(&mut self) { + let ConnectionState::Connecting { since } = self.connection_state + else { + return; + }; + + let username = self.peer.read().unwrap().username.clone(); + + if since.elapsed() > Duration::from_secs(20) { + error!( + "[peer_actor:{}] Connection timeout after 20 seconds", + username + ); + self.disconnect_with_error(io::Error::new( + io::ErrorKind::TimedOut, + "Connection timeout", + )); + return; + } + + let Some(ref stream) = self.stream else { + return; + }; + + match stream.peer_addr() { + Ok(_) => { + self.connection_state = ConnectionState::Connected; + self.on_connection_established(); + } + Err(ref e) if e.kind() == io::ErrorKind::NotConnected => {} + Err(e) => { + error!("[peer_actor:{}] Connection failed: {}", username, e); + self.disconnect_with_error(e); + } + } + } + + fn on_connection_established(&mut self) { + let peer = self.peer.read().unwrap(); + let username = peer.username.clone(); + let token = peer.token; + drop(peer); + + let Some(ref mut stream) = self.stream else { + return; + }; + + if let Some(token) = token { + let handshake_msg = MessageFactory::build_watch_user(token); + if let Err(e) = stream.write_all(&handshake_msg.get_data()) { + error!( + "[peer_actor:{}] Failed to send watch user handshake: {}", + username, e + ); + self.disconnect_with_error(e); + return; + } + } + + self.initialize_dispatcher(); + + let queued = std::mem::take(&mut self.queued_messages); + for msg in queued { + self.handle_message(msg); + } + + if let Some(ref handle) = self.self_handle { + handle.send(PeerMessage::ProcessRead).ok(); + } + + self.process_read(); + } } impl Actor for PeerActor { @@ -389,29 +535,12 @@ impl Actor for PeerActor { } fn on_start(&mut self) { - let username = self.peer.read().unwrap().username.clone(); - - // Initialize the dispatcher - self.initialize_dispatcher(); - - // Trigger initial read - if let Some(ref handle) = self.self_handle { - match handle.send(PeerMessage::ProcessRead) { - Ok(_) => {} - Err(e) => error!( - "[peer_actor:{}] FAILED to send ProcessRead: {}", - username, e - ), - } + if self.stream.is_none() { + self.initiate_connection(); } else { - error!( - "[peer_actor:{}] self_handle is None! Cannot send ProcessRead", - username - ); + self.connection_state = ConnectionState::Connected; + self.on_connection_established(); } - - // Also do an immediate read as fallback - self.process_read(); } fn on_stop(&mut self) { @@ -421,15 +550,20 @@ impl Actor for PeerActor { } fn tick(&mut self) { - // Periodically check for new data on the socket - if self.stream.is_some() { - self.process_read(); - } else { - let username = self.peer.read().unwrap().username.clone(); - trace!( - "[peer_actor:{}] tick() called but stream is None", - username - ); + match self.connection_state { + ConnectionState::Connecting { .. } => { + // Check if connection has been established + self.check_connection_status(); + } + ConnectionState::Connected => { + // Normal operation - process reads + if self.stream.is_some() { + self.process_read(); + } + } + ConnectionState::Disconnected => { + // Do nothing + } } } } diff --git a/src/actor/peer_registry.rs b/src/actor/peer_registry.rs index 992396e..a7dc1b7 100644 --- a/src/actor/peer_registry.rs +++ b/src/actor/peer_registry.rs @@ -1,16 +1,15 @@ use crate::actor::peer_actor::{PeerActor, PeerMessage}; use crate::actor::{ActorHandle, ActorSystem}; use crate::client::ClientOperation; +use crate::debug; use crate::message::MessageReader; use crate::peer::Peer; -use crate::{debug, trace}; use std::collections::HashMap; use std::net::TcpStream; use std::sync::mpsc::Sender; use std::sync::{Arc, Mutex}; -/// Registry for managing peer actors pub struct PeerRegistry { peers: Arc>>>, actor_system: Arc, @@ -29,45 +28,33 @@ impl PeerRegistry { } } - /// Spawn a new peer actor and register it pub fn register_peer( &self, peer: Peer, - stream: TcpStream, + stream: Option, reader: Option, ) -> Result, String> { let username = peer.username.clone(); - debug!("[peer_registry] Registering peer actor for {}", username); - let actor = PeerActor::new(peer, stream, reader, self.client_channel.clone()); - // Spawn the actor with initialization callback to set self_handle let handle = self.actor_system.spawn_with_handle(actor, |actor, handle| { actor.set_self_handle(handle); }); - // Store in registry let mut peers = self.peers.lock().unwrap(); peers.insert(username.clone(), handle.clone()); - trace!( - "[peer_registry] Peer actor {} registered successfully", - username - ); - Ok(handle) } - /// Get a peer actor handle by username pub fn get_peer(&self, username: &str) -> Option> { let peers = self.peers.lock().unwrap(); peers.get(username).cloned() } - /// Remove a peer actor from the registry pub fn remove_peer( &self, username: &str, @@ -82,25 +69,21 @@ impl PeerRegistry { handle } - /// Get all registered peer usernames pub fn get_all_usernames(&self) -> Vec { let peers = self.peers.lock().unwrap(); peers.keys().cloned().collect() } - /// Get count of registered peers pub fn count(&self) -> usize { let peers = self.peers.lock().unwrap(); peers.len() } - /// Check if a peer is registered pub fn contains(&self, username: &str) -> bool { let peers = self.peers.lock().unwrap(); peers.contains_key(username) } - /// Send a message to a peer by username pub fn send_to_peer( &self, username: &str, diff --git a/src/client.rs b/src/client.rs index e5adc61..b5aae32 100644 --- a/src/client.rs +++ b/src/client.rs @@ -318,82 +318,96 @@ impl Client { own_username: String, ) { thread::spawn(move || loop { - if let Ok(operation) = reader.recv() { - match operation { - ClientOperation::ConnectToPeer(peer) => { - trace!("[client] ConnectToPeer {:?}", peer); - Self::connect_to_peer( - peer, - client_context.clone(), - own_username.clone(), - None, - ); - } - ClientOperation::SearchResult(file_search) => { - trace!("[client] SearchResult {:?}", file_search); - client_context - .write() - .unwrap() - .search_results - .push(file_search.clone()); - } - ClientOperation::PeerDisconnected(username, error) => { - let context = client_context.read().unwrap(); - if let Some(ref registry) = context.peer_registry { - if let Some(handle) = - registry.remove_peer(&username) - { - let _ = handle.stop(); - } + match reader.recv() { + Ok(operation) => { + match operation { + ClientOperation::ConnectToPeer(peer) => { + let client_context_clone = client_context.clone(); + let own_username_clone = own_username.clone(); + + // Spawn in background to avoid blocking the receiver thread + thread::spawn(move || { + Self::connect_to_peer( + peer, + client_context_clone, + own_username_clone, + None, + ); + }); } - if let Some(error) = error { - error!( + ClientOperation::SearchResult(file_search) => { + trace!("[client] SearchResult {:?}", file_search); + client_context + .write() + .unwrap() + .search_results + .push(file_search.clone()); + } + ClientOperation::PeerDisconnected(username, error) => { + let context = client_context.read().unwrap(); + if let Some(ref registry) = context.peer_registry { + if let Some(handle) = + registry.remove_peer(&username) + { + let _ = handle.stop(); + } + } + if let Some(error) = error { + error!( "[client] Peer {} disconnected with error: {:?}", username, error ); - Self::process_failed_uploads( + Self::process_failed_uploads( + client_context.clone(), + &username, + None, + ); + } + } + ClientOperation::PierceFireWall(peer) => { + Self::pierce_firewall( + peer, client_context.clone(), - &username, - None, + own_username.clone(), ); } - } - ClientOperation::PierceFireWall(peer) => { - Self::pierce_firewall( + ClientOperation::DownloadFromPeer( + token, peer, - client_context.clone(), - own_username.clone(), - ); - } - ClientOperation::DownloadFromPeer(token, peer, allowed) => { - let maybe_download = { - let client_context = client_context.read().unwrap(); - client_context.download_tokens.get(&token).cloned() - }; - let own_username = own_username.clone(); - let client_context_clone = client_context.clone(); + allowed, + ) => { + let maybe_download = { + let client_context = + client_context.read().unwrap(); + client_context + .download_tokens + .get(&token) + .cloned() + }; + let own_username = own_username.clone(); + let client_context_clone = client_context.clone(); - trace!( + trace!( "[client] DownloadFromPeer token: {} peer: {:?}", token, peer ); - return match maybe_download { - Some(download) => { - thread::spawn(move || { - let download_peer = DownloadPeer::new( - download.username.clone(), - peer.host.clone(), - peer.port, - token, - allowed, - own_username, - ); - let filename: Option<&str> = download - .filename - .split('\\') - .next_back(); - match filename { + match maybe_download { + Some(download) => { + thread::spawn(move || { + let download_peer = DownloadPeer::new( + download.username.clone(), + peer.host.clone(), + peer.port, + token, + allowed, + own_username, + ); + let filename: Option<&str> = download + .filename + .split('\\') + .next_back(); + match filename { Some(filename) => { match download_peer.download_file( client_context_clone, @@ -415,164 +429,176 @@ impl Client { } None => error!("Cant find filename to save download: {:?}", download.filename), } - }); - } - None => { - error!( - "Can't find download with token {:?}", - token - ); - } - }; - } - ClientOperation::NewPeer(new_peer) => { - let peer_exists = client_context - .read() - .unwrap() - .peer_registry - .as_ref() - .map(|r| r.contains(&new_peer.username)) - .unwrap_or(false); - - if peer_exists { - debug!( - "Already connected to {}", - new_peer.username - ); - } else if let Some(server_sender) = - &client_context.read().unwrap().server_sender - { - server_sender - .send(ServerOperation::GetPeerAddress( - new_peer.username.clone(), - )) - .unwrap(); + }); + } + None => { + error!( + "Can't find download with token {:?}", + token + ); + } + }; } + ClientOperation::NewPeer(new_peer) => { + let peer_exists = client_context + .read() + .unwrap() + .peer_registry + .as_ref() + .map(|r| r.contains(&new_peer.username)) + .unwrap_or(false); - let addr = new_peer.tcp_stream.peer_addr().unwrap(); - let host = addr.ip().to_string(); - let port: u32 = addr.port().into(); + if peer_exists { + debug!( + "Already connected to {}", + new_peer.username + ); + } else if let Some(server_sender) = + &client_context.read().unwrap().server_sender + { + server_sender + .send(ServerOperation::GetPeerAddress( + new_peer.username.clone(), + )) + .unwrap(); + } - let peer = Peer { - username: new_peer.username.clone(), - connection_type: new_peer.connection_type, - host, - port, - token: Some(new_peer.token), - privileged: None, - obfuscated_port: None, - unknown: None, - }; + let addr = new_peer.tcp_stream.peer_addr().unwrap(); + let host = addr.ip().to_string(); + let port: u32 = addr.port().into(); - Self::connect_to_peer( - peer, - client_context.clone(), - own_username.clone(), - Some(new_peer.tcp_stream), - ); - } - ClientOperation::GetPeerAddressResponse { - username, - host, - port, - obfuscation_type, - obfuscated_port, - } => { - debug!( - "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", - username, host, port, obfuscation_type, obfuscated_port - ); - - let peer_exists = client_context - .read() - .unwrap() - .peer_registry - .as_ref() - .map(|r| r.contains(&username)) - .unwrap_or(false); - - if peer_exists { - // don't know if i should update? and or reconnect the peer - // debug!( - // "existing peer: {:?}, new peer details: - // username: {}, - // host: {}, - // port: {} - // obfuscation_type: {} - // obfuscated_port: {}", - // peer, - // username, - // host, - // port, - // obfuscation_type, - // obfuscated_port, - // ); - } else { - let peer = Peer::new( - username, - ConnectionType::P, + let peer = Peer { + username: new_peer.username.clone(), + connection_type: new_peer.connection_type, host, port, - None, - 0, - obfuscation_type.try_into().unwrap(), - obfuscated_port.try_into().unwrap(), - ); + token: Some(new_peer.token), + privileged: None, + obfuscated_port: None, + unknown: None, + }; + Self::connect_to_peer( peer, client_context.clone(), own_username.clone(), - None, + Some(new_peer.tcp_stream), ); } - } - ClientOperation::UpdateDownloadTokens( - transfer, - username, - ) => { - let mut context = client_context.write().unwrap(); + ClientOperation::GetPeerAddressResponse { + username, + host, + port, + obfuscation_type, + obfuscated_port, + } => { + debug!( + "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", + username, host, port, obfuscation_type, obfuscated_port + ); - let key_to_remove = context - .download_tokens - .iter() - .find_map(|(key, d)| { - if d.username == username - && d.filename == transfer.filename - { - Some((*key, d.clone())) - } else { - None - } - }); + let peer_exists = client_context + .read() + .unwrap() + .peer_registry + .as_ref() + .map(|r| r.contains(&username)) + .unwrap_or(false); - if let Some((key, download)) = key_to_remove { - trace!( + if peer_exists { + // don't know if i should update? and or reconnect the peer + // debug!( + // "existing peer: {:?}, new peer details: + // username: {}, + // host: {}, + // port: {} + // obfuscation_type: {} + // obfuscated_port: {}", + // peer, + // username, + // host, + // port, + // obfuscation_type, + // obfuscated_port, + // ); + } else { + let peer = Peer::new( + username, + ConnectionType::P, + host, + port, + None, + 0, + obfuscation_type.try_into().unwrap(), + obfuscated_port.try_into().unwrap(), + ); + let client_context_clone = + client_context.clone(); + let own_username_clone = own_username.clone(); + + // Spawn in background to avoid blocking the receiver thread + thread::spawn(move || { + Self::connect_to_peer( + peer, + client_context_clone, + own_username_clone, + None, + ); + }); + } + } + ClientOperation::UpdateDownloadTokens( + transfer, + username, + ) => { + let mut context = client_context.write().unwrap(); + + let key_to_remove = context + .download_tokens + .iter() + .find_map(|(key, d)| { + if d.username == username + && d.filename == transfer.filename + { + Some((*key, d.clone())) + } else { + None + } + }); + + if let Some((key, download)) = key_to_remove { + trace!( "[client] UpdateDownloadTokens found {key}, transfer: {:?}", transfer ); - context.download_tokens.insert( - transfer.token, - Download { - username: username.clone(), - filename: transfer.filename, - token: transfer.token, - size: transfer.size, - download_directory: download - .download_directory, - sender: download.sender.clone(), - }, + context.download_tokens.insert( + transfer.token, + Download { + username: username.clone(), + filename: transfer.filename, + token: transfer.token, + size: transfer.size, + download_directory: download + .download_directory, + sender: download.sender.clone(), + }, + ); + context.download_tokens.remove(&key); + } + } + ClientOperation::UploadFailed(username, filename) => { + Self::process_failed_uploads( + client_context.clone(), + &username, + Some(&filename), ); - context.download_tokens.remove(&key); } } - ClientOperation::UploadFailed(username, filename) => { - Self::process_failed_uploads( - client_context.clone(), - &username, - Some(&filename), - ); - } + } + Err(e) => { + error!("[client] Channel recv error: {:?}", e); + break; } } }); @@ -584,168 +610,70 @@ impl Client { own_username: String, stream: Option, ) { - let client_context2 = client_context.clone(); - let unlocked_context = client_context.read().unwrap(); + let client_context = client_context.clone(); - trace!("[client] connect_to_peer: {}", peer.username); - if let Some(_sender) = unlocked_context.sender.clone() { - let peer_clone = peer.clone(); - trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); - match peer.connection_type { - ConnectionType::P => { - let username = peer.username.clone(); + let peer_clone = peer.clone(); + trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); + match peer.connection_type { + ConnectionType::P => { + let username = peer.username.clone(); - let mut tcp_stream = match stream { - Some(s) => s, - None => { - let socket_addr = format!( - "{}:{}", - peer_clone.host, peer_clone.port - ) - .parse::(); - - match socket_addr { - Ok(addr) => { - match TcpStream::connect_timeout( - &addr, - Duration::from_secs(20), - ) { - Ok(s) => s, - Err(e) => { - trace!( - "Can't connect to {:?} {:?}:{:?} - {:?}", - username, peer_clone.host, peer_clone.port, e - ); - return; - } - } - } - Err(e) => { - trace!( - "Invalid socket address for {:?}: {:?}", - username, - e - ); - return; - } - } - } - }; - - // Send connection handshake message - use std::io::Write; - if let Some(token) = peer_clone.token { - // For indirect connections, send watch user message - let handshake_msg = crate::message::server::MessageFactory::build_watch_user(token); - if let Err(e) = - tcp_stream.write_all(&handshake_msg.get_data()) - { - trace!("Failed to send watch user handshake to {:?}: {:?}", username, e); - return; - } - trace!( - "[client] Sent watch user handshake for token: {}", - token - ); - } - - // Configure socket - if let Err(e) = tcp_stream - .set_read_timeout(Some(Duration::from_secs(5))) - { - trace!( - "Failed to set read timeout for {:?}: {:?}", - username, - e - ); - return; - } - if let Err(e) = tcp_stream - .set_write_timeout(Some(Duration::from_secs(5))) - { - trace!( - "Failed to set write timeout for {:?}: {:?}", - username, - e - ); - return; - } - if let Err(e) = tcp_stream.set_nodelay(true) { - trace!( - "Failed to set nodelay for {:?}: {:?}", - username, - e - ); - return; - } - - // Spawn peer actor - let context = client_context2.read().unwrap(); - if let Some(ref registry) = context.peer_registry { - match registry - .register_peer(peer_clone, tcp_stream, None) - { - Ok(_) => { - trace!( - "[client] peer actor spawned for: {}", - username - ); - } - Err(e) => { - trace!( - "Failed to spawn peer actor for {:?}: {:?}", - username, - e - ); - } - } - } else { - trace!("PeerRegistry not initialized"); - } - } - - ConnectionType::F => { - trace!( - "[client] downloading from: {}, {:?}", - peer.username, - peer.token - ); - let download_peer = DownloadPeer::new( - peer.username, - peer.host, - peer.port, - peer.token.unwrap(), - false, - own_username.clone(), - ); - - match download_peer.download_file( - client_context2.clone(), - None, - None, - ) { - Ok((download, filename)) => { - trace!( - "[client] downloaded {} bytes {:?} ", - filename, - download.size - ); - download - .sender - .send(DownloadStatus::Completed) - .unwrap(); - } + let context = client_context.read().unwrap(); + if let Some(ref registry) = context.peer_registry { + match registry.register_peer(peer_clone, stream, None) { + Ok(_) => (), Err(e) => { - trace!("[client] failed to download: {}", e); + trace!( + "Failed to spawn peer actor for {:?}: {:?}", + username, + e + ); } } - } - ConnectionType::D => { - error!("ConnectionType::D not implemented") + } else { + trace!("PeerRegistry not initialized"); } } - } else { - debug!("[client] Peer already connected: {}", peer.username); + + ConnectionType::F => { + trace!( + "[client] downloading from: {}, {:?}", + peer.username, + peer.token + ); + let download_peer = DownloadPeer::new( + peer.username, + peer.host, + peer.port, + peer.token.unwrap(), + false, + own_username.clone(), + ); + + match download_peer.download_file( + client_context.clone(), + None, + None, + ) { + Ok((download, filename)) => { + trace!( + "[client] downloaded {} bytes {:?} ", + filename, + download.size + ); + download + .sender + .send(DownloadStatus::Completed) + .unwrap(); + } + Err(e) => { + trace!("[client] failed to download: {}", e); + } + } + } + ConnectionType::D => { + error!("ConnectionType::D not implemented") + } } } fn pierce_firewall( diff --git a/src/main.rs b/src/main.rs index 55d15bd..bbc4458 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,7 +14,7 @@ fn main() { match client.login() { Ok(_) => { match client - .search("Eternity Alex Kassian", Duration::from_secs(10)) + .search("Eternity Alex Kassian", Duration::from_secs(30)) { Ok(results) => { if let Some(result) = diff --git a/src/peer/listen.rs b/src/peer/listen.rs index 5f90be5..080fc31 100644 --- a/src/peer/listen.rs +++ b/src/peer/listen.rs @@ -75,14 +75,7 @@ fn extract_download_from_buffer( if reader.buffer_len() == 0 { return None; } - let buffer = reader.get_buffer(); - trace!( - "[listener:{peer_ip}:{peer_port}] reader buffer has {} bytes. {:?}", - buffer.len(), - buffer - ); - let token = parse_token_from_buffer(&buffer, username)?; trace!( "[listener:{}] got transfer_token: {} from data chunk", @@ -107,18 +100,13 @@ fn handle_peer_connection( stream: TcpStream, reader: MessageReader, context: &ConnectionContext, - peer_ip: &str, - peer_port: u16, + _peer_ip: &str, + _peer_port: u16, ) { - debug!("[listener:{peer_ip}:{peer_port}] connection type is P, reader buffer has {} bytes", reader.buffer_len()); - - // Spawn peer actor directly using the peer registry let client_context = context.client_context.read().unwrap(); if let Some(ref registry) = client_context.peer_registry { - match registry.register_peer(peer.clone(), stream, Some(reader)) { - Ok(_) => { - trace!("[listener] peer actor spawned for: {}", peer.username); - } + match registry.register_peer(peer.clone(), Some(stream), Some(reader)) { + Ok(_) => (), Err(e) => { error!( "Failed to spawn peer actor for {:?}: {:?}", @@ -213,7 +201,7 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { ); let peer = Peer::new( - format!("{}:direct", init_data.username), // Add direct suffix to distinguish from regular peers + format!("{}:direct", init_data.username), // init_data.username.clone(), init_data.connection_type.clone(), peer_ip.clone(), diff --git a/src/server.rs b/src/server.rs index d8cc7d9..795773c 100644 --- a/src/server.rs +++ b/src/server.rs @@ -341,10 +341,6 @@ impl Server { if let Ok(operation) = server_channel.recv() { match operation { ServerOperation::ConnectToPeer(peer) => { - trace!( - "[server] → ConnectToPeer {}", - peer.username - ); if let Some(op) = match peer.connection_type { ConnectionType::P | ConnectionType::F => { Some(ClientOperation::ConnectToPeer( @@ -353,18 +349,7 @@ impl Server { } ConnectionType::D => None, } { - trace!( - "[server] → ConnectToPeer {} - op {:?}", - peer.username, - op - ); - - match client_channel.send(op) { - Ok(_) => {} - Err(e) => { - error!("[server] Failed to send: {}", e) - } - } + client_channel.send(op).unwrap(); } } ServerOperation::LoginStatus(message) => {