diff --git a/src/client.rs b/src/client.rs index fcce2e3..8f53f33 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,6 +1,7 @@ use crate::{ peer::{DefaultPeer, Peer}, server::{PeerAddress, Server}, + types::FileSearch, utils::{md5, thread_pool::ThreadPool}, }; use std::{ @@ -19,10 +20,12 @@ use std::{ const MAX_THREADS: usize = 100; pub enum ClientOperation { ConnectToPeer(Peer), + SearchResult(FileSearch), } struct ClientContext { peers: HashMap, sender: Option>, + search_results: Vec, thread_pool: ThreadPool, } impl ClientContext { @@ -30,6 +33,7 @@ impl ClientContext { Self { peers: HashMap::new(), sender: None, + search_results: Vec::new(), thread_pool: ThreadPool::new(MAX_THREADS), } } @@ -99,7 +103,7 @@ impl Client { } } - pub fn search(&self, query: &str, timeout: Duration) { + pub fn search(&self, query: &str, timeout: Duration) -> Vec { println!("Searching for {}", query); if let Some(server) = &self.server { let hash = md5::md5(query); @@ -115,9 +119,7 @@ impl Client { break; } } - if let Some(server) = &self.server { - println!("server {:?}", server) - } + return self.context.lock().unwrap().search_results.clone(); } fn listen_to_client_operations( @@ -130,6 +132,13 @@ impl Client { ClientOperation::ConnectToPeer(peer) => { Self::connect_to_peer(peer, client_context.clone()) } + ClientOperation::SearchResult(file_search) => { + client_context + .lock() + .unwrap() + .search_results + .push(file_search); + } } } }); diff --git a/src/main.rs b/src/main.rs index bf6b7a8..0b3ff89 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,7 @@ mod dispatcher; mod message; mod peer; mod server; +mod types; mod utils; use client::Client; @@ -23,7 +24,8 @@ fn main() { client.connect(); match client.login() { Ok(_) => { - client.search("Fantazia", Duration::from_secs(20)); + let results = client.search("Fantazia", Duration::from_secs(10)); + println!("Search results: {:?}", results); } Err(e) => { println!("Failed to login: {}", e); diff --git a/src/message/peer/file_search_response.rs b/src/message/peer/file_search_response.rs index 819dce8..05fb0fc 100644 --- a/src/message/peer/file_search_response.rs +++ b/src/message/peer/file_search_response.rs @@ -1,25 +1,10 @@ use crate::message::{Message, MessageHandler}; use crate::peer::PeerOperation; +use crate::types::{File, FileSearch}; use crate::utils::zlib::deflate; use std::collections::HashMap; use std::sync::mpsc::Sender; -#[derive(Debug)] -#[allow(dead_code)] -pub struct File { - pub username: String, - pub name: String, - pub size: i32, - pub attribs: HashMap, -} -#[derive(Debug)] -#[allow(dead_code)] -pub struct FileSearch { - pub token: String, - pub files: Vec, - pub slots: i8, - pub speed: i32, -} fn decompress_buffer(buffer: &[u8]) -> Result, String> { deflate(buffer) @@ -72,7 +57,6 @@ impl MessageHandler for FileSearchResponse { 9 } fn handle(&self, message: &mut Message, sender: Sender) { - // println!("{:?}", message); let file_search = FileSearch::new_from_message(message); sender diff --git a/src/peer/default_peer.rs b/src/peer/default_peer.rs index 7038889..2b2e9b0 100644 --- a/src/peer/default_peer.rs +++ b/src/peer/default_peer.rs @@ -1,14 +1,15 @@ use crate::dispatcher::MessageDispatcher; -use crate::message::peer::{FileSearch, FileSearchResponse}; +use crate::message::peer::FileSearchResponse; use crate::message::server::MessageFactory; use crate::message::{Handlers, Message, MessageReader}; +use crate::types::FileSearch; use std::sync::mpsc; use std::sync::mpsc::{Receiver, Sender}; use std::thread::{self, JoinHandle}; -use crate::peer::Peer; use crate::client::ClientOperation; +use crate::peer::Peer; use std::io::{self, Write}; use std::net::TcpStream; use std::net::ToSocketAddrs; @@ -18,6 +19,7 @@ use std::time::Duration; pub struct DefaultPeer { peer: Peer, peer_channel: Option>, + client_channel: Sender, // Store thread handles for proper lifecycle management read_thread: Option>, write_thread: Option>, @@ -29,10 +31,11 @@ pub enum PeerOperation { FileSearchResult(FileSearch), } impl DefaultPeer { - pub fn new(peer: Peer, _client_channel: Sender) -> Self { + pub fn new(peer: Peer, client_channel: Sender) -> Self { Self { peer, peer_channel: None, + client_channel, // Initialize handles as None read_thread: None, write_thread: None, @@ -68,7 +71,7 @@ impl DefaultPeer { let peer = self.peer.clone(); // Spawn the reader thread - let read_handle = thread::spawn(move || { + self.read_thread = Some(thread::spawn(move || { let mut handlers = Handlers::new(); handlers.register_handler(FileSearchResponse); @@ -95,29 +98,36 @@ impl DefaultPeer { match buffered_reader.extract_message() { Ok(Some(mut message)) => dispatcher.dispatch(&mut message), Err(e) => { - println!("Error extracting message in default peer: {}. Terminating read loop.", e); + println!( + "Error extracting message in default peer: {}. Terminating read loop.", + e + ); break; } Ok(None) => continue, } } - }); + })); - // Spawn the writer thread - let write_handle = thread::spawn(move || { + let client_channel = self.client_channel.clone(); + self.write_thread = Some(thread::spawn(move || { loop { match peer_reader.recv() { - Ok(operation) => match operation { - PeerOperation::SendMessage(message) => { - if let Err(e) = write_stream.write_all(&message.get_buffer()) { - eprintln!("Error writing message to stream: {}. Terminating write loop.", e); - break; + Ok(operation) => { + match operation { + PeerOperation::SendMessage(message) => { + if let Err(e) = write_stream.write_all(&message.get_buffer()) { + eprintln!("Error writing message to stream: {}. Terminating write loop.", e); + break; + } + } + PeerOperation::FileSearchResult(file_search) => { + client_channel + .send(ClientOperation::SearchResult(file_search)) + .unwrap(); } } - PeerOperation::FileSearchResult(file_search) => { - println!("{:?}", file_search) - } - }, + } Err(_) => { // The sender has been dropped, the peer is shutting down. println!("Peer channel closed. Terminating write loop."); @@ -125,8 +135,36 @@ impl DefaultPeer { } } } - }); - + })); + Ok(()) } } + +impl Drop for DefaultPeer { + fn drop(&mut self) { + // Signal threads to shut down (e.g., by dropping the sender half of the channel) + // For now, we'll just drop the peer_channel, which will cause the writer thread + // to exit its loop. A more robust solution might involve a shared AtomicBool + // or a dedicated shutdown channel. + self.peer_channel = None; + + // Join the read thread + if let Some(handle) = self.read_thread.take() { + println!("Joining read thread..."); + match handle.join() { + Ok(_) => println!("Read thread joined successfully."), + Err(e) => eprintln!("Read thread panicked: {:?}", e), + } + } + + // Join the write thread + if let Some(handle) = self.write_thread.take() { + println!("Joining write thread..."); + match handle.join() { + Ok(_) => println!("Write thread joined successfully."), + Err(e) => eprintln!("Write thread panicked: {:?}", e), + } + } + } +} diff --git a/src/types.rs b/src/types.rs new file mode 100644 index 0000000..590ae74 --- /dev/null +++ b/src/types.rs @@ -0,0 +1,20 @@ +use std::collections::HashMap; + +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub struct File { + pub username: String, + pub name: String, + pub size: i32, + pub attribs: HashMap, +} + +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub struct FileSearch { + pub token: String, + pub files: Vec, + pub slots: i8, + pub speed: i32, +} +