From 2584428170e1ad7bb97f0c2a3ccc656b85529dac Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 13:59:35 +0100 Subject: [PATCH] chore: linting --- src/actor/mod.rs | 48 +++++++++++----- src/actor/peer_actor.rs | 83 +++++++++++++++++++--------- src/actor/peer_registry.rs | 51 +++++++++++------ src/client.rs | 37 +++++++------ src/main.rs | 2 +- src/message/handlers.rs | 5 +- src/message/peer/transfer_request.rs | 4 +- src/peer/listen.rs | 13 ++--- 8 files changed, 156 insertions(+), 87 deletions(-) diff --git a/src/actor/mod.rs b/src/actor/mod.rs index 7587f7e..f56c2d8 100644 --- a/src/actor/mod.rs +++ b/src/actor/mod.rs @@ -5,8 +5,8 @@ use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; use std::time::{Duration, Instant}; -use crate::utils::thread_pool::ThreadPool; use crate::trace; +use crate::utils::thread_pool::ThreadPool; pub mod peer_actor; pub mod peer_registry; @@ -54,6 +54,7 @@ impl ActorHandle { pub(crate) enum ActorMessage { UserMessage(M), Stop, + #[allow(dead_code)] Tick, } @@ -70,9 +71,9 @@ impl ActorSystem { /// Spawn a new actor and return its handle pub fn spawn(&self, mut actor: A) -> ActorHandle { let (sender, receiver) = channel::>(); - let handle = ActorHandle { sender: sender.clone() }; - - let thread_pool = self.thread_pool.clone(); + let handle = ActorHandle { + sender: sender.clone(), + }; // Start the actor event loop on the thread pool self.thread_pool.execute(move || { @@ -86,23 +87,29 @@ impl ActorSystem { /// Spawn a new actor with initialization callback and return its handle /// The callback receives the actor handle before on_start is called - pub fn spawn_with_handle(&self, mut actor: A, init: F) -> ActorHandle + pub fn spawn_with_handle( + &self, + mut actor: A, + init: F, + ) -> ActorHandle where F: FnOnce(&mut A, ActorHandle) + Send + 'static, { let (sender, receiver) = channel::>(); - let handle = ActorHandle { sender: sender.clone() }; + let handle = ActorHandle { + sender: sender.clone(), + }; let handle_for_init = handle.clone(); - let thread_pool = self.thread_pool.clone(); - // Start the actor event loop on the thread pool self.thread_pool.execute(move || { trace!("[actor_system] Actor task starting on thread pool"); init(&mut actor, handle_for_init); trace!("[actor_system] Actor init callback completed"); actor.on_start(); - trace!("[actor_system] Actor on_start completed, starting event loop"); + trace!( + "[actor_system] Actor on_start completed, starting event loop" + ); Self::run_actor_loop(&mut actor, receiver); trace!("[actor_system] Actor event loop ended"); actor.on_stop(); @@ -113,7 +120,10 @@ impl ActorSystem { } /// Run the actor's message processing loop - fn run_actor_loop(actor: &mut A, receiver: Receiver>) { + fn run_actor_loop( + actor: &mut A, + receiver: Receiver>, + ) { trace!("[actor_system] run_actor_loop STARTED"); let tick_interval = Duration::from_millis(100); let mut last_tick = Instant::now(); @@ -125,16 +135,24 @@ impl ActorSystem { match receiver.recv_timeout(tick_interval) { Ok(ActorMessage::UserMessage(msg)) => { message_count += 1; - trace!("[actor_system] Processing message #{}", message_count); + trace!( + "[actor_system] Processing message #{}", + message_count + ); actor.handle(msg); } Ok(ActorMessage::Stop) => { - trace!("[actor_system] Received Stop message, breaking loop"); + trace!( + "[actor_system] Received Stop message, breaking loop" + ); break; } Ok(ActorMessage::Tick) => { tick_count += 1; - trace!("[actor_system] Received explicit Tick message #{}", tick_count); + trace!( + "[actor_system] Received explicit Tick message #{}", + tick_count + ); actor.tick(); } Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { @@ -149,7 +167,9 @@ impl ActorSystem { } } Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - trace!("[actor_system] Channel disconnected, breaking loop"); + trace!( + "[actor_system] Channel disconnected, breaking loop" + ); // Channel closed, actor should stop break; } diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index 7f5ac89..b0efbe5 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -2,8 +2,8 @@ use crate::actor::{Actor, ActorHandle}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::peer::{ - FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueResponse, TransferRequest, - TransferResponse, UploadFailedHandler, + FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueResponse, + TransferRequest, TransferResponse, UploadFailedHandler, }; use crate::message::server::MessageFactory; use crate::message::{Handlers, Message, MessageReader, MessageType}; @@ -81,7 +81,8 @@ impl PeerActor { /// Initialize the message dispatcher with handlers fn initialize_dispatcher(&mut self) { // Create a channel for the dispatcher to send messages to - let (dispatcher_sender, dispatcher_receiver) = std::sync::mpsc::channel::(); + let (dispatcher_sender, dispatcher_receiver) = + std::sync::mpsc::channel::(); self.dispatcher_receiver = Some(dispatcher_receiver); @@ -104,15 +105,16 @@ impl PeerActor { /// Process any messages from the dispatcher fn process_dispatcher_messages(&mut self) { // Collect all messages first to avoid borrow issues - let messages: Vec = if let Some(ref receiver) = self.dispatcher_receiver { - let mut msgs = Vec::new(); - while let Ok(msg) = receiver.try_recv() { - msgs.push(msg); - } - msgs - } else { - Vec::new() - }; + let messages: Vec = + if let Some(ref receiver) = self.dispatcher_receiver { + let mut msgs = Vec::new(); + while let Ok(msg) = receiver.try_recv() { + msgs.push(msg); + } + msgs + } else { + Vec::new() + }; // Process all collected messages for msg in messages { @@ -149,7 +151,9 @@ impl PeerActor { MessageFactory::build_transfer_response_message(transfer); if let Some(ref handle) = self.self_handle { - if let Err(e) = handle.send(PeerMessage::SendMessage(transfer_response)) { + if let Err(e) = + handle.send(PeerMessage::SendMessage(transfer_response)) + { error!("[peer_actor:{}] Failed to send TransferResponse message: {}", username, e); } } @@ -202,7 +206,8 @@ impl PeerActor { self.peer.write().unwrap().username = username; } PeerMessage::QueueUpload(filename) => { - let message = MessageFactory::build_queue_upload_message(&filename); + let message = + MessageFactory::build_queue_upload_message(&filename); self.send_message(message); } PeerMessage::RequestTransfer(download) => { @@ -214,7 +219,10 @@ impl PeerActor { } PeerMessage::ProcessRead => { let username = self.peer.read().unwrap().username.clone(); - trace!("[peer_actor:{}] Handling ProcessRead message", username); + trace!( + "[peer_actor:{}] Handling ProcessRead message", + username + ); self.process_read(); } } @@ -269,13 +277,19 @@ impl PeerActor { username, extracted_count, message - .get_message_name(MessageType::Peer, message.get_message_code() as u32) + .get_message_name( + MessageType::Peer, + message.get_message_code() as u32 + ) .map_err(|e| e.to_string()) ); if let Some(ref dispatcher) = self.dispatcher { dispatcher.dispatch(&mut message); } else { - warn!("[peer_actor:{}] No dispatcher available!", username); + warn!( + "[peer_actor:{}] No dispatcher available!", + username + ); } } Err(e) => { @@ -316,7 +330,9 @@ impl PeerActor { message .get_message_name( MessageType::Peer, - u32::from_le_bytes(message.get_slice(0, 4).try_into().unwrap()) + u32::from_le_bytes( + message.get_slice(0, 4).try_into().unwrap() + ) ) .map_err(|e| e.to_string()) ); @@ -372,17 +388,32 @@ impl Actor for PeerActor { // Trigger initial read if let Some(ref handle) = self.self_handle { - trace!("[peer_actor:{}] Sending initial ProcessRead message", username); + trace!( + "[peer_actor:{}] Sending initial ProcessRead message", + username + ); match handle.send(PeerMessage::ProcessRead) { - Ok(_) => trace!("[peer_actor:{}] ProcessRead message sent successfully", username), - Err(e) => error!("[peer_actor:{}] FAILED to send ProcessRead: {}", username, e), + Ok(_) => trace!( + "[peer_actor:{}] ProcessRead message sent successfully", + username + ), + Err(e) => error!( + "[peer_actor:{}] FAILED to send ProcessRead: {}", + username, e + ), } } else { - error!("[peer_actor:{}] self_handle is None! Cannot send ProcessRead", username); + error!( + "[peer_actor:{}] self_handle is None! Cannot send ProcessRead", + username + ); } // Also do an immediate read as fallback - trace!("[peer_actor:{}] Performing immediate process_read as fallback", username); + trace!( + "[peer_actor:{}] Performing immediate process_read as fallback", + username + ); self.process_read(); } @@ -398,8 +429,10 @@ impl Actor for PeerActor { self.process_read(); } else { let username = self.peer.read().unwrap().username.clone(); - trace!("[peer_actor:{}] tick() called but stream is None", username); + trace!( + "[peer_actor:{}] tick() called but stream is None", + username + ); } } } - diff --git a/src/actor/peer_registry.rs b/src/actor/peer_registry.rs index 02de074..5d99014 100644 --- a/src/actor/peer_registry.rs +++ b/src/actor/peer_registry.rs @@ -1,8 +1,8 @@ -use crate::actor::{ActorHandle, ActorSystem}; use crate::actor::peer_actor::{PeerActor, PeerMessage}; +use crate::actor::{ActorHandle, ActorSystem}; use crate::client::ClientOperation; -use crate::peer::Peer; use crate::message::MessageReader; +use crate::peer::Peer; use crate::{debug, trace}; use std::collections::HashMap; @@ -18,7 +18,10 @@ pub struct PeerRegistry { } impl PeerRegistry { - pub fn new(actor_system: Arc, client_channel: Sender) -> Self { + pub fn new( + actor_system: Arc, + client_channel: Sender, + ) -> Self { Self { peers: Arc::new(Mutex::new(HashMap::new())), actor_system, @@ -37,23 +40,23 @@ impl PeerRegistry { debug!("[peer_registry] Registering peer actor for {}", username); - let actor = PeerActor::new( - peer, - stream, - reader, - self.client_channel.clone(), - ); + let actor = + PeerActor::new(peer, stream, reader, self.client_channel.clone()); // Spawn the actor with initialization callback to set self_handle - let handle = self.actor_system.spawn_with_handle(actor, |actor, handle| { - actor.set_self_handle(handle); - }); + let handle = + self.actor_system.spawn_with_handle(actor, |actor, handle| { + actor.set_self_handle(handle); + }); // Store in registry let mut peers = self.peers.lock().unwrap(); peers.insert(username.clone(), handle.clone()); - trace!("[peer_registry] Peer actor {} registered successfully", username); + trace!( + "[peer_registry] Peer actor {} registered successfully", + username + ); Ok(handle) } @@ -65,7 +68,10 @@ impl PeerRegistry { } /// Remove a peer actor from the registry - pub fn remove_peer(&self, username: &str) -> Option> { + pub fn remove_peer( + &self, + username: &str, + ) -> Option> { let mut peers = self.peers.lock().unwrap(); let handle = peers.remove(username); @@ -95,15 +101,24 @@ impl PeerRegistry { } /// Send a message to a peer by username - pub fn send_to_peer(&self, username: &str, message: PeerMessage) -> Result<(), String> { - let handle = self.get_peer(username) - .ok_or_else(|| format!("Peer {} not found in registry", username))?; + pub fn send_to_peer( + &self, + username: &str, + message: PeerMessage, + ) -> Result<(), String> { + let handle = self.get_peer(username).ok_or_else(|| { + format!("Peer {} not found in registry", username) + })?; handle.send(message) } /// Queue an upload for a peer - pub fn queue_upload(&self, username: &str, filename: String) -> Result<(), String> { + pub fn queue_upload( + &self, + username: &str, + filename: String, + ) -> Result<(), String> { self.send_to_peer(username, PeerMessage::QueueUpload(filename)) } } diff --git a/src/client.rs b/src/client.rs index c87cc64..862269a 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,11 +1,8 @@ use crate::types::{DownloadResult, DownloadStatus}; use crate::{ - actor::{ActorSystem, peer_registry::PeerRegistry}, + actor::{peer_registry::PeerRegistry, ActorSystem}, error::{Result, SoulseekRs}, - peer::{ - listen::Listen, ConnectionType, DownloadPeer, NewPeer, - Peer, - }, + peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer}, server::{PeerAddress, Server, ServerOperation}, types::{Download, FileSearchResult}, utils::{md5, thread_pool::ThreadPool}, @@ -136,10 +133,8 @@ impl Client { { let mut ctx = self.context.write().unwrap(); ctx.sender = Some(sender.clone()); - let peer_registry = PeerRegistry::new( - ctx.actor_system.clone(), - sender.clone(), - ); + let peer_registry = + PeerRegistry::new(ctx.actor_system.clone(), sender.clone()); ctx.peer_registry = Some(peer_registry); } @@ -261,11 +256,14 @@ impl Client { let mut context = self.context.write().unwrap(); context.download_tokens.insert(token, download.clone()); - let download_initiated = if let Some(ref registry) = context.peer_registry { - registry.queue_upload(&username, download.filename.clone()).is_ok() - } else { - false - }; + let download_initiated = + if let Some(ref registry) = context.peer_registry { + registry + .queue_upload(&username, download.filename.clone()) + .is_ok() + } else { + false + }; drop(context); @@ -315,7 +313,9 @@ impl Client { ClientOperation::PeerDisconnected(username) => { let context = client_context.read().unwrap(); if let Some(ref registry) = context.peer_registry { - if let Some(handle) = registry.remove_peer(&username) { + if let Some(handle) = + registry.remove_peer(&username) + { let _ = handle.stop(); } } @@ -445,7 +445,9 @@ impl Client { username, host, port, obfuscation_type, obfuscated_port ); - let peer_exists = client_context.read().unwrap() + let peer_exists = client_context + .read() + .unwrap() .peer_registry .as_ref() .map(|r| r.contains(&username)) @@ -541,9 +543,8 @@ impl Client { let unlocked_context = client_context.read().unwrap(); trace!("[client] connect_to_peer: {}", peer.username); - if let Some(sender) = unlocked_context.sender.clone() { + if let Some(_sender) = unlocked_context.sender.clone() { let peer_clone = peer.clone(); - let sender_clone = sender; unlocked_context.thread_pool.execute(move || { trace!("[client] connecting to {}, with connection_type: {}, and token {:?}", peer.username, peer.connection_type, peer.token); match peer.connection_type { diff --git a/src/main.rs b/src/main.rs index b9d8880..64c7710 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,7 +6,7 @@ fn main() { PeerAddress::new(String::from("server.slsknet.org"), 2242), String::from("xxxxxx"), String::from("xxxxxx"), - false, + true, None, ); diff --git a/src/message/handlers.rs b/src/message/handlers.rs index 57f1e99..0f343c8 100644 --- a/src/message/handlers.rs +++ b/src/message/handlers.rs @@ -31,7 +31,10 @@ impl Handlers { self.handlers.insert(handler.get_code(), Box::new(handler)); self } - pub fn get_handler(&self, code: u8) -> Option<&(dyn MessageHandler + Send)> { + pub fn get_handler( + &self, + code: u8, + ) -> Option<&(dyn MessageHandler + Send)> { self.handlers.get(&code).map(|v| &**v) } } diff --git a/src/message/peer/transfer_request.rs b/src/message/peer/transfer_request.rs index 914bc0e..a4d7499 100644 --- a/src/message/peer/transfer_request.rs +++ b/src/message/peer/transfer_request.rs @@ -13,8 +13,6 @@ impl MessageHandler for TransferRequest { fn handle(&self, message: &mut Message, sender: Sender) { let transfer = Transfer::new_from_message(message); - sender - .send(PeerMessage::TransferRequest(transfer)) - .unwrap(); + sender.send(PeerMessage::TransferRequest(transfer)).unwrap(); } } diff --git a/src/peer/listen.rs b/src/peer/listen.rs index dd58b10..789a82f 100644 --- a/src/peer/listen.rs +++ b/src/peer/listen.rs @@ -15,6 +15,7 @@ const PEER_INIT_MESSAGE_CODE: u8 = 1; #[derive(Clone)] struct ConnectionContext { + #[allow(dead_code)] client_sender: Sender, client_context: Arc>, own_username: String, @@ -119,7 +120,10 @@ fn handle_peer_connection( trace!("[listener] peer actor spawned for: {}", peer.username); } Err(e) => { - error!("Failed to spawn peer actor for {:?}: {:?}", peer.username, e); + error!( + "Failed to spawn peer actor for {:?}: {:?}", + peer.username, e + ); } } } else { @@ -221,12 +225,7 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { match init_data.connection_type { ConnectionType::P => handle_peer_connection( - peer, - stream, - reader, - &context, - &peer_ip, - peer_port, + peer, stream, reader, &context, &peer_ip, peer_port, ), ConnectionType::F => handle_file_connection( peer,