1mod module;
2
3pub use crate::{
4 error::*, id_provider::SubscriptionIdProvider, RpcServerHandle,
5};
6use cfx_rpc_middlewares::{
7 load_throttling_manager, maybe_cors_layer, Logger, Metrics, Throttle,
8};
9pub use module::{CfxRpcModule, RpcModuleSelection};
10
11use blockgen::BlockGeneratorTestApi;
12use cfx_rpc_cfx_api::{
13 CfxDebugRpcServer, CfxFilterRpcServer, CfxRpcServer, DebugRpcServer,
14 PosRpcServer, PubSubApiServer, TestRpcServer, TraceServer, TxPoolServer,
15};
16use cfx_rpc_cfx_impl::{
17 CfxFilterHandler, CfxHandler, DebugHandler, PosHandler, PubSubHandler,
18 TestHandler, TraceHandler, TxPoolHandler,
19};
20use cfx_rpc_cfx_types::RpcImplConfiguration;
21use cfx_tasks::TaskExecutor;
22use cfxcore::{
23 block_data_manager::BlockDataManager, consensus::pos_handler::PosVerifier,
24 Notifications, SharedConsensusGraph, SharedSynchronizationService,
25 SharedTransactionPool,
26};
27use cfxcore_accounts::AccountProvider;
28use jsonrpsee::{
29 core::RegisterMethodError,
30 server::{
31 middleware::rpc::RpcServiceBuilder, IdProvider, ServerBuilder,
32 ServerConfigBuilder,
33 },
34 Methods, RpcModule,
35};
36use network::NetworkService;
37use parking_lot::{Condvar, Mutex};
38use std::{
39 collections::{HashMap, HashSet},
40 net::{Ipv4Addr, SocketAddr, SocketAddrV4},
41 sync::Arc,
42};
43use txgen::{DirectTransactionGenerator, TransactionGenerator};
44
45pub const DEFAULT_HTTP_PORT: u16 = 12537;
46pub const DEFAULT_WS_PORT: u16 = 12538;
47
48#[derive(Clone)]
49pub struct RpcModuleBuilder {
50 rpc_impl_config: RpcImplConfiguration,
51 consensus: SharedConsensusGraph,
52 sync: SharedSynchronizationService,
53 tx_pool: SharedTransactionPool,
54 executor: TaskExecutor,
55 data_man: Arc<BlockDataManager>,
56 network: Arc<NetworkService>,
57 pos_handler: Arc<PosVerifier>,
58 notifications: Arc<Notifications>,
59 accounts: Arc<AccountProvider>,
60 exit: Arc<(Mutex<bool>, Condvar)>,
61 block_gen: BlockGeneratorTestApi,
62 maybe_txgen: Option<Arc<TransactionGenerator>>,
63 maybe_direct_txgen: Option<Arc<Mutex<DirectTransactionGenerator>>>,
64}
65
66impl RpcModuleBuilder {
67 pub fn new(
68 rpc_impl_config: RpcImplConfiguration, consensus: SharedConsensusGraph,
69 sync: SharedSynchronizationService, tx_pool: SharedTransactionPool,
70 executor: TaskExecutor, data_man: Arc<BlockDataManager>,
71 network: Arc<NetworkService>, pos_handler: Arc<PosVerifier>,
72 notifications: Arc<Notifications>, accounts: Arc<AccountProvider>,
73 exit: Arc<(Mutex<bool>, Condvar)>, block_gen: BlockGeneratorTestApi,
74 maybe_txgen: Option<Arc<TransactionGenerator>>,
75 maybe_direct_txgen: Option<Arc<Mutex<DirectTransactionGenerator>>>,
76 ) -> Self {
77 Self {
78 rpc_impl_config,
79 consensus,
80 sync,
81 tx_pool,
82 executor,
83 data_man,
84 network,
85 pos_handler,
86 notifications,
87 accounts,
88 exit,
89 block_gen,
90 maybe_txgen,
91 maybe_direct_txgen,
92 }
93 }
94
95 pub fn build(
96 self, module_config: TransportRpcModuleConfig,
97 ) -> TransportRpcModules<()> {
98 let mut modules = TransportRpcModules::default();
99
100 if !module_config.is_empty() {
101 let TransportRpcModuleConfig { http, ws } = module_config.clone();
102
103 let Self {
104 rpc_impl_config,
105 consensus,
106 sync,
107 tx_pool,
108 executor,
109 data_man,
110 network,
111 pos_handler,
112 notifications,
113 accounts,
114 exit,
115 block_gen,
116 maybe_txgen,
117 maybe_direct_txgen,
118 } = self;
119
120 let mut registry = RpcRegistryInner::new(
121 rpc_impl_config,
122 consensus,
123 sync,
124 tx_pool,
125 executor,
126 data_man,
127 network,
128 pos_handler,
129 notifications,
130 accounts,
131 exit,
132 block_gen,
133 maybe_txgen,
134 maybe_direct_txgen,
135 );
136
137 modules.config = module_config;
138 modules.http = registry.maybe_module(http.as_ref());
139 modules.ws = registry.maybe_module(ws.as_ref());
140 }
141
142 modules
143 }
144}
145
146#[derive(Clone)]
147pub struct RpcRegistryInner {
148 rpc_impl_config: RpcImplConfiguration,
149 consensus: SharedConsensusGraph,
150 sync: SharedSynchronizationService,
151 tx_pool: SharedTransactionPool,
152 executor: TaskExecutor,
153 data_man: Arc<BlockDataManager>,
154 network: Arc<NetworkService>,
155 pos_handler: Arc<PosVerifier>,
156 notifications: Arc<Notifications>,
157 accounts: Arc<AccountProvider>,
158 exit: Arc<(Mutex<bool>, Condvar)>,
159 block_gen: BlockGeneratorTestApi,
160 maybe_txgen: Option<Arc<TransactionGenerator>>,
161 maybe_direct_txgen: Option<Arc<Mutex<DirectTransactionGenerator>>>,
162 modules: HashMap<CfxRpcModule, Methods>,
163}
164
165impl RpcRegistryInner {
166 pub fn new(
167 rpc_impl_config: RpcImplConfiguration, consensus: SharedConsensusGraph,
168 sync: SharedSynchronizationService, tx_pool: SharedTransactionPool,
169 executor: TaskExecutor, data_man: Arc<BlockDataManager>,
170 network: Arc<NetworkService>, pos_handler: Arc<PosVerifier>,
171 notifications: Arc<Notifications>, accounts: Arc<AccountProvider>,
172 exit: Arc<(Mutex<bool>, Condvar)>, block_gen: BlockGeneratorTestApi,
173 maybe_txgen: Option<Arc<TransactionGenerator>>,
174 maybe_direct_txgen: Option<Arc<Mutex<DirectTransactionGenerator>>>,
175 ) -> Self {
176 Self {
177 rpc_impl_config,
178 consensus,
179 sync,
180 tx_pool,
181 executor,
182 data_man,
183 network,
184 pos_handler,
185 notifications,
186 accounts,
187 exit,
188 block_gen,
189 maybe_txgen,
190 maybe_direct_txgen,
191 modules: Default::default(),
192 }
193 }
194
195 fn maybe_module(
196 &mut self, config: Option<&RpcModuleSelection>,
197 ) -> Option<RpcModule<()>> {
198 config.map(|config| self.module_for(config))
199 }
200
201 pub fn module_for(&mut self, config: &RpcModuleSelection) -> RpcModule<()> {
202 let mut module = RpcModule::new(());
203 let all_methods = self.cfx_methods(config.iter_selection());
204 for methods in all_methods {
205 module.merge(methods).expect("No conflicts");
206 }
207 module
208 }
209
210 pub fn cfx_methods(
211 &mut self, namespaces: impl Iterator<Item = CfxRpcModule>,
212 ) -> Vec<Methods> {
213 let namespaces: Vec<_> = namespaces.collect();
214
215 let namespace_methods = |namespace| {
216 self.modules
217 .entry(namespace)
218 .or_insert_with(|| match namespace {
219 CfxRpcModule::Debug => {
220 let mut methods = DebugHandler::new(
221 self.tx_pool.clone(),
222 self.consensus.clone(),
223 self.sync.clone(),
224 self.network.clone(),
225 self.accounts.clone(),
226 self.pos_handler.clone(),
227 self.exit.clone(),
228 )
229 .into_rpc();
230 methods
231 .merge(CfxDebugRpcServer::into_rpc(
232 CfxHandler::new(
233 self.rpc_impl_config.clone(),
234 self.consensus.clone(),
235 self.sync.clone(),
236 self.tx_pool.clone(),
237 self.accounts.clone(),
238 self.pos_handler.clone(),
239 self.block_gen.clone(),
240 ),
241 ))
242 .expect("No conflicts");
243 methods.into()
244 }
245 CfxRpcModule::Pos => {
246 let handler = PosHandler::new(
247 self.pos_handler.clone(),
248 self.data_man.clone(),
249 *self.network.get_network_type(),
250 self.consensus.clone(),
251 );
252 handler.into_rpc().into()
253 }
254 CfxRpcModule::Trace => {
255 let handler = TraceHandler::new(
256 *self.network.get_network_type(),
257 self.consensus.clone(),
258 );
259 handler.into_rpc().into()
260 }
261 CfxRpcModule::Txpool => TxPoolHandler::new(
262 self.tx_pool.clone(),
263 self.consensus.clone(),
264 *self.network.get_network_type(),
265 )
266 .into_rpc()
267 .into(),
268 CfxRpcModule::PubSub => PubSubHandler::new(
269 self.notifications.clone(),
270 self.executor.clone(),
271 self.consensus.clone(),
272 *self.network.get_network_type(),
273 )
274 .into_rpc()
275 .into(),
276 CfxRpcModule::Cfx => {
277 let mut module =
278 CfxRpcServer::into_rpc(CfxHandler::new(
279 self.rpc_impl_config.clone(),
280 self.consensus.clone(),
281 self.sync.clone(),
282 self.tx_pool.clone(),
283 self.accounts.clone(),
284 self.pos_handler.clone(),
285 self.block_gen.clone(),
286 ));
287 if self
288 .rpc_impl_config
289 .poll_lifetime_in_seconds
290 .is_some()
291 {
292 let filter_module =
293 CfxFilterHandler::new_with_task_executor(
294 self.consensus.clone(),
295 self.tx_pool.clone(),
296 self.notifications.epochs_ordered.clone(),
297 self.executor.clone(),
298 self.rpc_impl_config
299 .poll_lifetime_in_seconds
300 .unwrap(),
301 self.rpc_impl_config
302 .get_logs_filter_max_limit,
303 *self.network.get_network_type(),
304 )
305 .into_rpc();
306 module.merge(filter_module).expect("No conflicts");
307 }
308 module.into()
309 }
310 CfxRpcModule::Test => TestHandler::new(
311 self.exit.clone(),
312 self.consensus.clone(),
313 self.network.clone(),
314 self.pos_handler.clone(),
315 self.tx_pool.clone(),
316 self.accounts.clone(),
317 self.block_gen.clone(),
318 self.maybe_txgen.clone(),
319 self.maybe_direct_txgen.clone(),
320 self.sync.clone(),
321 )
322 .into_rpc()
323 .into(),
324 })
325 .clone()
326 };
327
328 namespaces.iter().copied().map(namespace_methods).collect()
329 }
330}
331
332#[derive(Debug)]
333pub struct RpcServerConfig {
334 http_server_config: Option<ServerConfigBuilder>,
335 http_cors_domains: Option<String>,
336 http_addr: Option<SocketAddr>,
337 ws_server_config: Option<ServerConfigBuilder>,
338 ws_cors_domains: Option<String>,
339 ws_addr: Option<SocketAddr>,
340}
341
342impl Default for RpcServerConfig {
343 fn default() -> Self {
344 Self {
345 http_server_config: None,
346 http_cors_domains: None,
347 http_addr: None,
348 ws_server_config: None,
349 ws_cors_domains: None,
350 ws_addr: None,
351 }
352 }
353}
354
355impl RpcServerConfig {
356 pub fn http(config: ServerConfigBuilder) -> Self {
357 Self::default().with_http(config)
358 }
359
360 pub fn ws(config: ServerConfigBuilder) -> Self {
361 Self::default().with_ws(config)
362 }
363
364 pub fn with_http(mut self, config: ServerConfigBuilder) -> Self {
365 self.http_server_config =
366 Some(config.set_id_provider(SubscriptionIdProvider::default()));
367 self
368 }
369
370 pub fn with_ws(mut self, config: ServerConfigBuilder) -> Self {
371 self.ws_server_config =
372 Some(config.set_id_provider(SubscriptionIdProvider::default()));
373 self
374 }
375
376 pub fn with_cors(self, cors_domain: Option<String>) -> Self {
377 self.with_http_cors(cors_domain.clone())
378 .with_ws_cors(cors_domain)
379 }
380
381 pub fn with_ws_cors(mut self, cors_domain: Option<String>) -> Self {
382 self.ws_cors_domains = cors_domain;
383 self
384 }
385
386 pub fn with_http_cors(mut self, cors_domain: Option<String>) -> Self {
387 self.http_cors_domains = cors_domain;
388 self
389 }
390
391 pub const fn with_http_address(mut self, addr: SocketAddr) -> Self {
392 self.http_addr = Some(addr);
393 self
394 }
395
396 pub const fn with_ws_address(mut self, addr: SocketAddr) -> Self {
397 self.ws_addr = Some(addr);
398 self
399 }
400
401 pub fn with_id_provider<I>(mut self, id_provider: I) -> Self
402 where I: IdProvider + Clone + 'static {
403 if let Some(http) = self.http_server_config.take() {
404 self.http_server_config =
405 Some(http.set_id_provider(id_provider.clone()));
406 }
407 if let Some(ws) = self.ws_server_config.take() {
408 self.ws_server_config =
409 Some(ws.set_id_provider(id_provider.clone()));
410 }
411
412 self
413 }
414
415 pub const fn has_server(&self) -> bool {
416 self.http_server_config.is_some() || self.ws_server_config.is_some()
417 }
418
419 pub const fn http_address(&self) -> Option<SocketAddr> { self.http_addr }
420
421 pub const fn ws_address(&self) -> Option<SocketAddr> { self.ws_addr }
422
423 pub async fn start(
424 self, modules: &TransportRpcModules,
425 throttling_conf_file: Option<String>, throttling_section: &str,
426 enable_metrics: bool,
427 ) -> Result<RpcServerHandle, RpcError<CfxRpcModule>> {
428 if !self.has_server() {
430 return Ok(RpcServerHandle {
431 http_local_addr: None,
432 ws_local_addr: None,
433 http: None,
434 ws: None,
435 });
436 }
437
438 let throttle_manager = load_throttling_manager(
439 throttling_conf_file.as_deref(),
440 throttling_section,
441 );
442 let rpc_middleware = RpcServiceBuilder::new()
443 .layer_fn(move |s| Throttle::new(throttle_manager.clone(), s))
444 .layer_fn(move |s| Metrics::new(s, enable_metrics))
445 .layer_fn(|s| Logger::new(s));
446
447 let http_socket_addr = self.http_addr.unwrap_or(SocketAddr::V4(
448 SocketAddrV4::new(Ipv4Addr::LOCALHOST, DEFAULT_HTTP_PORT),
449 ));
450
451 let ws_socket_addr = self.ws_addr.unwrap_or(SocketAddr::V4(
452 SocketAddrV4::new(Ipv4Addr::LOCALHOST, DEFAULT_WS_PORT),
453 ));
454
455 if self.http_addr == self.ws_addr
456 && self.http_server_config.is_some()
457 && self.ws_server_config.is_some()
458 {
459 modules.config.ensure_ws_http_identical()?;
460
461 let cors = match (
462 self.ws_cors_domains.as_ref(),
463 self.http_cors_domains.as_ref(),
464 ) {
465 (Some(ws_cors), Some(http_cors)) => {
466 if ws_cors.trim() != http_cors.trim() {
467 return Err(
468 WsHttpSamePortError::ConflictingCorsDomains {
469 http_cors_domains: Some(http_cors.clone()),
470 ws_cors_domains: Some(ws_cors.clone()),
471 }
472 .into(),
473 );
474 }
475 Some(ws_cors)
476 }
477 (a, b) => a.or(b),
478 }
479 .cloned();
480
481 if let Some(config) = self.http_server_config {
482 let server = ServerBuilder::new()
483 .set_http_middleware(
484 tower::ServiceBuilder::new()
485 .option_layer(maybe_cors_layer(cors)?),
486 )
487 .set_rpc_middleware(rpc_middleware)
488 .set_config(config.build())
489 .build(http_socket_addr)
490 .await
491 .map_err(|err| {
492 RpcError::server_error(
493 err,
494 ServerKind::WsHttp(http_socket_addr),
495 )
496 })?;
497 let addr = server.local_addr().map_err(|err| {
498 RpcError::server_error(
499 err,
500 ServerKind::WsHttp(http_socket_addr),
501 )
502 })?;
503 if let Some(module) =
504 modules.http.as_ref().or(modules.ws.as_ref())
505 {
506 let handle = server.start(module.clone());
507 return Ok(RpcServerHandle {
508 http_local_addr: Some(addr),
509 ws_local_addr: Some(addr),
510 http: Some(handle.clone()),
511 ws: Some(handle),
512 });
513 }
514
515 return Err(RpcError::Custom(
516 "No valid RpcModule found from modules".to_string(),
517 ));
518 }
519 }
520
521 let mut result = RpcServerHandle {
522 http_local_addr: None,
523 ws_local_addr: None,
524 http: None,
525 ws: None,
526 };
527
528 if let Some(config) = self.ws_server_config {
529 let server = ServerBuilder::new()
530 .set_config(config.ws_only().build())
531 .set_http_middleware(tower::ServiceBuilder::new().option_layer(
532 maybe_cors_layer(self.ws_cors_domains.clone())?,
533 ))
534 .set_rpc_middleware(rpc_middleware.clone())
535 .build(ws_socket_addr)
536 .await
537 .map_err(|err| {
538 RpcError::server_error(err, ServerKind::WS(ws_socket_addr))
539 })?;
540
541 let addr = server.local_addr().map_err(|err| {
542 RpcError::server_error(err, ServerKind::WS(ws_socket_addr))
543 })?;
544
545 let ws_local_addr = Some(addr);
546 let ws_handle = Some(
547 server.start(modules.ws.clone().expect("ws server error")),
548 );
549
550 result.ws = ws_handle;
551 result.ws_local_addr = ws_local_addr;
552 }
553
554 if let Some(config) = self.http_server_config {
555 let server = ServerBuilder::new()
556 .set_config(config.http_only().build())
557 .set_http_middleware(tower::ServiceBuilder::new().option_layer(
558 maybe_cors_layer(self.http_cors_domains.clone())?,
559 ))
560 .set_rpc_middleware(rpc_middleware)
561 .build(http_socket_addr)
562 .await
563 .map_err(|err| {
564 RpcError::server_error(
565 err,
566 ServerKind::Http(http_socket_addr),
567 )
568 })?;
569 let local_addr = server.local_addr().map_err(|err| {
570 RpcError::server_error(err, ServerKind::Http(http_socket_addr))
571 })?;
572 let http_local_addr = Some(local_addr);
573 let http_handle = Some(
574 server.start(modules.http.clone().expect("http server error")),
575 );
576
577 result.http = http_handle;
578 result.http_local_addr = http_local_addr;
579 }
580
581 Ok(result)
582 }
583}
584
585#[derive(Debug, Clone, Default, Eq, PartialEq)]
586pub struct TransportRpcModuleConfig {
587 pub http: Option<RpcModuleSelection>,
588 pub ws: Option<RpcModuleSelection>,
589}
590
591impl TransportRpcModuleConfig {
592 pub fn set_http(http: impl Into<RpcModuleSelection>) -> Self {
593 Self::default().with_http(http)
594 }
595
596 pub fn set_ws(ws: impl Into<RpcModuleSelection>) -> Self {
597 Self::default().with_ws(ws)
598 }
599
600 pub fn with_http(mut self, http: impl Into<RpcModuleSelection>) -> Self {
601 self.http = Some(http.into());
602 self
603 }
604
605 pub fn with_ws(mut self, ws: impl Into<RpcModuleSelection>) -> Self {
606 self.ws = Some(ws.into());
607 self
608 }
609
610 pub fn http_mut(&mut self) -> &mut Option<RpcModuleSelection> {
611 &mut self.http
612 }
613
614 pub fn ws_mut(&mut self) -> &mut Option<RpcModuleSelection> { &mut self.ws }
615
616 pub const fn is_empty(&self) -> bool {
617 self.http.is_none() && self.ws.is_none()
618 }
619
620 pub const fn http(&self) -> Option<&RpcModuleSelection> {
621 self.http.as_ref()
622 }
623
624 pub const fn ws(&self) -> Option<&RpcModuleSelection> { self.ws.as_ref() }
625
626 fn ensure_ws_http_identical(
627 &self,
628 ) -> Result<(), WsHttpSamePortError<CfxRpcModule>> {
629 if RpcModuleSelection::are_identical(
630 self.http.as_ref(),
631 self.ws.as_ref(),
632 ) {
633 Ok(())
634 } else {
635 let http_modules = self
636 .http
637 .as_ref()
638 .map(RpcModuleSelection::to_selection)
639 .unwrap_or_default();
640 let ws_modules = self
641 .ws
642 .as_ref()
643 .map(RpcModuleSelection::to_selection)
644 .unwrap_or_default();
645
646 let http_not_ws: HashSet<CfxRpcModule> =
647 http_modules.difference(&ws_modules).copied().collect();
648 let ws_not_http: HashSet<CfxRpcModule> =
649 ws_modules.difference(&http_modules).copied().collect();
650 let overlap: HashSet<CfxRpcModule> =
651 http_modules.intersection(&ws_modules).copied().collect();
652 let conflicting_modules = ConflictingModules {
654 overlap,
655 http_not_ws,
656 ws_not_http,
657 };
658 Err(WsHttpSamePortError::ConflictingModules(Box::new(
659 conflicting_modules,
660 )))
661 }
662 }
663}
664
665#[derive(Debug, Clone, Default)]
666pub struct TransportRpcModules<Context = ()> {
667 pub config: TransportRpcModuleConfig,
668 pub http: Option<RpcModule<Context>>,
669 pub ws: Option<RpcModule<Context>>,
670}
671
672impl TransportRpcModules {
673 pub const fn module_config(&self) -> &TransportRpcModuleConfig {
674 &self.config
675 }
676
677 pub fn merge_http(
678 &mut self, other: impl Into<Methods>,
679 ) -> Result<bool, RegisterMethodError> {
680 if let Some(ref mut http) = self.http {
681 return http.merge(other.into()).map(|_| true);
682 }
683 Ok(false)
684 }
685
686 pub fn merge_ws(
687 &mut self, other: impl Into<Methods>,
688 ) -> Result<bool, RegisterMethodError> {
689 if let Some(ref mut ws) = self.ws {
690 return ws.merge(other.into()).map(|_| true);
691 }
692 Ok(false)
693 }
694
695 pub fn merge_configured(
696 &mut self, other: impl Into<Methods>,
697 ) -> Result<(), RegisterMethodError> {
698 let other = other.into();
699 self.merge_http(other.clone())?;
700 self.merge_ws(other.clone())?;
701 Ok(())
702 }
703
704 pub fn remove_http_method(&mut self, method_name: &'static str) -> bool {
705 if let Some(http_module) = &mut self.http {
706 http_module.remove_method(method_name).is_some()
707 } else {
708 false
709 }
710 }
711
712 pub fn remove_ws_method(&mut self, method_name: &'static str) -> bool {
713 if let Some(ws_module) = &mut self.ws {
714 ws_module.remove_method(method_name).is_some()
715 } else {
716 false
717 }
718 }
719
720 pub fn remove_method_from_configured(
721 &mut self, method_name: &'static str,
722 ) -> bool {
723 let http_removed = self.remove_http_method(method_name);
724 let ws_removed = self.remove_ws_method(method_name);
725
726 http_removed || ws_removed
727 }
728}