feat: actor refactoring default peer

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<M: Send> {
+    pub(crate) sender: Sender<ActorMessage<M>>,
+}
+
+impl<M: Send> ActorHandle<M> {
+    /// 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<M> {
+    UserMessage(M),
+    Stop,
+    Tick,
+}
+
+/// Actor system that manages actor lifecycle
+pub struct ActorSystem {
+    thread_pool: Arc<ThreadPool>,
+}
+
+impl ActorSystem {
+    pub fn new(thread_pool: Arc<ThreadPool>) -> Self {
+        ActorSystem { thread_pool }
+    }
+
+    /// Spawn a new actor and return its handle
+    pub fn spawn<A: Actor>(&self, mut actor: A) -> ActorHandle<A::Message> {
+        let (sender, receiver) = channel::<ActorMessage<A::Message>>();
+        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<A: Actor, F>(&self, mut actor: A, init: F) -> ActorHandle<A::Message>
+    where
+        F: FnOnce(&mut A, ActorHandle<A::Message>) + Send + 'static,
+    {
+        let (sender, receiver) = channel::<ActorMessage<A::Message>>();
+        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<A: Actor>(actor: &mut A, receiver: Receiver<ActorMessage<A::Message>>) {
+        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<AtomicUsize>,
+    }
+
+    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();
+    }
+}

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<String>,
+    },
+    /// 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<RwLock<Peer>>,
+    stream: Option<TcpStream>,
+    reader: MessageReader,
+    client_channel: Sender<ClientOperation>,
+    self_handle: Option<ActorHandle<PeerMessage>>,
+    dispatcher: Option<MessageDispatcher<PeerMessage>>,
+    dispatcher_receiver: Option<Receiver<PeerMessage>>,
+}
+
+impl PeerActor {
+    /// Create a new PeerActor with an existing socket connection
+    pub fn new(
+        peer: Peer,
+        stream: TcpStream,
+        reader: Option<MessageReader>,
+        client_channel: Sender<ClientOperation>,
+    ) -> 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<PeerMessage>) {
+        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::<PeerMessage>();
+
+        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<PeerMessage> = 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);
+        }
+    }
+}
+

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<Mutex<HashMap<String, ActorHandle<PeerMessage>>>>,
+    actor_system: Arc<ActorSystem>,
+    client_channel: Sender<ClientOperation>,
+}
+
+impl PeerRegistry {
+    pub fn new(actor_system: Arc<ActorSystem>, client_channel: Sender<ClientOperation>) -> 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<MessageReader>,
+    ) -> Result<ActorHandle<PeerMessage>, 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<ActorHandle<PeerMessage>> {
+        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<ActorHandle<PeerMessage>> {
+        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<String> {
+        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(),
+        }
+    }
+}

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<String, DefaultPeer>,
+    pub peer_registry: Option<PeerRegistry>,
     sender: Option<Sender<ClientOperation>>,
     server_sender: Option<Sender<crate::server::ServerOperation>>,
     search_results: Vec<FileSearchResult>,
     pub download_tokens: HashMap<u32, Download>,
-    thread_pool: ThreadPool,
+    thread_pool: Arc<ThreadPool>,
+    actor_system: Arc<ActorSystem>,
 }
 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::<std::net::SocketAddr>();
 
-                            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");
                             }
                         }
 

src/lib.rs

@@ -1,4 +1,5 @@
 // Core modules
+pub mod actor;
 pub mod client;
 pub mod dispatcher;
 pub mod error;

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,
     );
 

src/message/handlers.rs

@@ -3,12 +3,12 @@ use std::collections::HashMap;
 use crate::message::Message;
 use std::sync::mpsc::Sender;
 
