client/
rpc_starter.rs

1// Copyright 2019 Conflux Foundation. All rights reserved.
2// Conflux is free software and distributed under GNU General Public License.
3// See http://www.gnu.org/licenses/
4
5use blockgen::BlockGeneratorTestApi;
6use cfx_config::Configuration;
7use cfx_rpc_builder::{
8    CfxRpcModule, CfxRpcModuleBuilder, CfxRpcModuleSelection,
9    CfxRpcServerConfig, CfxTransportRpcModuleConfig, CfxTransportRpcModules,
10    RpcModuleBuilder, RpcServerConfig, RpcServerHandle,
11    TransportRpcModuleConfig,
12};
13use cfx_rpc_cfx_api::{
14    CfxDebugRpcServer, CfxRpcServer, DebugRpcServer, PubSubApiServer,
15    TestRpcServer,
16};
17use cfx_rpc_cfx_types::apis::ApiSet;
18use cfx_tasks::TaskExecutor;
19use cfxcore::{
20    block_data_manager::BlockDataManager, consensus::pos_handler::PosVerifier,
21    LightQueryService, Notifications, SharedConsensusGraph,
22    SharedSynchronizationService, SharedTransactionPool,
23};
24use jsonrpsee::RpcModule;
25use log::{info, warn};
26use network::NetworkService;
27use parking_lot::{Condvar, Mutex};
28use std::sync::Arc;
29use txgen::{DirectTransactionGenerator, TransactionGenerator};
30
31// start espace rpc server v2(async)
32pub async fn launch_async_rpc_servers(
33    consensus: SharedConsensusGraph, sync: SharedSynchronizationService,
34    tx_pool: SharedTransactionPool, notifications: Arc<Notifications>,
35    executor: TaskExecutor, conf: &Configuration,
36) -> Result<Option<RpcServerHandle>, String> {
37    let http_config = conf.eth_http_config();
38    let ws_config = conf.eth_ws_config();
39    let apis = conf.raw_conf.public_evm_rpc_apis.clone();
40
41    let (transport_rpc_module_config, server_config) =
42        match (http_config.enabled, ws_config.enabled) {
43            (true, true) => {
44                let transport_rpc_module_config =
45                    TransportRpcModuleConfig::set_http(apis.clone())
46                        .with_ws(apis.clone());
47
48                let server_config =
49                    RpcServerConfig::http(conf.jsonrpsee_server_builder())
50                        .with_ws(conf.jsonrpsee_server_builder())
51                        .with_cors(http_config.cors.clone())
52                        .with_ws_cors(ws_config.cors.clone())
53                        .with_http_address(http_config.address)
54                        .with_ws_address(ws_config.address);
55                (transport_rpc_module_config, server_config)
56            }
57            (true, false) => {
58                let transport_rpc_module_config =
59                    TransportRpcModuleConfig::set_http(apis.clone());
60                let server_config =
61                    RpcServerConfig::http(conf.jsonrpsee_server_builder())
62                        .with_http_address(http_config.address)
63                        .with_cors(http_config.cors.clone());
64                (transport_rpc_module_config, server_config)
65            }
66            (false, true) => {
67                let transport_rpc_module_config =
68                    TransportRpcModuleConfig::set_ws(apis.clone());
69                let server_config =
70                    RpcServerConfig::ws(conf.jsonrpsee_server_builder())
71                        .with_ws_address(ws_config.address)
72                        .with_ws_cors(ws_config.cors.clone());
73                (transport_rpc_module_config, server_config)
74            }
75            _ => return Ok(None),
76        };
77
78    info!("Enabled evm async rpc modules: {:?}", apis.into_selection());
79    let rpc_conf = conf.rpc_impl_config();
80    let enable_metrics = rpc_conf.enable_metrics;
81
82    let rpc_module_builder = RpcModuleBuilder::new(
83        rpc_conf,
84        consensus,
85        sync,
86        tx_pool,
87        executor,
88        notifications,
89    );
90
91    let transport_rpc_modules =
92        rpc_module_builder.build(transport_rpc_module_config);
93
94    let throttling_conf_file = conf.raw_conf.throttling_conf.clone();
95    let server_handle = server_config
96        .start(&transport_rpc_modules, throttling_conf_file, enable_metrics)
97        .await
98        .map_err(|e| e.to_string())?;
99
100    Ok(Some(server_handle))
101}
102
103// start core space rpc server v2(async)
104pub async fn launch_cfx_async_rpc_servers(
105    consensus: SharedConsensusGraph, sync: SharedSynchronizationService,
106    tx_pool: SharedTransactionPool, data_man: Arc<BlockDataManager>,
107    network: Arc<NetworkService>, pos_handler: Arc<PosVerifier>,
108    notifications: Arc<Notifications>, executor: TaskExecutor,
109    accounts: Arc<cfxcore_accounts::AccountProvider>,
110    exit: Arc<(parking_lot::Mutex<bool>, parking_lot::Condvar)>,
111    block_gen: BlockGeneratorTestApi,
112    maybe_txgen: Option<Arc<TransactionGenerator>>,
113    maybe_direct_txgen: Option<Arc<Mutex<DirectTransactionGenerator>>>,
114    conf: &Configuration, apis: ApiSet, is_debug: bool,
115) -> Result<Option<RpcServerHandle>, String> {
116    let (http_config, ws_config) = if !is_debug {
117        (conf.http_config(), conf.ws_config())
118    } else {
119        (conf.local_http_config(), conf.local_ws_config())
120    };
121
122    let (transport_rpc_module_config, server_config) =
123        match (http_config.enabled, ws_config.enabled) {
124            (true, true) => {
125                let transport_rpc_module_config =
126                    CfxTransportRpcModuleConfig::set_http(apis.clone())
127                        .with_ws(apis.clone());
128
129                let server_config =
130                    CfxRpcServerConfig::http(conf.jsonrpsee_server_builder())
131                        .with_ws(conf.jsonrpsee_server_builder())
132                        .with_cors(http_config.cors.clone())
133                        .with_ws_cors(ws_config.cors.clone())
134                        .with_http_address(http_config.address)
135                        .with_ws_address(ws_config.address);
136                (transport_rpc_module_config, server_config)
137            }
138            (true, false) => {
139                let transport_rpc_module_config =
140                    CfxTransportRpcModuleConfig::set_http(apis.clone());
141                let server_config =
142                    CfxRpcServerConfig::http(conf.jsonrpsee_server_builder())
143                        .with_http_address(http_config.address)
144                        .with_cors(http_config.cors.clone());
145                (transport_rpc_module_config, server_config)
146            }
147            (false, true) => {
148                let transport_rpc_module_config =
149                    CfxTransportRpcModuleConfig::set_ws(apis.clone());
150                let server_config =
151                    CfxRpcServerConfig::ws(conf.jsonrpsee_server_builder())
152                        .with_ws_address(ws_config.address)
153                        .with_ws_cors(ws_config.cors.clone());
154                (transport_rpc_module_config, server_config)
155            }
156            _ => return Ok(None),
157        };
158
159    info!(
160        "Enabled cfx async rpc modules: {:?}",
161        CfxRpcModuleSelection::from(apis).into_selection()
162    );
163
164    let rpc_conf = conf.rpc_impl_config();
165    let enable_metrics = rpc_conf.enable_metrics;
166    let rpc_module_builder = CfxRpcModuleBuilder::new(
167        rpc_conf,
168        consensus,
169        sync,
170        tx_pool,
171        executor,
172        data_man,
173        network,
174        pos_handler,
175        notifications,
176        accounts,
177        exit,
178        block_gen,
179        maybe_txgen,
180        maybe_direct_txgen,
181    );
182
183    let transport_rpc_modules =
184        rpc_module_builder.build(transport_rpc_module_config);
185
186    let throttling_conf_file = conf.raw_conf.throttling_conf.clone();
187    let throttling_section = if is_debug { "rpc_local" } else { "rpc" };
188
189    let server_handle = server_config
190        .start(
191            &transport_rpc_modules,
192            throttling_conf_file,
193            throttling_section,
194            enable_metrics,
195        )
196        .await
197        .map_err(|e| e.to_string())?;
198
199    Ok(Some(server_handle))
200}
201
202// start core space light rpc server (async, jsonrpsee)
203pub async fn launch_cfx_light_async_rpc_servers(
204    consensus: SharedConsensusGraph, tx_pool: SharedTransactionPool,
205    data_man: Arc<BlockDataManager>, network: Arc<NetworkService>,
206    pos_handler: Arc<PosVerifier>,
207    accounts: Arc<cfxcore_accounts::AccountProvider>,
208    light: Arc<LightQueryService>, exit: Arc<(Mutex<bool>, Condvar)>,
209    executor: TaskExecutor, notifications: Arc<Notifications>,
210    conf: &Configuration, apis: ApiSet, is_debug: bool,
211) -> Result<Option<RpcServerHandle>, String> {
212    use cfx_rpc_cfx_impl::{
213        common::CommonRpcImpl,
214        light::{
215            LightCfxHandler, LightDebugHandler, LightTestHandler,
216            RpcImpl as LightRpcImpl,
217        },
218        PubSubHandler,
219    };
220
221    let (http_config, ws_config) = if is_debug {
222        (conf.local_http_config(), conf.local_ws_config())
223    } else {
224        (conf.http_config(), conf.ws_config())
225    };
226    let apis = CfxRpcModuleSelection::from(apis);
227
228    let server_config = match (http_config.enabled, ws_config.enabled) {
229        (true, true) => {
230            CfxRpcServerConfig::http(conf.jsonrpsee_server_builder())
231                .with_ws(conf.jsonrpsee_server_builder())
232                .with_cors(http_config.cors.clone())
233                .with_ws_cors(ws_config.cors.clone())
234                .with_http_address(http_config.address)
235                .with_ws_address(ws_config.address)
236        }
237        (true, false) => {
238            CfxRpcServerConfig::http(conf.jsonrpsee_server_builder())
239                .with_cors(http_config.cors.clone())
240                .with_http_address(http_config.address)
241        }
242        (false, true) => {
243            CfxRpcServerConfig::ws(conf.jsonrpsee_server_builder())
244                .with_ws_cors(ws_config.cors.clone())
245                .with_ws_address(ws_config.address)
246        }
247        _ => return Ok(None),
248    };
249
250    let common_rpc_impl = Arc::new(CommonRpcImpl::new(
251        exit,
252        consensus.clone(),
253        network.clone(),
254        tx_pool,
255        accounts.clone(),
256        pos_handler.clone(),
257    ));
258
259    let light_rpc_impl = Arc::new(LightRpcImpl::new(
260        light,
261        accounts,
262        consensus.clone(),
263        data_man,
264    ));
265
266    let mut module = RpcModule::new(());
267
268    for api in apis.iter_selection() {
269        match api {
270            CfxRpcModule::Cfx => {
271                let handler = LightCfxHandler::new(
272                    light_rpc_impl.clone(),
273                    common_rpc_impl.clone(),
274                );
275                module
276                    .merge(CfxRpcServer::into_rpc(handler))
277                    .expect("No conflicts for Cfx module");
278            }
279            CfxRpcModule::Debug => {
280                let cfx_debug_handler = LightCfxHandler::new(
281                    light_rpc_impl.clone(),
282                    common_rpc_impl.clone(),
283                );
284                module
285                    .merge(CfxDebugRpcServer::into_rpc(cfx_debug_handler))
286                    .expect("No conflicts for CfxDebug module");
287
288                let debug_handler =
289                    LightDebugHandler::new(common_rpc_impl.clone());
290                module
291                    .merge(DebugRpcServer::into_rpc(debug_handler))
292                    .expect("No conflicts for Debug module");
293            }
294            CfxRpcModule::Test => {
295                let handler = LightTestHandler::new(common_rpc_impl.clone());
296                module
297                    .merge(TestRpcServer::into_rpc(handler))
298                    .expect("No conflicts for Test module");
299            }
300            CfxRpcModule::PubSub => {
301                let pubsub_handler = PubSubHandler::new(
302                    notifications.clone(),
303                    executor.clone(),
304                    consensus.clone(),
305                    *network.get_network_type(),
306                );
307                module
308                    .merge(PubSubApiServer::into_rpc(pubsub_handler))
309                    .expect("No conflicts for PubSub module");
310            }
311            CfxRpcModule::Trace => {
312                warn!("Light nodes do not support trace RPC");
313            }
314            CfxRpcModule::Txpool => {
315                warn!("Light nodes do not support txpool RPC");
316            }
317            CfxRpcModule::Pos => {
318                warn!("Light nodes do not support PoS RPC");
319            }
320        }
321    }
322
323    info!(
324        "Enabled cfx light async rpc modules: {:?}",
325        apis.to_selection()
326    );
327
328    let mut transport_modules = CfxTransportRpcModules::default();
329
330    match (http_config.enabled, ws_config.enabled) {
331        (true, true) => {
332            transport_modules.config =
333                CfxTransportRpcModuleConfig::set_http(apis.clone())
334                    .with_ws(apis);
335            transport_modules.http = Some(module.clone());
336            transport_modules.ws = Some(module);
337        }
338        (true, false) => {
339            transport_modules.config =
340                CfxTransportRpcModuleConfig::set_http(apis);
341            transport_modules.http = Some(module);
342        }
343        (false, true) => {
344            transport_modules.config =
345                CfxTransportRpcModuleConfig::set_ws(apis);
346            transport_modules.ws = Some(module);
347        }
348        _ => unreachable!(),
349    }
350
351    let rpc_conf = conf.rpc_impl_config();
352    let enable_metrics = rpc_conf.enable_metrics;
353    let throttling_conf_file = conf.raw_conf.throttling_conf.clone();
354    let throttling_section = if is_debug { "rpc_local" } else { "rpc" };
355
356    let server_handle = server_config
357        .start(
358            &transport_modules,
359            throttling_conf_file,
360            throttling_section,
361            enable_metrics,
362        )
363        .await
364        .map_err(|e| e.to_string())?;
365
366    Ok(Some(server_handle))
367}
client/rpc_starter.rs

client/
rpc_starter.rs
