dumbgoy/soulseek-rs-modded
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chore: linting

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