-pub trait MessageHandler<Op> {
+pub trait MessageHandler<Op>: Send {
     fn get_code(&self) -> u8;
     fn handle(&self, message: &mut Message, sender: Sender<Op>);
 }
 pub struct Handlers<Op> {
-    handlers: HashMap<u8, Box<dyn MessageHandler<Op>>>,
+    handlers: HashMap<u8, Box<dyn MessageHandler<Op> + Send>>,
 }
 
 impl<Op> Default for Handlers<Op> {
@@ -31,7 +31,7 @@ impl<Op> Handlers<Op> {
         self.handlers.insert(handler.get_code(), Box::new(handler));
         self
     }
-    pub fn get_handler(&self, code: u8) -> Option<&dyn MessageHandler<Op>> {
+    pub fn get_handler(&self, code: u8) -> Option<&(dyn MessageHandler<Op> + Send)> {
         self.handlers.get(&code).map(|v| &**v)
     }
 }

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<PeerOperation> for FileSearchResponse {
+impl MessageHandler<PeerMessage> for FileSearchResponse {
     fn get_code(&self) -> u8 {
         9
     }
-    fn handle(&self, message: &mut Message, sender: Sender<PeerOperation>) {
+    fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
         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();
     }
 }

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<PeerOperation> for GetShareFileList {
+impl MessageHandler<PeerMessage> for GetShareFileList {
     fn get_code(&self) -> u8 {
         4
     }
-    fn handle(&self, _message: &mut Message, sender: Sender<PeerOperation>) {
+    fn handle(&self, _message: &mut Message, sender: Sender<PeerMessage>) {
         let message = MessageFactory::build_shared_folders_message(100, 800);
 
-        sender.send(PeerOperation::SendMessage(message)).unwrap();
+        sender.send(PeerMessage::SendMessage(message)).unwrap();
     }
 }

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<PeerOperation> for PeerInit {
+impl MessageHandler<PeerMessage> for PeerInit {
     fn get_code(&self) -> u8 {
         1
     }
 
-    fn handle(&self, message: &mut Message, sender: Sender<PeerOperation>) {
+    fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
         message.set_pointer(4);
         let _message_code = message.read_int8();
         let username = message.read_string();
@@ -25,6 +25,6 @@ impl MessageHandler<PeerOperation> for PeerInit {
             token
         );
 
-        sender.send(PeerOperation::SetUsername(username)).unwrap();
+        sender.send(PeerMessage::SetUsername(username)).unwrap();
     }
 }

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<PeerOperation> for PlaceInQueueResponse {
+impl MessageHandler<PeerMessage> for PlaceInQueueResponse {
     fn get_code(&self) -> u8 {
         43
     }
 
-    fn handle(&self, message: &mut Message, sender: Sender<PeerOperation>) {
+    fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
         let filename = message.read_string();
         let place = message.read_int32();
 
         sender
-            .send(PeerOperation::PlaceInQueueResponse { filename, place })
+            .send(PeerMessage::PlaceInQueueResponse { filename, place })
             .unwrap();
     }
 }

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<PeerOperation> for TransferRequest {
+impl MessageHandler<PeerMessage> for TransferRequest {
     fn get_code(&self) -> u8 {
         40
     }
-    fn handle(&self, message: &mut Message, sender: Sender<PeerOperation>) {
+    fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
         let transfer = Transfer::new_from_message(message);
 
         sender
-            .send(PeerOperation::TransferRequest(transfer))
+            .send(PeerMessage::TransferRequest(transfer))
             .unwrap();
     }
 }

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<PeerOperation> for TransferResponse {
+impl MessageHandler<PeerMessage> for TransferResponse {
     fn get_code(&self) -> u8 {
         41
     }
 
-    fn handle(&self, message: &mut Message, sender: Sender<PeerOperation>) {
+    fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
         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,

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<PeerOperation> for UploadFailedHandler {
+impl MessageHandler<PeerMessage> for UploadFailedHandler {
     fn get_code(&self) -> u8 {
         46
     }
-    fn handle(&self, message: &mut Message, _sender: Sender<PeerOperation>) {
+    fn handle(&self, message: &mut Message, _sender: Sender<PeerMessage>) {
         let upload_failed = UploadFailed::new_from_message(message);
         info!("Upload failed for ${}", upload_failed.filename);
     }

src/peer/default_peer.rs

@@ -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<RwLock<Peer>>,
-    peer_channel: Option<Sender<PeerOperation>>,
-    client_channel: Sender<ClientOperation>,
-    read_thread: Option<JoinHandle<()>>,
-    write_thread: Option<JoinHandle<()>>,
-}
-
-#[allow(dead_code)]
-pub enum PeerOperation {
-    SendMessage(Message),
-    FileSearchResult(FileSearchResult),
-    TransferRequest(Transfer),
-    TransferResponse {
-        token: u32,
-        allowed: bool,
-        reason: Option<String>,
-    },
-    PlaceInQueueResponse {
-        filename: String,
-        place: u32,
-    },
-    SetUsername(String),
-}
-
-impl DefaultPeer {
-    pub fn new(peer: Peer, client_channel: Sender<ClientOperation>) -> 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<MessageReader>,
-    ) -> Result<Self, io::Error> {
-        if let Some(token) = self.peer.read().unwrap().token {
-            let mut message: Vec<u8> = [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<Self, io::Error> {
-        {
-            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<MessageReader>,
-    ) -> Result<(), io::Error> {
-        let (peer_sender, peer_reader): (
-            Sender<PeerOperation>,
-            Receiver<PeerOperation>,
-        ) = 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),
-            }
-        }
-    }
-}

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<ClientOperation>,
+    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,
         ),

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;