refactor: move component communication to infra crate

crates/mempool_infra/src/component_definitions.rs

@@ -1,13 +1,32 @@
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
-use tokio::sync::mpsc::Sender;
+use tokio::sync::mpsc::{Receiver, Sender};
 
 #[async_trait]
 pub trait ComponentRequestHandler<Request, Response> {
     async fn handle_request(&mut self, request: Request) -> Response;
 }
 
+pub struct ComponentCommunication<T: Send + Sync> {
+    tx: Option<Sender<T>>,
+    rx: Option<Receiver<T>>,
+}
+
+impl<T: Send + Sync> ComponentCommunication<T> {
+    pub fn new(tx: Option<Sender<T>>, rx: Option<Receiver<T>>) -> Self {
+        Self { tx, rx }
+    }
+
+    pub fn take_tx(&self) -> Sender<T> {
+        self.tx.to_owned().expect("Sender should be available, could be taken only once")
+    }
+
+    pub fn take_rx(&mut self) -> Receiver<T> {
+        self.rx.take().expect("Receiver should be available, could be taken only once")
+    }
+}
+
 pub struct ComponentRequestAndResponseSender<Request, Response>
 where
     Request: Send + Sync,

crates/mempool_node/Cargo.toml

@@ -13,15 +13,15 @@ anyhow.workspace = true
 clap.workspace = true
 const_format.workspace = true
 futures.workspace = true
+papyrus_config.workspace = true
+serde.workspace = true
 starknet_gateway = { path = "../gateway", version = "0.0" }
-starknet_mempool_infra = { path = "../mempool_infra", version = "0.0" }
 starknet_mempool = { path = "../mempool", version = "0.0" }
+starknet_mempool_infra = { path = "../mempool_infra", version = "0.0" }
 starknet_mempool_types = { path = "../mempool_types", version = "0.0" }
-serde.workspace = true
-papyrus_config.workspace = true
 tokio.workspace = true
-tracing.workspace = true
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
+tracing.workspace = true
 validator.workspace = true
 
 [dev-dependencies]
@@ -30,4 +30,4 @@ assert_matches.workspace = true
 colored.workspace = true
 pretty_assertions.workspace = true
 serde_json.workspace = true
-test_utils = {path = "../test_utils"}
+test_utils = { path = "../test_utils" }

crates/mempool_node/src/communication.rs

@@ -1,26 +1,13 @@
 use std::sync::Arc;
 
+use starknet_mempool_infra::component_definitions::ComponentCommunication;
 use starknet_mempool_types::communication::{
     MempoolClientImpl, MempoolRequestAndResponseSender, SharedMempoolClient,
 };
 use tokio::sync::mpsc::{channel, Receiver, Sender};
 
 use crate::config::MempoolNodeConfig;
 
-pub struct ComponentCommunication<T: Send + Sync> {
-    tx: Option<Sender<T>>,
-    rx: Option<Receiver<T>>,
-}
-
-impl<T: Send + Sync> ComponentCommunication<T> {
-    fn take_tx(&self) -> Sender<T> {
-        self.tx.to_owned().expect("Sender should be available, could be taken only once")
-    }
-    fn take_rx(&mut self) -> Receiver<T> {
-        self.rx.take().expect("Receiver should be available, could be taken only once")
-    }
-}
-
 pub struct MempoolNodeCommunication {
     mempool_channel: ComponentCommunication<MempoolRequestAndResponseSender>,
 }
@@ -39,7 +26,7 @@ pub fn create_node_channels() -> MempoolNodeCommunication {
     let (tx_mempool, rx_mempool) =
         channel::<MempoolRequestAndResponseSender>(MEMPOOL_INVOCATIONS_QUEUE_SIZE);
     MempoolNodeCommunication {
-        mempool_channel: ComponentCommunication { tx: Some(tx_mempool), rx: Some(rx_mempool) },
+        mempool_channel: ComponentCommunication::new(Some(tx_mempool), Some(rx_mempool)),
     }
 }