Merge branch 'feat/server_actor'

src/actor/mod.rs

@@ -1,6 +1,3 @@
-// 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};
@@ -10,7 +7,14 @@ use crate::utils::thread_pool::ThreadPool;
 
 pub mod peer_actor;
 pub mod peer_registry;
+pub mod server_actor;
 
+#[derive(Debug, Clone)]
+pub enum ConnectionState {
+    Disconnected,
+    Connecting { since: Instant },
+    Connected,
+}
 /// Core actor trait - each actor processes messages
 pub trait Actor: Send + 'static {
     type Message: Send + Clone + 'static;
@@ -28,14 +32,12 @@ pub trait Actor: Send + 'static {
     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))

src/actor/peer_actor.rs

@@ -1,4 +1,4 @@
-use crate::actor::{Actor, ActorHandle};
+use crate::actor::{Actor, ActorHandle, ConnectionState};
 use crate::client::ClientOperation;
 use crate::dispatcher::MessageDispatcher;
 use crate::message::peer::{
@@ -17,13 +17,6 @@ use std::sync::mpsc::{Receiver, Sender};
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 
-#[derive(Debug, Clone)]
-enum ConnectionState {
-    Disconnected,
-    Connecting { since: Instant },
-    Connected,
-}
-
 #[derive(Debug, Clone)]
 pub enum PeerMessage {
     SendMessage(Message),
@@ -552,18 +545,14 @@ impl Actor for PeerActor {
     fn tick(&mut self) {
         match self.connection_state {
             ConnectionState::Connecting { .. } => {
-                // Check if connection has been established
                 self.check_connection_status();
             }
             ConnectionState::Connected => {
-                // Normal operation - process reads
                 if self.stream.is_some() {
                     self.process_read();
                 }
             }
-            ConnectionState::Disconnected => {
+            ConnectionState::Disconnected => {}
-                // Do nothing
-            }
         }
     }
 }

src/actor/server_actor.rs

@@ -0,0 +1,758 @@
+use crate::actor::{Actor, ActorHandle, ConnectionState};
+use crate::client::ClientOperation;
+use crate::dispatcher::MessageDispatcher;
+use crate::message::server::ConnectToPeerHandler;
+use crate::message::server::ExcludedSearchPhrasesHandler;
+use crate::message::server::FileSearchHandler;
+use crate::message::server::GetPeerAddressHandler;
+use crate::message::server::LoginHandler;
+use crate::message::server::MessageFactory;
+use crate::message::server::MessageUser;
+use crate::message::server::ParentMinSpeedHandler;
+use crate::message::server::ParentSpeedRatioHandler;
+use crate::message::server::PrivilegedUsersHandler;
+use crate::message::server::RoomListHandler;
+use crate::message::server::WishListIntervalHandler;
+use crate::message::{Handlers, MessageType};
+use crate::message::{Message, MessageReader};
+use crate::peer::ConnectionType;
+use crate::peer::Peer;
+
+use std::io::{self, Error, Write};
+use std::net::{TcpStream, ToSocketAddrs};
+use std::sync::mpsc::{Receiver, Sender};
+use std::sync::{Arc, RwLock};
+use std::time::{Duration, Instant};
+
+use crate::{debug, error, info, trace, warn, SoulseekRs};
+
+#[derive(Debug, Clone)]
+pub struct PeerAddress {
+    host: String,
+    port: u16,
+}
+
+impl PeerAddress {
+    pub fn new(host: String, port: u16) -> Self {
+        Self { host, port }
+    }
+
+    pub fn get_host(&self) -> &str {
+        &self.host
+    }
+
+    pub fn get_port(&self) -> u16 {
+        self.port
+    }
+}
+
+impl std::fmt::Display for PeerAddress {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "{}:{}", self.host, self.port)
+    }
+}
+
+#[derive(Debug)]
+pub struct Context {
+    pub logged_in: Option<bool>,
+    #[allow(dead_code)]
+    rooms: Rooms,
+}
+
+impl Default for Context {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Context {
+    pub fn new() -> Self {
+        Self {
+            #[allow(dead_code)]
+            rooms: Rooms::new(),
+            logged_in: Option::None,
+        }
+    }
+}
+#[derive(Debug, Clone)]
+pub struct UserMessage {
+    id: u32,
+    timestamp: u32,
+    username: String,
+    message: String,
+    new_message: bool,
+}
+impl UserMessage {
+    pub fn new(
+        id: u32,
+        timestamp: u32,
+        username: String,
+        message: String,
+        new_message: bool,
+    ) -> Self {
+        Self {
+            id,
+            timestamp,
+            username,
+            message,
+            new_message,
+        }
+    }
+    pub fn print(&self) {
+        debug!(
+            "Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}",
+            self.timestamp,
+            self.username,
+            self.id,
+            self.new_message,
+            self.message
+        );
+    }
+}
+
+#[allow(dead_code)]
+#[derive(Debug)]
+pub struct Rooms {
+    pub public_rooms: Vec<Room>,
+    pub owned_private_rooms: Vec<Room>,
+    pub private_rooms: Vec<Room>,
+    pub operated_private_rooms: Vec<Room>,
+}
+impl Rooms {
+    fn new() -> Self {
+        Self {
+            public_rooms: Vec::new(),
+            owned_private_rooms: Vec::new(),
+            private_rooms: Vec::new(),
+            operated_private_rooms: Vec::new(),
+        }
+    }
+
+    #[allow(dead_code)]
+    pub fn print(&self) {
+        info!("Public rooms ({}):", self.public_rooms.len());
+        for room in &self.public_rooms {
+            room.print();
+        }
+        info!("Owned private rooms ({}):", self.owned_private_rooms.len());
+        for room in &self.owned_private_rooms {
+            room.print();
+        }
+        info!("Private rooms ({}):", self.private_rooms.len());
+        for room in &self.private_rooms {
+            room.print();
+        }
+        info!(
+            "Operated private rooms ({}):",
+            self.operated_private_rooms.len()
+        );
+        for room in &self.operated_private_rooms {
+            room.print();
+        }
+    }
+}
+#[derive(Debug)]
+pub struct Room {
+    name: String,
+    number_of_users: i32,
+}
+impl Room {
+    #[allow(dead_code)]
+    pub fn new(name: String, number_of_users: i32) -> Self {
+        Self {
+            name,
+            number_of_users,
+        }
+    }
+    #[allow(dead_code)]
+    pub fn set_number_of_users(&mut self, number_of_users: i32) {
+        self.number_of_users = number_of_users;
+    }
+    pub fn print(&self) {
+        debug!(
+            "Room: {}, Number of users: {}",
+            self.name, self.number_of_users
+        );
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum ServerOperation {
+    ProcessRead,
+    LoginStatus(bool),
+    SendMessage(Message),
+    Login {
+        username: String,
+        password: String,
+        response: std::sync::mpsc::Sender<Result<bool, SoulseekRs>>,
+    },
+    FileSearch {
+        token: u32,
+        query: String,
+    },
+    #[allow(dead_code)]
+    ConnectToPeer(Peer),
+    PierceFirewall(u32),
+    GetPeerAddress(String),
+    GetPeerAddressResponse {
+        username: String,
+        host: String,
+        port: u32,
+        obfuscation_type: u32,
+        obfuscated_port: u16,
+    },
+}
+
+pub struct ServerActor {
+    address: PeerAddress,
+    context: Arc<RwLock<Context>>,
+    listen_port: u32,
+    enable_listen: bool,
+    stream: Option<TcpStream>,
+    connection_state: ConnectionState,
+    reader: MessageReader,
+    client_channel: Sender<ClientOperation>,
+    self_handle: Option<ActorHandle<ServerOperation>>,
+    dispatcher: Option<MessageDispatcher<ServerOperation>>,
+    dispatcher_receiver: Option<Receiver<ServerOperation>>,
+    dispatcher_sender: Option<Sender<ServerOperation>>,
+    queued_messages: Vec<ServerOperation>,
+}
+
+impl ServerActor {
+    pub fn new(
+        address: PeerAddress,
+        client_channel: Sender<ClientOperation>,
+        listen_port: u32,
+        enable_listen: bool,
+    ) -> Self {
+        Self {
+            address,
+            context: Arc::new(RwLock::new(Context::new())),
+            listen_port,
+            enable_listen,
+            stream: None,
+            connection_state: ConnectionState::Disconnected,
+            dispatcher: None,
+            dispatcher_receiver: None,
+            dispatcher_sender: None,
+            reader: MessageReader::new(),
+            client_channel,
+            self_handle: None,
+            queued_messages: Vec::new(),
+        }
+    }
+
+    pub fn get_address(&self) -> &PeerAddress {
+        &self.address
+    }
+
+    pub fn get_sender(&self) -> Option<&Sender<ServerOperation>> {
+        self.dispatcher_sender.as_ref()
+    }
+
+    fn initiate_connection(&mut self) -> bool {
+        let host = self.address.host.clone();
+        let port = self.address.port;
+
+        let addr_str = format!("{}:{}", host, port);
+
+        let mut socket_addrs = match addr_str.to_socket_addrs() {
+            Ok(addrs) => addrs,
+            Err(e) => {
+                error!("[server_actor] Failed to resolve address: {}", e);
+
+                self.disconnect_with_error(io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    e,
+                ));
+                return false;
+            }
+        };
+
+        let socket_addr = socket_addrs.next();
+
+        match socket_addr {
+            Some(addr) => {
+                if let Ok(stream) = TcpStream::connect(addr) {
+                    if let Err(e) = stream.set_nonblocking(true) {
+                        error!(
+                            "[server_actor] Failed to set non-blocking: {}",
+                            e
+                        );
+                        self.disconnect_with_error(e);
+                        return false;
+                    }
+
+                    stream.set_nodelay(true).ok();
+
+                    self.stream = Some(stream);
+                    self.connection_state = ConnectionState::Connecting {
+                        since: Instant::now(),
+                    };
+                    true
+                } else {
+                    match TcpStream::connect(addr) {
+                        Ok(stream) => {
+                            if let Err(e) = stream.set_nonblocking(true) {
+                                error!("[server_actor] Failed to set non-blocking: {}", e);
+                                self.disconnect_with_error(e);
+                                return false;
+                            }
+                            stream.set_nodelay(true).ok();
+                            self.stream = Some(stream);
+                            self.connection_state =
+                                ConnectionState::Connecting {
+                                    since: Instant::now(),
+                                };
+                            true
+                        }
+                        Err(e) => {
+                            self.disconnect_with_error(e);
+                            false
+                        }
+                    }
+                }
+            }
+            None => {
+                let error_msg =
+                    format!("No socket addresses found for {}:{}", host, port);
+                error!("[server_actor] {}", error_msg);
+                self.disconnect_with_error(io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    error_msg,
+                ));
+                false
+            }
+        }
+    }
+
+    pub fn set_self_handle(&mut self, handle: ActorHandle<ServerOperation>) {
+        self.self_handle = Some(handle);
+    }
+
+    fn initialize_dispatcher(&mut self) {
+        let (dispatcher_sender, dispatcher_receiver) =
+            std::sync::mpsc::channel::<ServerOperation>();
+
+        self.dispatcher_receiver = Some(dispatcher_receiver);
+        self.dispatcher_sender = Some(dispatcher_sender.clone());
+
+        self.client_channel
+            .send(ClientOperation::SetServerSender(dispatcher_sender.clone()))
+            .unwrap();
+
+        let mut handlers = Handlers::new();
+
+        handlers.register_handler(LoginHandler);
+        handlers.register_handler(RoomListHandler);
+        handlers.register_handler(ExcludedSearchPhrasesHandler);
+        handlers.register_handler(PrivilegedUsersHandler);
+        handlers.register_handler(MessageUser);
+        handlers.register_handler(WishListIntervalHandler);
+        handlers.register_handler(ParentMinSpeedHandler);
+        handlers.register_handler(ParentSpeedRatioHandler);
+        handlers.register_handler(PrivilegedUsersHandler);
+        handlers.register_handler(FileSearchHandler);
+        handlers.register_handler(GetPeerAddressHandler);
+        handlers.register_handler(ConnectToPeerHandler);
+
+        self.dispatcher = Some(MessageDispatcher::new(
+            "server_actor".into(),
+            dispatcher_sender,
+            handlers,
+        ));
+    }
+
+    fn process_dispatcher_messages(&mut self) {
+        let messages: Vec<ServerOperation> = self
+            .dispatcher_receiver
+            .as_ref()
+            .map_or_else(Vec::new, |receiver| {
+                let mut msgs = Vec::new();
+                while let Ok(msg) = receiver.try_recv() {
+                    msgs.push(msg);
+                }
+                msgs
+            });
+
+        messages
+            .iter()
+            .for_each(|msg| self.handle_message(msg.clone()));
+    }
+
+    pub fn login(
+        &mut self,
+        username: &str,
+        password: &str,
+    ) -> Result<bool, SoulseekRs> {
+        self.queue_message(MessageFactory::build_login_message(
+            username, password,
+        ));
+        let context = self.context.clone();
+        let mut logged_in;
+        let timeout = Duration::from_secs(5);
+        let start = Instant::now();
+
+        loop {
+            if start.elapsed() > timeout {
+                warn!("Timeout waiting for login response");
+                return Err(SoulseekRs::Timeout);
+            }
+
+            {
+                logged_in = context.read().unwrap().logged_in
+            }
+
+            if logged_in.is_some() {
+                break;
+            }
+        }
+
+        if logged_in.unwrap() {
+            info!("Logged in as {}", username);
+            self.queue_message(MessageFactory::build_shared_folders_message(
+                1, 499,
+            ));
+            self.queue_message(MessageFactory::build_no_parent_message());
+            self.queue_message(MessageFactory::build_set_status_message(2));
+            if self.enable_listen {
+                self.queue_message(
+                    MessageFactory::build_set_wait_port_message(
+                        self.listen_port,
+                    ),
+                );
+            }
+        }
+
+        Ok(logged_in.unwrap())
+    }
+
+    pub fn file_search(&mut self, token: u32, query: &str) {
+        self.queue_message(MessageFactory::build_file_search_message(
+            token, query,
+        ));
+    }
+
+    fn handle_message(&mut self, msg: ServerOperation) {
+        if !matches!(self.connection_state, ConnectionState::Connected) {
+            match &msg {
+                ServerOperation::ProcessRead => {
+                    // Always process read operations
+                }
+                _ => {
+                    // Queue all other messages when not connected
+                    self.queued_messages.push(msg);
+                    return;
+                }
+            }
+        }
+
+        match msg {
+            ServerOperation::ConnectToPeer(peer) => {
+                if let Some(op) = match peer.connection_type {
+                    ConnectionType::P | ConnectionType::F => {
+                        Some(ClientOperation::ConnectToPeer(peer.clone()))
+                    }
+                    ConnectionType::D => None,
+                } {
+                    self.client_channel.send(op).unwrap();
+                }
+            }
+            ServerOperation::LoginStatus(message) => {
+                self.context.write().unwrap().logged_in = Some(message);
+            }
+            ServerOperation::PierceFirewall(token) => {
+                self.send_message(
+                    MessageFactory::build_pierce_firewall_message(token),
+                );
+            }
+            ServerOperation::SendMessage(message) => {
+                self.send_message(message);
+            }
+            ServerOperation::GetPeerAddress(username) => {
+                self.send_message(MessageFactory::build_get_peer_address(
+                    &username,
+                ));
+            }
+            ServerOperation::GetPeerAddressResponse {
+                username,
+                host,
+                port,
+                obfuscation_type,
+                obfuscated_port,
+            } => {
+                debug!(
+                    "[server_actor] Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})"
+                    , username, host, port, obfuscation_type, obfuscated_port
+                );
+
+                if let Err(e) = self.client_channel.send(
+                    ClientOperation::GetPeerAddressResponse {
+                        username,
+                        host,
+                        port,
+                        obfuscation_type,
+                        obfuscated_port,
+                    },
+                ) {
+                    error!("[server_actor] Error forwarding GetPeerAddress response to client: {}", e);
+                }
+            }
+            ServerOperation::ProcessRead => {
+                self.process_read();
+            }
+            ServerOperation::Login {
+                username,
+                password,
+                response,
+            } => {
+                self.queue_message(MessageFactory::build_login_message(
+                    &username, &password,
+                ));
+
+                let start = std::time::Instant::now();
+                let timeout = Duration::from_secs(5);
+
+                let context = self.context.clone();
+                std::thread::spawn(move || loop {
+                    if start.elapsed() >= timeout {
+                        let _ = response.send(Err(SoulseekRs::Timeout));
+                        break;
+                    }
+
+                    if let Some(logged_in) = context.read().unwrap().logged_in {
+                        let result = if logged_in {
+                            Ok(true)
+                        } else {
+                            Err(SoulseekRs::AuthenticationFailed)
+                        };
+                        let _ = response.send(result);
+                        break;
+                    }
+
+                    std::thread::sleep(Duration::from_millis(100));
+                });
+            }
+            ServerOperation::FileSearch { token, query } => {
+                self.file_search(token, &query);
+            }
+        }
+    }
+
+    fn process_read(&mut self) {
+        if self.reader.buffer_len() > 0 {
+            self.extract_and_process_messages();
+        }
+
+        {
+            let stream = match self.stream.as_mut() {
+                Some(s) => s,
+                None => 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 => {
+                    debug!("[server_actor] Read operation timed out",);
+                }
+                Err(e) => {
+                    error!(
+                        "[server_actor] Error reading from server: {} (kind: {:?}). Disconnecting.",
+                         e, e.kind()
+                    );
+                    self.disconnect_with_error(e);
+                    return;
+                }
+            }
+        }
+        self.extract_and_process_messages();
+    }
+
+    fn extract_and_process_messages(&mut self) {
+        let mut extracted_count = 0;
+        loop {
+            match self.reader.extract_message() {
+                Ok(Some(mut message)) => {
+                    extracted_count += 1;
+                    trace!(
+                        "[server_actor] ← Message #{}: {:?}",
+                        extracted_count,
+                        message
+                            .get_message_name(
+                                MessageType::Server,
+                                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!("[server_actor] No dispatcher available!",);
+                    }
+                }
+                Err(e) => {
+                    warn!( "[server_actor] Error extracting message: {}. Disconnecting.", e);
+                    self.disconnect_with_error(e);
+                    return;
+                }
+                Ok(None) => {
+                    break;
+                }
+            }
+        }
+
+        self.process_dispatcher_messages();
+    }
+
+    fn queue_message(&mut self, message: Message) {
+        if let Some(sender) = &self.dispatcher_sender {
+            match sender.send(ServerOperation::SendMessage(message)) {
+                Ok(_) => {}
+                Err(e) => error!("Failed to send: {}", e),
+            }
+        } else {
+            self.queued_messages
+                .push(ServerOperation::SendMessage(message));
+        }
+    }
+
+    fn send_message(&mut self, message: Message) {
+        let stream = match self.stream.as_mut() {
+            Some(s) => s,
+            None => {
+                error!("[server_actor] Cannot send message: stream is None");
+                return;
+            }
+        };
+
+        trace!(
+            "[server_actor] ➡ {:?}",
+            message
+                .get_message_name(
+                    MessageType::Server,
+                    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!(
+                "[server_actor] Error writing message: {}. Disconnecting.",
+                e
+            );
+            self.disconnect_with_error(e);
+            return;
+        }
+
+        if let Err(e) = stream.flush() {
+            error!(
+                "[server_actor] Error flushing stream: {}. Disconnecting.",
+                e
+            );
+            self.disconnect_with_error(e);
+        }
+    }
+
+    fn disconnect_with_error(&mut self, _error: Error) {
+        debug!("[server_actor] disconnect");
+
+        self.stream.take();
+    }
+
+    fn disconnect(&mut self) {
+        debug!("[server_actor] disconnected");
+
+        self.stream.take();
+    }
+
+    fn check_connection_status(&mut self) {
+        let ConnectionState::Connecting { since } = self.connection_state
+        else {
+            return;
+        };
+
+        if since.elapsed() > Duration::from_secs(20) {
+            error!("[server_actor] Connection timeout after 20 seconds");
+            self.disconnect_with_error(io::Error::new(
+                io::ErrorKind::TimedOut,
+                "Connection timeout",
+            ));
+            return;
+        }
+
+        let Some(ref stream) = self.stream else {
+            return;
+        };
+
+        match stream.peer_addr() {
+            Ok(_) => {
+                self.connection_state = ConnectionState::Connected;
+                self.on_connection_established();
+            }
+            Err(ref e) if e.kind() == io::ErrorKind::NotConnected => {}
+            Err(e) => {
+                error!("[server_actor] Connection failed: {}", e);
+                self.disconnect_with_error(e);
+            }
+        }
+    }
+
+    fn on_connection_established(&mut self) {
+        let Some(_) = self.stream else {
+            panic!("Stream should be available here")
+        };
+
+        self.initialize_dispatcher();
+
+        let queued = std::mem::take(&mut self.queued_messages);
+        for msg in queued {
+            self.handle_message(msg);
+        }
+
+        if let Some(ref handle) = self.self_handle {
+            handle.send(ServerOperation::ProcessRead).ok();
+        }
+
+        self.process_read();
+    }
+}
+
+impl Actor for ServerActor {
+    type Message = ServerOperation;
+
+    fn handle(&mut self, msg: Self::Message) {
+        self.handle_message(msg);
+    }
+
+    fn on_start(&mut self) {
+        if self.stream.is_none() {
+            self.initiate_connection();
+        } else {
+            self.connection_state = ConnectionState::Connected;
+            self.on_connection_established();
+        }
+    }
+
+    fn on_stop(&mut self) {
+        trace!("[server_actor] actor stopping");
+        self.disconnect();
+    }
+
+    fn tick(&mut self) {
+        match self.connection_state {
+            ConnectionState::Connecting { .. } => {
+                self.check_connection_status();
+            }
+            ConnectionState::Connected => {
+                if self.stream.is_some() {
+                    self.process_read();
+                }
+            }
+            ConnectionState::Disconnected => {}
+        }
+    }
+}

src/client.rs

@@ -1,9 +1,10 @@
+use crate::actor::server_actor::{PeerAddress, ServerActor, ServerOperation};
+use crate::actor::ActorHandle;
 use crate::types::{DownloadResult, DownloadStatus};
 use crate::{
     actor::{peer_registry::PeerRegistry, ActorSystem},
     error::{Result, SoulseekRs},
-    peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer},
+    peer::{ConnectionType, DownloadPeer, NewPeer, Peer},
-    server::{PeerAddress, Server, ServerOperation},
     types::{Download, FileSearchResult},
     utils::{md5, thread_pool::ThreadPool},
     Transfer,
@@ -43,11 +44,12 @@ pub enum ClientOperation {
         obfuscated_port: u16,
     },
     UploadFailed(String, String),
+    SetServerSender(Sender<ServerOperation>),
 }
 pub struct ClientContext {
     pub peer_registry: Option<PeerRegistry>,
     sender: Option<Sender<ClientOperation>>,
-    server_sender: Option<Sender<crate::server::ServerOperation>>,
+    server_sender: Option<Sender<ServerOperation>>,
     search_results: Vec<FileSearchResult>,
     pub download_tokens: HashMap<u32, Download>,
     actor_system: Arc<ActorSystem>,
@@ -84,7 +86,7 @@ pub struct Client {
     address: PeerAddress,
     username: String,
     password: String,
-    server: Option<Server>,
+    server_handle: Option<ActorHandle<ServerOperation>>,
     context: Arc<RwLock<ClientContext>>,
 }
 
@@ -97,15 +99,14 @@ impl Client {
         listen_port: Option<u32>,
     ) -> Self {
         crate::utils::logger::init();
-        let context = Arc::new(RwLock::new(ClientContext::new()));
         Self {
             enable_listen,
             listen_port: listen_port.unwrap_or(DEFALT_LISTEN_PORT),
             address,
             username,
             password,
-            server: None,
+            context: Arc::new(RwLock::new(ClientContext::new())),
-            context,
+            server_handle: None,
         }
     }
     pub fn with_defaults(
@@ -122,72 +123,50 @@ impl Client {
             Receiver<ClientOperation>,
         ) = mpsc::channel();
 
-        let listen_port = self.listen_port;
+        let mut ctx = self.context.write().unwrap();
-        let context = self.context.clone();
+        ctx.sender = Some(sender.clone());
-        let own_username = self.username.clone();
+        let peer_registry =
+            PeerRegistry::new(ctx.actor_system.clone(), sender.clone());
+        ctx.peer_registry = Some(peer_registry);
 
-        // Initialize PeerRegistry with the actor system
+        let server_actor = ServerActor::new(
-        {
-            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(
             self.address.clone(),
             sender,
             self.listen_port,
             self.enable_listen,
-        ) {
+        );
-            Ok(server) => {
-                info!(
-                    "Connected to server at {}:{}",
-                    server.get_address().get_host(),
-                    server.get_address().get_port()
-                );
 
-                if self.enable_listen {
+        self.server_handle = Some(ctx.actor_system.spawn_with_handle(
-                    thread::spawn(move || {
+            server_actor,
-                        Listen::start(
+            |actor, handle| {
-                            listen_port,
+                actor.set_self_handle(handle);
-                            client_sender.clone(),
+            },
-                            context.clone(),
+        ));
-                            own_username,
-                        );
-                    });
-                }
-                let mut unlocked_context = self.context.write().unwrap();
-                unlocked_context.server_sender =
-                    Some(server.get_sender().clone());
 
-                Self::listen_to_client_operations(
+        Self::listen_to_client_operations(
-                    message_reader,
+            message_reader,
-                    self.context.clone(),
+            self.context.clone(),
-                    self.username.clone(),
+            self.username.clone(),
-                );
+        );
-                Some(server)
-            }
-            Err(e) => {
-                error!("Error connecting to server: {}", e);
-                None
-            }
-        };
     }
 
     pub fn login(&self) -> Result<bool> {
         info!("Logging in as {}", self.username);
-        if let Some(server) = &self.server {
+        if let Some(handle) = &self.server_handle {
-            return if (server.login(&self.username, &self.password))? {
+            let (tx, rx) = std::sync::mpsc::channel();
-                Ok(true)
+            let _ = handle.send(ServerOperation::Login {
-            } else {
+                username: self.username.clone(),
-                Err(SoulseekRs::AuthenticationFailed)
+                password: self.password.clone(),
-            };
+                response: tx,
+            });
+
+            match rx.recv() {
+                Ok(result) => result,
+                Err(_) => Err(SoulseekRs::Timeout),
+            }
+        } else {
+            Err(SoulseekRs::NotConnected)
         }
-        Err(SoulseekRs::NotConnected)
     }
 
     #[allow(dead_code)]
@@ -206,10 +185,13 @@ impl Client {
         timeout: Duration,
     ) -> Result<Vec<FileSearchResult>> {
         info!("Searching for {}", query);
-        if let Some(server) = &self.server {
+        if let Some(handle) = &self.server_handle {
             let hash = md5::md5(query);
             let token = u32::from_str_radix(&hash[0..5], 16)?;
-            server.file_search(token, query);
+            let _ = handle.send(ServerOperation::FileSearch {
+                token,
+                query: query.to_string(),
+            });
         } else {
             return Err(SoulseekRs::NotConnected);
         }
@@ -325,7 +307,6 @@ impl Client {
                             let client_context_clone = client_context.clone();
                             let own_username_clone = own_username.clone();
 
-                            // Spawn in background to avoid blocking the receiver thread
                             thread::spawn(move || {
                                 Self::connect_to_peer(
                                     peer,
@@ -536,7 +517,6 @@ impl Client {
                                     client_context.clone();
                                 let own_username_clone = own_username.clone();
 
-                                // Spawn in background to avoid blocking the receiver thread
                                 thread::spawn(move || {
                                     Self::connect_to_peer(
                                         peer,
@@ -594,6 +574,11 @@ impl Client {
                                 Some(&filename),
                             );
                         }
+                        ClientOperation::SetServerSender(sender) => {
+                            client_context.write().unwrap().server_sender =
+                                Some(sender);
+                            debug!("[client] Server sender initialized");
+                        }
                     }
                 }
                 Err(e) => {

src/lib.rs

@@ -5,23 +5,23 @@ pub mod dispatcher;
 pub mod error;
 pub mod message;
 pub mod peer;
-pub mod server;
 pub mod types;
 #[macro_use]
 pub mod utils;
+
 // Prelude module for commonly used items
 pub mod prelude {
+    pub use crate::actor::server_actor::PeerAddress;
     pub use crate::types::{
         DownloadResult, DownloadStatus, File, FileSearchResult, Transfer,
     };
     pub use crate::{debug, error, info, trace, warn};
-    pub use crate::{Client, PeerAddress};
 }
 
 // Re-export commonly used types
+pub use actor::server_actor::PeerAddress;
 pub use client::Client;
 pub use error::{Result, SoulseekRs};
-pub use server::PeerAddress;
 pub use types::{
     DownloadResult, DownloadStatus, File, FileSearchResult, Transfer,
 };

src/main.rs

@@ -46,7 +46,7 @@ fn main() {
             }
         }
         Err(e) => {
-            eprintln!("Failed to login: {e}");
+            eprintln!("Failed to login: {}", e);
         }
     }
 }

src/message/server/connect_to_peer.rs

@@ -1,6 +1,6 @@
 use crate::message::{Message, MessageHandler};
 use crate::peer::Peer;
-use crate::server::ServerOperation;
+use crate::actor::server_actor::ServerOperation;
 use std::sync::mpsc::Sender;
 pub struct ConnectToPeerHandler;
 

src/message/server/excluded_search_phrases.rs

@@ -1,7 +1,7 @@
 use crate::debug;
 use crate::{
+    actor::server_actor::ServerOperation,
     message::{Message, MessageHandler},
-    server::ServerOperation,
 };
 use std::sync::mpsc::Sender;
 

src/message/server/file_search.rs

@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
 
 use crate::{
     message::handlers::MessageHandler, message::Message,
-    server::ServerOperation,
+    actor::server_actor::ServerOperation,
 };
 
 pub struct FileSearchHandler;

src/message/server/get_peer_address.rs

@@ -1,5 +1,5 @@
 use crate::message::{Message, MessageHandler};
-use crate::server::ServerOperation;
+use crate::actor::server_actor::ServerOperation;
 use std::sync::mpsc::Sender;
 
 pub struct GetPeerAddressHandler;

src/message/server/login.rs

@@ -1,8 +1,8 @@
-use crate::{debug, info};
+use crate::{
+    actor::server_actor::ServerOperation, debug, info, message::Message,
+};
 use std::sync::mpsc::Sender;
 
-use crate::{message::Message, server::ServerOperation};
-
 use crate::message::MessageHandler;
 
 pub struct LoginHandler;

src/message/server/message_user.rs

@@ -1,8 +1,6 @@
+use crate::actor::server_actor::{ServerOperation, UserMessage};
 use crate::info;
-use crate::{
+use crate::message::{Message, MessageHandler};
-    message::{Message, MessageHandler},
-    server::{ServerOperation, UserMessage},
-};
 
 use std::sync::mpsc::Sender;
 

src/message/server/parent_min_speed.rs

@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
 
 use crate::{
     message::{Message, MessageHandler},
-    server::ServerOperation,
+    actor::server_actor::ServerOperation,
 };
 
 pub struct ParentMinSpeedHandler;

src/message/server/parent_speed_ratio.rs

@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
 
 use crate::{
     message::{Message, MessageHandler},
-    server::ServerOperation,
+    actor::server_actor::ServerOperation,
 };
 
 pub struct ParentSpeedRatioHandler;

src/message/server/privileged_users.rs

@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
 
 use crate::{
     message::{Message, MessageHandler},
-    server::ServerOperation,
+    actor::server_actor::ServerOperation,
 };
 
 pub struct PrivilegedUsersHandler;

src/message/server/room_list.rs

@@ -1,6 +1,6 @@
 use crate::{
+    actor::server_actor::ServerOperation,
     message::{Message, MessageHandler},
-    server::ServerOperation,
 };
 use std::sync::mpsc::Sender;
 

src/message/server/wish_list_interval.rs

@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
 
 use crate::{
     message::{Message, MessageHandler},
-    server::ServerOperation,
+    actor::server_actor::ServerOperation,
 };
 
 pub struct WishListIntervalHandler;

src/server.rs

@@ -1,500 +0,0 @@
-use crate::client::ClientOperation;
-use crate::dispatcher::MessageDispatcher;
-use crate::error::{Result, SoulseekRs};
-use crate::message::server::ConnectToPeerHandler;
-use crate::message::server::ExcludedSearchPhrasesHandler;
-use crate::message::server::FileSearchHandler;
-use crate::message::server::GetPeerAddressHandler;
-use crate::message::server::LoginHandler;
-use crate::message::server::MessageFactory;
-use crate::message::server::MessageUser;
-use crate::message::server::ParentMinSpeedHandler;
-use crate::message::server::ParentSpeedRatioHandler;
-use crate::message::server::PrivilegedUsersHandler;
-use crate::message::server::RoomListHandler;
-use crate::message::server::WishListIntervalHandler;
-use crate::message::Message;
-use crate::message::MessageReader;
-use crate::message::{Handlers, MessageType};
-use crate::peer::ConnectionType;
-use crate::peer::Peer;
-
-use std::io::{self, Write};
-use std::net::{TcpStream, ToSocketAddrs};
-use std::sync::mpsc;
-use std::sync::mpsc::{Receiver, Sender};
-use std::sync::{Arc, Barrier, RwLock};
-use std::thread::{self};
-use std::time::{Duration, Instant};
-
-use crate::{debug, error, info, trace, warn};
-
-#[derive(Debug, Clone)]
-pub struct PeerAddress {
-    host: String,
-    port: u16,
-}
-
-impl PeerAddress {
-    pub fn new(host: String, port: u16) -> Self {
-        Self { host, port }
-    }
-
-    pub fn get_host(&self) -> &str {
-        &self.host
-    }
-
-    pub fn get_port(&self) -> u16 {
-        self.port
-    }
-}
-
-#[derive(Debug)]
-pub struct Context {
-    pub logged_in: Option<bool>,
-    #[allow(dead_code)]
-    rooms: Rooms,
-}
-
-impl Default for Context {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl Context {
-    pub fn new() -> Self {
-        Self {
-            #[allow(dead_code)]
-            rooms: Rooms::new(),
-            logged_in: Option::None,
-        }
-    }
-}
-#[derive(Debug, Clone)]
-pub struct UserMessage {
-    id: u32,
-    timestamp: u32,
-    username: String,
-    message: String,
-    new_message: bool,
-}
-impl UserMessage {
-    pub fn new(
-        id: u32,
-        timestamp: u32,
-        username: String,
-        message: String,
-        new_message: bool,
-    ) -> Self {
-        Self {
-            id,
-            timestamp,
-            username,
-            message,
-            new_message,
-        }
-    }
-    pub fn print(&self) {
-        debug!(
-            "Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}",
-            self.timestamp,
-            self.username,
-            self.id,
-            self.new_message,
-            self.message
-        );
-    }
-}
-
-#[allow(dead_code)]
-#[derive(Debug)]
-pub struct Rooms {
-    pub public_rooms: Vec<Room>,
-    pub owned_private_rooms: Vec<Room>,
-    pub private_rooms: Vec<Room>,
-    pub operated_private_rooms: Vec<Room>,
-}
-impl Rooms {
-    fn new() -> Self {
-        Self {
-            public_rooms: Vec::new(),
-            owned_private_rooms: Vec::new(),
-            private_rooms: Vec::new(),
-            operated_private_rooms: Vec::new(),
-        }
-    }
-
-    #[allow(dead_code)]
-    pub fn print(&self) {
-        info!("Public rooms ({}):", self.public_rooms.len());
-        for room in &self.public_rooms {
-            room.print();
-        }
-        info!("Owned private rooms ({}):", self.owned_private_rooms.len());
-        for room in &self.owned_private_rooms {
-            room.print();
-        }
-        info!("Private rooms ({}):", self.private_rooms.len());
-        for room in &self.private_rooms {
-            room.print();
-        }
-        info!(
-            "Operated private rooms ({}):",
-            self.operated_private_rooms.len()
-        );
-        for room in &self.operated_private_rooms {
-            room.print();
-        }
-    }
-}
-#[derive(Debug)]
-pub struct Room {
-    name: String,
-    number_of_users: i32,
-}
-impl Room {
-    #[allow(dead_code)]
-    pub fn new(name: String, number_of_users: i32) -> Self {
-        Self {
-            name,
-            number_of_users,
-        }
-    }
-    #[allow(dead_code)]
-    pub fn set_number_of_users(&mut self, number_of_users: i32) {
-        self.number_of_users = number_of_users;
-    }
-    pub fn print(&self) {
-        debug!(
-            "Room: {}, Number of users: {}",
-            self.name, self.number_of_users
-        );
-    }
-}
-
-pub enum ServerOperation {
-    LoginStatus(bool),
-    SendMessage(Message),
-    #[allow(dead_code)]
-    ConnectToPeer(Peer),
-    PierceFirewall(u32),
-    GetPeerAddress(String),
-    GetPeerAddressResponse {
-        username: String,
-        host: String,
-        port: u32,
-        obfuscation_type: u32,
-        obfuscated_port: u16,
-    },
-}
-
-#[derive(Debug)]
-pub struct Server {
-    address: PeerAddress,
-    sender: Sender<ServerOperation>,
-    context: Arc<RwLock<Context>>,
-    listen_port: u32,
-    enable_listen: bool,
-}
-impl Server {
-    pub fn new(
-        address: PeerAddress,
-        client_channel: Sender<ClientOperation>,
-        listen_port: u32,
-        enable_listen: bool,
-    ) -> std::result::Result<Self, io::Error> {
-        let (sender, server_channel): (
-            Sender<ServerOperation>,
-            Receiver<ServerOperation>,
-        ) = mpsc::channel();
-
-        let context = Arc::new(RwLock::new(Context::new()));
-
-        let mut server = Self {
-            address,
-            context,
-            sender,
-            listen_port,
-            enable_listen,
-        };
-
-        server.start_read_write_loops(server_channel, client_channel)?;
-        Ok(server)
-    }
-
-    pub fn get_address(&self) -> &PeerAddress {
-        &self.address
-    }
-
-    pub fn get_sender(&self) -> &Sender<ServerOperation> {
-        &self.sender
-    }
-
-    /// Start reading and writing loops in separate threads
-    fn start_read_write_loops(
-        &mut self,
-        server_channel: Receiver<ServerOperation>,
-        client_channel: Sender<ClientOperation>,
-    ) -> std::result::Result<(), io::Error> {
-        let socket_address =
-            format!("{}:{}", self.address.host, self.address.port)
-                .to_socket_addrs()?
-                .next()
-                .ok_or_else(|| {
-                    io::Error::new(
-                        io::ErrorKind::InvalidInput,
-                        "Invalid address",
-                    )
-                })?;
-
-        info!(
-            "Connecting to server at {}:{}",
-            self.address.host, self.address.port
-        );
-
-        let stream = TcpStream::connect_timeout(
-            &socket_address,
-            Duration::from_secs(10),
-        )?;
-
-        stream.set_read_timeout(Some(Duration::from_secs(5)))?;
-        stream.set_write_timeout(Some(Duration::from_secs(5)))?;
-        stream.set_nodelay(true)?;
-
-        let mut read_stream = stream.try_clone()?;
-        let mut write_stream = stream.try_clone()?;
-
-        let barrier = Arc::new(Barrier::new(3));
-        let read_barrier = barrier.clone();
-        let write_barrier = barrier.clone();
-        let done_barrier = barrier.clone();
-        let sender = self.sender.clone();
-
-        thread::spawn(move || {
-            read_barrier.wait();
-
-            let mut handlers = Handlers::new();
-
-            handlers.register_handler(LoginHandler);
-            handlers.register_handler(RoomListHandler);
-            handlers.register_handler(ExcludedSearchPhrasesHandler);
-            handlers.register_handler(PrivilegedUsersHandler);
-            handlers.register_handler(MessageUser);
-            handlers.register_handler(WishListIntervalHandler);
-            handlers.register_handler(ParentMinSpeedHandler);
-            handlers.register_handler(ParentSpeedRatioHandler);
-            handlers.register_handler(PrivilegedUsersHandler);
-            handlers.register_handler(FileSearchHandler);
-            handlers.register_handler(GetPeerAddressHandler);
-            handlers.register_handler(ConnectToPeerHandler);
-
-            let dispatcher =
-                MessageDispatcher::new("server".to_string(), sender, handlers);
-
-            let mut buffered_reader = MessageReader::new();
-            loop {
-                match buffered_reader.read_from_socket(&mut read_stream) {
-                    Ok(_) => {}
-                    Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue
-                    }
-                    Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
-                        debug!("Read operation timed out");
-                        continue;
-                    }
-                    Err(e) => {
-                        error!("Error reading from server: {}", e);
-                        break;
-                    }
-                }
-
-                loop {
-                    match buffered_reader.extract_message() {
-                        Ok(Some(mut message)) => {
-                            trace!(
-                                "[server] ← {:?} - {}",
-                                message
-                                    .get_message_name(
-                                        MessageType::Server,
-                                        message.get_message_code_u32()
-                                    )
-                                    .map_err(|e| e.to_string()),
-                                message.get_message_code_u32()
-                            );
-                            dispatcher.dispatch(&mut message)
-                        }
-                        Err(e) => {
-                            warn!("Error extracting message: {}", e);
-                            break;
-                        }
-                        Ok(None) => break,
-                    }
-                }
-            }
-        });
-
-        let context = self.context.clone();
-        thread::spawn(move || {
-            write_barrier.wait();
-            loop {
-                if let Ok(operation) = server_channel.recv() {
-                    match operation {
-                        ServerOperation::ConnectToPeer(peer) => {
-                            if let Some(op) = match peer.connection_type {
-                                ConnectionType::P | ConnectionType::F => {
-                                    Some(ClientOperation::ConnectToPeer(
-                                        peer.clone(),
-                                    ))
-                                }
-                                ConnectionType::D => None,
-                            } {
-                                client_channel.send(op).unwrap();
-                            }
-                        }
-                        ServerOperation::LoginStatus(message) => {
-                            context.write().unwrap().logged_in = Some(message);
-                        }
-                        ServerOperation::PierceFirewall(token) => {
-                            if let Err(e) = write_stream.write_all(
-                                &MessageFactory::build_pierce_firewall_message(
-                                    token,
-                                )
-                                .get_buffer(),
-                            ) {
-                                error!(
-                                    "Error writing PierceFirewall message: {}",
-                                    e
-                                );
-                                break;
-                            }
-                        }
-                        ServerOperation::SendMessage(message) => {
-                            match write_stream.write_all(&message.get_buffer())
-                            {
-                                Ok(_) => trace!(
-                                    "[server] → {:?} - {:}",
-                                    message
-                                        .get_message_name(
-                                            MessageType::Server,
-                                            message.get_message_code_send()
-                                                as u32
-                                        )
-                                        .map_err(|e| e.to_string()),
-                                    message.get_message_code_send() as u32
-                                ),
-                                Err(e) => error!(
-                                    "Error writing message to server: {}",
-                                    e
-                                ),
-                            }
-                        }
-                        ServerOperation::GetPeerAddress(username) => {
-                            if let Err(e) = write_stream.write_all(
-                                &MessageFactory::build_get_peer_address(
-                                    &username,
-                                )
-                                .get_buffer(),
-                            ) {
-                                error!("Error writing get_peeer_address message: {}", e);
-                                break;
-                            }
-                        }
-                        ServerOperation::GetPeerAddressResponse {
-                            username,
-                            host,
-                            port,
-                            obfuscation_type,
-                            obfuscated_port,
-                        } => {
-                            debug!(
-                                "Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})",
-                                username, host, port, obfuscation_type, obfuscated_port
-                            );
-
-                            // Forward the response to the client channel
-                            if let Err(e) = client_channel.send(
-                                ClientOperation::GetPeerAddressResponse {
-                                    username,
-                                    host,
-                                    port,
-                                    obfuscation_type,
-                                    obfuscated_port,
-                                },
-                            ) {
-                                error!("Error forwarding GetPeerAddress response to client: {}", e);
-                            }
-                        }
-                    }
-                }
-            }
-        });
-        done_barrier.wait();
-        Ok(())
-    }
-
-    fn queue_message(&self, message: Message) {
-        match self.sender.send(ServerOperation::SendMessage(message)) {
-            Ok(_) => {}
-            Err(e) => error!("Failed to send: {}", e),
-        }
-    }
-    pub fn login(&self, username: &str, password: &str) -> Result<bool> {
-        // Send the login message
-        self.queue_message(MessageFactory::build_login_message(
-            username, password,
-        ));
-        let context = self.context.clone();
-        let mut logged_in;
-        let timeout = Duration::from_secs(5);
-        let start = Instant::now();
-
-        // wait till server says your logged in or not
-        loop {
-            if start.elapsed() > timeout {
-                warn!("Timeout waiting for login response");
-                return Err(SoulseekRs::Timeout);
-            }
-
-            {
-                logged_in = context.read().unwrap().logged_in
-            }
-
-            if logged_in.is_some() {
-                break;
-            }
-        }
-
-        if logged_in.unwrap() {
-            info!("Logged in as {}", username);
-            self.queue_message(MessageFactory::build_shared_folders_message(
-                1, 499,
-            ));
-            self.queue_message(MessageFactory::build_no_parent_message());
-            self.queue_message(MessageFactory::build_set_status_message(2));
-            if self.enable_listen {
-                self.queue_message(
-                    MessageFactory::build_set_wait_port_message(
-                        self.listen_port,
-                    ),
-                );
-            }
-        }
-
-        Ok(logged_in.unwrap())
-    }
-
-    pub fn file_search(&self, token: u32, query: &str) {
-        self.queue_message(MessageFactory::build_file_search_message(
-            token, query,
-        ));
-    }
-
-    #[allow(dead_code)]
-    pub fn pierce_firewall(&self, token: u32) {
-        match self.sender.send(ServerOperation::PierceFirewall(token)) {
-            Ok(_) => {}
-            Err(e) => error!("Failed to send PierceFirewall: {}", e),
-        }
-    }
-}