chore: formatting
This commit is contained in:
Michel de Graaf
@@ -1,5 +1,5 @@
|
||||
use std::sync::mpsc::{channel, Receiver, Sender};
|
||||
use std::sync::Arc;
|
||||
use std::sync::mpsc::{Receiver, Sender, channel};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::trace;
|
||||
@@ -161,7 +161,10 @@ impl ActorSystem {
|
||||
}
|
||||
}
|
||||
}
|
||||
trace!("[actor_system] run_actor_loop ENDED - processed {} messages, {} ticks", message_count, tick_count);
|
||||
trace!(
|
||||
"[actor_system] run_actor_loop ENDED - processed {} messages, {} ticks",
|
||||
message_count, tick_count
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -159,9 +159,12 @@ impl PeerActor {
|
||||
if let Some(ref handle) = self.self_handle
|
||||
&& let Err(e) =
|
||||
handle.send(PeerMessage::SendMessage(transfer_response))
|
||||
{
|
||||
error!("[peer:{}] Failed to send TransferResponse message: {}", username, e);
|
||||
}
|
||||
{
|
||||
error!(
|
||||
"[peer:{}] Failed to send TransferResponse message: {}",
|
||||
username, e
|
||||
);
|
||||
}
|
||||
}
|
||||
PeerMessage::TransferResponse {
|
||||
token,
|
||||
@@ -177,15 +180,15 @@ impl PeerActor {
|
||||
if !allowed {
|
||||
if let Some(reason_text) = reason {
|
||||
debug!(
|
||||
"[peer:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...",
|
||||
username, reason_text, token
|
||||
);
|
||||
"[peer:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...",
|
||||
username, reason_text, token
|
||||
);
|
||||
}
|
||||
} else {
|
||||
debug!(
|
||||
"[peer:{}] Transfer allowed, ready to connect with token {:}",
|
||||
username, token
|
||||
);
|
||||
"[peer:{}] Transfer allowed, ready to connect with token {:}",
|
||||
username, token
|
||||
);
|
||||
self.client_channel
|
||||
.send(ClientOperation::DownloadFromPeer(
|
||||
token,
|
||||
@@ -258,7 +261,9 @@ impl PeerActor {
|
||||
let username = self.peer.read().unwrap().username.clone();
|
||||
error!(
|
||||
"[peer:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.",
|
||||
username, e, e.kind()
|
||||
username,
|
||||
e,
|
||||
e.kind()
|
||||
);
|
||||
self.disconnect_with_error(e);
|
||||
return;
|
||||
|
||||
@@ -24,7 +24,7 @@ use std::sync::mpsc::{Receiver, Sender};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::{debug, error, info, trace, warn, SoulseekRs};
|
||||
use crate::{SoulseekRs, debug, error, info, trace, warn};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct PeerAddress {
|
||||
@@ -483,8 +483,8 @@ impl ServerActor {
|
||||
obfuscated_port,
|
||||
} => {
|
||||
debug!(
|
||||
"[server] Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})"
|
||||
, username, host, port, obfuscation_type, obfuscated_port
|
||||
"[server] Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})",
|
||||
username, host, port, obfuscation_type, obfuscated_port
|
||||
);
|
||||
|
||||
if let Err(e) = self.client_channel.send(
|
||||
@@ -496,7 +496,10 @@ impl ServerActor {
|
||||
obfuscated_port,
|
||||
},
|
||||
) {
|
||||
error!("[server] Error forwarding GetPeerAddress response to client: {}", e);
|
||||
error!(
|
||||
"[server] Error forwarding GetPeerAddress response to client: {}",
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
ServerMessage::ProcessRead => {
|
||||
@@ -515,23 +518,27 @@ impl ServerActor {
|
||||
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;
|
||||
}
|
||||
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;
|
||||
}
|
||||
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));
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
});
|
||||
}
|
||||
ServerMessage::FileSearch { token, query } => {
|
||||
@@ -560,7 +567,8 @@ impl ServerActor {
|
||||
Err(e) => {
|
||||
error!(
|
||||
"[server] Error reading from server: {} (kind: {:?}). Disconnecting.",
|
||||
e, e.kind()
|
||||
e,
|
||||
e.kind()
|
||||
);
|
||||
self.disconnect_with_error(e);
|
||||
return;
|
||||
|
||||
+336
-290
@@ -1,27 +1,27 @@
|
||||
use crate::actor::server_actor::{PeerAddress, ServerActor, ServerMessage};
|
||||
use crate::actor::ActorHandle;
|
||||
use crate::actor::server_actor::{PeerAddress, ServerActor, ServerMessage};
|
||||
use crate::types::DownloadStatus;
|
||||
use crate::utils::logger;
|
||||
use crate::{
|
||||
actor::{peer_registry::PeerRegistry, ActorSystem},
|
||||
Transfer,
|
||||
actor::{ActorSystem, peer_registry::PeerRegistry},
|
||||
error::{Result, SoulseekRs},
|
||||
peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer},
|
||||
peer::{ConnectionType, DownloadPeer, NewPeer, Peer, listen::Listen},
|
||||
types::{Download, Search, SearchResult},
|
||||
utils::{md5, thread_pool::ThreadPool},
|
||||
Transfer,
|
||||
};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
net::TcpStream,
|
||||
sync::{
|
||||
RwLock,
|
||||
atomic::{AtomicBool, Ordering},
|
||||
mpsc::{Receiver, Sender},
|
||||
RwLock,
|
||||
},
|
||||
thread::{self, sleep},
|
||||
};
|
||||
use std::{
|
||||
sync::{mpsc, Arc},
|
||||
sync::{Arc, mpsc},
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
@@ -286,9 +286,10 @@ impl Client {
|
||||
pub fn remove_peer(&self, username: &str) {
|
||||
let context = self.context.read().unwrap();
|
||||
if let Some(ref registry) = context.peer_registry
|
||||
&& let Some(handle) = registry.remove_peer(username) {
|
||||
let _ = handle.stop();
|
||||
}
|
||||
&& let Some(handle) = registry.remove_peer(username)
|
||||
{
|
||||
let _ = handle.stop();
|
||||
}
|
||||
}
|
||||
|
||||
pub fn search(
|
||||
@@ -334,10 +335,11 @@ impl Client {
|
||||
|
||||
// Check if cancelled
|
||||
if let Some(ref flag) = cancel_flag
|
||||
&& flag.load(Ordering::Relaxed) {
|
||||
info!("Search cancelled by user");
|
||||
break;
|
||||
}
|
||||
&& flag.load(Ordering::Relaxed)
|
||||
{
|
||||
info!("Search cancelled by user");
|
||||
break;
|
||||
}
|
||||
|
||||
// Check if timeout reached
|
||||
if start.elapsed() >= timeout {
|
||||
@@ -470,225 +472,12 @@ impl Client {
|
||||
client_context: Arc<RwLock<ClientContext>>,
|
||||
own_username: String,
|
||||
) {
|
||||
thread::spawn(move || loop {
|
||||
match reader.recv() {
|
||||
Ok(operation) => {
|
||||
match operation {
|
||||
ClientOperation::ConnectToPeer(peer) => {
|
||||
let client_context_clone = client_context.clone();
|
||||
let own_username_clone = own_username.clone();
|
||||
|
||||
thread::spawn(move || {
|
||||
Self::connect_to_peer(
|
||||
peer,
|
||||
client_context_clone,
|
||||
own_username_clone,
|
||||
None,
|
||||
);
|
||||
});
|
||||
}
|
||||
ClientOperation::SearchResult(search_result) => {
|
||||
trace!("[client] SearchResult {:?}", search_result);
|
||||
let mut context = client_context.write().unwrap();
|
||||
let result_token = search_result.token;
|
||||
|
||||
// Find the search with matching token
|
||||
for search in context.searches.values_mut() {
|
||||
if search.token == result_token {
|
||||
search.results.push(search_result);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
ClientOperation::PeerDisconnected(username, error) => {
|
||||
let context = client_context.read().unwrap();
|
||||
if let Some(ref registry) = context.peer_registry
|
||||
&& let Some(handle) =
|
||||
registry.remove_peer(&username)
|
||||
{
|
||||
let _ = handle.stop();
|
||||
}
|
||||
if let Some(error) = error {
|
||||
warn!(
|
||||
"[client] Peer {} disconnected with error: {:?}",
|
||||
username, error
|
||||
);
|
||||
Self::process_failed_uploads(
|
||||
client_context.clone(),
|
||||
&username,
|
||||
None,
|
||||
);
|
||||
}
|
||||
}
|
||||
ClientOperation::PierceFireWall(peer) => {
|
||||
Self::pierce_firewall(
|
||||
peer,
|
||||
client_context.clone(),
|
||||
own_username.clone(),
|
||||
);
|
||||
}
|
||||
ClientOperation::DownloadFromPeer(
|
||||
token,
|
||||
peer,
|
||||
allowed,
|
||||
) => {
|
||||
let maybe_download = {
|
||||
let client_context =
|
||||
client_context.read().unwrap();
|
||||
client_context
|
||||
.get_download_by_token(token)
|
||||
.cloned()
|
||||
};
|
||||
let own_username = own_username.clone();
|
||||
let client_context_clone = client_context.clone();
|
||||
|
||||
trace!(
|
||||
"[client] DownloadFromPeer token: {} peer: {:?}",
|
||||
token,
|
||||
peer
|
||||
);
|
||||
match maybe_download {
|
||||
Some(download) => {
|
||||
thread::spawn(move || {
|
||||
let download_peer = DownloadPeer::new(
|
||||
download.username.clone(),
|
||||
peer.host.clone(),
|
||||
peer.port,
|
||||
token,
|
||||
allowed,
|
||||
own_username,
|
||||
);
|
||||
let filename: Option<&str> = download
|
||||
.filename
|
||||
.split('\\')
|
||||
.next_back();
|
||||
match filename {
|
||||
Some(filename) => {
|
||||
match download_peer.download_file(
|
||||
client_context_clone.clone(),
|
||||
Some(download.clone()),
|
||||
None
|
||||
) {
|
||||
Ok((download, filename)) => {
|
||||
let _ = download.sender.send(DownloadStatus::Completed);
|
||||
client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Completed);
|
||||
info!("Successfully downloaded {} bytes to {}", download.size, filename);
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = download.sender.send(DownloadStatus::Failed);
|
||||
client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Failed);
|
||||
error!(
|
||||
"Failed to download file '{}' from {}:{} (token: {}) - Error: {}",
|
||||
filename, peer.host, peer.port, download.token, e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
None => error!("Cant find filename to save download: {:?}", download.filename),
|
||||
}
|
||||
});
|
||||
}
|
||||
None => {
|
||||
error!(
|
||||
"Can't find download with token {:?}",
|
||||
token
|
||||
);
|
||||
}
|
||||
};
|
||||
}
|
||||
ClientOperation::NewPeer(new_peer) => {
|
||||
let peer_exists = client_context
|
||||
.read()
|
||||
.unwrap()
|
||||
.peer_registry
|
||||
.as_ref()
|
||||
.map(|r| r.contains(&new_peer.username))
|
||||
.unwrap_or(false);
|
||||
|
||||
if peer_exists {
|
||||
debug!(
|
||||
"Already connected to {}",
|
||||
new_peer.username
|
||||
);
|
||||
} else if let Some(server_sender) =
|
||||
&client_context.read().unwrap().server_sender
|
||||
{
|
||||
server_sender
|
||||
.send(ServerMessage::GetPeerAddress(
|
||||
new_peer.username.clone(),
|
||||
))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let addr = new_peer.tcp_stream.peer_addr().unwrap();
|
||||
let host = addr.ip().to_string();
|
||||
let port: u32 = addr.port().into();
|
||||
|
||||
let peer = Peer {
|
||||
username: new_peer.username.clone(),
|
||||
connection_type: new_peer.connection_type,
|
||||
host,
|
||||
port,
|
||||
token: Some(new_peer.token),
|
||||
privileged: None,
|
||||
obfuscated_port: None,
|
||||
unknown: None,
|
||||
};
|
||||
|
||||
Self::connect_to_peer(
|
||||
peer,
|
||||
client_context.clone(),
|
||||
own_username.clone(),
|
||||
Some(new_peer.tcp_stream),
|
||||
);
|
||||
}
|
||||
ClientOperation::GetPeerAddressResponse {
|
||||
username,
|
||||
host,
|
||||
port,
|
||||
obfuscation_type,
|
||||
obfuscated_port,
|
||||
} => {
|
||||
debug!(
|
||||
"Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})",
|
||||
username, host, port, obfuscation_type, obfuscated_port
|
||||
);
|
||||
|
||||
let peer_exists = client_context
|
||||
.read()
|
||||
.unwrap()
|
||||
.peer_registry
|
||||
.as_ref()
|
||||
.map(|r| r.contains(&username))
|
||||
.unwrap_or(false);
|
||||
|
||||
if peer_exists {
|
||||
// don't know if i should update? and or reconnect the peer
|
||||
// debug!(
|
||||
// "existing peer: {:?}, new peer details:
|
||||
// username: {},
|
||||
// host: {},
|
||||
// port: {}
|
||||
// obfuscation_type: {}
|
||||
// obfuscated_port: {}",
|
||||
// peer,
|
||||
// username,
|
||||
// host,
|
||||
// port,
|
||||
// obfuscation_type,
|
||||
// obfuscated_port,
|
||||
// );
|
||||
} else {
|
||||
let peer = Peer::new(
|
||||
username,
|
||||
ConnectionType::P,
|
||||
host,
|
||||
port,
|
||||
None,
|
||||
0,
|
||||
obfuscation_type.try_into().unwrap(),
|
||||
obfuscated_port.try_into().unwrap(),
|
||||
);
|
||||
thread::spawn(move || {
|
||||
loop {
|
||||
match reader.recv() {
|
||||
Ok(operation) => {
|
||||
match operation {
|
||||
ClientOperation::ConnectToPeer(peer) => {
|
||||
let client_context_clone =
|
||||
client_context.clone();
|
||||
let own_username_clone = own_username.clone();
|
||||
@@ -702,62 +491,319 @@ impl Client {
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
ClientOperation::UpdateDownloadTokens(
|
||||
transfer,
|
||||
username,
|
||||
) => {
|
||||
let mut context = client_context.write().unwrap();
|
||||
|
||||
let download_to_update =
|
||||
context.get_downloads().iter().find_map(|d| {
|
||||
if d.username == username
|
||||
&& d.filename == transfer.filename
|
||||
{
|
||||
Some((d.token, d.clone()))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
|
||||
if let Some((old_token, download)) =
|
||||
download_to_update
|
||||
{
|
||||
ClientOperation::SearchResult(search_result) => {
|
||||
trace!(
|
||||
"[client] UpdateDownloadTokens found {old_token}, transfer: {:?}",
|
||||
transfer
|
||||
);
|
||||
"[client] SearchResult {:?}",
|
||||
search_result
|
||||
);
|
||||
let mut context =
|
||||
client_context.write().unwrap();
|
||||
let result_token = search_result.token;
|
||||
|
||||
context.add_download(Download {
|
||||
username: username.clone(),
|
||||
filename: transfer.filename,
|
||||
token: transfer.token,
|
||||
size: transfer.size,
|
||||
download_directory: download
|
||||
.download_directory,
|
||||
status: download.status.clone(),
|
||||
sender: download.sender.clone(),
|
||||
});
|
||||
context.remove_download(old_token);
|
||||
// Find the search with matching token
|
||||
for search in context.searches.values_mut() {
|
||||
if search.token == result_token {
|
||||
search.results.push(search_result);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
ClientOperation::PeerDisconnected(
|
||||
username,
|
||||
error,
|
||||
) => {
|
||||
let context = client_context.read().unwrap();
|
||||
if let Some(ref registry) =
|
||||
context.peer_registry
|
||||
&& let Some(handle) =
|
||||
registry.remove_peer(&username)
|
||||
{
|
||||
let _ = handle.stop();
|
||||
}
|
||||
if let Some(error) = error {
|
||||
warn!(
|
||||
"[client] Peer {} disconnected with error: {:?}",
|
||||
username, error
|
||||
);
|
||||
Self::process_failed_uploads(
|
||||
client_context.clone(),
|
||||
&username,
|
||||
None,
|
||||
);
|
||||
}
|
||||
}
|
||||
ClientOperation::PierceFireWall(peer) => {
|
||||
Self::pierce_firewall(
|
||||
peer,
|
||||
client_context.clone(),
|
||||
own_username.clone(),
|
||||
);
|
||||
}
|
||||
ClientOperation::DownloadFromPeer(
|
||||
token,
|
||||
peer,
|
||||
allowed,
|
||||
) => {
|
||||
let maybe_download = {
|
||||
let client_context =
|
||||
client_context.read().unwrap();
|
||||
client_context
|
||||
.get_download_by_token(token)
|
||||
.cloned()
|
||||
};
|
||||
let own_username = own_username.clone();
|
||||
let client_context_clone =
|
||||
client_context.clone();
|
||||
|
||||
trace!(
|
||||
"[client] DownloadFromPeer token: {} peer: {:?}",
|
||||
token, peer
|
||||
);
|
||||
match maybe_download {
|
||||
Some(download) => {
|
||||
thread::spawn(move || {
|
||||
let download_peer =
|
||||
DownloadPeer::new(
|
||||
download.username.clone(),
|
||||
peer.host.clone(),
|
||||
peer.port,
|
||||
token,
|
||||
allowed,
|
||||
own_username,
|
||||
);
|
||||
let filename: Option<&str> =
|
||||
download
|
||||
.filename
|
||||
.split('\\')
|
||||
.next_back();
|
||||
match filename {
|
||||
Some(filename) => {
|
||||
match download_peer
|
||||
.download_file(
|
||||
client_context_clone
|
||||
.clone(),
|
||||
Some(download.clone()),
|
||||
None,
|
||||
) {
|
||||
Ok((
|
||||
download,
|
||||
filename,
|
||||
)) => {
|
||||
let _ = download.sender.send(DownloadStatus::Completed);
|
||||
client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Completed);
|
||||
info!(
|
||||
"Successfully downloaded {} bytes to {}",
|
||||
download.size,
|
||||
filename
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = download.sender.send(DownloadStatus::Failed);
|
||||
client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Failed);
|
||||
error!(
|
||||
"Failed to download file '{}' from {}:{} (token: {}) - Error: {}",
|
||||
filename,
|
||||
peer.host,
|
||||
peer.port,
|
||||
download.token,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
None => error!(
|
||||
"Cant find filename to save download: {:?}",
|
||||
download.filename
|
||||
),
|
||||
}
|
||||
});
|
||||
}
|
||||
None => {
|
||||
error!(
|
||||
"Can't find download with token {:?}",
|
||||
token
|
||||
);
|
||||
}
|
||||
};
|
||||
}
|
||||
ClientOperation::NewPeer(new_peer) => {
|
||||
let peer_exists = client_context
|
||||
.read()
|
||||
.unwrap()
|
||||
.peer_registry
|
||||
.as_ref()
|
||||
.map(|r| r.contains(&new_peer.username))
|
||||
.unwrap_or(false);
|
||||
|
||||
if peer_exists {
|
||||
debug!(
|
||||
"Already connected to {}",
|
||||
new_peer.username
|
||||
);
|
||||
} else if let Some(server_sender) =
|
||||
&client_context
|
||||
.read()
|
||||
.unwrap()
|
||||
.server_sender
|
||||
{
|
||||
server_sender
|
||||
.send(ServerMessage::GetPeerAddress(
|
||||
new_peer.username.clone(),
|
||||
))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let addr =
|
||||
new_peer.tcp_stream.peer_addr().unwrap();
|
||||
let host = addr.ip().to_string();
|
||||
let port: u32 = addr.port().into();
|
||||
|
||||
let peer = Peer {
|
||||
username: new_peer.username.clone(),
|
||||
connection_type: new_peer.connection_type,
|
||||
host,
|
||||
port,
|
||||
token: Some(new_peer.token),
|
||||
privileged: None,
|
||||
obfuscated_port: None,
|
||||
unknown: None,
|
||||
};
|
||||
|
||||
Self::connect_to_peer(
|
||||
peer,
|
||||
client_context.clone(),
|
||||
own_username.clone(),
|
||||
Some(new_peer.tcp_stream),
|
||||
);
|
||||
}
|
||||
ClientOperation::GetPeerAddressResponse {
|
||||
username,
|
||||
host,
|
||||
port,
|
||||
obfuscation_type,
|
||||
obfuscated_port,
|
||||
} => {
|
||||
debug!(
|
||||
"Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})",
|
||||
username,
|
||||
host,
|
||||
port,
|
||||
obfuscation_type,
|
||||
obfuscated_port
|
||||
);
|
||||
|
||||
let peer_exists = client_context
|
||||
.read()
|
||||
.unwrap()
|
||||
.peer_registry
|
||||
.as_ref()
|
||||
.map(|r| r.contains(&username))
|
||||
.unwrap_or(false);
|
||||
|
||||
if peer_exists {
|
||||
// don't know if i should update? and or reconnect the peer
|
||||
// debug!(
|
||||
// "existing peer: {:?}, new peer details:
|
||||
// username: {},
|
||||
// host: {},
|
||||
// port: {}
|
||||
// obfuscation_type: {}
|
||||
// obfuscated_port: {}",
|
||||
// peer,
|
||||
// username,
|
||||
// host,
|
||||
// port,
|
||||
// obfuscation_type,
|
||||
// obfuscated_port,
|
||||
// );
|
||||
} else {
|
||||
let peer = Peer::new(
|
||||
username,
|
||||
ConnectionType::P,
|
||||
host,
|
||||
port,
|
||||
None,
|
||||
0,
|
||||
obfuscation_type.try_into().unwrap(),
|
||||
obfuscated_port.try_into().unwrap(),
|
||||
);
|
||||
let client_context_clone =
|
||||
client_context.clone();
|
||||
let own_username_clone =
|
||||
own_username.clone();
|
||||
|
||||
thread::spawn(move || {
|
||||
Self::connect_to_peer(
|
||||
peer,
|
||||
client_context_clone,
|
||||
own_username_clone,
|
||||
None,
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
ClientOperation::UpdateDownloadTokens(
|
||||
transfer,
|
||||
username,
|
||||
) => {
|
||||
let mut context =
|
||||
client_context.write().unwrap();
|
||||
|
||||
let download_to_update = context
|
||||
.get_downloads()
|
||||
.iter()
|
||||
.find_map(|d| {
|
||||
if d.username == username
|
||||
&& d.filename == transfer.filename
|
||||
{
|
||||
Some((d.token, d.clone()))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
|
||||
if let Some((old_token, download)) =
|
||||
download_to_update
|
||||
{
|
||||
trace!(
|
||||
"[client] UpdateDownloadTokens found {old_token}, transfer: {:?}",
|
||||
transfer
|
||||
);
|
||||
|
||||
context.add_download(Download {
|
||||
username: username.clone(),
|
||||
filename: transfer.filename,
|
||||
token: transfer.token,
|
||||
size: transfer.size,
|
||||
download_directory: download
|
||||
.download_directory,
|
||||
status: download.status.clone(),
|
||||
sender: download.sender.clone(),
|
||||
});
|
||||
context.remove_download(old_token);
|
||||
}
|
||||
}
|
||||
ClientOperation::UploadFailed(
|
||||
username,
|
||||
filename,
|
||||
) => {
|
||||
Self::process_failed_uploads(
|
||||
client_context.clone(),
|
||||
&username,
|
||||
Some(&filename),
|
||||
);
|
||||
}
|
||||
ClientOperation::SetServerSender(sender) => {
|
||||
client_context.write().unwrap().server_sender =
|
||||
Some(sender);
|
||||
debug!("[client] Server sender initialized");
|
||||
}
|
||||
}
|
||||
ClientOperation::UploadFailed(username, filename) => {
|
||||
Self::process_failed_uploads(
|
||||
client_context.clone(),
|
||||
&username,
|
||||
Some(&filename),
|
||||
);
|
||||
}
|
||||
ClientOperation::SetServerSender(sender) => {
|
||||
client_context.write().unwrap().server_sender =
|
||||
Some(sender);
|
||||
debug!("[client] Server sender initialized");
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("[client] Channel recv error: {:?}", e);
|
||||
break;
|
||||
Err(e) => {
|
||||
error!("[client] Channel recv error: {:?}", e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -772,7 +818,10 @@ impl Client {
|
||||
let client_context = client_context.clone();
|
||||
|
||||
let peer_clone = peer.clone();
|
||||
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 {
|
||||
ConnectionType::P => {
|
||||
let username = peer.username.clone();
|
||||
@@ -784,8 +833,7 @@ impl Client {
|
||||
Err(e) => {
|
||||
trace!(
|
||||
"Failed to spawn peer actor for {:?}: {:?}",
|
||||
username,
|
||||
e
|
||||
username, e
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -797,8 +845,7 @@ impl Client {
|
||||
ConnectionType::F => {
|
||||
trace!(
|
||||
"[client] downloading from: {}, {:?}",
|
||||
peer.username,
|
||||
peer.token
|
||||
peer.username, peer.token
|
||||
);
|
||||
let download_peer = DownloadPeer::new(
|
||||
peer.username,
|
||||
@@ -817,8 +864,7 @@ impl Client {
|
||||
Ok((download, filename)) => {
|
||||
trace!(
|
||||
"[client] downloaded {} bytes {:?} ",
|
||||
filename,
|
||||
download.size
|
||||
filename, download.size
|
||||
);
|
||||
let _ = download.sender.send(DownloadStatus::Completed);
|
||||
client_context
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
use crate::message::{handlers::Handlers, Message};
|
||||
use crate::message::{Message, handlers::Handlers};
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::warn;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
use crate::{
|
||||
message::{server::MessageFactory, Message, MessageHandler},
|
||||
message::{Message, MessageHandler, server::MessageFactory},
|
||||
peer::PeerMessage,
|
||||
};
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
@@ -20,9 +20,7 @@ impl MessageHandler<PeerMessage> for PeerInit {
|
||||
let token = message.read_int32();
|
||||
trace!(
|
||||
"PeerInit: username: {}, connection_type: {}, token: {}",
|
||||
username,
|
||||
connection_type,
|
||||
token
|
||||
username, connection_type, token
|
||||
);
|
||||
|
||||
sender.send(PeerMessage::SetUsername(username)).unwrap();
|
||||
|
||||
@@ -2,8 +2,8 @@ use crate::{debug, info};
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::{
|
||||
actor::server_actor::ServerMessage, message::handlers::MessageHandler,
|
||||
message::Message,
|
||||
actor::server_actor::ServerMessage, message::Message,
|
||||
message::handlers::MessageHandler,
|
||||
};
|
||||
|
||||
pub struct FileSearchHandler;
|
||||
|
||||
@@ -195,8 +195,7 @@ impl DownloadPeer {
|
||||
) -> Result<(), DownloadError> {
|
||||
trace!(
|
||||
"[download_peer:{}] performing handshake no_pierce: {}",
|
||||
self.username,
|
||||
self.no_pierce
|
||||
self.username, self.no_pierce
|
||||
);
|
||||
|
||||
if !self.no_pierce {
|
||||
@@ -231,8 +230,7 @@ impl DownloadPeer {
|
||||
|
||||
trace!(
|
||||
"[download_peer:{}] got token: {} from data chunk",
|
||||
self.username,
|
||||
token_u32
|
||||
self.username, token_u32
|
||||
);
|
||||
|
||||
stream
|
||||
@@ -277,8 +275,7 @@ impl DownloadPeer {
|
||||
Ok(0) => {
|
||||
trace!(
|
||||
"[download_peer:{}] connection closed by peer. bytes read: {}",
|
||||
self.username,
|
||||
processor.total_bytes
|
||||
self.username, processor.total_bytes
|
||||
);
|
||||
break;
|
||||
}
|
||||
@@ -294,9 +291,7 @@ impl DownloadPeer {
|
||||
)?;
|
||||
trace!(
|
||||
"[download_peer:{}] got download info for token: {} - filename: {}",
|
||||
self.username,
|
||||
self.token,
|
||||
new_download.filename
|
||||
self.username, self.token, new_download.filename
|
||||
);
|
||||
download = Some(new_download);
|
||||
processor.received = true;
|
||||
@@ -307,30 +302,30 @@ impl DownloadPeer {
|
||||
chunk_counter += 1;
|
||||
|
||||
if let Some(ref dl) = download
|
||||
&& chunk_counter % PROGRESS_UPDATE_CHUNKS == 0 {
|
||||
let elapsed =
|
||||
last_update_time.elapsed().as_secs_f64();
|
||||
let bytes_since_last_update =
|
||||
PROGRESS_UPDATE_CHUNKS * READ_BUFFER_SIZE;
|
||||
let speed = if elapsed > 0.0 {
|
||||
bytes_since_last_update as f64 / elapsed
|
||||
} else {
|
||||
0.0
|
||||
};
|
||||
&& chunk_counter % PROGRESS_UPDATE_CHUNKS == 0
|
||||
{
|
||||
let elapsed = last_update_time.elapsed().as_secs_f64();
|
||||
let bytes_since_last_update =
|
||||
PROGRESS_UPDATE_CHUNKS * READ_BUFFER_SIZE;
|
||||
let speed = if elapsed > 0.0 {
|
||||
bytes_since_last_update as f64 / elapsed
|
||||
} else {
|
||||
0.0
|
||||
};
|
||||
|
||||
let status = DownloadStatus::InProgress {
|
||||
bytes_downloaded: processor.total_bytes as u64,
|
||||
total_bytes: dl.size,
|
||||
speed_bytes_per_sec: speed,
|
||||
};
|
||||
let _ = dl.sender.send(status.clone());
|
||||
client_context
|
||||
.write()
|
||||
.unwrap()
|
||||
.update_download_with_status(dl.token, status);
|
||||
let status = DownloadStatus::InProgress {
|
||||
bytes_downloaded: processor.total_bytes as u64,
|
||||
total_bytes: dl.size,
|
||||
speed_bytes_per_sec: speed,
|
||||
};
|
||||
let _ = dl.sender.send(status.clone());
|
||||
client_context
|
||||
.write()
|
||||
.unwrap()
|
||||
.update_download_with_status(dl.token, status);
|
||||
|
||||
last_update_time = Instant::now();
|
||||
}
|
||||
last_update_time = Instant::now();
|
||||
}
|
||||
|
||||
let expected_size = download
|
||||
.as_ref()
|
||||
@@ -486,7 +481,9 @@ mod tests {
|
||||
"file.mp3"
|
||||
);
|
||||
assert_eq!(
|
||||
FileManager::extract_filename_from_path("@@bhfrv\\Soulseek Downloads\\complete\\Beatport Top Deep House (2021)\\michel test file.mp3"),
|
||||
FileManager::extract_filename_from_path(
|
||||
"@@bhfrv\\Soulseek Downloads\\complete\\Beatport Top Deep House (2021)\\michel test file.mp3"
|
||||
),
|
||||
"michel test file.mp3"
|
||||
);
|
||||
assert_eq!(
|
||||
|
||||
@@ -9,7 +9,7 @@ use crate::client::{ClientContext, ClientOperation};
|
||||
use crate::message::{Message, MessageReader};
|
||||
use crate::peer::{ConnectionType, DownloadPeer, Peer};
|
||||
use crate::types::Download;
|
||||
use crate::{debug, error, info, trace, DownloadStatus};
|
||||
use crate::{DownloadStatus, debug, error, info, trace};
|
||||
|
||||
const PEER_INIT_MESSAGE_CODE: u8 = 1;
|
||||
|
||||
@@ -79,8 +79,7 @@ fn extract_download_from_buffer(
|
||||
let token = parse_token_from_buffer(&buffer, username)?;
|
||||
trace!(
|
||||
"[listener:{}] got transfer_token: {} from data chunk",
|
||||
username,
|
||||
token
|
||||
username, token
|
||||
);
|
||||
|
||||
let context = client_context.read().unwrap();
|
||||
@@ -88,7 +87,10 @@ fn extract_download_from_buffer(
|
||||
|
||||
if download.is_none() {
|
||||
let download_tokens = context.get_download_tokens();
|
||||
trace!("[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}", token, download_tokens);
|
||||
trace!(
|
||||
"[listener:{peer_ip}:{peer_port}] download token not found: {:?}, download tokens: {:?}",
|
||||
token, download_tokens
|
||||
);
|
||||
}
|
||||
|
||||
download
|
||||
@@ -129,8 +131,7 @@ fn handle_file_connection(
|
||||
) {
|
||||
trace!(
|
||||
"[client] DownloadFromPeer token: {} peer: {:?}",
|
||||
token,
|
||||
peer
|
||||
token, peer
|
||||
);
|
||||
|
||||
let download = extract_download_from_buffer(
|
||||
@@ -198,7 +199,9 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
|
||||
};
|
||||
|
||||
let Some(init_data) = parse_peer_init_message(message) else {
|
||||
error!("[listener:{peer_ip}:{peer_port}] Invalid or unknown peer init message");
|
||||
error!(
|
||||
"[listener:{peer_ip}:{peer_port}] Invalid or unknown peer init message"
|
||||
);
|
||||
return;
|
||||
};
|
||||
|
||||
@@ -241,7 +244,9 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
|
||||
});
|
||||
}
|
||||
ConnectionType::D => {
|
||||
debug!("[listener:{peer_ip}:{peer_port}] connection type is D, not supported yet, closing connection. ");
|
||||
debug!(
|
||||
"[listener:{peer_ip}:{peer_port}] connection type is D, not supported yet, closing connection. "
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,8 +2,8 @@ use std::env;
|
||||
use std::fs::{File, OpenOptions};
|
||||
use std::io::Write;
|
||||
use std::sync::{
|
||||
atomic::{AtomicBool, Ordering},
|
||||
Mutex, Once,
|
||||
atomic::{AtomicBool, Ordering},
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
|
||||
@@ -168,10 +168,11 @@ pub fn log(level: LogLevel, message: &str) {
|
||||
}
|
||||
// But ALSO write to file if configured (file logging bypasses buffering)
|
||||
if let Ok(mut log_file) = LOG_FILE.lock()
|
||||
&& let Some(file) = log_file.as_mut() {
|
||||
let _ = writeln!(file, "{}", formatted_message_plain);
|
||||
let _ = file.flush();
|
||||
}
|
||||
&& let Some(file) = log_file.as_mut()
|
||||
{
|
||||
let _ = writeln!(file, "{}", formatted_message_plain);
|
||||
let _ = file.flush();
|
||||
}
|
||||
} else {
|
||||
// Write to file if configured, otherwise to stderr
|
||||
if let Ok(mut log_file) = LOG_FILE.lock() {
|
||||
|
||||
@@ -369,7 +369,7 @@ fn bit_padding(input: &str) -> Vec<u8> {
|
||||
//check if bit length % 512 is 448 (64 less than 512)
|
||||
while (input_vector.len() * 8) % 512 != 448 {
|
||||
input_vector.push(0_u8); // push in another 8-bit 0 padded value until the correct
|
||||
// result is reached;
|
||||
// result is reached;
|
||||
}
|
||||
|
||||
let length_bits_as_u8_array = split_u64_to_u8_array(bit_length);
|
||||
@@ -467,8 +467,7 @@ fn all_alphanumeric_correct_hash() {
|
||||
|
||||
#[test]
|
||||
fn repeated_numbers_correct_hash() {
|
||||
let repeated_number_str =
|
||||
"12345678901234567890123456789012345678901234567890123456789012345678901234567890";
|
||||
let repeated_number_str = "12345678901234567890123456789012345678901234567890123456789012345678901234567890";
|
||||
|
||||
assert_eq!("57edf4a22be3c955ac49da2e2107b67a", md5(repeated_number_str));
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{mpsc, Arc, Mutex};
|
||||
use std::sync::{Arc, Mutex, mpsc};
|
||||
use std::thread;
|
||||
|
||||
type Job = Box<dyn FnOnce() + Send + 'static>;
|
||||
@@ -89,37 +89,39 @@ impl Worker {
|
||||
_active_threads: Arc<AtomicUsize>,
|
||||
_total_threads: usize,
|
||||
) -> Worker {
|
||||
let thread = thread::spawn(move || loop {
|
||||
let message = receiver.lock().unwrap().recv();
|
||||
match message {
|
||||
Ok(Message::NewJob(job)) => {
|
||||
// let active =
|
||||
// active_threads.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
// trace!(
|
||||
// "Thread {} started job (active: {}/{})",
|
||||
// id,
|
||||
// active,
|
||||
// total_threads
|
||||
// );
|
||||
let thread = thread::spawn(move || {
|
||||
loop {
|
||||
let message = receiver.lock().unwrap().recv();
|
||||
match message {
|
||||
Ok(Message::NewJob(job)) => {
|
||||
// let active =
|
||||
// active_threads.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
// trace!(
|
||||
// "Thread {} started job (active: {}/{})",
|
||||
// id,
|
||||
// active,
|
||||
// total_threads
|
||||
// );
|
||||
|
||||
job();
|
||||
job();
|
||||
|
||||
// let active =
|
||||
// active_threads.fetch_sub(1, Ordering::SeqCst) - 1;
|
||||
// trace!(
|
||||
// "Thread {} finished job (active: {}/{})",
|
||||
// id,
|
||||
// active,
|
||||
// total_threads
|
||||
// );
|
||||
}
|
||||
Ok(Message::Terminate) => {
|
||||
// Received termination signal
|
||||
break;
|
||||
}
|
||||
Err(_) => {
|
||||
// Channel disconnected
|
||||
break;
|
||||
// let active =
|
||||
// active_threads.fetch_sub(1, Ordering::SeqCst) - 1;
|
||||
// trace!(
|
||||
// "Thread {} finished job (active: {}/{})",
|
||||
// id,
|
||||
// active,
|
||||
// total_threads
|
||||
// );
|
||||
}
|
||||
Ok(Message::Terminate) => {
|
||||
// Received termination signal
|
||||
break;
|
||||
}
|
||||
Err(_) => {
|
||||
// Channel disconnected
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user