feat: adding empty server and refactoring ComponentServer

commit-id:689882a6

crates/gateway/Cargo.toml

@@ -12,6 +12,7 @@ workspace = true
 testing = []
 
 [dependencies]
+async-trait.workspace = true
 axum.workspace = true
 blockifier.workspace = true
 cairo-lang-starknet-classes.workspace = true
@@ -22,6 +23,7 @@ reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 starknet_api.workspace = true
+starknet_mempool_infra = { path = "../mempool_infra", version = "0.0" }
 starknet_mempool_types = { path = "../mempool_types", version = "0.0" }
 starknet_sierra_compile = { path = "../starknet_sierra_compile", version = "0.0" }
 thiserror.workspace = true

crates/gateway/src/gateway.rs

@@ -3,12 +3,15 @@ use std::net::SocketAddr;
 use std::panic;
 use std::sync::Arc;
 
+use async_trait::async_trait;
 use axum::extract::State;
 use axum::routing::{get, post};
 use axum::{Json, Router};
 use blockifier::execution::contract_class::{ClassInfo, ContractClass, ContractClassV1};
 use starknet_api::rpc_transaction::{RPCDeclareTransaction, RPCTransaction};
 use starknet_api::transaction::TransactionHash;
+use starknet_mempool_infra::component_runner::{ComponentRunner, ComponentStartError};
+use starknet_mempool_infra::empty_server::{create_empty_server, EmptyServer};
 use starknet_mempool_types::communication::SharedMempoolClient;
 use starknet_mempool_types::mempool_types::{Account, MempoolInput};
 use starknet_sierra_compile::compile::{compile_sierra_to_casm, CompilationUtilError};
@@ -177,3 +180,18 @@ pub fn create_gateway(
     let state_reader_factory = Arc::new(RpcStateReaderFactory { config: rpc_state_reader_config });
     Gateway::new(config, state_reader_factory, client)
 }
+
+pub type GatewayCommunicationServer = EmptyServer<Gateway>;
+
+pub fn create_gateway_server(gateway: Gateway) -> GatewayCommunicationServer {
+    create_empty_server(gateway)
+}
+
+#[async_trait]
+impl ComponentRunner for Gateway {
+    async fn start(&mut self) -> Result<(), ComponentStartError> {
+        // TODO(Lev, 23/07/2024): Implement the real logic.
+        println!("Gateway::start()");
+        Ok(())
+    }
+}

crates/gateway/src/gateway_test.rs

@@ -11,6 +11,7 @@ use starknet_api::rpc_transaction::RPCTransaction;
 use starknet_api::transaction::TransactionHash;
 use starknet_mempool::communication::create_mempool_server;
 use starknet_mempool::mempool::Mempool;
+use starknet_mempool_infra::component_server::CommunicationServer;
 use starknet_mempool_types::communication::{MempoolClientImpl, MempoolRequestAndResponseSender};
 use tokio::sync::mpsc::channel;
 use tokio::task;
@@ -76,6 +77,7 @@ async fn test_add_tx(
 ) {
     // TODO(Tsabary): wrap creation of channels in dedicated functions, take channel capacity from
     // config.
+
     let (tx_mempool, rx_mempool) =
         channel::<MempoolRequestAndResponseSender>(MEMPOOL_INVOCATIONS_QUEUE_SIZE);
     let mut mempool_server = create_mempool_server(mempool, rx_mempool);

crates/mempool_infra/src/component_server.rs

@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use tokio::sync::mpsc::Receiver;
 
 use crate::component_definitions::{ComponentRequestAndResponseSender, ComponentRequestHandler};
@@ -25,7 +26,7 @@ where
         Self { component, rx }
     }
 
-    pub async fn start(&mut self) {
+    pub async fn request_response_loop(&mut self) {
         while let Some(request_and_res_tx) = self.rx.recv().await {
             let request = request_and_res_tx.request;
             let tx = request_and_res_tx.tx;
@@ -36,3 +37,22 @@ where
         }
     }
 }
+
+#[async_trait]
+pub trait CommunicationServer: Send + Sync {
+    async fn start(&mut self);
+}
+
+#[async_trait]
+impl<Component, Request, Response> CommunicationServer
+    for ComponentServer<Component, Request, Response>
+where
+    Component: ComponentRequestHandler<Request, Response> + Send + Sync,
+    Request: Send + Sync,
+    Response: Send + Sync,
+{
+    async fn start(&mut self) {
+        self.request_response_loop().await;
+        println!("ComponentServer completed unexpectedly.");
+    }
+}

crates/mempool_infra/src/empty_server.rs

@@ -0,0 +1,33 @@
+use async_trait::async_trait;
+
+// use tokio::sync::mpsc::channel;
+
+// use crate::component_definitions::{ComponentRequestAndResponseSender,
+// ComponentRequestHandler}; use crate::component_runner::{ComponentRunner,
+// ComponentStartError};
+use crate::component_runner::ComponentRunner;
+use crate::component_server::CommunicationServer;
+
+pub struct EmptyServer<T: ComponentRunner + Send> {
+    component: T,
+}
+
+impl<T: ComponentRunner + Send + Sync> EmptyServer<T> {
+    pub fn new(component: T) -> Self {
+        Self { component }
+    }
+}
+
+#[async_trait]
+impl<T: ComponentRunner + Send + Sync> CommunicationServer for EmptyServer<T> {
+    async fn start(&mut self) {
+        match self.component.start().await {
+            Ok(_) => println!("ComponentServer::start() completed."),
+            Err(err) => println!("ComponentServer::start() failed: {:?}", err),
+        }
+    }
+}
+
+pub fn create_empty_server<T: ComponentRunner + Send + Sync>(component: T) -> EmptyServer<T> {
+    EmptyServer::new(component)
+}

crates/mempool_infra/src/lib.rs

@@ -2,3 +2,4 @@ pub mod component_client;
 pub mod component_definitions;
 pub mod component_runner;
 pub mod component_server;
+pub mod empty_server;

crates/mempool_infra/tests/component_server_client_test.rs

@@ -6,7 +6,7 @@ use starknet_mempool_infra::component_client::ComponentClient;
 use starknet_mempool_infra::component_definitions::{
     ComponentRequestAndResponseSender, ComponentRequestHandler,
 };
-use starknet_mempool_infra::component_server::ComponentServer;
+use starknet_mempool_infra::component_server::{CommunicationServer, ComponentServer};
 use tokio::sync::mpsc::{channel, Sender};
 use tokio::task;
 

crates/tests-integration/Cargo.toml

@@ -22,6 +22,7 @@ starknet_api.workspace = true
 starknet_client.workspace = true
 starknet_gateway = { path = "../gateway", version = "0.0", features = ["testing"] }
 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" }
 starknet_task_executor = { path = "../task_executor", version = "0.0" }
 strum.workspace = true

crates/tests-integration/src/integration_test_setup.rs

@@ -7,6 +7,7 @@ use starknet_gateway::config::GatewayNetworkConfig;
 use starknet_gateway::errors::GatewayError;
 use starknet_mempool::communication::create_mempool_server;
 use starknet_mempool::mempool::Mempool;
+use starknet_mempool_infra::component_server::CommunicationServer;
 use starknet_mempool_types::communication::{
     MempoolClient, MempoolClientImpl, MempoolRequestAndResponseSender,
 };