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

chore: refactor to use RwLock

Browse Source
This commit is contained in:
michel
2025-11-01 14:57:27 +01:00
committed by Michel de Graaf
parent 4932386056
commit ff30db3e29
5 changed files with 47 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client.rs
+22 -22
View File
@@ -14,7 +14,7 @@ use std::{
net::TcpStream,
sync::{
mpsc::{Receiver, Sender},
Mutex,
RwLock,
},
thread::{self, sleep},
};
@@ -73,7 +73,7 @@ pub struct Client {
username: String,
password: String,
server: Option<Server>,
context: Arc<Mutex<ClientContext>>,
context: Arc<RwLock<ClientContext>>,
}
impl Client {
@@ -83,10 +83,10 @@ impl Client {
password: String,
) -> Self {
crate::utils::logger::init();
let context = Arc::new(Mutex::new(ClientContext::new()));
let context = Arc::new(RwLock::new(ClientContext::new()));
debug!(
"ThreadPool initialized with {} threads",
context.lock().unwrap().thread_pool.thread_count()
context.read().unwrap().thread_pool.thread_count()
);
Self {
address,
@@ -103,7 +103,7 @@ impl Client {
Receiver<ClientOperation>,
) = mpsc::channel();
self.context.lock().unwrap().sender = Some(sender.clone());
self.context.write().unwrap().sender = Some(sender.clone());
let client_sender = sender.clone();
@@ -118,7 +118,7 @@ impl Client {
thread::spawn(move || {
Listen::start(2234, client_sender.clone());
});
let mut unlocked_context = self.context.lock().unwrap();
let mut unlocked_context = self.context.write().unwrap();
unlocked_context.server_sender =
Some(server.get_sender().clone());
@@ -153,7 +153,7 @@ impl Client {
#[allow(dead_code)]
pub fn remove_peer(&self, username: &str) {
let mut context = self.context.lock().unwrap();
let mut context = self.context.write().unwrap();
if let Some(peer) = context.peers.remove(username) {
drop(peer);
}
@@ -180,7 +180,7 @@ impl Client {
break;
}
}
Ok(self.context.lock().unwrap().search_results.clone())
Ok(self.context.read().unwrap().search_results.clone())
}
pub fn download(
@@ -206,7 +206,7 @@ impl Client {
download_directory,
};
let mut context = self.context.lock().unwrap();
let mut context = self.context.write().unwrap();
context.download_tokens.insert(token, download.clone());
let download_initiated = context
.peers
@@ -242,7 +242,7 @@ impl Client {
fn listen_to_client_operations(
reader: Receiver<ClientOperation>,
client_context: Arc<Mutex<ClientContext>>,
client_context: Arc<RwLock<ClientContext>>,
own_username: String,
) {
thread::spawn(move || loop {
@@ -259,13 +259,13 @@ impl Client {
ClientOperation::SearchResult(file_search) => {
trace!("[client] SearchResult {:?}", file_search);
client_context
.lock()
.write()
.unwrap()
.search_results
.push(file_search.clone());
}
ClientOperation::PeerDisconnected(username) => {
let mut context = client_context.lock().unwrap();
let mut context = client_context.write().unwrap();
if let Some(peer) = context.peers.remove(&username) {
drop(peer); // Explicitly drop to trigger cleanup
}
@@ -279,7 +279,7 @@ impl Client {
}
ClientOperation::DownloadFromPeer(token, peer) => {
let maybe_download = {
let client_context = client_context.lock().unwrap();
let client_context = client_context.read().unwrap();
client_context.download_tokens.get(&token).cloned()
};
let own_username = own_username.clone();
@@ -332,7 +332,7 @@ impl Client {
}
ClientOperation::NewPeer(new_peer) => {
if client_context
.lock()
.read()
.unwrap()
.peers
.contains_key(&new_peer.username)
@@ -342,7 +342,7 @@ impl Client {
new_peer.username
);
} else if let Some(server_sender) =
&client_context.lock().unwrap().server_sender
&client_context.read().unwrap().server_sender
{
server_sender
.send(ServerOperation::GetPeerAddress(
@@ -386,7 +386,7 @@ impl Client {
);
if let Some(_peer) =
client_context.lock().unwrap().peers.get(&username)
client_context.read().unwrap().peers.get(&username)
{
// don't know if i should update? and or reconnect the peer
// debug!(
@@ -426,7 +426,7 @@ impl Client {
transfer,
username,
) => {
let mut context = client_context.lock().unwrap();
let mut context = client_context.write().unwrap();
let key_to_remove = context
.download_tokens
@@ -468,12 +468,12 @@ impl Client {
fn connect_to_peer(
peer: Peer,
client_context: Arc<Mutex<ClientContext>>,
client_context: Arc<RwLock<ClientContext>>,
own_username: String,
stream: Option<TcpStream>,
) {
let client_context2 = client_context.clone();
let unlocked_context = client_context.lock().unwrap();
let unlocked_context = client_context.read().unwrap();
trace!("[client] connect_to_peer: {}", peer.username);
if let Some(sender) = unlocked_context.sender.clone() {
@@ -494,7 +494,7 @@ impl Client {
match connect_result {
Ok(p) => {
trace!("[client] connected to: {}", peer.username);
client_context2.lock().unwrap().peers.insert(peer.username, p);
client_context2.write().unwrap().peers.insert(peer.username, p);
}
Err(e) => {
trace!(
@@ -545,12 +545,12 @@ impl Client {
}
fn pierce_firewall(
peer: Peer,
client_context: Arc<Mutex<ClientContext>>,
client_context: Arc<RwLock<ClientContext>>,
own_username: String,
) {
debug!("Piercing firewall for peer: {:?}", peer);
let context = client_context.lock().unwrap();
let context = client_context.read().unwrap();
if let Some(server_sender) = &context.server_sender {
if let Some(token) = peer.token {
match server_sender.send(ServerOperation::PierceFirewall(token))
src/message/mod.rs
+1 -1
View File
@@ -8,7 +8,7 @@ pub mod server;
pub use handlers::{Handlers, MessageHandler};
pub use message_reader::MessageReader;
use std::{str, usize};
use std::str;
#[derive(Debug, PartialEq)]
#[allow(dead_code)]
src/peer/download_peer.rs
+19 -20
View File
@@ -6,8 +6,8 @@ use std::fs::{self, File};
use std::io::{self, BufWriter, Read, Write};
use std::net::TcpStream;
use std::net::ToSocketAddrs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::path::Path;
use std::sync::{Arc, RwLock};
use std::time::Duration;
#[derive(Debug)]
@@ -22,10 +22,10 @@ struct FileManager;
#[allow(dead_code)]
impl FileManager {
fn expand_path(mut path: &str) -> String {
if path.starts_with('~') {
fn expand_path(path: &str) -> String {
if let Some(stripped) = path.strip_prefix('~') {
let home = env::var("HOME").expect("HOME not set");
format!("{}{}", home, &path[1..])
format!("{}{}", home, stripped)
} else {
path.to_string()
}
@@ -244,7 +244,7 @@ impl DownloadPeer {
pub fn download_file(
self,
client_context: Arc<Mutex<ClientContext>>,
client_context: Arc<RwLock<ClientContext>>,
mut expected_size: Option<usize>,
mut download_path: Option<String>,
) -> Result<(usize, String), io::Error> {
@@ -289,7 +289,7 @@ impl DownloadPeer {
);
download_info = client_context
.lock()
.read()
.unwrap()
.download_tokens
.get(&token_u32)
@@ -392,12 +392,12 @@ mod tests {
use std::fs;
use std::io::{Read, Write};
use std::net::TcpListener;
use std::sync::{Arc, Mutex};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::Duration;
pub fn build_test_server() -> (u16, Arc<Mutex<Vec<Vec<u8>>>>) {
let messages = Arc::new(Mutex::new(Vec::<Vec<u8>>::new()));
pub fn build_test_server() -> (u16, Arc<RwLock<Vec<Vec<u8>>>>) {
let messages = Arc::new(RwLock::new(Vec::<Vec<u8>>::new()));
let messages_clone = Arc::clone(&messages);
// Find an available port by binding to port 0
@@ -412,7 +412,7 @@ mod tests {
match stream.read(&mut buffer) {
Ok(bytes_read) if bytes_read > 0 => {
let data = buffer[..bytes_read].to_vec();
messages_clone.lock().unwrap().push(data);
messages_clone.write().unwrap().push(data);
}
_ => {}
}
@@ -440,7 +440,7 @@ mod tests {
false,
"own_username".to_string(),
);
let dummy_context = Arc::new(Mutex::new(ClientContext::new()));
let dummy_context = Arc::new(RwLock::new(ClientContext::new()));
let _ = download_peer
.download_file(dummy_context, None, None)
.unwrap();
@@ -448,7 +448,7 @@ mod tests {
// Give the client time to send messages
thread::sleep(Duration::from_millis(10));
let received_messages = messages.lock().unwrap();
let received_messages = messages.read().unwrap();
assert_eq!(received_messages.len(), 1);
assert_eq!(received_messages[0], vec![0, 0, 0, 0, 33, 0, 0, 0]);
}
@@ -465,7 +465,7 @@ mod tests {
true, // no_pierce = true, should send init message
"own_username".to_string(),
);
let dummy_context = Arc::new(Mutex::new(ClientContext::new()));
let dummy_context = Arc::new(RwLock::new(ClientContext::new()));
let _ = download_peer
.download_file(dummy_context, None, None)
.unwrap();
@@ -473,7 +473,7 @@ mod tests {
// Give the client time to send messages
thread::sleep(Duration::from_millis(10));
let received_messages = messages.lock().unwrap();
let received_messages = messages.read().unwrap();
assert_eq!(received_messages.len(), 1); // Expect one init message
assert_eq!(
received_messages[0],
@@ -489,7 +489,7 @@ mod tests {
#[test]
pub fn test_download_file() {
let test_data = b"test file content";
let messages = Arc::new(Mutex::new(Vec::<Vec<u8>>::new()));
let messages = Arc::new(RwLock::new(Vec::<Vec<u8>>::new()));
let messages_clone = Arc::clone(&messages);
// Create a test server that sends test data
@@ -509,7 +509,7 @@ mod tests {
if let Ok(bytes_read) = stream.read(&mut buffer) {
if bytes_read > 0 {
messages_clone
.lock()
.write()
.unwrap()
.push(buffer[..bytes_read].to_vec());
}
@@ -537,7 +537,7 @@ mod tests {
"test_user".to_string(),
);
let dummy_context = Arc::new(Mutex::new(ClientContext::new()));
let dummy_context = Arc::new(RwLock::new(ClientContext::new()));
let result = download_peer.download_file(
dummy_context,
Some(test_data.len()),
@@ -654,8 +654,7 @@ mod tests {
let mut processor = StreamProcessor::new(false, 123);
let data = b"test data";
let result = processor.process_data_chunk(data);
assert!(result.is_ok());
processor.process_data_chunk(data);
assert_eq!(processor.total_bytes, data.len());
assert_eq!(processor.buffer, data);
}
src/server.rs
+5 -5
View File
@@ -23,7 +23,7 @@ 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, Mutex};
use std::sync::{Arc, Barrier, RwLock};
use std::thread::{self};
use std::time::{Duration, Instant};
@@ -193,7 +193,7 @@ pub enum ServerOperation {
pub struct Server {
address: PeerAddress,
sender: Sender<ServerOperation>,
context: Arc<Mutex<Context>>,
context: Arc<RwLock<Context>>,
}
impl Server {
pub fn new(
@@ -205,7 +205,7 @@ impl Server {
Receiver<ServerOperation>,
) = mpsc::channel();
let context = Arc::new(Mutex::new(Context::new()));
let context = Arc::new(RwLock::new(Context::new()));
let mut server = Self {
address,
@@ -353,7 +353,7 @@ impl Server {
}
}
ServerOperation::LoginStatus(message) => {
context.lock().unwrap().logged_in = Some(message);
context.write().unwrap().logged_in = Some(message);
}
ServerOperation::PierceFirewall(token) => {
if let Err(e) = write_stream.write_all(
@@ -457,7 +457,7 @@ impl Server {
}
{
logged_in = context.lock().unwrap().logged_in
logged_in = context.read().unwrap().logged_in
}
if logged_in.is_some() {
test_download.mp3
-1
View File
@@ -1 +0,0 @@
test file content