kabu_rpc_handler/
web_actor.rs

1use crate::router::router;
2use axum::Router;
3use kabu_core_components::Component;
4
5use kabu_core_blockchain::{Blockchain, BlockchainState};
6use kabu_rpc_state::AppState;
7use kabu_storage_db::DbPool;
8use reth_ethereum_primitives::EthPrimitives;
9use revm::{DatabaseCommit, DatabaseRef};
10use std::net::SocketAddr;
11use tokio::net::TcpListener;
12use tokio_util::sync::CancellationToken;
13use tower_http::trace::{DefaultMakeSpan, TraceLayer};
14use tracing::info;
15
16pub async fn start_web_server_worker<S, DB>(
17    host: String,
18    extra_router: Router<S>,
19    bc: Blockchain,
20    state: BlockchainState<DB, EthPrimitives>,
21    db_pool: DbPool,
22    shutdown_token: CancellationToken,
23) -> eyre::Result<()>
24where
25    DB: DatabaseRef<Error = kabu_evm_db::KabuDBError> + DatabaseCommit + Send + Sync + Clone + Default + 'static,
26    S: Clone + Send + Sync + 'static,
27    Router: From<Router<S>>,
28{
29    let app_state = AppState { db: db_pool, bc, state };
30    let router = router(app_state);
31    let router = router.merge(extra_router);
32
33    // logging
34    let router = router.layer(TraceLayer::new_for_http().make_span_with(DefaultMakeSpan::default().include_headers(true)));
35
36    info!("Webserver listening on {}", &host);
37    let listener = TcpListener::bind(host).await?;
38    axum::serve(listener, router.into_make_service_with_connect_info::<SocketAddr>())
39        .with_graceful_shutdown(async move {
40            shutdown_token.cancelled().await;
41            info!("Shutting down webserver...");
42        })
43        .await?;
44
45    Ok(())
46}
47
48pub struct WebServerComponent<S, DB: Clone + Send + Sync + 'static> {
49    host: String,
50    extra_router: Router<S>,
51    shutdown_token: CancellationToken,
52    db_pool: DbPool,
53    bc: Option<Blockchain>,
54    state: Option<BlockchainState<DB, EthPrimitives>>,
55}
56
57impl<S, DB> WebServerComponent<S, DB>
58where
59    DB: DatabaseRef<Error = kabu_evm_db::KabuDBError> + Send + Sync + Clone + Default + 'static,
60    S: Clone + Send + Sync + 'static,
61    Router: From<Router<S>>,
62{
63    pub fn new(host: String, extra_router: Router<S>, db_pool: DbPool, shutdown_token: CancellationToken) -> Self {
64        Self { host, extra_router, shutdown_token, db_pool, bc: None, state: None }
65    }
66
67    pub fn on_bc(self, bc: &Blockchain, state: &BlockchainState<DB, EthPrimitives>) -> Self {
68        Self { bc: Some(bc.clone()), state: Some(state.clone()), ..self }
69    }
70}
71
72impl<S, DB> Component for WebServerComponent<S, DB>
73where
74    S: Clone + Send + Sync + 'static,
75    DB: DatabaseRef<Error = kabu_evm_db::KabuDBError> + DatabaseCommit + Send + Sync + Clone + Default + 'static,
76    Router: From<Router<S>>,
77{
78    fn spawn(self, executor: reth_tasks::TaskExecutor) -> eyre::Result<()> {
79        let name = self.name();
80        let bc = self.bc.ok_or_else(|| eyre::eyre!("Blockchain not set"))?;
81        let state = self.state.ok_or_else(|| eyre::eyre!("State not set"))?;
82
83        executor.spawn_critical(name, async move {
84            if let Err(e) = start_web_server_worker(self.host, self.extra_router, bc, state, self.db_pool, self.shutdown_token).await {
85                tracing::error!("Web server worker failed: {}", e);
86            }
87        });
88        Ok(())
89    }
90
91    fn name(&self) -> &'static str {
92        "WebServerComponent"
93    }
94}
kabu_rpc_handler/web_actor.rs

kabu_rpc_handler/
web_actor.rs
