feat: implement default peer message processing

src/client.rs

@@ -1,5 +1,5 @@
 use crate::{
-    peer::{default_peer::DefaultPeer, peer::Peer},
+    peer::{DefaultPeer, Peer},
     server::{PeerAddress, Server},
     utils::md5,
 };

src/message/handlers.rs

@@ -9,48 +9,46 @@ use super::server::{
     WishListIntervalHandler,
 };
 
-pub trait MessageHandler {
+pub trait MessageHandler<Op> {
     fn get_code(&self) -> u8;
-    fn handle(&self, message: &mut Message, sender: Sender<ServerOperation>);
+    fn handle(&self, message: &mut Message, sender: Sender<Op>);
 }
-pub struct Handlers {
+pub struct Handlers<Op> {
-    handlers: HashMap<u8, Box<dyn MessageHandler>>,
+    handlers: HashMap<u8, Box<dyn MessageHandler<Op>>>,
 }
 
-impl Handlers {
+impl<Op> Handlers<Op> {
     pub fn new() -> Self {
         Self {
             handlers: HashMap::new(),
         }
     }
+
     pub fn new_with_server_handlers() -> Self {
         let mut handlers = Self::new();
-        handlers.add_server_handlers();
+        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(FileSearch);
+        handlers.register_handler(ConnectToPeerHandler);
         handlers
     }
 
-    fn add_server_handlers(&mut self) {
+    /// Register a handler for this particular `Op`.
-        self.register_handler(LoginHandler);
+    pub fn register_handler<H>(&mut self, handler: H) -> &mut Self
-        self.register_handler(RoomListHandler);
+    where
-        self.register_handler(ExcludedSearchPhrasesHandler);
+        H: 'static + MessageHandler<Op> + Send + Sync,
-        self.register_handler(PrivilegedUsersHandler);
+    {
-        self.register_handler(MessageUser);
-        self.register_handler(WishListIntervalHandler);
-        self.register_handler(ParentMinSpeedHandler);
-        self.register_handler(ParentSpeedRatioHandler);
-        self.register_handler(PrivilegedUsersHandler);
-        self.register_handler(FileSearch);
-        self.register_handler(ConnectToPeerHandler);
-    }
-
-    pub fn register_handler<H: 'static + MessageHandler + Send + Sync>(
-        &mut self,
-        handler: H,
-    ) -> &mut Self {
         self.handlers.insert(handler.get_code(), Box::new(handler));
         self
     }
-    pub fn get_handler(&self, code: u8) -> Option<&Box<dyn MessageHandler>> {
+    pub fn get_handler(&self, code: u8) -> Option<&Box<dyn MessageHandler<Op>>> {
         self.handlers.get(&code)
     }
 }

src/message/server/connect_to_peer.rs

@@ -1,4 +1,5 @@
 use crate::message::{Message, MessageHandler};
+use crate::peer::Peer;
 use crate::server::ServerOperation;
 use std::sync::mpsc::Sender;
 pub struct ConnectToPeerHandler;
@@ -6,35 +7,8 @@ impl MessageHandler for ConnectToPeerHandler {
     fn get_code(&self) -> u8 {
         18
     }
-    fn handle(&self, _message: &mut Message, _sender: Sender<ServerOperation>) {
+    fn handle(&self, message: &mut Message, sender: Sender<ServerOperation>) {
-        println!("Handling ConnectToPeer message");
+        let peer = Peer::new_from_message(message);
-        // let username = message.read_string();
+        sender.send(ServerOperation::ConnectToPeer(peer)).unwrap();
-        // let connection_type = message.read_string();
-        // let mut ip: Vec<i8> = vec![];
-        // for _ in 0..4 {
-        //     ip.push(message.read_int8());
-        // }
-        // let host: String = format!("{}.{}.{}.{}", ip[3], ip[2], ip[1], ip[0]);
-        //
-        // let (port, token, privileged, unknown, obfuscated_port) = (
-        //     message.read_int32(),
-        //     message.read_int32(),
-        //     message.read_int8(),
-        //     message.read_int8(),
-        //     message.read_int8(),
-        // );
-        //
-        // let peer = Peer::new(
-        //     username,
-        //     connection_type,
-        //     host,
-        //     port,
-        //     token,
-        //     privileged,
-        //     unknown,
-        //     obfuscated_port,
-        // );
-        //
-        // sender.send(ServerOperation::ConnectToPeer(peer)).unwrap();
     }
 }

src/message/server/login.rs

@@ -1,13 +1,14 @@
 use std::sync::mpsc::Sender;
 
-use crate::{
+use crate::{message::Message, server::ServerOperation};
-    message::{Message, MessageHandler},
+
-    server::ServerOperation,
+// IMPORTANT: note the generic trait is now `MessageHandler<ServerOperation>`
-};
+use crate::message::MessageHandler;
 
 pub struct LoginHandler;
 
-impl MessageHandler for LoginHandler {
+// Implement `MessageHandler` specifically for `ServerOperation`.
+impl MessageHandler<ServerOperation> for LoginHandler {
     fn get_code(&self) -> u8 {
         1
     }
@@ -21,7 +22,7 @@ impl MessageHandler for LoginHandler {
 
         println!("Login successful");
         let greeting = message.read_string();
-        println!("Server geeting: {:?}", greeting);
+        println!("Server greeting: {:?}", greeting);
 
         sender.send(ServerOperation::LoginStatus(true)).unwrap();
     }

src/peer/default_peer.rs

@@ -31,7 +31,6 @@ impl DefaultPeer {
         }
     }
     pub fn connect(mut self) -> Result<Self, io::Error> {
-        // println!("Connecting to peer {:?}", self.peer);
         let socket_address = format!("{}:{}", self.peer.host, self.peer.port)
             .to_socket_addrs()?
             .next()
@@ -45,7 +44,6 @@ impl DefaultPeer {
             stream
                 .write_all(&MessageFactory::build_watch_user(&token).get_data())
                 .unwrap();
-            // self.queue_message(build_watch_user(&token));
         }
         self.start_read_write_loops(stream).unwrap();
 
@@ -102,14 +100,6 @@ impl DefaultPeer {
             }
         });
 
-        // thread::spawn(move || loop {
-        //     thread::sleep(Duration::from_secs(1));
-        //     // println!(
-        //     //     "Default peer ({}) - stream state: connected={}",
-        //     //     px.host,
-        //     //     stream.peer_addr().is_ok()
-        //     // );
-        // });
         thread::spawn(move || {
             write_barrier.wait();
             loop {
@@ -131,12 +121,4 @@ impl DefaultPeer {
         done_barrier.wait();
         Ok(())
     }
-
-    // pub fn queue_message(&self, message: Message) {
-    //     if let Some(sender) = self.peer_channel.clone() {
-    //         if let Err(e) = sender.send(PeerOperation::SendMessage(message)) {
-    //             eprintln!("Failed to send: {}", e);
-    //         }
-    //     }
-    // }
 }

src/peer/mod.rs

@@ -1,3 +1,4 @@
 pub mod default_peer;
 pub mod listen;
 pub mod peer;
+pub use self::{default_peer::*, listen::*, peer::*};