fix: packet processor structure, now is a trait

Author: Bordi00

Cargo.lock

@@ -4,6 +4,8 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-tokio = { version = "1.47.1", features = ["full"] }
 wg_internal = { git = "https://github.com/WGL-2024/WGL_repo_2024.git", features = ["debug"] }
 crossbeam-channel = "0.5.15"
+serde = { version = "1.0.219", features = ["derive"] }
+anyhow = "1.0.99"
+tokio = { version = "1.47.1", features = ["full"] }

Cargo.toml

@@ -5,22 +5,22 @@ use wg_internal::{network::NodeId, packet::Fragment};
 
 
 #[derive(Debug, Default)]
-pub struct FragmentAssembler<'a> {
-    pub fragments: HashMap<(u64, NodeId), Vec<&'a Fragment>>, // session_id -> data buffer
+pub struct FragmentAssembler {
+    pub fragments: HashMap<(u64, NodeId), Vec<Fragment>>, // session_id -> data buffer
     pub expected_fragments: HashMap<(u64, NodeId), u64>, // session_id -> total_fragments
     pub received_fragments: HashMap<(u64, NodeId), Vec<bool>>, // session_id -> received status
 }
 
-impl<'a> FragmentAssembler<'a> {
-    pub fn add_fragment(&mut self, fragment: &'a Fragment, session_id: u64, sender: NodeId) -> Option<Vec<u8>> {
+impl FragmentAssembler {
+    pub fn add_fragment(&mut self, fragment: Fragment, session_id: u64, sender: NodeId) -> Option<Vec<u8>> {
         let communication_id = ( session_id, sender );
         #[allow(clippy::cast_possible_truncation)]
         let index = fragment.fragment_index as usize;
 
         if let Vacant(entry) = self.fragments.entry(communication_id) {
-            entry.insert(vec![fragment]);
             self.expected_fragments.insert(communication_id, fragment.total_n_fragments);
             self.received_fragments.insert(communication_id, vec![false; index]);
+            entry.insert(vec![fragment]);
         }
 
         {

src/assembler.rs

@@ -2,6 +2,7 @@ pub mod network;
 pub mod types;
 pub mod assembler;
 pub mod routing_handler;
+pub mod packet_processor;
 
 pub use routing_handler::RoutingHandler;
 pub use assembler::FragmentAssembler;

src/lib.rs

@@ -0,0 +1,67 @@
+use crate::{network::NetworkError, FragmentAssembler, RoutingHandler};
+use std::any::Any;
+
+use crossbeam_channel::{select_biased, Receiver};
+use wg_internal::packet::{Packet, PacketType};
+
+pub trait Processor {
+    fn controller_recv(&self) -> &Receiver<Box<dyn Any>>;
+    fn packet_recv(&self) -> &Receiver<Packet>;
+    fn assembler(&mut self) -> &mut FragmentAssembler;
+    fn routing_header(&mut self) -> &mut RoutingHandler;
+
+    fn handle_msg(&mut self, msg: Vec<u8>);
+    fn handle_command(&mut self, cmd: Box<dyn Any>);
+
+    /// Handles a packet in a standard way
+    /// # Errors
+    /// returns an Errors if handling fails
+    fn handle_packet(&mut self, pkt: Packet) -> Result<(), NetworkError> {
+        let router = self.routing_header();
+        match pkt.pack_type {
+            PacketType::MsgFragment(fragment) => {
+                if let Some(msg) = self.assembler().add_fragment(
+                    fragment,
+                    pkt.session_id,
+                    pkt.routing_header.hops[0],
+                ) {
+                    self.handle_msg(msg);
+                }
+            }
+            PacketType::Ack(ack) => {
+                router.handle_ack(&ack, pkt.session_id, pkt.routing_header.hops[0]);
+            }
+            PacketType::Nack(nack) => {
+                router.handle_nack(&nack, pkt.session_id, pkt.routing_header.hops[0])?;
+            }
+            PacketType::FloodRequest(flood_request) => {
+                router.handle_flood_request(flood_request, pkt.session_id)?;
+            }
+            PacketType::FloodResponse(flood_response) => {
+                router.handle_flood_response(&flood_response);
+            }
+        }
+        Ok(())
+    }
+
+    fn run(&mut self) {
+        loop {
+            select_biased! {
+                recv(self.controller_recv()) -> cmd => {
+                    if let Ok(cmd) = cmd {
+                        self.handle_command(cmd);
+                    }
+                }
+
+                recv(self.packet_recv()) -> pkt => {
+                    if let Ok(pkt) = pkt {
+                        match self.handle_packet(pkt) {
+                            Ok(()) => {},
+                            Err(_) => return
+                        }
+                    }
+                }
+            }
+        }
+    }
+}