From 3dc667fe3049ae042a68b48d8ba5c1b85433e63c Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 13:54:44 +0100 Subject: [PATCH] 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;