feat: adding main module

commit-id:8df0f3c6

crates/gateway/src/gateway.rs

@@ -65,7 +65,7 @@ impl Gateway {
         Gateway { config, app_state }
     }
 
-    pub async fn run(self) -> Result<(), GatewayRunError> {
+    pub async fn run(&mut self) -> Result<(), GatewayRunError> {
         // Parses the bind address from GatewayConfig, returning an error for invalid addresses.
         let GatewayNetworkConfig { ip, port } = self.config.network_config;
         let addr = SocketAddr::new(ip, port);
@@ -75,11 +75,11 @@ impl Gateway {
         Ok(axum::Server::bind(&addr).serve(app.into_make_service()).await?)
     }
 
-    pub fn app(self) -> Router {
+    pub fn app(&self) -> Router {
         Router::new()
             .route("/is_alive", get(is_alive))
             .route("/add_tx", post(add_tx))
-            .with_state(self.app_state)
+            .with_state(self.app_state.clone())
     }
 }
 
@@ -194,8 +194,10 @@ pub fn create_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(())
+        match self.run().await {
+            Ok(_) => Ok(()),
+            Err(_) => Err(ComponentStartError::InternalComponentError),
+        }
     }
 }

crates/mempool_node/Cargo.toml

@@ -9,8 +9,10 @@ license.workspace = true
 workspace = true
 
 [dependencies]
+anyhow.workspace = true
 clap.workspace = true
 const_format.workspace = true
+futures.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" }

crates/mempool_node/src/main.rs

@@ -1,5 +1,33 @@
+use std::env::args;
+use std::process::exit;
+
+use papyrus_config::validators::config_validate;
+use papyrus_config::ConfigError;
+use starknet_mempool_node::communication::{create_node_channels, create_node_clients};
+use starknet_mempool_node::components::create_components;
+use starknet_mempool_node::config::MempoolNodeConfig;
+use starknet_mempool_node::servers::{create_servers, run_server_components};
+
 #[tokio::main]
-async fn main() {
-    let my_string = "Main function placeholder";
-    println!("{}", my_string);
+async fn main() -> anyhow::Result<()> {
+    let config = MempoolNodeConfig::load_and_process(args().collect());
+    if let Err(ConfigError::CommandInput(clap_err)) = config {
+        clap_err.exit();
+    }
+
+    let config = config?;
+    if let Err(error) = config_validate(&config) {
+        println!("Error: {}", error);
+        exit(1);
+    }
+
+    let channels = create_node_channels();
+    let clients = create_node_clients(&config, &channels);
+    let components = create_components(&config, &clients);
+    let servers = create_servers(&config, channels, components);
+
+    println!("Info: Starting components!");
+    run_server_components(&config, servers).await?;
+
+    Ok(())
 }

crates/mempool_node/src/servers.rs

@@ -1,3 +1,7 @@
+use std::future::pending;
+use std::pin::Pin;
+
+use futures::{Future, FutureExt};
 use starknet_gateway::communication::create_gateway_server;
 use starknet_mempool::communication::create_mempool_server;
 use starknet_mempool_infra::component_server::ComponentServerStarter;
@@ -38,4 +42,51 @@ pub fn create_servers(
     Servers { gateway: gateway_server, mempool: mempool_server }
 }
 
-// TODO (Lev): Implement the run server components function.
+pub async fn run_server_components(
+    config: &MempoolNodeConfig,
+    servers: Servers,
+) -> anyhow::Result<()> {
+    // Gateway component.
+    // let gateway: Gateway;
+    let gateway_future =
+        get_server_future("Gateway", config.components.gateway_component.execute, servers.gateway);
+
+    // Mempool component.
+    let mempool_future =
+        get_server_future("Mempool", config.components.mempool_component.execute, servers.mempool);
+
+    let gateway_handle = tokio::spawn(gateway_future);
+    let mempool_handle = tokio::spawn(mempool_future);
+
+    tokio::select! {
+        res = gateway_handle => {
+            println!("Error: Gateway Server stopped.");
+            res?
+        }
+        res = mempool_handle => {
+            println!("Error: Mempool Server stopped.");
+            res?
+        }
+    };
+    println!("Error: Servers ended with unexpected Ok.");
+
+    Ok(())
+}
+
+pub fn get_server_future(
+    name: &str,
+    execute_flag: bool,
+    server: Option<Box<dyn ComponentServerStarter>>,
+) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+    let server_future = match execute_flag {
+        true => {
+            let mut server = match server {
+                Some(server) => server,
+                _ => panic!("{} component is not initialized.", name),
+            };
+            async move { server.start().await }.boxed()
+        }
+        false => pending().boxed(),
+    };
+    server_future
+}

crates/tests-integration/src/integration_test_setup.rs

@@ -42,7 +42,7 @@ impl IntegrationTestSetup {
             channel::<MempoolRequestAndResponseSender>(MEMPOOL_INVOCATIONS_QUEUE_SIZE);
         // Build and run gateway; initialize a gateway client.
         let gateway_mempool_client = MempoolClientImpl::new(tx_mempool.clone());
-        let gateway =
+        let mut gateway =
             create_gateway(Arc::new(gateway_mempool_client), n_initialized_account_contracts).await;
         let GatewayNetworkConfig { ip, port } = gateway.config.network_config;
         let gateway_client = GatewayClient::new(SocketAddr::from((ip, port)));