feat: add download feedback via channel

This commit is contained in:
Michel de Graaf
2025-11-03 16:37:21 +01:00
parent a9d0101a58
commit c346c56297
3 changed files with 81 additions and 167 deletions
+22 -17
View File
@@ -1,3 +1,4 @@
use crate::types::{DownloadResult, DownloadStatus};
use crate::{ use crate::{
error::{Result, SoulseekRs}, error::{Result, SoulseekRs},
peer::{ peer::{
@@ -204,20 +205,23 @@ impl Client {
size: u64, size: u64,
download_directory: String, download_directory: String,
) -> Result<crate::types::DownloadResult> { ) -> Result<crate::types::DownloadResult> {
use crate::types::{DownloadResult, DownloadStatus};
use std::time::{Duration, Instant};
let start_time = Instant::now(); let start_time = Instant::now();
let hash = md5::md5(&filename); let hash = md5::md5(&filename);
let token = u32::from_str_radix(&hash[0..5], 16)?; let token = u32::from_str_radix(&hash[0..5], 16)?;
let (download_sender, download_receivier): (
Sender<DownloadStatus>,
Receiver<DownloadStatus>,
) = mpsc::channel();
let download = Download { let download = Download {
username: username.clone(), username: username.clone(),
filename: filename.clone(), filename: filename.clone(),
token, token,
size, size,
download_directory, download_directory,
sender: download_sender,
}; };
let mut context = self.context.write().unwrap(); let mut context = self.context.write().unwrap();
@@ -230,9 +234,6 @@ impl Client {
drop(context); drop(context);
let timeout = Duration::from_secs(150);
let check_interval = Duration::from_millis(100);
if !download_initiated { if !download_initiated {
return Ok(DownloadResult { return Ok(DownloadResult {
filename, filename,
@@ -242,14 +243,12 @@ impl Client {
}); });
} }
while start_time.elapsed() < timeout { let status = download_receivier.recv().unwrap();
std::thread::sleep(check_interval);
}
Ok(DownloadResult { Ok(DownloadResult {
filename, filename,
username, username,
status: DownloadStatus::TimedOut, status,
elapsed_time: start_time.elapsed(), elapsed_time: start_time.elapsed(),
}) })
} }
@@ -299,6 +298,11 @@ impl Client {
let own_username = own_username.clone(); let own_username = own_username.clone();
let client_context_clone = client_context.clone(); let client_context_clone = client_context.clone();
trace!(
"[client] DownloadFromPeer token: {} peer: {:?}",
token,
peer
);
match maybe_download { match maybe_download {
Some(download) => { Some(download) => {
thread::spawn(move || { thread::spawn(move || {
@@ -318,13 +322,14 @@ impl Client {
Some(filename) => { Some(filename) => {
match download_peer.download_file( match download_peer.download_file(
client_context_clone, client_context_clone,
Some(download.size as usize), Some(download.clone())
Some(download.download_directory)
) { ) {
Ok((bytes, filename)) => { Ok((download, filename)) => {
info!("Successfully downloaded {} bytes to {}", bytes, filename); download.sender.send(DownloadStatus::Completed).unwrap();
info!("Successfully downloaded {} bytes to {}", download.size, filename);
} }
Err(e) => { Err(e) => {
download.sender.send(DownloadStatus::Failed).unwrap();
error!( error!(
"Failed to download file '{}' from {}:{} (token: {}) - Error: {}", "Failed to download file '{}' from {}:{} (token: {}) - Error: {}",
filename, peer.host, peer.port, download.token, e filename, peer.host, peer.port, download.token, e
@@ -470,6 +475,7 @@ impl Client {
size: transfer.size, size: transfer.size,
download_directory: download download_directory: download
.download_directory, .download_directory,
sender: download.sender.clone(),
}, },
); );
context.download_tokens.remove(&key); context.download_tokens.remove(&key);
@@ -537,11 +543,10 @@ impl Client {
match download_peer match download_peer
.download_file( .download_file(
client_context2.clone(), client_context2.clone(),
None,
None None
) { ) {
Ok((bytes, filename)) => { Ok((download, filename)) => {
trace!("[client] downloaded {} bytes {:?} ", filename,bytes); trace!("[client] downloaded {} bytes {:?} ", filename, download.size);
} }
Err(e) => { Err(e) => {
trace!("[client] failed to download: {}", e); trace!("[client] failed to download: {}", e);
+57 -149
View File
@@ -1,6 +1,7 @@
use crate::client::ClientContext; use crate::client::ClientContext;
use crate::message::server::MessageFactory; use crate::message::server::MessageFactory;
use crate::trace; use crate::trace;
use crate::types::Download;
use std::env; use std::env;
use std::fs::{self, File}; use std::fs::{self, File};
use std::io::{self, BufWriter, Read, Write}; use std::io::{self, BufWriter, Read, Write};
@@ -80,24 +81,14 @@ impl FileManager {
} }
fn create_download_path_from_filename( fn create_download_path_from_filename(
output_directory: Option<&str>, output_directory: String,
remote_username: &str, filename: String,
token: u32,
filename: Option<&str>,
) -> String { ) -> String {
let base_dir = output_directory.unwrap_or("/tmp"); format!(
"{}/{}",
let mut expanded_base_dir = FileManager::expand_path(base_dir); output_directory.trim_end_matches('/'),
if expanded_base_dir.ends_with('/') { Self::extract_filename_from_path(&filename)
expanded_base_dir.pop(); )
}
if let Some(filename) = filename {
let clean_filename = Self::extract_filename_from_path(filename);
format!("{expanded_base_dir}/{clean_filename}")
} else {
format!("{expanded_base_dir}/{remote_username}_{token}")
}
} }
} }
@@ -245,9 +236,8 @@ impl DownloadPeer {
pub fn download_file( pub fn download_file(
self, self,
client_context: Arc<RwLock<ClientContext>>, client_context: Arc<RwLock<ClientContext>>,
mut expected_size: Option<usize>, mut download: Option<Download>,
mut download_path: Option<String>, ) -> Result<(Download, String), io::Error> {
) -> Result<(usize, String), io::Error> {
let mut stream = self.establish_connection()?; let mut stream = self.establish_connection()?;
trace!("[download_peer:{}] connected", self.username); trace!("[download_peer:{}] connected", self.username);
@@ -256,7 +246,6 @@ impl DownloadPeer {
let mut processor = StreamProcessor::new(self.no_pierce, self.token); let mut processor = StreamProcessor::new(self.no_pierce, self.token);
let mut read_buffer = [1u8; 8192]; let mut read_buffer = [1u8; 8192];
let mut download_info: Option<crate::types::Download> = None;
trace!( trace!(
"[download_peer:{}] Starting to read data from peer", "[download_peer:{}] Starting to read data from peer",
@@ -265,12 +254,13 @@ impl DownloadPeer {
loop { loop {
match stream.read(&mut read_buffer) { match stream.read(&mut read_buffer) {
// connection closed
Ok(0) => { Ok(0) => {
trace!( trace!(
"[download_peer:{}] connection closed by peer", "[download_peer:{}] connection closed by peer",
self.username self.username
); );
break; // Connection closed break;
} }
Ok(bytes_read) => { Ok(bytes_read) => {
let data = &read_buffer[..bytes_read]; let data = &read_buffer[..bytes_read];
@@ -288,30 +278,27 @@ impl DownloadPeer {
token token
); );
download_info = client_context let read = client_context.read().unwrap();
.read()
.unwrap()
.download_tokens
.get(&token_u32)
.cloned();
match &download_info { let download_info =
Some(download) => { read.download_tokens.get(&token_u32);
match download_info {
Some(d) => {
trace!( trace!(
"[download_peer:{}] got download info for token: {} - filename: {}", "[download_peer:{}] got download info for token: {} - filename: {}",
self.username, self.username,
token_u32, token_u32,
download.filename d.filename
); );
expected_size = Some(download.size as usize); trace!("[download_peer:{}] setting download info for token: {}",
download_path = self.username,
Some(download.download_directory.clone()); token_u32);
download = Some(d.clone());
} }
None => { None => {
trace!( panic!(
"[download_peer:{}] no download info for token: {}", "No download info for token {token_u32}"
self.username,
self.token
); );
} }
} }
@@ -321,7 +308,9 @@ impl DownloadPeer {
processor.process_data_chunk(data); processor.process_data_chunk(data);
if !processor.should_continue(expected_size) { if !processor.should_continue(Some(
download.clone().unwrap().size as usize,
)) {
break; break;
} }
} }
@@ -334,45 +323,39 @@ impl DownloadPeer {
"[download_peer:{}] finished reading data from peer", "[download_peer:{}] finished reading data from peer",
self.username self.username
); );
let download = download.clone().unwrap_or_else(|| {
let output_directory = download_path.as_ref().map(|path| { panic!(
if Path::new(path).is_dir() { "[download_peer] No download info found for token: {}",
path.as_str() self.token
} else if let Some(parent) = Path::new(path).parent() { )
parent.to_str().unwrap_or("/tmp")
} else {
"/tmp"
}
}); });
let filename = download_info.as_ref().map(|d| d.filename.as_str()); let download_directory = download.download_directory.clone();
let mut expaned_path = FileManager::expand_path(&download_directory);
let final_path = if let Some(output_path) = &download_path { if !Path::new(&expaned_path).is_dir() {
if Path::new(output_path).is_dir() || output_path.ends_with('/') { expaned_path = Path::new(&expaned_path)
// output_path is a directory, append filename .parent()
FileManager::create_download_path_from_filename( .unwrap()
Some(output_path), .to_str()
&self.username, .unwrap()
self.token, .to_string()
filename,
)
} else {
output_path.clone()
} }
} else {
FileManager::create_download_path_from_filename( let final_path = FileManager::create_download_path_from_filename(
output_directory, expaned_path,
&self.username, download.filename.clone(),
self.token, );
filename,
)
};
if let Some(parent) = Path::new(&final_path).parent() { if let Some(parent) = Path::new(&final_path).parent() {
fs::create_dir_all(parent)?; fs::create_dir_all(parent)?;
} }
fs::write(&final_path, &processor.buffer)?; fs::write(&final_path, &processor.buffer).unwrap_or_else(|e| {
panic!(
"[download_peer] Failed to write to file {}: {}",
final_path, e
)
});
trace!( trace!(
"[download_peer:{}] download completed successfully: {} bytes, saved to: {}", "[download_peer:{}] download completed successfully: {} bytes, saved to: {}",
@@ -381,7 +364,7 @@ impl DownloadPeer {
final_path final_path
); );
Ok((processor.total_bytes, final_path)) Ok((download.clone(), final_path))
} }
} }
@@ -390,7 +373,7 @@ mod tests {
use super::{DownloadPeer, FileManager, StreamProcessor}; use super::{DownloadPeer, FileManager, StreamProcessor};
use crate::client::ClientContext; use crate::client::ClientContext;
use std::fs; use std::fs;
use std::io::{Read, Write}; use std::io::Read;
use std::net::TcpListener; use std::net::TcpListener;
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::thread; use std::thread;
@@ -441,9 +424,7 @@ mod tests {
"own_username".to_string(), "own_username".to_string(),
); );
let dummy_context = Arc::new(RwLock::new(ClientContext::new())); let dummy_context = Arc::new(RwLock::new(ClientContext::new()));
let _ = download_peer let _ = download_peer.download_file(dummy_context).unwrap();
.download_file(dummy_context, None, None)
.unwrap();
// Give the client time to send messages // Give the client time to send messages
thread::sleep(Duration::from_millis(10)); thread::sleep(Duration::from_millis(10));
@@ -466,9 +447,7 @@ mod tests {
"own_username".to_string(), "own_username".to_string(),
); );
let dummy_context = Arc::new(RwLock::new(ClientContext::new())); let dummy_context = Arc::new(RwLock::new(ClientContext::new()));
let _ = download_peer let _ = download_peer.download_file(dummy_context).unwrap();
.download_file(dummy_context, None, None)
.unwrap();
// Give the client time to send messages // Give the client time to send messages
thread::sleep(Duration::from_millis(10)); thread::sleep(Duration::from_millis(10));
@@ -486,77 +465,6 @@ mod tests {
// No messages expected when no_pierce is false // No messages expected when no_pierce is false
} }
#[test]
pub fn test_download_file() {
let test_data = b"test file content";
let messages = Arc::new(RwLock::new(Vec::<Vec<u8>>::new()));
let messages_clone = Arc::clone(&messages);
// Create a test server that sends test data
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
thread::spawn(move || {
for stream in listener.incoming() {
match stream {
Ok(mut stream) => {
// Set non-blocking to avoid hanging
stream.set_nonblocking(true).unwrap();
// Read handshake data
let mut buffer = [0u8; 1024];
std::thread::sleep(Duration::from_millis(10));
if let Ok(bytes_read) = stream.read(&mut buffer) {
if bytes_read > 0 {
messages_clone
.write()
.unwrap()
.push(buffer[..bytes_read].to_vec());
}
}
// Send pierce token (4 bytes) then test data
stream.set_nonblocking(false).unwrap();
let _ = stream.write_all(&[42, 0, 0, 0]);
std::thread::sleep(Duration::from_millis(10));
let _ = stream.write_all(test_data);
}
Err(_) => break,
}
}
});
thread::sleep(Duration::from_millis(10));
let download_peer = DownloadPeer::new(
"remote_user".to_string(),
"127.0.0.1".to_string(),
u32::from(port),
42,
false,
"test_user".to_string(),
);
let dummy_context = Arc::new(RwLock::new(ClientContext::new()));
let result = download_peer.download_file(
dummy_context,
Some(test_data.len()),
Some("test_download.mp3".to_string()),
);
assert!(result.is_ok());
let (bytes_downloaded, file_path) = result.unwrap();
assert_eq!(bytes_downloaded, test_data.len());
assert_eq!(file_path, "test_download.mp3");
// Verify the file was written correctly
let downloaded_data = fs::read("test_download.mp3").unwrap();
assert_eq!(downloaded_data, test_data);
// Clean up test file
let _ = fs::remove_file("test_download.mp3");
}
#[test] #[test]
fn test_file_manager_create_paths_with_custom_path() { fn test_file_manager_create_paths_with_custom_path() {
let paths = FileManager::create_download_paths( let paths = FileManager::create_download_paths(
+2 -1
View File
@@ -1,4 +1,4 @@
use std::collections::HashMap; use std::{collections::HashMap, sync::mpsc::Sender};
use crate::{error::Result, message::Message, utils::zlib::deflate}; use crate::{error::Result, message::Message, utils::zlib::deflate};
@@ -88,6 +88,7 @@ pub struct Download {
pub token: u32, pub token: u32,
pub size: u64, pub size: u64,
pub download_directory: String, pub download_directory: String,
pub sender: Sender<DownloadStatus>,
} }
#[derive(Debug, Clone)] #[derive(Debug, Clone)]