refactor(l1): abstract request logic into PeerHandler

crates/networking/p2p/kademlia.rs

@@ -1,12 +1,15 @@
+use std::sync::Arc;
+
 use crate::{
     discv4::messages::FindNodeRequest,
-    peer_channels::PeerChannels,
     rlpx::p2p::Capability,
     types::{Node, NodeRecord},
+    RLPxMessage,
 };
 use ethrex_core::{H256, H512, U256};
 use sha3::{Digest, Keccak256};
 use tokio::sync::mpsc::UnboundedSender;
+use tokio::sync::{mpsc, Mutex};
 use tracing::{debug, info};
 
 pub const MAX_NODES_PER_BUCKET: usize = 16;
@@ -395,6 +398,34 @@ impl PeerData {
     }
 }
 
+pub const MAX_MESSAGES_IN_PEER_CHANNEL: usize = 25;
+
+#[derive(Debug, Clone)]
+/// Holds the respective sender and receiver ends of the communication channels bewteen the peer data and its active connection
+pub struct PeerChannels {
+    pub(crate) sender: mpsc::Sender<RLPxMessage>,
+    pub(crate) receiver: Arc<Mutex<mpsc::Receiver<RLPxMessage>>>,
+}
+
+impl PeerChannels {
+    /// Sets up the communication channels for the peer
+    /// Returns the channel endpoints to send to the active connection's listen loop
+    pub(crate) fn create() -> (Self, mpsc::Sender<RLPxMessage>, mpsc::Receiver<RLPxMessage>) {
+        let (sender, connection_receiver) =
+            mpsc::channel::<RLPxMessage>(MAX_MESSAGES_IN_PEER_CHANNEL);
+        let (connection_sender, receiver) =
+            mpsc::channel::<RLPxMessage>(MAX_MESSAGES_IN_PEER_CHANNEL);
+        (
+            Self {
+                sender,
+                receiver: Arc::new(Mutex::new(receiver)),
+            },
+            connection_sender,
+            connection_receiver,
+        )
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/networking/p2p/net.rs

@@ -26,7 +26,7 @@ use types::Node;
 pub mod bootnode;
 pub(crate) mod discv4;
 pub(crate) mod kademlia;
-pub mod peer_channels;
+pub mod peer_handler;
 pub mod rlpx;
 pub(crate) mod snap;
 pub mod sync;