diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index d732146..bec2d16 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,9 +1,18 @@ name: Release on: - push: - branches: - - master + workflow_dispatch: + inputs: + release_type: + description: 'Release type (patch, minor, major)' + required: false + default: 'auto' + type: choice + options: + - auto + - patch + - minor + - major permissions: contents: write @@ -42,30 +51,59 @@ jobs: env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} run: | - cat > .releaserc.json << 'EOF' - { - "branches": ["master"], - "plugins": [ - "@semantic-release/commit-analyzer", - "@semantic-release/release-notes-generator" - ], - "dryRun": true - } + # If a specific release type is requested (not 'auto'), always proceed + if [ "${{ github.event.inputs.release_type }}" != "auto" ] && [ -n "${{ github.event.inputs.release_type }}" ]; then + echo "Manual release type specified: ${{ github.event.inputs.release_type }}" + echo "should_release=true" >> $GITHUB_OUTPUT + + # Get current version from Cargo.toml + CURRENT_VERSION=$(grep '^version = ' Cargo.toml | head -1 | cut -d'"' -f2) + echo "Current version: $CURRENT_VERSION" + + # Calculate next version based on release type + IFS='.' read -r major minor patch <<< "$CURRENT_VERSION" + case "${{ github.event.inputs.release_type }}" in + major) + VERSION="$((major + 1)).0.0" + ;; + minor) + VERSION="${major}.$((minor + 1)).0" + ;; + patch) + VERSION="${major}.${minor}.$((patch + 1))" + ;; + esac + echo "version=$VERSION" >> $GITHUB_OUTPUT + echo "Manual version to release: $VERSION" + else + # Auto mode - use semantic-release to determine if release is needed + cat > .releaserc.json << 'EOF' + { + "branches": ["master"], + "plugins": [ + "@semantic-release/commit-analyzer", + "@semantic-release/release-notes-generator" + ], + "dryRun": true + } EOF - OUTPUT=$(npx semantic-release --dry-run 2>&1 || true) - echo "$OUTPUT" + OUTPUT=$(npx semantic-release --dry-run 2>&1 || true) + echo "$OUTPUT" - if echo "$OUTPUT" | grep -q "There is no relevant changes"; then - echo "should_release=false" >> $GITHUB_OUTPUT - else - VERSION=$(echo "$OUTPUT" | grep -oP "next release version is \K[0-9]+\.[0-9]+\.[0-9]+" | head -1) - if [ -n "$VERSION" ]; then - echo "should_release=true" >> $GITHUB_OUTPUT - echo "version=$VERSION" >> $GITHUB_OUTPUT - echo "Version to release: $VERSION" - else + if echo "$OUTPUT" | grep -q "There is no relevant changes"; then echo "should_release=false" >> $GITHUB_OUTPUT + echo "No relevant changes found for automatic release" + else + VERSION=$(echo "$OUTPUT" | grep -oP "next release version is \K[0-9]+\.[0-9]+\.[0-9]+" | head -1) + if [ -n "$VERSION" ]; then + echo "should_release=true" >> $GITHUB_OUTPUT + echo "version=$VERSION" >> $GITHUB_OUTPUT + echo "Auto version to release: $VERSION" + else + echo "should_release=false" >> $GITHUB_OUTPUT + echo "Could not determine version for automatic release" + fi fi fi @@ -191,40 +229,97 @@ jobs: - name: Create semantic-release config run: | - cat > .releaserc.json << 'EOF' - { - "branches": ["master"], - "plugins": [ - "@semantic-release/commit-analyzer", - "@semantic-release/release-notes-generator", - "@semantic-release/changelog", - [ - "@semantic-release/exec", - { - "prepareCmd": "sed -i 's/^version = \".*\"/version = \"${nextRelease.version}\"/' Cargo.toml" - } - ], - [ - "@semantic-release/git", - { - "assets": ["CHANGELOG.md", "Cargo.toml"], - "message": "chore(release): ${nextRelease.version} [skip ci]\n\n${nextRelease.notes}" - } - ], - [ - "@semantic-release/github", - { - "assets": [ - {"path": "release-assets/*.tar.gz"}, - {"path": "release-assets/*.zip"} - ] - } + # Check if we need to force a specific version for manual releases + if [ "${{ github.event.inputs.release_type }}" != "auto" ] && [ -n "${{ github.event.inputs.release_type }}" ]; then + # For manual releases, we'll bypass semantic-release version detection + cat > .releaserc.json << 'EOF' + { + "branches": ["master"], + "plugins": [ + [ + "@semantic-release/commit-analyzer", + { + "releaseRules": [ + {"breaking": true, "release": "${{ github.event.inputs.release_type }}"}, + {"revert": true, "release": "patch"}, + {"type": "feat", "release": "${{ github.event.inputs.release_type }}"}, + {"type": "fix", "release": "${{ github.event.inputs.release_type }}"}, + {"type": "perf", "release": "${{ github.event.inputs.release_type }}"}, + {"scope": "no-release", "release": false} + ] + } + ], + "@semantic-release/release-notes-generator", + "@semantic-release/changelog", + [ + "@semantic-release/exec", + { + "prepareCmd": "sed -i 's/^version = \".*\"/version = \"${{ needs.check-release.outputs.version }}\"/' Cargo.toml" + } + ], + [ + "@semantic-release/git", + { + "assets": ["CHANGELOG.md", "Cargo.toml"], + "message": "chore(release): ${{ needs.check-release.outputs.version }} [skip ci]\n\n${nextRelease.notes}" + } + ], + [ + "@semantic-release/github", + { + "assets": [ + {"path": "release-assets/*.tar.gz"}, + {"path": "release-assets/*.zip"} + ] + } + ] ] - ] - } + } EOF + else + # For automatic releases, use standard semantic-release + cat > .releaserc.json << 'EOF' + { + "branches": ["master"], + "plugins": [ + "@semantic-release/commit-analyzer", + "@semantic-release/release-notes-generator", + "@semantic-release/changelog", + [ + "@semantic-release/exec", + { + "prepareCmd": "sed -i 's/^version = \".*\"/version = \"${nextRelease.version}\"/' Cargo.toml" + } + ], + [ + "@semantic-release/git", + { + "assets": ["CHANGELOG.md", "Cargo.toml"], + "message": "chore(release): ${nextRelease.version} [skip ci]\n\n${nextRelease.notes}" + } + ], + [ + "@semantic-release/github", + { + "assets": [ + {"path": "release-assets/*.tar.gz"}, + {"path": "release-assets/*.zip"} + ] + } + ] + ] + } + EOF + fi - name: Run semantic-release env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - run: npx semantic-release + run: | + if [ "${{ github.event.inputs.release_type }}" != "auto" ] && [ -n "${{ github.event.inputs.release_type }}" ]; then + # For manual releases, we need to create the release directly + echo "Creating manual release version ${{ needs.check-release.outputs.version }}" + npx semantic-release --no-ci + else + npx semantic-release + fi diff --git a/Cargo.lock b/Cargo.lock index 53fe6dc..4bc772f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,4 +4,4 @@ version = 4 [[package]] name = "soulseek-rs" -version = "1.0.0" +version = "0.1.0" diff --git a/README.md b/README.md index c81b1d0..dcda68f 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Soulseek-rs 🦀 -# ⚠️WARNING THIS IS UNDER DEVELOPMENT AND NOT READY FOR USE ⚠️ +## ⚠️WARNING THIS IS UNDER DEVELOPMENT AND NOT READY FOR USE ⚠️ Souleek-rs is an experimental Soulseek client built in Rust. It's under development and not yet ready for use. Soulseek is a closed-source P2P diff --git a/src/actor/mod.rs b/src/actor/mod.rs index 0b619ad..b2ee98d 100644 --- a/src/actor/mod.rs +++ b/src/actor/mod.rs @@ -1,6 +1,3 @@ -// src/actor/mod.rs -// Lightweight actor system using the existing thread pool - use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -10,7 +7,14 @@ use crate::utils::thread_pool::ThreadPool; pub mod peer_actor; pub mod peer_registry; +pub mod server_actor; +#[derive(Debug, Clone)] +pub enum ConnectionState { + Disconnected, + Connecting { since: Instant }, + Connected, +} /// Core actor trait - each actor processes messages pub trait Actor: Send + 'static { type Message: Send + Clone + 'static; @@ -28,14 +32,12 @@ pub trait Actor: Send + 'static { fn tick(&mut self) {} } -/// Handle to send messages to an actor #[derive(Clone)] pub struct ActorHandle { pub(crate) sender: Sender>, } impl ActorHandle { - /// Send a message to the actor (non-blocking) pub fn send(&self, msg: M) -> Result<(), String> { self.sender .send(ActorMessage::UserMessage(msg)) diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index 36cf6e0..7d27a30 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -1,4 +1,4 @@ -use crate::actor::{Actor, ActorHandle}; +use crate::actor::{Actor, ActorHandle, ConnectionState}; use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; use crate::message::peer::{ @@ -17,13 +17,6 @@ use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; -#[derive(Debug, Clone)] -enum ConnectionState { - Disconnected, - Connecting { since: Instant }, - Connected, -} - #[derive(Debug, Clone)] pub enum PeerMessage { SendMessage(Message), @@ -552,18 +545,14 @@ impl Actor for PeerActor { fn tick(&mut self) { match self.connection_state { ConnectionState::Connecting { .. } => { - // Check if connection has been established self.check_connection_status(); } ConnectionState::Connected => { - // Normal operation - process reads if self.stream.is_some() { self.process_read(); } } - ConnectionState::Disconnected => { - // Do nothing - } + ConnectionState::Disconnected => {} } } } diff --git a/src/actor/server_actor.rs b/src/actor/server_actor.rs new file mode 100644 index 0000000..ecc89ee --- /dev/null +++ b/src/actor/server_actor.rs @@ -0,0 +1,758 @@ +use crate::actor::{Actor, ActorHandle, ConnectionState}; +use crate::client::ClientOperation; +use crate::dispatcher::MessageDispatcher; +use crate::message::server::ConnectToPeerHandler; +use crate::message::server::ExcludedSearchPhrasesHandler; +use crate::message::server::FileSearchHandler; +use crate::message::server::GetPeerAddressHandler; +use crate::message::server::LoginHandler; +use crate::message::server::MessageFactory; +use crate::message::server::MessageUser; +use crate::message::server::ParentMinSpeedHandler; +use crate::message::server::ParentSpeedRatioHandler; +use crate::message::server::PrivilegedUsersHandler; +use crate::message::server::RoomListHandler; +use crate::message::server::WishListIntervalHandler; +use crate::message::{Handlers, MessageType}; +use crate::message::{Message, MessageReader}; +use crate::peer::ConnectionType; +use crate::peer::Peer; + +use std::io::{self, Error, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::sync::mpsc::{Receiver, Sender}; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + +use crate::{debug, error, info, trace, warn, SoulseekRs}; + +#[derive(Debug, Clone)] +pub struct PeerAddress { + host: String, + port: u16, +} + +impl PeerAddress { + pub fn new(host: String, port: u16) -> Self { + Self { host, port } + } + + pub fn get_host(&self) -> &str { + &self.host + } + + pub fn get_port(&self) -> u16 { + self.port + } +} + +impl std::fmt::Display for PeerAddress { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + write!(f, "{}:{}", self.host, self.port) + } +} + +#[derive(Debug)] +pub struct Context { + pub logged_in: Option, + #[allow(dead_code)] + rooms: Rooms, +} + +impl Default for Context { + fn default() -> Self { + Self::new() + } +} + +impl Context { + pub fn new() -> Self { + Self { + #[allow(dead_code)] + rooms: Rooms::new(), + logged_in: Option::None, + } + } +} +#[derive(Debug, Clone)] +pub struct UserMessage { + id: u32, + timestamp: u32, + username: String, + message: String, + new_message: bool, +} +impl UserMessage { + pub fn new( + id: u32, + timestamp: u32, + username: String, + message: String, + new_message: bool, + ) -> Self { + Self { + id, + timestamp, + username, + message, + new_message, + } + } + pub fn print(&self) { + debug!( + "Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}", + self.timestamp, + self.username, + self.id, + self.new_message, + self.message + ); + } +} + +#[allow(dead_code)] +#[derive(Debug)] +pub struct Rooms { + pub public_rooms: Vec, + pub owned_private_rooms: Vec, + pub private_rooms: Vec, + pub operated_private_rooms: Vec, +} +impl Rooms { + fn new() -> Self { + Self { + public_rooms: Vec::new(), + owned_private_rooms: Vec::new(), + private_rooms: Vec::new(), + operated_private_rooms: Vec::new(), + } + } + + #[allow(dead_code)] + pub fn print(&self) { + info!("Public rooms ({}):", self.public_rooms.len()); + for room in &self.public_rooms { + room.print(); + } + info!("Owned private rooms ({}):", self.owned_private_rooms.len()); + for room in &self.owned_private_rooms { + room.print(); + } + info!("Private rooms ({}):", self.private_rooms.len()); + for room in &self.private_rooms { + room.print(); + } + info!( + "Operated private rooms ({}):", + self.operated_private_rooms.len() + ); + for room in &self.operated_private_rooms { + room.print(); + } + } +} +#[derive(Debug)] +pub struct Room { + name: String, + number_of_users: i32, +} +impl Room { + #[allow(dead_code)] + pub fn new(name: String, number_of_users: i32) -> Self { + Self { + name, + number_of_users, + } + } + #[allow(dead_code)] + pub fn set_number_of_users(&mut self, number_of_users: i32) { + self.number_of_users = number_of_users; + } + pub fn print(&self) { + debug!( + "Room: {}, Number of users: {}", + self.name, self.number_of_users + ); + } +} + +#[derive(Debug, Clone)] +pub enum ServerOperation { + ProcessRead, + LoginStatus(bool), + SendMessage(Message), + Login { + username: String, + password: String, + response: std::sync::mpsc::Sender>, + }, + FileSearch { + token: u32, + query: String, + }, + #[allow(dead_code)] + ConnectToPeer(Peer), + PierceFirewall(u32), + GetPeerAddress(String), + GetPeerAddressResponse { + username: String, + host: String, + port: u32, + obfuscation_type: u32, + obfuscated_port: u16, + }, +} + +pub struct ServerActor { + address: PeerAddress, + context: Arc>, + listen_port: u32, + enable_listen: bool, + stream: Option, + connection_state: ConnectionState, + reader: MessageReader, + client_channel: Sender, + self_handle: Option>, + dispatcher: Option>, + dispatcher_receiver: Option>, + dispatcher_sender: Option>, + queued_messages: Vec, +} + +impl ServerActor { + pub fn new( + address: PeerAddress, + client_channel: Sender, + listen_port: u32, + enable_listen: bool, + ) -> Self { + Self { + address, + context: Arc::new(RwLock::new(Context::new())), + listen_port, + enable_listen, + stream: None, + connection_state: ConnectionState::Disconnected, + dispatcher: None, + dispatcher_receiver: None, + dispatcher_sender: None, + reader: MessageReader::new(), + client_channel, + self_handle: None, + queued_messages: Vec::new(), + } + } + + pub fn get_address(&self) -> &PeerAddress { + &self.address + } + + pub fn get_sender(&self) -> Option<&Sender> { + self.dispatcher_sender.as_ref() + } + + fn initiate_connection(&mut self) -> bool { + let host = self.address.host.clone(); + let port = self.address.port; + + let addr_str = format!("{}:{}", host, port); + + let mut socket_addrs = match addr_str.to_socket_addrs() { + Ok(addrs) => addrs, + Err(e) => { + error!("[server_actor] Failed to resolve address: {}", e); + + self.disconnect_with_error(io::Error::new( + io::ErrorKind::InvalidInput, + e, + )); + return false; + } + }; + + let socket_addr = socket_addrs.next(); + + match socket_addr { + Some(addr) => { + if let Ok(stream) = TcpStream::connect(addr) { + if let Err(e) = stream.set_nonblocking(true) { + error!( + "[server_actor] Failed to set non-blocking: {}", + e + ); + self.disconnect_with_error(e); + return false; + } + + stream.set_nodelay(true).ok(); + + self.stream = Some(stream); + self.connection_state = ConnectionState::Connecting { + since: Instant::now(), + }; + true + } else { + match TcpStream::connect(addr) { + Ok(stream) => { + if let Err(e) = stream.set_nonblocking(true) { + error!("[server_actor] Failed to set non-blocking: {}", e); + self.disconnect_with_error(e); + return false; + } + stream.set_nodelay(true).ok(); + self.stream = Some(stream); + self.connection_state = + ConnectionState::Connecting { + since: Instant::now(), + }; + true + } + Err(e) => { + self.disconnect_with_error(e); + false + } + } + } + } + None => { + let error_msg = + format!("No socket addresses found for {}:{}", host, port); + error!("[server_actor] {}", error_msg); + self.disconnect_with_error(io::Error::new( + io::ErrorKind::InvalidInput, + error_msg, + )); + false + } + } + } + + pub fn set_self_handle(&mut self, handle: ActorHandle) { + self.self_handle = Some(handle); + } + + fn initialize_dispatcher(&mut self) { + let (dispatcher_sender, dispatcher_receiver) = + std::sync::mpsc::channel::(); + + self.dispatcher_receiver = Some(dispatcher_receiver); + self.dispatcher_sender = Some(dispatcher_sender.clone()); + + self.client_channel + .send(ClientOperation::SetServerSender(dispatcher_sender.clone())) + .unwrap(); + + let mut handlers = Handlers::new(); + + handlers.register_handler(LoginHandler); + handlers.register_handler(RoomListHandler); + handlers.register_handler(ExcludedSearchPhrasesHandler); + handlers.register_handler(PrivilegedUsersHandler); + handlers.register_handler(MessageUser); + handlers.register_handler(WishListIntervalHandler); + handlers.register_handler(ParentMinSpeedHandler); + handlers.register_handler(ParentSpeedRatioHandler); + handlers.register_handler(PrivilegedUsersHandler); + handlers.register_handler(FileSearchHandler); + handlers.register_handler(GetPeerAddressHandler); + handlers.register_handler(ConnectToPeerHandler); + + self.dispatcher = Some(MessageDispatcher::new( + "server_actor".into(), + dispatcher_sender, + handlers, + )); + } + + fn process_dispatcher_messages(&mut self) { + let messages: Vec = self + .dispatcher_receiver + .as_ref() + .map_or_else(Vec::new, |receiver| { + let mut msgs = Vec::new(); + while let Ok(msg) = receiver.try_recv() { + msgs.push(msg); + } + msgs + }); + + messages + .iter() + .for_each(|msg| self.handle_message(msg.clone())); + } + + pub fn login( + &mut self, + username: &str, + password: &str, + ) -> Result { + self.queue_message(MessageFactory::build_login_message( + username, password, + )); + let context = self.context.clone(); + let mut logged_in; + let timeout = Duration::from_secs(5); + let start = Instant::now(); + + loop { + if start.elapsed() > timeout { + warn!("Timeout waiting for login response"); + return Err(SoulseekRs::Timeout); + } + + { + logged_in = context.read().unwrap().logged_in + } + + if logged_in.is_some() { + break; + } + } + + if logged_in.unwrap() { + info!("Logged in as {}", username); + self.queue_message(MessageFactory::build_shared_folders_message( + 1, 499, + )); + self.queue_message(MessageFactory::build_no_parent_message()); + self.queue_message(MessageFactory::build_set_status_message(2)); + if self.enable_listen { + self.queue_message( + MessageFactory::build_set_wait_port_message( + self.listen_port, + ), + ); + } + } + + Ok(logged_in.unwrap()) + } + + pub fn file_search(&mut self, token: u32, query: &str) { + self.queue_message(MessageFactory::build_file_search_message( + token, query, + )); + } + + fn handle_message(&mut self, msg: ServerOperation) { + if !matches!(self.connection_state, ConnectionState::Connected) { + match &msg { + ServerOperation::ProcessRead => { + // Always process read operations + } + _ => { + // Queue all other messages when not connected + self.queued_messages.push(msg); + return; + } + } + } + + match msg { + ServerOperation::ConnectToPeer(peer) => { + if let Some(op) = match peer.connection_type { + ConnectionType::P | ConnectionType::F => { + Some(ClientOperation::ConnectToPeer(peer.clone())) + } + ConnectionType::D => None, + } { + self.client_channel.send(op).unwrap(); + } + } + ServerOperation::LoginStatus(message) => { + self.context.write().unwrap().logged_in = Some(message); + } + ServerOperation::PierceFirewall(token) => { + self.send_message( + MessageFactory::build_pierce_firewall_message(token), + ); + } + ServerOperation::SendMessage(message) => { + self.send_message(message); + } + ServerOperation::GetPeerAddress(username) => { + self.send_message(MessageFactory::build_get_peer_address( + &username, + )); + } + ServerOperation::GetPeerAddressResponse { + username, + host, + port, + obfuscation_type, + obfuscated_port, + } => { + debug!( + "[server_actor] Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})" + , username, host, port, obfuscation_type, obfuscated_port + ); + + if let Err(e) = self.client_channel.send( + ClientOperation::GetPeerAddressResponse { + username, + host, + port, + obfuscation_type, + obfuscated_port, + }, + ) { + error!("[server_actor] Error forwarding GetPeerAddress response to client: {}", e); + } + } + ServerOperation::ProcessRead => { + self.process_read(); + } + ServerOperation::Login { + username, + password, + response, + } => { + self.queue_message(MessageFactory::build_login_message( + &username, &password, + )); + + let start = std::time::Instant::now(); + 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; + } + + 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)); + }); + } + ServerOperation::FileSearch { token, query } => { + self.file_search(token, &query); + } + } + } + + fn process_read(&mut self) { + if self.reader.buffer_len() > 0 { + self.extract_and_process_messages(); + } + + { + let stream = match self.stream.as_mut() { + Some(s) => s, + None => return, + }; + + match self.reader.read_from_socket(stream) { + Ok(()) => {} + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {} + Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { + debug!("[server_actor] Read operation timed out",); + } + Err(e) => { + error!( + "[server_actor] Error reading from server: {} (kind: {:?}). Disconnecting.", + e, e.kind() + ); + self.disconnect_with_error(e); + return; + } + } + } + self.extract_and_process_messages(); + } + + fn extract_and_process_messages(&mut self) { + let mut extracted_count = 0; + loop { + match self.reader.extract_message() { + Ok(Some(mut message)) => { + extracted_count += 1; + trace!( + "[server_actor] ← Message #{}: {:?}", + extracted_count, + message + .get_message_name( + MessageType::Server, + 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!("[server_actor] No dispatcher available!",); + } + } + Err(e) => { + warn!( "[server_actor] Error extracting message: {}. Disconnecting.", e); + self.disconnect_with_error(e); + return; + } + Ok(None) => { + break; + } + } + } + + self.process_dispatcher_messages(); + } + + fn queue_message(&mut self, message: Message) { + if let Some(sender) = &self.dispatcher_sender { + match sender.send(ServerOperation::SendMessage(message)) { + Ok(_) => {} + Err(e) => error!("Failed to send: {}", e), + } + } else { + self.queued_messages + .push(ServerOperation::SendMessage(message)); + } + } + + fn send_message(&mut self, message: Message) { + let stream = match self.stream.as_mut() { + Some(s) => s, + None => { + error!("[server_actor] Cannot send message: stream is None"); + return; + } + }; + + trace!( + "[server_actor] ➡ {:?}", + message + .get_message_name( + MessageType::Server, + u32::from_le_bytes( + message.get_slice(0, 4).try_into().unwrap() + ) + ) + .map_err(|e| e.to_string()) + ); + + if let Err(e) = stream.write_all(&message.get_buffer()) { + error!( + "[server_actor] Error writing message: {}. Disconnecting.", + e + ); + self.disconnect_with_error(e); + return; + } + + if let Err(e) = stream.flush() { + error!( + "[server_actor] Error flushing stream: {}. Disconnecting.", + e + ); + self.disconnect_with_error(e); + } + } + + fn disconnect_with_error(&mut self, _error: Error) { + debug!("[server_actor] disconnect"); + + self.stream.take(); + } + + fn disconnect(&mut self) { + debug!("[server_actor] disconnected"); + + self.stream.take(); + } + + fn check_connection_status(&mut self) { + let ConnectionState::Connecting { since } = self.connection_state + else { + return; + }; + + if since.elapsed() > Duration::from_secs(20) { + error!("[server_actor] Connection timeout after 20 seconds"); + self.disconnect_with_error(io::Error::new( + io::ErrorKind::TimedOut, + "Connection timeout", + )); + return; + } + + let Some(ref stream) = self.stream else { + return; + }; + + match stream.peer_addr() { + Ok(_) => { + self.connection_state = ConnectionState::Connected; + self.on_connection_established(); + } + Err(ref e) if e.kind() == io::ErrorKind::NotConnected => {} + Err(e) => { + error!("[server_actor] Connection failed: {}", e); + self.disconnect_with_error(e); + } + } + } + + fn on_connection_established(&mut self) { + let Some(_) = self.stream else { + panic!("Stream should be available here") + }; + + self.initialize_dispatcher(); + + let queued = std::mem::take(&mut self.queued_messages); + for msg in queued { + self.handle_message(msg); + } + + if let Some(ref handle) = self.self_handle { + handle.send(ServerOperation::ProcessRead).ok(); + } + + self.process_read(); + } +} + +impl Actor for ServerActor { + type Message = ServerOperation; + + fn handle(&mut self, msg: Self::Message) { + self.handle_message(msg); + } + + fn on_start(&mut self) { + if self.stream.is_none() { + self.initiate_connection(); + } else { + self.connection_state = ConnectionState::Connected; + self.on_connection_established(); + } + } + + fn on_stop(&mut self) { + trace!("[server_actor] actor stopping"); + self.disconnect(); + } + + fn tick(&mut self) { + match self.connection_state { + ConnectionState::Connecting { .. } => { + self.check_connection_status(); + } + ConnectionState::Connected => { + if self.stream.is_some() { + self.process_read(); + } + } + ConnectionState::Disconnected => {} + } + } +} diff --git a/src/client.rs b/src/client.rs index b5aae32..aa5cbcd 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,9 +1,10 @@ +use crate::actor::server_actor::{PeerAddress, ServerActor, ServerOperation}; +use crate::actor::ActorHandle; use crate::types::{DownloadResult, DownloadStatus}; use crate::{ actor::{peer_registry::PeerRegistry, ActorSystem}, error::{Result, SoulseekRs}, - peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer}, - server::{PeerAddress, Server, ServerOperation}, + peer::{ConnectionType, DownloadPeer, NewPeer, Peer}, types::{Download, FileSearchResult}, utils::{md5, thread_pool::ThreadPool}, Transfer, @@ -43,11 +44,12 @@ pub enum ClientOperation { obfuscated_port: u16, }, UploadFailed(String, String), + SetServerSender(Sender), } pub struct ClientContext { pub peer_registry: Option, sender: Option>, - server_sender: Option>, + server_sender: Option>, search_results: Vec, pub download_tokens: HashMap, actor_system: Arc, @@ -84,7 +86,7 @@ pub struct Client { address: PeerAddress, username: String, password: String, - server: Option, + server_handle: Option>, context: Arc>, } @@ -97,15 +99,14 @@ impl Client { listen_port: Option, ) -> Self { crate::utils::logger::init(); - let context = Arc::new(RwLock::new(ClientContext::new())); Self { enable_listen, listen_port: listen_port.unwrap_or(DEFALT_LISTEN_PORT), address, username, password, - server: None, - context, + context: Arc::new(RwLock::new(ClientContext::new())), + server_handle: None, } } pub fn with_defaults( @@ -122,72 +123,50 @@ impl Client { Receiver, ) = mpsc::channel(); - let listen_port = self.listen_port; - let context = self.context.clone(); - let own_username = self.username.clone(); + let mut ctx = self.context.write().unwrap(); + ctx.sender = Some(sender.clone()); + let peer_registry = + PeerRegistry::new(ctx.actor_system.clone(), sender.clone()); + ctx.peer_registry = Some(peer_registry); - // Initialize PeerRegistry with the actor system - { - let mut ctx = self.context.write().unwrap(); - ctx.sender = Some(sender.clone()); - let peer_registry = - PeerRegistry::new(ctx.actor_system.clone(), sender.clone()); - ctx.peer_registry = Some(peer_registry); - } - - let client_sender = sender.clone(); - - self.server = match Server::new( + let server_actor = ServerActor::new( self.address.clone(), sender, self.listen_port, self.enable_listen, - ) { - Ok(server) => { - info!( - "Connected to server at {}:{}", - server.get_address().get_host(), - server.get_address().get_port() - ); + ); - if self.enable_listen { - thread::spawn(move || { - Listen::start( - listen_port, - client_sender.clone(), - context.clone(), - own_username, - ); - }); - } - let mut unlocked_context = self.context.write().unwrap(); - unlocked_context.server_sender = - Some(server.get_sender().clone()); + self.server_handle = Some(ctx.actor_system.spawn_with_handle( + server_actor, + |actor, handle| { + actor.set_self_handle(handle); + }, + )); - Self::listen_to_client_operations( - message_reader, - self.context.clone(), - self.username.clone(), - ); - Some(server) - } - Err(e) => { - error!("Error connecting to server: {}", e); - None - } - }; + Self::listen_to_client_operations( + message_reader, + self.context.clone(), + self.username.clone(), + ); } pub fn login(&self) -> Result { info!("Logging in as {}", self.username); - if let Some(server) = &self.server { - return if (server.login(&self.username, &self.password))? { - Ok(true) - } else { - Err(SoulseekRs::AuthenticationFailed) - }; + if let Some(handle) = &self.server_handle { + let (tx, rx) = std::sync::mpsc::channel(); + let _ = handle.send(ServerOperation::Login { + username: self.username.clone(), + password: self.password.clone(), + response: tx, + }); + + match rx.recv() { + Ok(result) => result, + Err(_) => Err(SoulseekRs::Timeout), + } + } else { + Err(SoulseekRs::NotConnected) } - Err(SoulseekRs::NotConnected) } #[allow(dead_code)] @@ -206,10 +185,13 @@ impl Client { timeout: Duration, ) -> Result> { info!("Searching for {}", query); - if let Some(server) = &self.server { + if let Some(handle) = &self.server_handle { let hash = md5::md5(query); let token = u32::from_str_radix(&hash[0..5], 16)?; - server.file_search(token, query); + let _ = handle.send(ServerOperation::FileSearch { + token, + query: query.to_string(), + }); } else { return Err(SoulseekRs::NotConnected); } @@ -325,7 +307,6 @@ impl Client { let client_context_clone = client_context.clone(); let own_username_clone = own_username.clone(); - // Spawn in background to avoid blocking the receiver thread thread::spawn(move || { Self::connect_to_peer( peer, @@ -536,7 +517,6 @@ impl Client { client_context.clone(); let own_username_clone = own_username.clone(); - // Spawn in background to avoid blocking the receiver thread thread::spawn(move || { Self::connect_to_peer( peer, @@ -594,6 +574,11 @@ impl Client { Some(&filename), ); } + ClientOperation::SetServerSender(sender) => { + client_context.write().unwrap().server_sender = + Some(sender); + debug!("[client] Server sender initialized"); + } } } Err(e) => { diff --git a/src/lib.rs b/src/lib.rs index 7bedec5..966c18e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,23 +5,23 @@ pub mod dispatcher; pub mod error; pub mod message; pub mod peer; -pub mod server; pub mod types; #[macro_use] pub mod utils; + // Prelude module for commonly used items pub mod prelude { + pub use crate::actor::server_actor::PeerAddress; pub use crate::types::{ DownloadResult, DownloadStatus, File, FileSearchResult, Transfer, }; pub use crate::{debug, error, info, trace, warn}; - pub use crate::{Client, PeerAddress}; } // Re-export commonly used types +pub use actor::server_actor::PeerAddress; pub use client::Client; pub use error::{Result, SoulseekRs}; -pub use server::PeerAddress; pub use types::{ DownloadResult, DownloadStatus, File, FileSearchResult, Transfer, }; diff --git a/src/main.rs b/src/main.rs index bbc4458..aef5167 100644 --- a/src/main.rs +++ b/src/main.rs @@ -46,7 +46,7 @@ fn main() { } } Err(e) => { - eprintln!("Failed to login: {e}"); + eprintln!("Failed to login: {}", e); } } } diff --git a/src/message/server/connect_to_peer.rs b/src/message/server/connect_to_peer.rs index ca7550d..995df0c 100644 --- a/src/message/server/connect_to_peer.rs +++ b/src/message/server/connect_to_peer.rs @@ -1,6 +1,6 @@ use crate::message::{Message, MessageHandler}; use crate::peer::Peer; -use crate::server::ServerOperation; +use crate::actor::server_actor::ServerOperation; use std::sync::mpsc::Sender; pub struct ConnectToPeerHandler; diff --git a/src/message/server/excluded_search_phrases.rs b/src/message/server/excluded_search_phrases.rs index d221f8c..23f03cc 100644 --- a/src/message/server/excluded_search_phrases.rs +++ b/src/message/server/excluded_search_phrases.rs @@ -1,7 +1,7 @@ use crate::debug; use crate::{ + actor::server_actor::ServerOperation, message::{Message, MessageHandler}, - server::ServerOperation, }; use std::sync::mpsc::Sender; diff --git a/src/message/server/file_search.rs b/src/message/server/file_search.rs index 016e384..d3ed46b 100644 --- a/src/message/server/file_search.rs +++ b/src/message/server/file_search.rs @@ -3,7 +3,7 @@ use std::sync::mpsc::Sender; use crate::{ message::handlers::MessageHandler, message::Message, - server::ServerOperation, + actor::server_actor::ServerOperation, }; pub struct FileSearchHandler; diff --git a/src/message/server/get_peer_address.rs b/src/message/server/get_peer_address.rs index 625c7d9..d2025e0 100644 --- a/src/message/server/get_peer_address.rs +++ b/src/message/server/get_peer_address.rs @@ -1,5 +1,5 @@ use crate::message::{Message, MessageHandler}; -use crate::server::ServerOperation; +use crate::actor::server_actor::ServerOperation; use std::sync::mpsc::Sender; pub struct GetPeerAddressHandler; diff --git a/src/message/server/login.rs b/src/message/server/login.rs index fc0396b..ef8b544 100644 --- a/src/message/server/login.rs +++ b/src/message/server/login.rs @@ -1,8 +1,8 @@ -use crate::{debug, info}; +use crate::{ + actor::server_actor::ServerOperation, debug, info, message::Message, +}; use std::sync::mpsc::Sender; -use crate::{message::Message, server::ServerOperation}; - use crate::message::MessageHandler; pub struct LoginHandler; diff --git a/src/message/server/message_user.rs b/src/message/server/message_user.rs index 7943512..189fbbc 100644 --- a/src/message/server/message_user.rs +++ b/src/message/server/message_user.rs @@ -1,8 +1,6 @@ +use crate::actor::server_actor::{ServerOperation, UserMessage}; use crate::info; -use crate::{ - message::{Message, MessageHandler}, - server::{ServerOperation, UserMessage}, -}; +use crate::message::{Message, MessageHandler}; use std::sync::mpsc::Sender; diff --git a/src/message/server/parent_min_speed.rs b/src/message/server/parent_min_speed.rs index dfae298..bca413b 100644 --- a/src/message/server/parent_min_speed.rs +++ b/src/message/server/parent_min_speed.rs @@ -3,7 +3,7 @@ use std::sync::mpsc::Sender; use crate::{ message::{Message, MessageHandler}, - server::ServerOperation, + actor::server_actor::ServerOperation, }; pub struct ParentMinSpeedHandler; diff --git a/src/message/server/parent_speed_ratio.rs b/src/message/server/parent_speed_ratio.rs index 9c6415a..1ffe506 100644 --- a/src/message/server/parent_speed_ratio.rs +++ b/src/message/server/parent_speed_ratio.rs @@ -3,7 +3,7 @@ use std::sync::mpsc::Sender; use crate::{ message::{Message, MessageHandler}, - server::ServerOperation, + actor::server_actor::ServerOperation, }; pub struct ParentSpeedRatioHandler; diff --git a/src/message/server/privileged_users.rs b/src/message/server/privileged_users.rs index a25dc31..f24759a 100644 --- a/src/message/server/privileged_users.rs +++ b/src/message/server/privileged_users.rs @@ -3,7 +3,7 @@ use std::sync::mpsc::Sender; use crate::{ message::{Message, MessageHandler}, - server::ServerOperation, + actor::server_actor::ServerOperation, }; pub struct PrivilegedUsersHandler; diff --git a/src/message/server/room_list.rs b/src/message/server/room_list.rs index 98c1315..784dffb 100644 --- a/src/message/server/room_list.rs +++ b/src/message/server/room_list.rs @@ -1,6 +1,6 @@ use crate::{ + actor::server_actor::ServerOperation, message::{Message, MessageHandler}, - server::ServerOperation, }; use std::sync::mpsc::Sender; diff --git a/src/message/server/wish_list_interval.rs b/src/message/server/wish_list_interval.rs index 3ffc956..cbea5fd 100644 --- a/src/message/server/wish_list_interval.rs +++ b/src/message/server/wish_list_interval.rs @@ -3,7 +3,7 @@ use std::sync::mpsc::Sender; use crate::{ message::{Message, MessageHandler}, - server::ServerOperation, + actor::server_actor::ServerOperation, }; pub struct WishListIntervalHandler; diff --git a/src/server.rs b/src/server.rs deleted file mode 100644 index 795773c..0000000 --- a/src/server.rs +++ /dev/null @@ -1,500 +0,0 @@ -use crate::client::ClientOperation; -use crate::dispatcher::MessageDispatcher; -use crate::error::{Result, SoulseekRs}; -use crate::message::server::ConnectToPeerHandler; -use crate::message::server::ExcludedSearchPhrasesHandler; -use crate::message::server::FileSearchHandler; -use crate::message::server::GetPeerAddressHandler; -use crate::message::server::LoginHandler; -use crate::message::server::MessageFactory; -use crate::message::server::MessageUser; -use crate::message::server::ParentMinSpeedHandler; -use crate::message::server::ParentSpeedRatioHandler; -use crate::message::server::PrivilegedUsersHandler; -use crate::message::server::RoomListHandler; -use crate::message::server::WishListIntervalHandler; -use crate::message::Message; -use crate::message::MessageReader; -use crate::message::{Handlers, MessageType}; -use crate::peer::ConnectionType; -use crate::peer::Peer; - -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, RwLock}; -use std::thread::{self}; -use std::time::{Duration, Instant}; - -use crate::{debug, error, info, trace, warn}; - -#[derive(Debug, Clone)] -pub struct PeerAddress { - host: String, - port: u16, -} - -impl PeerAddress { - pub fn new(host: String, port: u16) -> Self { - Self { host, port } - } - - pub fn get_host(&self) -> &str { - &self.host - } - - pub fn get_port(&self) -> u16 { - self.port - } -} - -#[derive(Debug)] -pub struct Context { - pub logged_in: Option, - #[allow(dead_code)] - rooms: Rooms, -} - -impl Default for Context { - fn default() -> Self { - Self::new() - } -} - -impl Context { - pub fn new() -> Self { - Self { - #[allow(dead_code)] - rooms: Rooms::new(), - logged_in: Option::None, - } - } -} -#[derive(Debug, Clone)] -pub struct UserMessage { - id: u32, - timestamp: u32, - username: String, - message: String, - new_message: bool, -} -impl UserMessage { - pub fn new( - id: u32, - timestamp: u32, - username: String, - message: String, - new_message: bool, - ) -> Self { - Self { - id, - timestamp, - username, - message, - new_message, - } - } - pub fn print(&self) { - debug!( - "Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}", - self.timestamp, - self.username, - self.id, - self.new_message, - self.message - ); - } -} - -#[allow(dead_code)] -#[derive(Debug)] -pub struct Rooms { - pub public_rooms: Vec, - pub owned_private_rooms: Vec, - pub private_rooms: Vec, - pub operated_private_rooms: Vec, -} -impl Rooms { - fn new() -> Self { - Self { - public_rooms: Vec::new(), - owned_private_rooms: Vec::new(), - private_rooms: Vec::new(), - operated_private_rooms: Vec::new(), - } - } - - #[allow(dead_code)] - pub fn print(&self) { - info!("Public rooms ({}):", self.public_rooms.len()); - for room in &self.public_rooms { - room.print(); - } - info!("Owned private rooms ({}):", self.owned_private_rooms.len()); - for room in &self.owned_private_rooms { - room.print(); - } - info!("Private rooms ({}):", self.private_rooms.len()); - for room in &self.private_rooms { - room.print(); - } - info!( - "Operated private rooms ({}):", - self.operated_private_rooms.len() - ); - for room in &self.operated_private_rooms { - room.print(); - } - } -} -#[derive(Debug)] -pub struct Room { - name: String, - number_of_users: i32, -} -impl Room { - #[allow(dead_code)] - pub fn new(name: String, number_of_users: i32) -> Self { - Self { - name, - number_of_users, - } - } - #[allow(dead_code)] - pub fn set_number_of_users(&mut self, number_of_users: i32) { - self.number_of_users = number_of_users; - } - pub fn print(&self) { - debug!( - "Room: {}, Number of users: {}", - self.name, self.number_of_users - ); - } -} - -pub enum ServerOperation { - LoginStatus(bool), - SendMessage(Message), - #[allow(dead_code)] - ConnectToPeer(Peer), - PierceFirewall(u32), - GetPeerAddress(String), - GetPeerAddressResponse { - username: String, - host: String, - port: u32, - obfuscation_type: u32, - obfuscated_port: u16, - }, -} - -#[derive(Debug)] -pub struct Server { - address: PeerAddress, - sender: Sender, - context: Arc>, - listen_port: u32, - enable_listen: bool, -} -impl Server { - pub fn new( - address: PeerAddress, - client_channel: Sender, - listen_port: u32, - enable_listen: bool, - ) -> std::result::Result { - let (sender, server_channel): ( - Sender, - Receiver, - ) = mpsc::channel(); - - let context = Arc::new(RwLock::new(Context::new())); - - let mut server = Self { - address, - context, - sender, - listen_port, - enable_listen, - }; - - server.start_read_write_loops(server_channel, client_channel)?; - Ok(server) - } - - pub fn get_address(&self) -> &PeerAddress { - &self.address - } - - pub fn get_sender(&self) -> &Sender { - &self.sender - } - - /// Start reading and writing loops in separate threads - fn start_read_write_loops( - &mut self, - server_channel: Receiver, - client_channel: Sender, - ) -> std::result::Result<(), io::Error> { - let socket_address = - format!("{}:{}", self.address.host, self.address.port) - .to_socket_addrs()? - .next() - .ok_or_else(|| { - io::Error::new( - io::ErrorKind::InvalidInput, - "Invalid address", - ) - })?; - - info!( - "Connecting to server at {}:{}", - self.address.host, self.address.port - ); - - let stream = TcpStream::connect_timeout( - &socket_address, - Duration::from_secs(10), - )?; - - stream.set_read_timeout(Some(Duration::from_secs(5)))?; - stream.set_write_timeout(Some(Duration::from_secs(5)))?; - stream.set_nodelay(true)?; - - let mut read_stream = stream.try_clone()?; - let mut write_stream = stream.try_clone()?; - - let barrier = Arc::new(Barrier::new(3)); - let read_barrier = barrier.clone(); - let write_barrier = barrier.clone(); - let done_barrier = barrier.clone(); - let sender = self.sender.clone(); - - thread::spawn(move || { - read_barrier.wait(); - - let mut handlers = Handlers::new(); - - handlers.register_handler(LoginHandler); - handlers.register_handler(RoomListHandler); - handlers.register_handler(ExcludedSearchPhrasesHandler); - handlers.register_handler(PrivilegedUsersHandler); - handlers.register_handler(MessageUser); - handlers.register_handler(WishListIntervalHandler); - handlers.register_handler(ParentMinSpeedHandler); - handlers.register_handler(ParentSpeedRatioHandler); - handlers.register_handler(PrivilegedUsersHandler); - handlers.register_handler(FileSearchHandler); - handlers.register_handler(GetPeerAddressHandler); - handlers.register_handler(ConnectToPeerHandler); - - let dispatcher = - MessageDispatcher::new("server".to_string(), sender, handlers); - - let mut buffered_reader = MessageReader::new(); - loop { - match buffered_reader.read_from_socket(&mut read_stream) { - Ok(_) => {} - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { - continue - } - Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { - debug!("Read operation timed out"); - continue; - } - Err(e) => { - error!("Error reading from server: {}", e); - break; - } - } - - loop { - match buffered_reader.extract_message() { - Ok(Some(mut message)) => { - trace!( - "[server] ← {:?} - {}", - message - .get_message_name( - MessageType::Server, - message.get_message_code_u32() - ) - .map_err(|e| e.to_string()), - message.get_message_code_u32() - ); - dispatcher.dispatch(&mut message) - } - Err(e) => { - warn!("Error extracting message: {}", e); - break; - } - Ok(None) => break, - } - } - } - }); - - let context = self.context.clone(); - thread::spawn(move || { - write_barrier.wait(); - loop { - if let Ok(operation) = server_channel.recv() { - match operation { - ServerOperation::ConnectToPeer(peer) => { - if let Some(op) = match peer.connection_type { - ConnectionType::P | ConnectionType::F => { - Some(ClientOperation::ConnectToPeer( - peer.clone(), - )) - } - ConnectionType::D => None, - } { - client_channel.send(op).unwrap(); - } - } - ServerOperation::LoginStatus(message) => { - context.write().unwrap().logged_in = Some(message); - } - ServerOperation::PierceFirewall(token) => { - if let Err(e) = write_stream.write_all( - &MessageFactory::build_pierce_firewall_message( - token, - ) - .get_buffer(), - ) { - error!( - "Error writing PierceFirewall message: {}", - e - ); - break; - } - } - ServerOperation::SendMessage(message) => { - match write_stream.write_all(&message.get_buffer()) - { - Ok(_) => trace!( - "[server] → {:?} - {:}", - message - .get_message_name( - MessageType::Server, - message.get_message_code_send() - as u32 - ) - .map_err(|e| e.to_string()), - message.get_message_code_send() as u32 - ), - Err(e) => error!( - "Error writing message to server: {}", - e - ), - } - } - ServerOperation::GetPeerAddress(username) => { - if let Err(e) = write_stream.write_all( - &MessageFactory::build_get_peer_address( - &username, - ) - .get_buffer(), - ) { - error!("Error writing get_peeer_address message: {}", e); - break; - } - } - ServerOperation::GetPeerAddressResponse { - username, - host, - port, - obfuscation_type, - obfuscated_port, - } => { - debug!( - "Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})", - username, host, port, obfuscation_type, obfuscated_port - ); - - // Forward the response to the client channel - if let Err(e) = client_channel.send( - ClientOperation::GetPeerAddressResponse { - username, - host, - port, - obfuscation_type, - obfuscated_port, - }, - ) { - error!("Error forwarding GetPeerAddress response to client: {}", e); - } - } - } - } - } - }); - done_barrier.wait(); - Ok(()) - } - - fn queue_message(&self, message: Message) { - match self.sender.send(ServerOperation::SendMessage(message)) { - Ok(_) => {} - Err(e) => error!("Failed to send: {}", e), - } - } - pub fn login(&self, username: &str, password: &str) -> Result { - // Send the login message - self.queue_message(MessageFactory::build_login_message( - username, password, - )); - let context = self.context.clone(); - let mut logged_in; - let timeout = Duration::from_secs(5); - let start = Instant::now(); - - // wait till server says your logged in or not - loop { - if start.elapsed() > timeout { - warn!("Timeout waiting for login response"); - return Err(SoulseekRs::Timeout); - } - - { - logged_in = context.read().unwrap().logged_in - } - - if logged_in.is_some() { - break; - } - } - - if logged_in.unwrap() { - info!("Logged in as {}", username); - self.queue_message(MessageFactory::build_shared_folders_message( - 1, 499, - )); - self.queue_message(MessageFactory::build_no_parent_message()); - self.queue_message(MessageFactory::build_set_status_message(2)); - if self.enable_listen { - self.queue_message( - MessageFactory::build_set_wait_port_message( - self.listen_port, - ), - ); - } - } - - Ok(logged_in.unwrap()) - } - - pub fn file_search(&self, token: u32, query: &str) { - self.queue_message(MessageFactory::build_file_search_message( - token, query, - )); - } - - #[allow(dead_code)] - pub fn pierce_firewall(&self, token: u32) { - match self.sender.send(ServerOperation::PierceFirewall(token)) { - Ok(_) => {} - Err(e) => error!("Failed to send PierceFirewall: {}", e), - } - } -}