chore: make message handlers and dispatcher generic

This commit is contained in:
michel
2025-01-06 22:49:19 +01:00
committed by Michel de Graaf
parent 766554134d
commit 066ddadaed
26 changed files with 787 additions and 121 deletions
+15 -8
View File
@@ -1,5 +1,7 @@
use crate::dispatcher::MessageDispatcher;
use crate::message::peer::{FileSearch, FileSearchResponse};
use crate::message::server::MessageFactory;
use crate::message::{Message, MessageReader};
use crate::message::{Handlers, Message, MessageReader};
use std::sync::mpsc;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, Barrier};
@@ -21,6 +23,7 @@ pub struct DefaultPeer {
#[allow(dead_code)]
pub enum PeerOperation {
SendMessage(Message),
FileSearchResult(FileSearch),
}
impl DefaultPeer {
pub fn new(peer: Peer, _client_channel: Sender<ClientOperation>) -> Self {
@@ -53,8 +56,8 @@ impl DefaultPeer {
fn start_read_write_loops(&mut self, stream: TcpStream) -> Result<(), io::Error> {
let (peer_sender, peer_reader): (Sender<PeerOperation>, Receiver<PeerOperation>) =
mpsc::channel();
self.peer_channel = Some(peer_sender);
let sender = peer_sender.clone();
let barrier = Arc::new(Barrier::new(3));
let read_barrier = barrier.clone();
let write_barrier = barrier.clone();
@@ -66,6 +69,12 @@ impl DefaultPeer {
let peer = self.peer.clone();
thread::spawn(move || {
read_barrier.wait();
let mut handlers = Handlers::new();
handlers.register_handler(FileSearchResponse);
let dispatcher = MessageDispatcher::new(sender, handlers);
let mut buffered_reader = MessageReader::new();
loop {
match buffered_reader.read_from_socket(&mut read_stream) {
@@ -86,12 +95,7 @@ impl DefaultPeer {
}
match buffered_reader.extract_message() {
Ok(Some(message)) => {
println!(
"Received message in default peer: {:?}",
message.get_message_code_u32()
);
}
Ok(Some(mut message)) => dispatcher.dispatch(&mut message),
Err(e) => {
println!("Error extracting message in default peer: {}", e)
}
@@ -114,6 +118,9 @@ impl DefaultPeer {
}
}
}
PeerOperation::FileSearchResult(file_search) => {
println!("{:?}", file_search)
}
}
}
}
+24 -3
View File
@@ -1,11 +1,13 @@
use std::sync::mpsc::Sender;
use std::{io, net::TcpListener};
use crate::message::MessageReader;
use crate::server::ServerOperation;
pub struct Listen {}
impl Listen {
pub fn new(port: u32) {
pub fn new(port: u32, server_channel: Sender<ServerOperation>) {
println!("starting listener on port {port}");
let listener = TcpListener::bind(format!("0.0.0.0:{port}")).unwrap();
for stream in listener.incoming() {
@@ -28,9 +30,28 @@ impl Listen {
}
match buffered_reader.extract_message() {
Ok(Some(message)) => {
println!("Received message: {:?}", message.get_message_code_u32());
Ok(Some(mut message)) => {
println!("Received message: {:?}", message.get_message_code());
println!("{:?}", message.get_data());
message.print_hex();
if message.get_message_code() == 1 {
let size = message.read_int32();
message.set_pointer(8);
let typex = message.read_string();
// let user = message.read_string();
// // let connection_type: ConnectionType =
// // message.read_string().parse().unwrap();
// // let token = message.read_int32();
//
println!("type: {:?}", typex);
println!("user: {:?}", size);
// server_channel
// .send(ServerOperation::ConnectToPeer(peer))
// .unwrap();
}
}
Err(e) => {
println!("Error extracting message: {}", e)
+1 -1
View File
@@ -1,4 +1,4 @@
pub mod default_peer;
pub mod listen;
pub mod peer;
pub use self::{default_peer::*, listen::*, peer::*};
pub use self::{default_peer::*, peer::*};