1#![doc(
15 html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
16 html_favicon_url = "https://avatars0.githubusercontent.com/u/97369466?s=256",
17 issue_tracker_base_url = "https://github.com/paradigmxyz/reth/issues/"
18)]
19#![cfg_attr(not(test), warn(unused_crate_dependencies))]
20#![cfg_attr(docsrs, feature(doc_cfg))]
21
22use crate::{auth::AuthRpcModule, error::WsHttpSamePortError, metrics::RpcRequestMetrics};
23use alloy_network::{Ethereum, IntoWallet};
24use alloy_provider::{fillers::RecommendedFillers, Provider, ProviderBuilder};
25use core::marker::PhantomData;
26use error::{ConflictingModules, RpcError, ServerKind};
27use http::{header::AUTHORIZATION, HeaderMap};
28use jsonrpsee::{
29 core::RegisterMethodError,
30 server::{middleware::rpc::RpcServiceBuilder, AlreadyStoppedError, IdProvider, ServerHandle},
31 Methods, RpcModule,
32};
33use reth_chainspec::{ChainSpecProvider, EthereumHardforks};
34use reth_consensus::FullConsensus;
35use reth_engine_primitives::{ConsensusEngineEvent, ConsensusEngineHandle};
36use reth_evm::ConfigureEvm;
37use reth_network_api::{noop::NoopNetwork, NetworkInfo, Peers};
38use reth_payload_primitives::PayloadTypes;
39use reth_primitives_traits::{NodePrimitives, TxTy};
40use reth_rpc::{
41 AdminApi, DebugApi, EngineEthApi, EthApi, EthApiBuilder, EthBundle, MinerApi, NetApi,
42 OtterscanApi, RPCApi, RethApi, TraceApi, TxPoolApi, Web3Api,
43};
44use reth_rpc_api::servers::*;
45use reth_rpc_engine_api::RethEngineApi;
46use reth_rpc_eth_api::{
47 helpers::{
48 pending_block::PendingEnvBuilder, Call, EthApiSpec, EthTransactions, LoadPendingBlock,
49 TraceExt,
50 },
51 node::RpcNodeCoreAdapter,
52 EthApiServer, EthApiTypes, FullEthApiServer, FullEthApiTypes, RpcBlock, RpcConvert,
53 RpcConverter, RpcHeader, RpcNodeCore, RpcReceipt, RpcTransaction, RpcTxReq,
54};
55use reth_rpc_eth_types::{receipt::EthReceiptConverter, EthConfig, EthSubscriptionIdProvider};
56use reth_rpc_layer::{
57 AuthLayer, Claims, CompressionLayer, DecompressionLayer, JwtAuthValidator, JwtSecret,
58};
59pub use reth_rpc_server_types::RethRpcModule;
60use reth_storage_api::{
61 BlockReader, ChangeSetReader, FullRpcProvider, NodePrimitivesProvider, StateProviderFactory,
62};
63use reth_tasks::{pool::BlockingTaskGuard, Runtime};
64use reth_tokio_util::EventSender;
65use reth_transaction_pool::{noop::NoopTransactionPool, TransactionPool};
66use serde::{Deserialize, Serialize};
67use std::{
68 collections::HashMap,
69 fmt::Debug,
70 net::{Ipv4Addr, SocketAddr, SocketAddrV4},
71 time::{Duration, SystemTime, UNIX_EPOCH},
72};
73use tower_http::cors::CorsLayer;
74
75pub use cors::CorsDomainError;
76
77pub use jsonrpsee::server::ServerBuilder;
79use jsonrpsee::server::ServerConfigBuilder;
80pub use reth_ipc::server::{
81 Builder as IpcServerBuilder, RpcServiceBuilder as IpcRpcServiceBuilder,
82};
83pub use reth_rpc_server_types::{constants, RpcModuleSelection};
84pub use tower::layer::util::{Identity, Stack};
85
86pub mod auth;
88
89pub mod config;
91
92pub mod middleware;
94
95mod cors;
97
98pub mod error;
100
101pub mod eth;
103pub use eth::EthHandlers;
104
105mod metrics;
107use crate::middleware::RethRpcMiddleware;
108pub use metrics::{MeteredBatchRequestsFuture, MeteredRequestFuture, RpcRequestMetricsService};
109use reth_chain_state::{
110 CanonStateSubscriptions, ForkChoiceSubscriptions, PersistedBlockSubscriptions,
111};
112use reth_rpc::eth::sim_bundle::EthSimBundle;
113
114pub mod rate_limiter;
116
117#[derive(Debug, Clone)]
121pub struct RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, Consensus> {
122 provider: Provider,
124 pool: Pool,
126 network: Network,
128 executor: Option<Runtime>,
130 evm_config: EvmConfig,
132 consensus: Consensus,
134 _primitives: PhantomData<N>,
136}
137
138impl<N, Provider, Pool, Network, EvmConfig, Consensus>
141 RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, Consensus>
142{
143 pub const fn new(
145 provider: Provider,
146 pool: Pool,
147 network: Network,
148 executor: Runtime,
149 evm_config: EvmConfig,
150 consensus: Consensus,
151 ) -> Self {
152 Self {
153 provider,
154 pool,
155 network,
156 executor: Some(executor),
157 evm_config,
158 consensus,
159 _primitives: PhantomData,
160 }
161 }
162
163 pub fn with_provider<P>(
165 self,
166 provider: P,
167 ) -> RpcModuleBuilder<N, P, Pool, Network, EvmConfig, Consensus> {
168 let Self { pool, network, executor, evm_config, consensus, _primitives, .. } = self;
169 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
170 }
171
172 pub fn with_pool<P>(
174 self,
175 pool: P,
176 ) -> RpcModuleBuilder<N, Provider, P, Network, EvmConfig, Consensus> {
177 let Self { provider, network, executor, evm_config, consensus, _primitives, .. } = self;
178 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
179 }
180
181 pub fn with_noop_pool(
187 self,
188 ) -> RpcModuleBuilder<N, Provider, NoopTransactionPool, Network, EvmConfig, Consensus> {
189 let Self { provider, executor, network, evm_config, consensus, _primitives, .. } = self;
190 RpcModuleBuilder {
191 provider,
192 executor,
193 network,
194 evm_config,
195 pool: NoopTransactionPool::default(),
196 consensus,
197 _primitives,
198 }
199 }
200
201 pub fn with_network<Net>(
203 self,
204 network: Net,
205 ) -> RpcModuleBuilder<N, Provider, Pool, Net, EvmConfig, Consensus> {
206 let Self { provider, pool, executor, evm_config, consensus, _primitives, .. } = self;
207 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
208 }
209
210 pub fn with_noop_network(
216 self,
217 ) -> RpcModuleBuilder<N, Provider, Pool, NoopNetwork, EvmConfig, Consensus> {
218 let Self { provider, pool, executor, evm_config, consensus, _primitives, .. } = self;
219 RpcModuleBuilder {
220 provider,
221 pool,
222 executor,
223 network: NoopNetwork::default(),
224 evm_config,
225 consensus,
226 _primitives,
227 }
228 }
229
230 pub fn with_executor(self, executor: Runtime) -> Self {
232 let Self { pool, network, provider, evm_config, consensus, _primitives, .. } = self;
233 Self {
234 provider,
235 network,
236 pool,
237 executor: Some(executor),
238 evm_config,
239 consensus,
240 _primitives,
241 }
242 }
243
244 pub fn with_evm_config<E>(
246 self,
247 evm_config: E,
248 ) -> RpcModuleBuilder<N, Provider, Pool, Network, E, Consensus> {
249 let Self { provider, pool, executor, network, consensus, _primitives, .. } = self;
250 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
251 }
252
253 pub fn with_consensus<C>(
255 self,
256 consensus: C,
257 ) -> RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, C> {
258 let Self { provider, network, pool, executor, evm_config, _primitives, .. } = self;
259 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
260 }
261
262 #[expect(clippy::type_complexity)]
264 pub fn eth_api_builder<ChainSpec>(
265 &self,
266 ) -> EthApiBuilder<
267 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>,
268 RpcConverter<Ethereum, EvmConfig, EthReceiptConverter<ChainSpec>>,
269 >
270 where
271 Provider: Clone,
272 Pool: Clone,
273 Network: Clone,
274 EvmConfig: Clone,
275 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>:
276 RpcNodeCore<Provider: ChainSpecProvider<ChainSpec = ChainSpec>, Evm = EvmConfig>,
277 {
278 EthApiBuilder::new(
279 self.provider.clone(),
280 self.pool.clone(),
281 self.network.clone(),
282 self.evm_config.clone(),
283 )
284 }
285
286 #[expect(clippy::type_complexity)]
292 pub fn bootstrap_eth_api<ChainSpec>(
293 &self,
294 ) -> EthApi<
295 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>,
296 RpcConverter<Ethereum, EvmConfig, EthReceiptConverter<ChainSpec>>,
297 >
298 where
299 Provider: Clone,
300 Pool: Clone,
301 Network: Clone,
302 EvmConfig: ConfigureEvm + Clone,
303 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>:
304 RpcNodeCore<Provider: ChainSpecProvider<ChainSpec = ChainSpec>, Evm = EvmConfig>,
305 RpcConverter<Ethereum, EvmConfig, EthReceiptConverter<ChainSpec>>: RpcConvert,
306 (): PendingEnvBuilder<EvmConfig>,
307 {
308 self.eth_api_builder().build()
309 }
310}
311
312impl<N, Provider, Pool, Network, EvmConfig, Consensus>
313 RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, Consensus>
314where
315 N: NodePrimitives,
316 Provider: FullRpcProvider<Block = N::Block, Receipt = N::Receipt, Header = N::BlockHeader>
317 + CanonStateSubscriptions<Primitives = N>
318 + ForkChoiceSubscriptions<Header = N::BlockHeader>
319 + PersistedBlockSubscriptions
320 + ChangeSetReader,
321 Pool: TransactionPool + Clone + 'static,
322 Network: NetworkInfo + Peers + Clone + 'static,
323 EvmConfig: ConfigureEvm<Primitives = N> + 'static,
324 Consensus: FullConsensus<N> + Clone + 'static,
325{
326 pub fn build_with_auth_server<EthApi, Payload>(
333 self,
334 module_config: TransportRpcModuleConfig,
335 engine: impl IntoEngineApiRpcModule,
336 eth: EthApi,
337 engine_events: EventSender<ConsensusEngineEvent<N>>,
338 beacon_engine_handle: ConsensusEngineHandle<Payload>,
339 ) -> (
340 TransportRpcModules,
341 AuthRpcModule,
342 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>,
343 )
344 where
345 EthApi: FullEthApiServer<Provider = Provider, Pool = Pool>,
346 Payload: PayloadTypes,
347 {
348 let config = module_config.config.clone().unwrap_or_default();
349
350 let mut registry = self.into_registry(config, eth, engine_events);
351 let modules = registry.create_transport_rpc_modules(module_config);
352 let auth_module = registry.create_auth_module(engine, beacon_engine_handle);
353
354 (modules, auth_module, registry)
355 }
356
357 pub fn into_registry<EthApi>(
362 self,
363 config: RpcModuleConfig,
364 eth: EthApi,
365 engine_events: EventSender<ConsensusEngineEvent<N>>,
366 ) -> RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
367 where
368 EthApi: FullEthApiServer<Provider = Provider, Pool = Pool>,
369 {
370 let Self { provider, pool, network, executor, consensus, evm_config, .. } = self;
371 let executor =
372 executor.expect("RpcModuleBuilder requires a Runtime to be set via `with_executor`");
373 RpcRegistryInner::new(
374 provider,
375 pool,
376 network,
377 executor,
378 consensus,
379 config,
380 evm_config,
381 eth,
382 engine_events,
383 )
384 }
385
386 pub fn build<EthApi>(
389 self,
390 module_config: TransportRpcModuleConfig,
391 eth: EthApi,
392 engine_events: EventSender<ConsensusEngineEvent<N>>,
393 ) -> TransportRpcModules<()>
394 where
395 EthApi: FullEthApiServer<Provider = Provider, Pool = Pool>,
396 {
397 if module_config.is_empty() {
398 TransportRpcModules::default()
399 } else {
400 let config = module_config.config.clone().unwrap_or_default();
401 let mut registry = self.into_registry(config, eth, engine_events);
402 registry.create_transport_rpc_modules(module_config)
403 }
404 }
405}
406
407impl<N: NodePrimitives> Default for RpcModuleBuilder<N, (), (), (), (), ()> {
408 fn default() -> Self {
409 Self {
410 provider: (),
411 pool: (),
412 network: (),
413 executor: None,
414 evm_config: (),
415 consensus: (),
416 _primitives: PhantomData,
417 }
418 }
419}
420
421#[derive(Debug, Default, Clone, Eq, PartialEq, Serialize, Deserialize)]
423pub struct RpcModuleConfig {
424 eth: EthConfig,
426}
427
428impl RpcModuleConfig {
431 pub fn builder() -> RpcModuleConfigBuilder {
433 RpcModuleConfigBuilder::default()
434 }
435
436 pub const fn new(eth: EthConfig) -> Self {
438 Self { eth }
439 }
440
441 pub const fn eth(&self) -> &EthConfig {
443 &self.eth
444 }
445
446 pub const fn eth_mut(&mut self) -> &mut EthConfig {
448 &mut self.eth
449 }
450}
451
452#[derive(Clone, Debug, Default)]
454pub struct RpcModuleConfigBuilder {
455 eth: Option<EthConfig>,
456}
457
458impl RpcModuleConfigBuilder {
461 pub fn eth(mut self, eth: EthConfig) -> Self {
463 self.eth = Some(eth);
464 self
465 }
466
467 pub fn build(self) -> RpcModuleConfig {
469 let Self { eth } = self;
470 RpcModuleConfig { eth: eth.unwrap_or_default() }
471 }
472
473 pub const fn get_eth(&self) -> Option<&EthConfig> {
475 self.eth.as_ref()
476 }
477
478 pub const fn eth_mut(&mut self) -> &mut Option<EthConfig> {
480 &mut self.eth
481 }
482
483 pub fn eth_mut_or_default(&mut self) -> &mut EthConfig {
485 self.eth.get_or_insert_with(EthConfig::default)
486 }
487}
488
489#[derive(Debug)]
491pub struct RpcRegistryInner<Provider, Pool, Network, EthApi: EthApiTypes, EvmConfig, Consensus> {
492 provider: Provider,
493 pool: Pool,
494 network: Network,
495 executor: Runtime,
496 evm_config: EvmConfig,
497 consensus: Consensus,
498 eth: EthHandlers<EthApi>,
500 blocking_pool_guard: BlockingTaskGuard,
502 modules: HashMap<RethRpcModule, Methods>,
504 eth_config: EthConfig,
506 engine_events:
508 EventSender<ConsensusEngineEvent<<EthApi::RpcConvert as RpcConvert>::Primitives>>,
509}
510
511impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
514 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
515where
516 N: NodePrimitives,
517 Provider: StateProviderFactory
518 + CanonStateSubscriptions<Primitives = N>
519 + BlockReader<Block = N::Block, Receipt = N::Receipt>
520 + Clone
521 + Unpin
522 + 'static,
523 Pool: Send + Sync + Clone + 'static,
524 Network: Clone + 'static,
525 EthApi: FullEthApiTypes + 'static,
526 EvmConfig: ConfigureEvm<Primitives = N>,
527{
528 #[expect(clippy::too_many_arguments)]
530 pub fn new(
531 provider: Provider,
532 pool: Pool,
533 network: Network,
534 executor: Runtime,
535 consensus: Consensus,
536 config: RpcModuleConfig,
537 evm_config: EvmConfig,
538 eth_api: EthApi,
539 engine_events: EventSender<
540 ConsensusEngineEvent<<EthApi::Provider as NodePrimitivesProvider>::Primitives>,
541 >,
542 ) -> Self
543 where
544 EvmConfig: ConfigureEvm<Primitives = N>,
545 {
546 let blocking_pool_guard = BlockingTaskGuard::new(config.eth.max_tracing_requests);
547
548 let eth = EthHandlers::bootstrap(config.eth.clone(), executor.clone(), eth_api);
549
550 Self {
551 provider,
552 pool,
553 network,
554 eth,
555 executor,
556 consensus,
557 modules: Default::default(),
558 blocking_pool_guard,
559 eth_config: config.eth,
560 evm_config,
561 engine_events,
562 }
563 }
564}
565
566impl<Provider, Pool, Network, EthApi, Evm, Consensus>
567 RpcRegistryInner<Provider, Pool, Network, EthApi, Evm, Consensus>
568where
569 EthApi: EthApiTypes,
570{
571 pub const fn eth_api(&self) -> &EthApi {
573 &self.eth.api
574 }
575
576 pub const fn eth_handlers(&self) -> &EthHandlers<EthApi> {
578 &self.eth
579 }
580
581 pub const fn pool(&self) -> &Pool {
583 &self.pool
584 }
585
586 pub const fn tasks(&self) -> &Runtime {
588 &self.executor
589 }
590
591 pub const fn provider(&self) -> &Provider {
593 &self.provider
594 }
595
596 pub const fn evm_config(&self) -> &Evm {
598 &self.evm_config
599 }
600
601 pub fn methods(&self) -> Vec<Methods> {
603 self.modules.values().cloned().collect()
604 }
605
606 pub fn module(&self) -> RpcModule<()> {
608 let mut module = RpcModule::new(());
609 for methods in self.modules.values().cloned() {
610 module.merge(methods).expect("No conflicts");
611 }
612 module
613 }
614}
615
616impl<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
617 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
618where
619 Network: NetworkInfo + Clone + 'static,
620 EthApi: EthApiTypes,
621 Provider: BlockReader + ChainSpecProvider<ChainSpec: EthereumHardforks>,
622 EvmConfig: ConfigureEvm,
623{
624 pub fn admin_api(&self) -> AdminApi<Network, Provider::ChainSpec, Pool>
626 where
627 Network: Peers,
628 Pool: TransactionPool + Clone + 'static,
629 {
630 AdminApi::new(self.network.clone(), self.provider.chain_spec(), self.pool.clone())
631 }
632
633 pub fn web3_api(&self) -> Web3Api<Network> {
635 Web3Api::new(self.network.clone())
636 }
637
638 pub fn register_admin(&mut self) -> &mut Self
640 where
641 Network: Peers,
642 Pool: TransactionPool + Clone + 'static,
643 {
644 let adminapi = self.admin_api();
645 self.modules.insert(RethRpcModule::Admin, adminapi.into_rpc().into());
646 self
647 }
648
649 pub fn register_web3(&mut self) -> &mut Self {
651 let web3api = self.web3_api();
652 self.modules.insert(RethRpcModule::Web3, web3api.into_rpc().into());
653 self
654 }
655}
656
657impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
658 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
659where
660 N: NodePrimitives,
661 Provider: FullRpcProvider<
662 Header = N::BlockHeader,
663 Block = N::Block,
664 Receipt = N::Receipt,
665 Transaction = N::SignedTx,
666 > + ChangeSetReader
667 + CanonStateSubscriptions<Primitives = N>
668 + ForkChoiceSubscriptions<Header = N::BlockHeader>
669 + PersistedBlockSubscriptions,
670 Network: NetworkInfo + Peers + Clone + 'static,
671 EthApi: EthApiServer<
672 RpcTxReq<EthApi::NetworkTypes>,
673 RpcTransaction<EthApi::NetworkTypes>,
674 RpcBlock<EthApi::NetworkTypes>,
675 RpcReceipt<EthApi::NetworkTypes>,
676 RpcHeader<EthApi::NetworkTypes>,
677 TxTy<N>,
678 > + EthApiTypes,
679 EvmConfig: ConfigureEvm<Primitives = N> + 'static,
680{
681 pub fn register_eth(&mut self) -> &mut Self {
687 let eth_api = self.eth_api().clone();
688 self.modules.insert(RethRpcModule::Eth, eth_api.into_rpc().into());
689 self
690 }
691
692 pub fn register_ots(&mut self) -> &mut Self
698 where
699 EthApi: TraceExt + EthTransactions<Primitives = N>,
700 {
701 let otterscan_api = self.otterscan_api();
702 self.modules.insert(RethRpcModule::Ots, otterscan_api.into_rpc().into());
703 self
704 }
705
706 pub fn register_debug(&mut self) -> &mut Self
712 where
713 EthApi: EthTransactions + TraceExt,
714 {
715 let debug_api = self.debug_api();
716 self.modules.insert(RethRpcModule::Debug, debug_api.into_rpc().into());
717 self
718 }
719
720 pub fn register_trace(&mut self) -> &mut Self
726 where
727 EthApi: TraceExt,
728 {
729 let trace_api = self.trace_api();
730 self.modules.insert(RethRpcModule::Trace, trace_api.into_rpc().into());
731 self
732 }
733
734 pub fn register_net(&mut self) -> &mut Self
742 where
743 EthApi: EthApiSpec + 'static,
744 {
745 let netapi = self.net_api();
746 self.modules.insert(RethRpcModule::Net, netapi.into_rpc().into());
747 self
748 }
749
750 pub fn register_reth(&mut self) -> &mut Self {
758 let rethapi = self.reth_api();
759 self.modules.insert(RethRpcModule::Reth, rethapi.into_rpc().into());
760 self
761 }
762
763 pub fn otterscan_api(&self) -> OtterscanApi<EthApi> {
769 let eth_api = self.eth_api().clone();
770 OtterscanApi::new(eth_api)
771 }
772}
773
774impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
775 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
776where
777 N: NodePrimitives,
778 Provider: FullRpcProvider<
779 Block = N::Block,
780 Header = N::BlockHeader,
781 Transaction = N::SignedTx,
782 Receipt = N::Receipt,
783 > + ChangeSetReader,
784 Network: NetworkInfo + Peers + Clone + 'static,
785 EthApi: EthApiTypes,
786 EvmConfig: ConfigureEvm<Primitives = N>,
787{
788 pub fn trace_api(&self) -> TraceApi<EthApi> {
794 TraceApi::new(
795 self.eth_api().clone(),
796 self.blocking_pool_guard.clone(),
797 self.eth_config.clone(),
798 )
799 }
800
801 pub fn bundle_api(&self) -> EthBundle<EthApi>
807 where
808 EthApi: EthTransactions + LoadPendingBlock + Call,
809 {
810 let eth_api = self.eth_api().clone();
811 EthBundle::new(eth_api, self.blocking_pool_guard.clone())
812 }
813
814 pub fn debug_api(&self) -> DebugApi<EthApi>
820 where
821 EthApi: FullEthApiTypes,
822 {
823 DebugApi::new(
824 self.eth_api().clone(),
825 self.blocking_pool_guard.clone(),
826 self.tasks(),
827 self.engine_events.new_listener(),
828 )
829 }
830
831 pub fn net_api(&self) -> NetApi<Network, EthApi>
837 where
838 EthApi: EthApiSpec + 'static,
839 {
840 let eth_api = self.eth_api().clone();
841 NetApi::new(self.network.clone(), eth_api)
842 }
843
844 pub fn reth_api(&self) -> RethApi<Provider, EvmConfig> {
846 RethApi::new(
847 self.provider.clone(),
848 self.evm_config.clone(),
849 self.blocking_pool_guard.clone(),
850 self.executor.clone(),
851 )
852 }
853}
854
855impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
856 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
857where
858 N: NodePrimitives,
859 Provider: FullRpcProvider<Block = N::Block>
860 + CanonStateSubscriptions<Primitives = N>
861 + ForkChoiceSubscriptions<Header = N::BlockHeader>
862 + PersistedBlockSubscriptions
863 + ChangeSetReader,
864 Pool: TransactionPool + Clone + 'static,
865 Network: NetworkInfo + Peers + Clone + 'static,
866 EthApi: FullEthApiServer,
867 EvmConfig: ConfigureEvm<Primitives = N> + 'static,
868 Consensus: FullConsensus<N> + Clone + 'static,
869{
870 pub fn create_auth_module<Payload>(
877 &self,
878 engine_api: impl IntoEngineApiRpcModule,
879 beacon_engine_handle: ConsensusEngineHandle<Payload>,
880 ) -> AuthRpcModule
881 where
882 Payload: PayloadTypes,
883 {
884 let mut module = engine_api.into_rpc_module();
885
886 let reth_engine_api = RethEngineApi::new(beacon_engine_handle);
888 module
889 .merge(RethEngineApiServer::into_rpc(reth_engine_api).remove_context())
890 .expect("No conflicting methods");
891
892 let eth_handlers = self.eth_handlers();
894 let engine_eth = EngineEthApi::new(eth_handlers.api.clone(), eth_handlers.filter.clone());
895
896 module.merge(engine_eth.into_rpc()).expect("No conflicting methods");
897
898 AuthRpcModule { inner: module }
899 }
900
901 fn maybe_module(&mut self, config: Option<&RpcModuleSelection>) -> Option<RpcModule<()>> {
903 config.map(|config| self.module_for(config))
904 }
905
906 pub fn create_transport_rpc_modules(
910 &mut self,
911 config: TransportRpcModuleConfig,
912 ) -> TransportRpcModules<()> {
913 let mut modules = TransportRpcModules::default();
914 let http = self.maybe_module(config.http.as_ref());
915 let ws = self.maybe_module(config.ws.as_ref());
916 let ipc = self.maybe_module(config.ipc.as_ref());
917
918 modules.config = config;
919 modules.http = http;
920 modules.ws = ws;
921 modules.ipc = ipc;
922 modules
923 }
924
925 pub fn module_for(&mut self, config: &RpcModuleSelection) -> RpcModule<()> {
928 let mut module = RpcModule::new(());
929 let all_methods = self.reth_methods(config.iter_selection());
930 for methods in all_methods {
931 module.merge(methods).expect("No conflicts");
932 }
933 module
934 }
935
936 pub fn reth_methods(
945 &mut self,
946 namespaces: impl Iterator<Item = RethRpcModule>,
947 ) -> Vec<Methods> {
948 let EthHandlers { api: eth_api, filter: eth_filter, pubsub: eth_pubsub, .. } =
949 self.eth_handlers().clone();
950
951 let namespaces: Vec<_> = namespaces.collect();
953 namespaces
954 .iter()
955 .map(|namespace| {
956 self.modules
957 .entry(namespace.clone())
958 .or_insert_with(|| match namespace.clone() {
959 RethRpcModule::Admin => AdminApi::new(
960 self.network.clone(),
961 self.provider.chain_spec(),
962 self.pool.clone(),
963 )
964 .into_rpc()
965 .into(),
966 RethRpcModule::Debug => DebugApi::new(
967 eth_api.clone(),
968 self.blocking_pool_guard.clone(),
969 &self.executor,
970 self.engine_events.new_listener(),
971 )
972 .into_rpc()
973 .into(),
974 RethRpcModule::Eth => {
975 let mut module = eth_api.clone().into_rpc();
977 module.merge(eth_filter.clone().into_rpc()).expect("No conflicts");
978 module.merge(eth_pubsub.clone().into_rpc()).expect("No conflicts");
979 module
980 .merge(
981 EthBundle::new(
982 eth_api.clone(),
983 self.blocking_pool_guard.clone(),
984 )
985 .into_rpc(),
986 )
987 .expect("No conflicts");
988
989 module.into()
990 }
991 RethRpcModule::Net => {
992 NetApi::new(self.network.clone(), eth_api.clone()).into_rpc().into()
993 }
994 RethRpcModule::Trace => TraceApi::new(
995 eth_api.clone(),
996 self.blocking_pool_guard.clone(),
997 self.eth_config.clone(),
998 )
999 .into_rpc()
1000 .into(),
1001 RethRpcModule::Web3 => Web3Api::new(self.network.clone()).into_rpc().into(),
1002 RethRpcModule::Txpool => TxPoolApi::new(
1003 self.eth.api.pool().clone(),
1004 dyn_clone::clone(self.eth.api.converter()),
1005 )
1006 .into_rpc()
1007 .into(),
1008 RethRpcModule::Rpc => RPCApi::new(
1009 namespaces
1010 .iter()
1011 .map(|module| (module.to_string(), "1.0".to_string()))
1012 .collect(),
1013 )
1014 .into_rpc()
1015 .into(),
1016 RethRpcModule::Ots => OtterscanApi::new(eth_api.clone()).into_rpc().into(),
1017 RethRpcModule::Reth => RethApi::new(
1018 self.provider.clone(),
1019 self.evm_config.clone(),
1020 self.blocking_pool_guard.clone(),
1021 self.executor.clone(),
1022 )
1023 .into_rpc()
1024 .into(),
1025 RethRpcModule::Miner => MinerApi::default().into_rpc().into(),
1026 RethRpcModule::Mev => {
1027 EthSimBundle::new(eth_api.clone(), self.blocking_pool_guard.clone())
1028 .into_rpc()
1029 .into()
1030 }
1031 RethRpcModule::Flashbots |
1035 RethRpcModule::Testing |
1036 RethRpcModule::Other(_) => Default::default(),
1037 })
1038 .clone()
1039 })
1040 .collect::<Vec<_>>()
1041 }
1042}
1043
1044impl<Provider, Pool, Network, EthApi, EvmConfig, Consensus> Clone
1045 for RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
1046where
1047 EthApi: EthApiTypes,
1048 Provider: Clone,
1049 Pool: Clone,
1050 Network: Clone,
1051 EvmConfig: Clone,
1052 Consensus: Clone,
1053{
1054 fn clone(&self) -> Self {
1055 Self {
1056 provider: self.provider.clone(),
1057 pool: self.pool.clone(),
1058 network: self.network.clone(),
1059 executor: self.executor.clone(),
1060 evm_config: self.evm_config.clone(),
1061 consensus: self.consensus.clone(),
1062 eth: self.eth.clone(),
1063 blocking_pool_guard: self.blocking_pool_guard.clone(),
1064 modules: self.modules.clone(),
1065 eth_config: self.eth_config.clone(),
1066 engine_events: self.engine_events.clone(),
1067 }
1068 }
1069}
1070
1071#[derive(Debug)]
1083pub struct RpcServerConfig<RpcMiddleware = Identity> {
1084 http_server_config: Option<ServerConfigBuilder>,
1086 http_cors_domains: Option<String>,
1088 http_addr: Option<SocketAddr>,
1090 http_disable_compression: bool,
1092 http_compression_algorithms: Option<Vec<String>>,
1096 http_decompression_algorithms: Option<Vec<String>>,
1100 http_max_request_body_size: Option<u32>,
1102 ws_server_config: Option<ServerConfigBuilder>,
1104 ws_cors_domains: Option<String>,
1106 ws_addr: Option<SocketAddr>,
1108 ipc_server_config: Option<IpcServerBuilder<Identity, Identity>>,
1110 ipc_endpoint: Option<String>,
1112 jwt_secret: Option<JwtSecret>,
1114 rpc_metrics_enabled: bool,
1116 rpc_middleware: RpcMiddleware,
1118}
1119
1120impl Default for RpcServerConfig<Identity> {
1123 fn default() -> Self {
1125 Self {
1126 http_server_config: None,
1127 http_cors_domains: None,
1128 http_addr: None,
1129 http_disable_compression: false,
1130 http_compression_algorithms: None,
1131 http_decompression_algorithms: None,
1132 http_max_request_body_size: None,
1133 ws_server_config: None,
1134 ws_cors_domains: None,
1135 ws_addr: None,
1136 ipc_server_config: None,
1137 ipc_endpoint: None,
1138 jwt_secret: None,
1139 rpc_metrics_enabled: true,
1140 rpc_middleware: Default::default(),
1141 }
1142 }
1143}
1144
1145impl RpcServerConfig {
1146 pub fn http(config: ServerConfigBuilder) -> Self {
1148 Self::default().with_http(config)
1149 }
1150
1151 pub fn ws(config: ServerConfigBuilder) -> Self {
1153 Self::default().with_ws(config)
1154 }
1155
1156 pub fn ipc(config: IpcServerBuilder<Identity, Identity>) -> Self {
1158 Self::default().with_ipc(config)
1159 }
1160
1161 pub fn with_http(mut self, config: ServerConfigBuilder) -> Self {
1166 self.http_server_config =
1167 Some(config.set_id_provider(EthSubscriptionIdProvider::default()));
1168 self
1169 }
1170
1171 pub fn with_ws(mut self, config: ServerConfigBuilder) -> Self {
1176 self.ws_server_config = Some(config.set_id_provider(EthSubscriptionIdProvider::default()));
1177 self
1178 }
1179
1180 pub fn with_ipc(mut self, config: IpcServerBuilder<Identity, Identity>) -> Self {
1185 self.ipc_server_config = Some(config.set_id_provider(EthSubscriptionIdProvider::default()));
1186 self
1187 }
1188}
1189
1190impl<RpcMiddleware> RpcServerConfig<RpcMiddleware> {
1191 pub fn set_rpc_middleware<T>(self, rpc_middleware: T) -> RpcServerConfig<T> {
1193 RpcServerConfig {
1194 http_server_config: self.http_server_config,
1195 http_cors_domains: self.http_cors_domains,
1196 http_addr: self.http_addr,
1197 http_disable_compression: self.http_disable_compression,
1198 http_compression_algorithms: self.http_compression_algorithms,
1199 http_decompression_algorithms: self.http_decompression_algorithms,
1200 http_max_request_body_size: self.http_max_request_body_size,
1201 ws_server_config: self.ws_server_config,
1202 ws_cors_domains: self.ws_cors_domains,
1203 ws_addr: self.ws_addr,
1204 ipc_server_config: self.ipc_server_config,
1205 ipc_endpoint: self.ipc_endpoint,
1206 jwt_secret: self.jwt_secret,
1207 rpc_metrics_enabled: self.rpc_metrics_enabled,
1208 rpc_middleware,
1209 }
1210 }
1211
1212 pub const fn with_rpc_metrics_enabled(mut self, enabled: bool) -> Self {
1214 self.rpc_metrics_enabled = enabled;
1215 self
1216 }
1217
1218 pub fn with_cors(self, cors_domain: Option<String>) -> Self {
1220 self.with_http_cors(cors_domain.clone()).with_ws_cors(cors_domain)
1221 }
1222
1223 pub fn with_ws_cors(mut self, cors_domain: Option<String>) -> Self {
1225 self.ws_cors_domains = cors_domain;
1226 self
1227 }
1228
1229 pub fn with_http_cors(mut self, cors_domain: Option<String>) -> Self {
1231 self.http_cors_domains = cors_domain;
1232 self
1233 }
1234
1235 pub const fn with_http_disable_compression(mut self, http_disable_compression: bool) -> Self {
1237 self.http_disable_compression = http_disable_compression;
1238 self
1239 }
1240
1241 pub fn with_http_compression_algorithms(mut self, algos: Option<Vec<String>>) -> Self {
1246 self.http_compression_algorithms = algos;
1247 self
1248 }
1249
1250 pub fn with_http_decompression(mut self, algos: Option<Vec<String>>, max_size: u32) -> Self {
1254 self.http_decompression_algorithms = algos;
1255 self.http_max_request_body_size = Some(max_size);
1256 self
1257 }
1258
1259 pub const fn with_http_address(mut self, addr: SocketAddr) -> Self {
1264 self.http_addr = Some(addr);
1265 self
1266 }
1267
1268 pub const fn with_ws_address(mut self, addr: SocketAddr) -> Self {
1273 self.ws_addr = Some(addr);
1274 self
1275 }
1276
1277 pub fn with_id_provider<I>(mut self, id_provider: I) -> Self
1281 where
1282 I: IdProvider + Clone + 'static,
1283 {
1284 if let Some(config) = self.http_server_config {
1285 self.http_server_config = Some(config.set_id_provider(id_provider.clone()));
1286 }
1287 if let Some(config) = self.ws_server_config {
1288 self.ws_server_config = Some(config.set_id_provider(id_provider.clone()));
1289 }
1290 if let Some(ipc) = self.ipc_server_config {
1291 self.ipc_server_config = Some(ipc.set_id_provider(id_provider));
1292 }
1293
1294 self
1295 }
1296
1297 pub fn with_ipc_endpoint(mut self, path: impl Into<String>) -> Self {
1301 self.ipc_endpoint = Some(path.into());
1302 self
1303 }
1304
1305 pub const fn with_jwt_secret(mut self, secret: Option<JwtSecret>) -> Self {
1307 self.jwt_secret = secret;
1308 self
1309 }
1310
1311 pub fn with_tokio_runtime(mut self, tokio_runtime: Option<tokio::runtime::Handle>) -> Self {
1313 let Some(tokio_runtime) = tokio_runtime else { return self };
1314 if let Some(http_server_config) = self.http_server_config {
1315 self.http_server_config =
1316 Some(http_server_config.custom_tokio_runtime(tokio_runtime.clone()));
1317 }
1318 if let Some(ws_server_config) = self.ws_server_config {
1319 self.ws_server_config =
1320 Some(ws_server_config.custom_tokio_runtime(tokio_runtime.clone()));
1321 }
1322 if let Some(ipc_server_config) = self.ipc_server_config {
1323 self.ipc_server_config = Some(ipc_server_config.custom_tokio_runtime(tokio_runtime));
1324 }
1325 self
1326 }
1327
1328 pub const fn has_server(&self) -> bool {
1332 self.http_server_config.is_some() ||
1333 self.ws_server_config.is_some() ||
1334 self.ipc_server_config.is_some()
1335 }
1336
1337 pub const fn http_address(&self) -> Option<SocketAddr> {
1339 self.http_addr
1340 }
1341
1342 pub const fn ws_address(&self) -> Option<SocketAddr> {
1344 self.ws_addr
1345 }
1346
1347 pub fn ipc_endpoint(&self) -> Option<String> {
1349 self.ipc_endpoint.clone()
1350 }
1351
1352 pub const fn rpc_metrics_enabled(&self) -> bool {
1354 self.rpc_metrics_enabled
1355 }
1356
1357 fn maybe_cors_layer(cors: Option<String>) -> Result<Option<CorsLayer>, CorsDomainError> {
1359 cors.as_deref().map(cors::create_cors_layer).transpose()
1360 }
1361
1362 fn maybe_jwt_layer(jwt_secret: Option<JwtSecret>) -> Option<AuthLayer<JwtAuthValidator>> {
1364 jwt_secret.map(|secret| AuthLayer::new(JwtAuthValidator::new(secret)))
1365 }
1366
1367 fn maybe_compression_layer(
1370 disable_compression: bool,
1371 algos: Option<&[String]>,
1372 ) -> Option<CompressionLayer> {
1373 if disable_compression {
1374 None
1375 } else {
1376 match algos {
1377 None => Some(CompressionLayer::new()),
1379 Some(algos) => Some(CompressionLayer::with_algorithms(algos)),
1380 }
1381 }
1382 }
1383
1384 fn maybe_decompression_layer(
1390 max_request_body_size: Option<u32>,
1391 algos: Option<&[String]>,
1392 ) -> Option<DecompressionLayer> {
1393 let algos = algos?;
1394 if algos.is_empty() {
1395 return None;
1396 }
1397
1398 let max = max_request_body_size?;
1399 Some(DecompressionLayer::new(algos, max as usize))
1400 }
1401
1402 pub async fn start(self, modules: &TransportRpcModules) -> Result<RpcServerHandle, RpcError>
1408 where
1409 RpcMiddleware: RethRpcMiddleware,
1410 {
1411 let mut http_handle = None;
1412 let mut ws_handle = None;
1413 let mut ipc_handle = None;
1414
1415 let http_socket_addr = self.http_addr.unwrap_or(SocketAddr::V4(SocketAddrV4::new(
1416 Ipv4Addr::LOCALHOST,
1417 constants::DEFAULT_HTTP_RPC_PORT,
1418 )));
1419
1420 let ws_socket_addr = self.ws_addr.unwrap_or(SocketAddr::V4(SocketAddrV4::new(
1421 Ipv4Addr::LOCALHOST,
1422 constants::DEFAULT_WS_RPC_PORT,
1423 )));
1424
1425 let rpc_metrics_enabled = self.rpc_metrics_enabled;
1426 let ipc_path =
1427 self.ipc_endpoint.clone().unwrap_or_else(|| constants::DEFAULT_IPC_ENDPOINT.into());
1428
1429 if let Some(builder) = self.ipc_server_config {
1430 let ipc = builder
1431 .set_rpc_middleware(
1432 IpcRpcServiceBuilder::new().option_layer(
1433 rpc_metrics_enabled
1434 .then(|| modules.ipc.as_ref().map(RpcRequestMetrics::ipc))
1435 .flatten(),
1436 ),
1437 )
1438 .build(ipc_path);
1439 ipc_handle = Some(ipc.start(modules.ipc.clone().expect("ipc server error")).await?);
1440 }
1441
1442 if self.http_addr == self.ws_addr &&
1444 self.http_server_config.is_some() &&
1445 self.ws_server_config.is_some()
1446 {
1447 let cors = match (self.ws_cors_domains.as_ref(), self.http_cors_domains.as_ref()) {
1448 (Some(ws_cors), Some(http_cors)) => {
1449 if ws_cors.trim() != http_cors.trim() {
1450 return Err(WsHttpSamePortError::ConflictingCorsDomains {
1451 http_cors_domains: Some(http_cors.clone()),
1452 ws_cors_domains: Some(ws_cors.clone()),
1453 }
1454 .into());
1455 }
1456 Some(ws_cors)
1457 }
1458 (a, b) => a.or(b),
1459 }
1460 .cloned();
1461
1462 modules.config.ensure_ws_http_identical()?;
1464
1465 if let Some(config) = self.http_server_config {
1466 let server = ServerBuilder::new()
1467 .set_http_middleware(
1468 tower::ServiceBuilder::new()
1469 .option_layer(Self::maybe_cors_layer(cors)?)
1470 .option_layer(Self::maybe_jwt_layer(self.jwt_secret))
1471 .option_layer(Self::maybe_decompression_layer(
1472 self.http_max_request_body_size,
1473 self.http_decompression_algorithms.as_deref(),
1474 ))
1475 .option_layer(Self::maybe_compression_layer(
1476 self.http_disable_compression,
1477 self.http_compression_algorithms.as_deref(),
1478 )),
1479 )
1480 .set_rpc_middleware(
1481 RpcServiceBuilder::default()
1482 .option_layer(
1483 rpc_metrics_enabled
1484 .then(|| {
1485 modules
1486 .http
1487 .as_ref()
1488 .or(modules.ws.as_ref())
1489 .map(RpcRequestMetrics::same_port)
1490 })
1491 .flatten(),
1492 )
1493 .layer(self.rpc_middleware.clone()),
1494 )
1495 .set_config(config.build())
1496 .build(http_socket_addr)
1497 .await
1498 .map_err(|err| {
1499 RpcError::server_error(err, ServerKind::WsHttp(http_socket_addr))
1500 })?;
1501 let addr = server.local_addr().map_err(|err| {
1502 RpcError::server_error(err, ServerKind::WsHttp(http_socket_addr))
1503 })?;
1504 if let Some(module) = modules.http.as_ref().or(modules.ws.as_ref()) {
1505 let handle = server.start(module.clone());
1506 http_handle = Some(handle.clone());
1507 ws_handle = Some(handle);
1508 }
1509 return Ok(RpcServerHandle {
1510 http_local_addr: Some(addr),
1511 ws_local_addr: Some(addr),
1512 http: http_handle,
1513 ws: ws_handle,
1514 ipc_endpoint: self.ipc_endpoint.clone(),
1515 ipc: ipc_handle,
1516 jwt_secret: self.jwt_secret,
1517 });
1518 }
1519 }
1520
1521 let mut ws_local_addr = None;
1522 let mut ws_server = None;
1523 let mut http_local_addr = None;
1524 let mut http_server = None;
1525
1526 if let Some(config) = self.ws_server_config {
1527 let server = ServerBuilder::new()
1528 .set_config(config.ws_only().build())
1529 .set_http_middleware(
1530 tower::ServiceBuilder::new()
1531 .option_layer(Self::maybe_cors_layer(self.ws_cors_domains.clone())?)
1532 .option_layer(Self::maybe_jwt_layer(self.jwt_secret)),
1533 )
1534 .set_rpc_middleware(
1535 RpcServiceBuilder::default()
1536 .option_layer(
1537 rpc_metrics_enabled
1538 .then(|| modules.ws.as_ref().map(RpcRequestMetrics::ws))
1539 .flatten(),
1540 )
1541 .layer(self.rpc_middleware.clone()),
1542 )
1543 .build(ws_socket_addr)
1544 .await
1545 .map_err(|err| RpcError::server_error(err, ServerKind::WS(ws_socket_addr)))?;
1546
1547 let addr = server
1548 .local_addr()
1549 .map_err(|err| RpcError::server_error(err, ServerKind::WS(ws_socket_addr)))?;
1550
1551 ws_local_addr = Some(addr);
1552 ws_server = Some(server);
1553 }
1554
1555 if let Some(config) = self.http_server_config {
1556 let server = ServerBuilder::new()
1557 .set_config(config.http_only().build())
1558 .set_http_middleware(
1559 tower::ServiceBuilder::new()
1560 .option_layer(Self::maybe_cors_layer(self.http_cors_domains.clone())?)
1561 .option_layer(Self::maybe_jwt_layer(self.jwt_secret))
1562 .option_layer(Self::maybe_decompression_layer(
1563 self.http_max_request_body_size,
1564 self.http_decompression_algorithms.as_deref(),
1565 ))
1566 .option_layer(Self::maybe_compression_layer(
1567 self.http_disable_compression,
1568 self.http_compression_algorithms.as_deref(),
1569 )),
1570 )
1571 .set_rpc_middleware(
1572 RpcServiceBuilder::default()
1573 .option_layer(
1574 rpc_metrics_enabled
1575 .then(|| modules.http.as_ref().map(RpcRequestMetrics::http))
1576 .flatten(),
1577 )
1578 .layer(self.rpc_middleware.clone()),
1579 )
1580 .build(http_socket_addr)
1581 .await
1582 .map_err(|err| RpcError::server_error(err, ServerKind::Http(http_socket_addr)))?;
1583 let local_addr = server
1584 .local_addr()
1585 .map_err(|err| RpcError::server_error(err, ServerKind::Http(http_socket_addr)))?;
1586 http_local_addr = Some(local_addr);
1587 http_server = Some(server);
1588 }
1589
1590 http_handle = http_server
1591 .map(|http_server| http_server.start(modules.http.clone().expect("http server error")));
1592 ws_handle = ws_server
1593 .map(|ws_server| ws_server.start(modules.ws.clone().expect("ws server error")));
1594 Ok(RpcServerHandle {
1595 http_local_addr,
1596 ws_local_addr,
1597 http: http_handle,
1598 ws: ws_handle,
1599 ipc_endpoint: self.ipc_endpoint.clone(),
1600 ipc: ipc_handle,
1601 jwt_secret: self.jwt_secret,
1602 })
1603 }
1604}
1605
1606#[derive(Debug, Clone, Default, Eq, PartialEq)]
1618pub struct TransportRpcModuleConfig {
1619 http: Option<RpcModuleSelection>,
1621 ws: Option<RpcModuleSelection>,
1623 ipc: Option<RpcModuleSelection>,
1625 config: Option<RpcModuleConfig>,
1627}
1628
1629impl TransportRpcModuleConfig {
1632 pub fn set_http(http: impl Into<RpcModuleSelection>) -> Self {
1634 Self::default().with_http(http)
1635 }
1636
1637 pub fn set_ws(ws: impl Into<RpcModuleSelection>) -> Self {
1639 Self::default().with_ws(ws)
1640 }
1641
1642 pub fn set_ipc(ipc: impl Into<RpcModuleSelection>) -> Self {
1644 Self::default().with_ipc(ipc)
1645 }
1646
1647 pub fn with_http(mut self, http: impl Into<RpcModuleSelection>) -> Self {
1649 self.http = Some(http.into());
1650 self
1651 }
1652
1653 pub fn with_ws(mut self, ws: impl Into<RpcModuleSelection>) -> Self {
1655 self.ws = Some(ws.into());
1656 self
1657 }
1658
1659 pub fn with_ipc(mut self, ipc: impl Into<RpcModuleSelection>) -> Self {
1661 self.ipc = Some(ipc.into());
1662 self
1663 }
1664
1665 pub fn with_config(mut self, config: RpcModuleConfig) -> Self {
1667 self.config = Some(config);
1668 self
1669 }
1670
1671 pub const fn http_mut(&mut self) -> &mut Option<RpcModuleSelection> {
1673 &mut self.http
1674 }
1675
1676 pub const fn ws_mut(&mut self) -> &mut Option<RpcModuleSelection> {
1678 &mut self.ws
1679 }
1680
1681 pub const fn ipc_mut(&mut self) -> &mut Option<RpcModuleSelection> {
1683 &mut self.ipc
1684 }
1685
1686 pub const fn config_mut(&mut self) -> &mut Option<RpcModuleConfig> {
1688 &mut self.config
1689 }
1690
1691 pub const fn is_empty(&self) -> bool {
1693 self.http.is_none() && self.ws.is_none() && self.ipc.is_none()
1694 }
1695
1696 pub const fn http(&self) -> Option<&RpcModuleSelection> {
1698 self.http.as_ref()
1699 }
1700
1701 pub const fn ws(&self) -> Option<&RpcModuleSelection> {
1703 self.ws.as_ref()
1704 }
1705
1706 pub const fn ipc(&self) -> Option<&RpcModuleSelection> {
1708 self.ipc.as_ref()
1709 }
1710
1711 pub const fn config(&self) -> Option<&RpcModuleConfig> {
1713 self.config.as_ref()
1714 }
1715
1716 pub fn contains_any(&self, module: &RethRpcModule) -> bool {
1718 self.contains_http(module) || self.contains_ws(module) || self.contains_ipc(module)
1719 }
1720
1721 pub fn contains_http(&self, module: &RethRpcModule) -> bool {
1723 self.http.as_ref().is_some_and(|http| http.contains(module))
1724 }
1725
1726 pub fn contains_ws(&self, module: &RethRpcModule) -> bool {
1728 self.ws.as_ref().is_some_and(|ws| ws.contains(module))
1729 }
1730
1731 pub fn contains_ipc(&self, module: &RethRpcModule) -> bool {
1733 self.ipc.as_ref().is_some_and(|ipc| ipc.contains(module))
1734 }
1735
1736 fn ensure_ws_http_identical(&self) -> Result<(), WsHttpSamePortError> {
1739 if RpcModuleSelection::are_identical(self.http.as_ref(), self.ws.as_ref()) {
1740 Ok(())
1741 } else {
1742 let http_modules =
1743 self.http.as_ref().map(RpcModuleSelection::to_selection).unwrap_or_default();
1744 let ws_modules =
1745 self.ws.as_ref().map(RpcModuleSelection::to_selection).unwrap_or_default();
1746
1747 let http_not_ws = http_modules.difference(&ws_modules).cloned().collect();
1748 let ws_not_http = ws_modules.difference(&http_modules).cloned().collect();
1749 let overlap = http_modules.intersection(&ws_modules).cloned().collect();
1750
1751 Err(WsHttpSamePortError::ConflictingModules(Box::new(ConflictingModules {
1752 overlap,
1753 http_not_ws,
1754 ws_not_http,
1755 })))
1756 }
1757 }
1758}
1759
1760#[derive(Debug, Clone, Default)]
1762pub struct TransportRpcModules<Context = ()> {
1763 config: TransportRpcModuleConfig,
1765 http: Option<RpcModule<Context>>,
1767 ws: Option<RpcModule<Context>>,
1769 ipc: Option<RpcModule<Context>>,
1771}
1772
1773impl TransportRpcModules {
1776 pub fn with_config(mut self, config: TransportRpcModuleConfig) -> Self {
1779 self.config = config;
1780 self
1781 }
1782
1783 pub fn with_http(mut self, http: RpcModule<()>) -> Self {
1786 self.http = Some(http);
1787 self
1788 }
1789
1790 pub fn with_ws(mut self, ws: RpcModule<()>) -> Self {
1793 self.ws = Some(ws);
1794 self
1795 }
1796
1797 pub fn with_ipc(mut self, ipc: RpcModule<()>) -> Self {
1800 self.ipc = Some(ipc);
1801 self
1802 }
1803
1804 pub const fn module_config(&self) -> &TransportRpcModuleConfig {
1806 &self.config
1807 }
1808
1809 pub fn merge_if_module_configured(
1814 &mut self,
1815 module: RethRpcModule,
1816 other: impl Into<Methods>,
1817 ) -> Result<(), RegisterMethodError> {
1818 let other = other.into();
1819 if self.module_config().contains_http(&module) {
1820 self.merge_http(other.clone())?;
1821 }
1822 if self.module_config().contains_ws(&module) {
1823 self.merge_ws(other.clone())?;
1824 }
1825 if self.module_config().contains_ipc(&module) {
1826 self.merge_ipc(other)?;
1827 }
1828
1829 Ok(())
1830 }
1831
1832 pub fn merge_if_module_configured_with<F>(
1839 &mut self,
1840 module: RethRpcModule,
1841 f: F,
1842 ) -> Result<(), RegisterMethodError>
1843 where
1844 F: FnOnce() -> Methods,
1845 {
1846 if !self.module_config().contains_any(&module) {
1848 return Ok(());
1849 }
1850 self.merge_if_module_configured(module, f())
1851 }
1852
1853 pub fn merge_http(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
1859 if let Some(ref mut http) = self.http {
1860 return http.merge(other.into()).map(|_| true)
1861 }
1862 Ok(false)
1863 }
1864
1865 pub fn merge_ws(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
1871 if let Some(ref mut ws) = self.ws {
1872 return ws.merge(other.into()).map(|_| true)
1873 }
1874 Ok(false)
1875 }
1876
1877 pub fn merge_ipc(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
1883 if let Some(ref mut ipc) = self.ipc {
1884 return ipc.merge(other.into()).map(|_| true)
1885 }
1886 Ok(false)
1887 }
1888
1889 pub fn merge_configured(
1893 &mut self,
1894 other: impl Into<Methods>,
1895 ) -> Result<(), RegisterMethodError> {
1896 let other = other.into();
1897 self.merge_http(other.clone())?;
1898 self.merge_ws(other.clone())?;
1899 self.merge_ipc(other)?;
1900 Ok(())
1901 }
1902
1903 pub fn methods_by_module(&self, module: RethRpcModule) -> Methods {
1907 self.methods_by(|name| name.starts_with(module.as_str()))
1908 }
1909
1910 pub fn methods_by<F>(&self, mut filter: F) -> Methods
1914 where
1915 F: FnMut(&str) -> bool,
1916 {
1917 let mut methods = Methods::new();
1918
1919 let mut f =
1921 |name: &str, mm: &Methods| filter(name) && !mm.method_names().any(|m| m == name);
1922
1923 if let Some(m) = self.http_methods(|name| f(name, &methods)) {
1924 let _ = methods.merge(m);
1925 }
1926 if let Some(m) = self.ws_methods(|name| f(name, &methods)) {
1927 let _ = methods.merge(m);
1928 }
1929 if let Some(m) = self.ipc_methods(|name| f(name, &methods)) {
1930 let _ = methods.merge(m);
1931 }
1932 methods
1933 }
1934
1935 pub fn http_methods<F>(&self, filter: F) -> Option<Methods>
1939 where
1940 F: FnMut(&str) -> bool,
1941 {
1942 self.http.as_ref().map(|module| methods_by(module, filter))
1943 }
1944
1945 pub fn ws_methods<F>(&self, filter: F) -> Option<Methods>
1949 where
1950 F: FnMut(&str) -> bool,
1951 {
1952 self.ws.as_ref().map(|module| methods_by(module, filter))
1953 }
1954
1955 pub fn ipc_methods<F>(&self, filter: F) -> Option<Methods>
1959 where
1960 F: FnMut(&str) -> bool,
1961 {
1962 self.ipc.as_ref().map(|module| methods_by(module, filter))
1963 }
1964
1965 pub fn remove_http_method(&mut self, method_name: &'static str) -> bool {
1973 if let Some(http_module) = &mut self.http {
1974 http_module.remove_method(method_name).is_some()
1975 } else {
1976 false
1977 }
1978 }
1979
1980 pub fn remove_http_methods(&mut self, methods: impl IntoIterator<Item = &'static str>) {
1982 for name in methods {
1983 self.remove_http_method(name);
1984 }
1985 }
1986
1987 pub fn remove_ws_method(&mut self, method_name: &'static str) -> bool {
1995 if let Some(ws_module) = &mut self.ws {
1996 ws_module.remove_method(method_name).is_some()
1997 } else {
1998 false
1999 }
2000 }
2001
2002 pub fn remove_ws_methods(&mut self, methods: impl IntoIterator<Item = &'static str>) {
2004 for name in methods {
2005 self.remove_ws_method(name);
2006 }
2007 }
2008
2009 pub fn remove_ipc_method(&mut self, method_name: &'static str) -> bool {
2017 if let Some(ipc_module) = &mut self.ipc {
2018 ipc_module.remove_method(method_name).is_some()
2019 } else {
2020 false
2021 }
2022 }
2023
2024 pub fn remove_ipc_methods(&mut self, methods: impl IntoIterator<Item = &'static str>) {
2026 for name in methods {
2027 self.remove_ipc_method(name);
2028 }
2029 }
2030
2031 pub fn remove_method_from_configured(&mut self, method_name: &'static str) -> bool {
2035 let http_removed = self.remove_http_method(method_name);
2036 let ws_removed = self.remove_ws_method(method_name);
2037 let ipc_removed = self.remove_ipc_method(method_name);
2038
2039 http_removed || ws_removed || ipc_removed
2040 }
2041
2042 pub fn rename(
2046 &mut self,
2047 old_name: &'static str,
2048 new_method: impl Into<Methods>,
2049 ) -> Result<(), RegisterMethodError> {
2050 self.remove_method_from_configured(old_name);
2052
2053 self.merge_configured(new_method)
2055 }
2056
2057 pub fn replace_http(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
2064 let other = other.into();
2065 self.remove_http_methods(other.method_names());
2066 self.merge_http(other)
2067 }
2068
2069 pub fn replace_ipc(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
2076 let other = other.into();
2077 self.remove_ipc_methods(other.method_names());
2078 self.merge_ipc(other)
2079 }
2080
2081 pub fn replace_ws(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
2088 let other = other.into();
2089 self.remove_ws_methods(other.method_names());
2090 self.merge_ws(other)
2091 }
2092
2093 pub fn replace_configured(
2097 &mut self,
2098 other: impl Into<Methods>,
2099 ) -> Result<bool, RegisterMethodError> {
2100 let other = other.into();
2101 self.replace_http(other.clone())?;
2102 self.replace_ws(other.clone())?;
2103 self.replace_ipc(other)?;
2104 Ok(true)
2105 }
2106
2107 pub fn add_or_replace_http(
2111 &mut self,
2112 other: impl Into<Methods>,
2113 ) -> Result<bool, RegisterMethodError> {
2114 let other = other.into();
2115 self.remove_http_methods(other.method_names());
2116 self.merge_http(other)
2117 }
2118
2119 pub fn add_or_replace_ws(
2123 &mut self,
2124 other: impl Into<Methods>,
2125 ) -> Result<bool, RegisterMethodError> {
2126 let other = other.into();
2127 self.remove_ws_methods(other.method_names());
2128 self.merge_ws(other)
2129 }
2130
2131 pub fn add_or_replace_ipc(
2135 &mut self,
2136 other: impl Into<Methods>,
2137 ) -> Result<bool, RegisterMethodError> {
2138 let other = other.into();
2139 self.remove_ipc_methods(other.method_names());
2140 self.merge_ipc(other)
2141 }
2142
2143 pub fn add_or_replace_configured(
2145 &mut self,
2146 other: impl Into<Methods>,
2147 ) -> Result<(), RegisterMethodError> {
2148 let other = other.into();
2149 self.add_or_replace_http(other.clone())?;
2150 self.add_or_replace_ws(other.clone())?;
2151 self.add_or_replace_ipc(other)?;
2152 Ok(())
2153 }
2154 pub fn add_or_replace_if_module_configured(
2157 &mut self,
2158 module: RethRpcModule,
2159 other: impl Into<Methods>,
2160 ) -> Result<(), RegisterMethodError> {
2161 let other = other.into();
2162 if self.module_config().contains_http(&module) {
2163 self.add_or_replace_http(other.clone())?;
2164 }
2165 if self.module_config().contains_ws(&module) {
2166 self.add_or_replace_ws(other.clone())?;
2167 }
2168 if self.module_config().contains_ipc(&module) {
2169 self.add_or_replace_ipc(other)?;
2170 }
2171 Ok(())
2172 }
2173}
2174
2175fn methods_by<T, F>(module: &RpcModule<T>, mut filter: F) -> Methods
2177where
2178 F: FnMut(&str) -> bool,
2179{
2180 let mut methods = Methods::new();
2181 let method_names = module.method_names().filter(|name| filter(name));
2182
2183 for name in method_names {
2184 if let Some(matched_method) = module.method(name).cloned() {
2185 let _ = methods.verify_and_insert(name, matched_method);
2186 }
2187 }
2188
2189 methods
2190}
2191
2192#[derive(Clone, Debug)]
2197#[must_use = "Server stops if dropped"]
2198pub struct RpcServerHandle {
2199 http_local_addr: Option<SocketAddr>,
2201 ws_local_addr: Option<SocketAddr>,
2202 http: Option<ServerHandle>,
2203 ws: Option<ServerHandle>,
2204 ipc_endpoint: Option<String>,
2205 ipc: Option<jsonrpsee::server::ServerHandle>,
2206 jwt_secret: Option<JwtSecret>,
2207}
2208
2209impl RpcServerHandle {
2212 fn bearer_token(&self) -> Option<String> {
2214 self.jwt_secret.as_ref().map(|secret| {
2215 format!(
2216 "Bearer {}",
2217 secret
2218 .encode(&Claims {
2219 iat: (SystemTime::now().duration_since(UNIX_EPOCH).unwrap() +
2220 Duration::from_secs(60))
2221 .as_secs(),
2222 exp: None,
2223 })
2224 .unwrap()
2225 )
2226 })
2227 }
2228 pub const fn http_local_addr(&self) -> Option<SocketAddr> {
2230 self.http_local_addr
2231 }
2232
2233 pub const fn ws_local_addr(&self) -> Option<SocketAddr> {
2235 self.ws_local_addr
2236 }
2237
2238 pub fn stop(self) -> Result<(), AlreadyStoppedError> {
2240 if let Some(handle) = self.http {
2241 handle.stop()?
2242 }
2243
2244 if let Some(handle) = self.ws {
2245 handle.stop()?
2246 }
2247
2248 if let Some(handle) = self.ipc {
2249 handle.stop()?
2250 }
2251
2252 Ok(())
2253 }
2254
2255 pub fn ipc_endpoint(&self) -> Option<String> {
2257 self.ipc_endpoint.clone()
2258 }
2259
2260 pub fn http_url(&self) -> Option<String> {
2262 self.http_local_addr.map(|addr| format!("http://{addr}"))
2263 }
2264
2265 pub fn ws_url(&self) -> Option<String> {
2267 self.ws_local_addr.map(|addr| format!("ws://{addr}"))
2268 }
2269
2270 pub fn http_client(&self) -> Option<jsonrpsee::http_client::HttpClient> {
2272 let url = self.http_url()?;
2273
2274 let client = if let Some(token) = self.bearer_token() {
2275 jsonrpsee::http_client::HttpClientBuilder::default()
2276 .set_headers(HeaderMap::from_iter([(AUTHORIZATION, token.parse().unwrap())]))
2277 .build(url)
2278 } else {
2279 jsonrpsee::http_client::HttpClientBuilder::default().build(url)
2280 };
2281
2282 client.expect("failed to create http client").into()
2283 }
2284
2285 pub async fn ws_client(&self) -> Option<jsonrpsee::ws_client::WsClient> {
2287 let url = self.ws_url()?;
2288 let mut builder = jsonrpsee::ws_client::WsClientBuilder::default();
2289
2290 if let Some(token) = self.bearer_token() {
2291 let headers = HeaderMap::from_iter([(AUTHORIZATION, token.parse().unwrap())]);
2292 builder = builder.set_headers(headers);
2293 }
2294
2295 let client = builder.build(url).await.expect("failed to create ws client");
2296 Some(client)
2297 }
2298
2299 pub fn eth_http_provider(
2301 &self,
2302 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static> {
2303 self.new_http_provider_for()
2304 }
2305
2306 pub fn eth_http_provider_with_wallet<W>(
2309 &self,
2310 wallet: W,
2311 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static>
2312 where
2313 W: IntoWallet<alloy_network::Ethereum, NetworkWallet: Clone + Unpin + 'static>,
2314 {
2315 let rpc_url = self.http_url()?;
2316 let provider =
2317 ProviderBuilder::new().wallet(wallet).connect_http(rpc_url.parse().expect("valid url"));
2318 Some(provider)
2319 }
2320
2321 pub fn new_http_provider_for<N>(&self) -> Option<impl Provider<N> + Clone + Unpin + 'static>
2326 where
2327 N: RecommendedFillers<RecommendedFillers: Unpin>,
2328 {
2329 let rpc_url = self.http_url()?;
2330 let provider = ProviderBuilder::default()
2331 .with_recommended_fillers()
2332 .connect_http(rpc_url.parse().expect("valid url"));
2333 Some(provider)
2334 }
2335
2336 pub async fn eth_ws_provider(
2338 &self,
2339 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static> {
2340 self.new_ws_provider_for().await
2341 }
2342
2343 pub async fn eth_ws_provider_with_wallet<W>(
2346 &self,
2347 wallet: W,
2348 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static>
2349 where
2350 W: IntoWallet<alloy_network::Ethereum, NetworkWallet: Clone + Unpin + 'static>,
2351 {
2352 let rpc_url = self.ws_url()?;
2353 let provider = ProviderBuilder::new()
2354 .wallet(wallet)
2355 .connect(&rpc_url)
2356 .await
2357 .expect("failed to create ws client");
2358 Some(provider)
2359 }
2360
2361 pub async fn new_ws_provider_for<N>(&self) -> Option<impl Provider<N> + Clone + Unpin + 'static>
2366 where
2367 N: RecommendedFillers<RecommendedFillers: Unpin>,
2368 {
2369 let rpc_url = self.ws_url()?;
2370 let provider = ProviderBuilder::default()
2371 .with_recommended_fillers()
2372 .connect(&rpc_url)
2373 .await
2374 .expect("failed to create ws client");
2375 Some(provider)
2376 }
2377
2378 pub async fn eth_ipc_provider(
2380 &self,
2381 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static> {
2382 self.new_ipc_provider_for().await
2383 }
2384
2385 pub async fn new_ipc_provider_for<N>(
2390 &self,
2391 ) -> Option<impl Provider<N> + Clone + Unpin + 'static>
2392 where
2393 N: RecommendedFillers<RecommendedFillers: Unpin>,
2394 {
2395 let rpc_url = self.ipc_endpoint()?;
2396 let provider = ProviderBuilder::default()
2397 .with_recommended_fillers()
2398 .connect(&rpc_url)
2399 .await
2400 .expect("failed to create ipc client");
2401 Some(provider)
2402 }
2403}
2404
2405#[cfg(test)]
2406mod tests {
2407 use super::*;
2408
2409 #[test]
2410 fn parse_eth_call_bundle_selection() {
2411 let selection = "eth,admin,debug".parse::<RpcModuleSelection>().unwrap();
2412 assert_eq!(
2413 selection,
2414 RpcModuleSelection::Selection(
2415 [RethRpcModule::Eth, RethRpcModule::Admin, RethRpcModule::Debug,].into()
2416 )
2417 );
2418 }
2419
2420 #[test]
2421 fn parse_rpc_module_selection() {
2422 let selection = "all".parse::<RpcModuleSelection>().unwrap();
2423 assert_eq!(selection, RpcModuleSelection::All);
2424 }
2425
2426 #[test]
2427 fn parse_rpc_module_selection_none() {
2428 let selection = "none".parse::<RpcModuleSelection>().unwrap();
2429 assert_eq!(selection, RpcModuleSelection::Selection(Default::default()));
2430 }
2431
2432 #[test]
2433 fn parse_rpc_unique_module_selection() {
2434 let selection = "eth,admin,eth,net".parse::<RpcModuleSelection>().unwrap();
2435 assert_eq!(
2436 selection,
2437 RpcModuleSelection::Selection(
2438 [RethRpcModule::Eth, RethRpcModule::Admin, RethRpcModule::Net,].into()
2439 )
2440 );
2441 }
2442
2443 #[test]
2444 fn identical_selection() {
2445 assert!(RpcModuleSelection::are_identical(
2446 Some(&RpcModuleSelection::All),
2447 Some(&RpcModuleSelection::All),
2448 ));
2449 assert!(!RpcModuleSelection::are_identical(
2450 Some(&RpcModuleSelection::All),
2451 Some(&RpcModuleSelection::Standard),
2452 ));
2453 assert!(RpcModuleSelection::are_identical(
2454 Some(&RpcModuleSelection::Selection(RpcModuleSelection::Standard.to_selection())),
2455 Some(&RpcModuleSelection::Standard),
2456 ));
2457 assert!(RpcModuleSelection::are_identical(
2458 Some(&RpcModuleSelection::Selection([RethRpcModule::Eth].into())),
2459 Some(&RpcModuleSelection::Selection([RethRpcModule::Eth].into())),
2460 ));
2461 assert!(RpcModuleSelection::are_identical(
2462 None,
2463 Some(&RpcModuleSelection::Selection(Default::default())),
2464 ));
2465 assert!(RpcModuleSelection::are_identical(
2466 Some(&RpcModuleSelection::Selection(Default::default())),
2467 None,
2468 ));
2469 assert!(RpcModuleSelection::are_identical(None, None));
2470 }
2471
2472 #[test]
2473 fn test_rpc_module_str() {
2474 macro_rules! assert_rpc_module {
2475 ($($s:expr => $v:expr,)*) => {
2476 $(
2477 let val: RethRpcModule = $s.parse().unwrap();
2478 assert_eq!(val, $v);
2479 assert_eq!(val.to_string(), $s);
2480 )*
2481 };
2482 }
2483 assert_rpc_module!
2484 (
2485 "admin" => RethRpcModule::Admin,
2486 "debug" => RethRpcModule::Debug,
2487 "eth" => RethRpcModule::Eth,
2488 "net" => RethRpcModule::Net,
2489 "trace" => RethRpcModule::Trace,
2490 "web3" => RethRpcModule::Web3,
2491 "rpc" => RethRpcModule::Rpc,
2492 "ots" => RethRpcModule::Ots,
2493 "reth" => RethRpcModule::Reth,
2494 );
2495 }
2496
2497 #[test]
2498 fn test_default_selection() {
2499 let selection = RpcModuleSelection::Standard.to_selection();
2500 assert_eq!(selection, [RethRpcModule::Eth, RethRpcModule::Net, RethRpcModule::Web3].into())
2501 }
2502
2503 #[test]
2504 fn test_create_rpc_module_config() {
2505 let selection = vec!["eth", "admin"];
2506 let config = RpcModuleSelection::try_from_selection(selection).unwrap();
2507 assert_eq!(
2508 config,
2509 RpcModuleSelection::Selection([RethRpcModule::Eth, RethRpcModule::Admin].into())
2510 );
2511 }
2512
2513 #[test]
2514 fn test_configure_transport_config() {
2515 let config = TransportRpcModuleConfig::default()
2516 .with_http([RethRpcModule::Eth, RethRpcModule::Admin]);
2517 assert_eq!(
2518 config,
2519 TransportRpcModuleConfig {
2520 http: Some(RpcModuleSelection::Selection(
2521 [RethRpcModule::Eth, RethRpcModule::Admin].into()
2522 )),
2523 ws: None,
2524 ipc: None,
2525 config: None,
2526 }
2527 )
2528 }
2529
2530 #[test]
2531 fn test_configure_transport_config_none() {
2532 let config = TransportRpcModuleConfig::default().with_http(Vec::<RethRpcModule>::new());
2533 assert_eq!(
2534 config,
2535 TransportRpcModuleConfig {
2536 http: Some(RpcModuleSelection::Selection(Default::default())),
2537 ws: None,
2538 ipc: None,
2539 config: None,
2540 }
2541 )
2542 }
2543
2544 fn create_test_module() -> RpcModule<()> {
2545 let mut module = RpcModule::new(());
2546 module.register_method("anything", |_, _, _| "succeed").unwrap();
2547 module
2548 }
2549
2550 #[test]
2551 fn test_remove_http_method() {
2552 let mut modules =
2553 TransportRpcModules { http: Some(create_test_module()), ..Default::default() };
2554 assert!(modules.remove_http_method("anything"));
2556
2557 assert!(!modules.remove_http_method("non_existent_method"));
2559
2560 assert!(modules.http.as_ref().unwrap().method("anything").is_none());
2562 }
2563
2564 #[test]
2565 fn test_remove_ws_method() {
2566 let mut modules =
2567 TransportRpcModules { ws: Some(create_test_module()), ..Default::default() };
2568
2569 assert!(modules.remove_ws_method("anything"));
2571
2572 assert!(!modules.remove_ws_method("non_existent_method"));
2574
2575 assert!(modules.ws.as_ref().unwrap().method("anything").is_none());
2577 }
2578
2579 #[test]
2580 fn test_remove_ipc_method() {
2581 let mut modules =
2582 TransportRpcModules { ipc: Some(create_test_module()), ..Default::default() };
2583
2584 assert!(modules.remove_ipc_method("anything"));
2586
2587 assert!(!modules.remove_ipc_method("non_existent_method"));
2589
2590 assert!(modules.ipc.as_ref().unwrap().method("anything").is_none());
2592 }
2593
2594 #[test]
2595 fn test_remove_method_from_configured() {
2596 let mut modules = TransportRpcModules {
2597 http: Some(create_test_module()),
2598 ws: Some(create_test_module()),
2599 ipc: Some(create_test_module()),
2600 ..Default::default()
2601 };
2602
2603 assert!(modules.remove_method_from_configured("anything"));
2605
2606 assert!(!modules.remove_method_from_configured("anything"));
2608
2609 assert!(!modules.remove_method_from_configured("non_existent_method"));
2611
2612 assert!(modules.http.as_ref().unwrap().method("anything").is_none());
2614 assert!(modules.ws.as_ref().unwrap().method("anything").is_none());
2615 assert!(modules.ipc.as_ref().unwrap().method("anything").is_none());
2616 }
2617
2618 #[test]
2619 fn test_transport_rpc_module_rename() {
2620 let mut modules = TransportRpcModules {
2621 http: Some(create_test_module()),
2622 ws: Some(create_test_module()),
2623 ipc: Some(create_test_module()),
2624 ..Default::default()
2625 };
2626
2627 assert!(modules.http.as_ref().unwrap().method("anything").is_some());
2629 assert!(modules.ws.as_ref().unwrap().method("anything").is_some());
2630 assert!(modules.ipc.as_ref().unwrap().method("anything").is_some());
2631
2632 assert!(modules.http.as_ref().unwrap().method("something").is_none());
2634 assert!(modules.ws.as_ref().unwrap().method("something").is_none());
2635 assert!(modules.ipc.as_ref().unwrap().method("something").is_none());
2636
2637 let mut other_module = RpcModule::new(());
2639 other_module.register_method("something", |_, _, _| "fails").unwrap();
2640
2641 modules.rename("anything", other_module).expect("rename failed");
2643
2644 assert!(modules.http.as_ref().unwrap().method("anything").is_none());
2646 assert!(modules.ws.as_ref().unwrap().method("anything").is_none());
2647 assert!(modules.ipc.as_ref().unwrap().method("anything").is_none());
2648
2649 assert!(modules.http.as_ref().unwrap().method("something").is_some());
2651 assert!(modules.ws.as_ref().unwrap().method("something").is_some());
2652 assert!(modules.ipc.as_ref().unwrap().method("something").is_some());
2653 }
2654
2655 #[test]
2656 fn test_replace_http_method() {
2657 let mut modules =
2658 TransportRpcModules { http: Some(create_test_module()), ..Default::default() };
2659
2660 let mut other_module = RpcModule::new(());
2661 other_module.register_method("something", |_, _, _| "fails").unwrap();
2662
2663 assert!(modules.replace_http(other_module.clone()).unwrap());
2664
2665 assert!(modules.http.as_ref().unwrap().method("something").is_some());
2666
2667 other_module.register_method("anything", |_, _, _| "fails").unwrap();
2668 assert!(modules.replace_http(other_module.clone()).unwrap());
2669
2670 assert!(modules.http.as_ref().unwrap().method("anything").is_some());
2671 }
2672 #[test]
2673 fn test_replace_ipc_method() {
2674 let mut modules =
2675 TransportRpcModules { ipc: Some(create_test_module()), ..Default::default() };
2676
2677 let mut other_module = RpcModule::new(());
2678 other_module.register_method("something", |_, _, _| "fails").unwrap();
2679
2680 assert!(modules.replace_ipc(other_module.clone()).unwrap());
2681
2682 assert!(modules.ipc.as_ref().unwrap().method("something").is_some());
2683
2684 other_module.register_method("anything", |_, _, _| "fails").unwrap();
2685 assert!(modules.replace_ipc(other_module.clone()).unwrap());
2686
2687 assert!(modules.ipc.as_ref().unwrap().method("anything").is_some());
2688 }
2689 #[test]
2690 fn test_replace_ws_method() {
2691 let mut modules =
2692 TransportRpcModules { ws: Some(create_test_module()), ..Default::default() };
2693
2694 let mut other_module = RpcModule::new(());
2695 other_module.register_method("something", |_, _, _| "fails").unwrap();
2696
2697 assert!(modules.replace_ws(other_module.clone()).unwrap());
2698
2699 assert!(modules.ws.as_ref().unwrap().method("something").is_some());
2700
2701 other_module.register_method("anything", |_, _, _| "fails").unwrap();
2702 assert!(modules.replace_ws(other_module.clone()).unwrap());
2703
2704 assert!(modules.ws.as_ref().unwrap().method("anything").is_some());
2705 }
2706
2707 #[test]
2708 fn test_replace_configured() {
2709 let mut modules = TransportRpcModules {
2710 http: Some(create_test_module()),
2711 ws: Some(create_test_module()),
2712 ipc: Some(create_test_module()),
2713 ..Default::default()
2714 };
2715 let mut other_module = RpcModule::new(());
2716 other_module.register_method("something", |_, _, _| "fails").unwrap();
2717
2718 assert!(modules.replace_configured(other_module).unwrap());
2719
2720 assert!(modules.http.as_ref().unwrap().method("something").is_some());
2722 assert!(modules.ipc.as_ref().unwrap().method("something").is_some());
2723 assert!(modules.ws.as_ref().unwrap().method("something").is_some());
2724
2725 assert!(modules.http.as_ref().unwrap().method("anything").is_some());
2726 assert!(modules.ipc.as_ref().unwrap().method("anything").is_some());
2727 assert!(modules.ws.as_ref().unwrap().method("anything").is_some());
2728 }
2729
2730 #[test]
2731 fn test_add_or_replace_if_module_configured() {
2732 let config = TransportRpcModuleConfig::default()
2734 .with_http([RethRpcModule::Eth])
2735 .with_ws([RethRpcModule::Eth]);
2736
2737 let mut http_module = RpcModule::new(());
2739 http_module.register_method("eth_existing", |_, _, _| "original").unwrap();
2740
2741 let mut ws_module = RpcModule::new(());
2743 ws_module.register_method("eth_existing", |_, _, _| "original").unwrap();
2744
2745 let ipc_module = RpcModule::new(());
2747
2748 let mut modules = TransportRpcModules {
2750 config,
2751 http: Some(http_module),
2752 ws: Some(ws_module),
2753 ipc: Some(ipc_module),
2754 };
2755
2756 let mut new_module = RpcModule::new(());
2758 new_module.register_method("eth_existing", |_, _, _| "replaced").unwrap(); new_module.register_method("eth_new", |_, _, _| "added").unwrap(); let new_methods: Methods = new_module.into();
2761
2762 let result = modules.add_or_replace_if_module_configured(RethRpcModule::Eth, new_methods);
2764 assert!(result.is_ok(), "Function should succeed");
2765
2766 let http = modules.http.as_ref().unwrap();
2768 assert!(http.method("eth_existing").is_some());
2769 assert!(http.method("eth_new").is_some());
2770
2771 let ws = modules.ws.as_ref().unwrap();
2773 assert!(ws.method("eth_existing").is_some());
2774 assert!(ws.method("eth_new").is_some());
2775
2776 let ipc = modules.ipc.as_ref().unwrap();
2778 assert!(ipc.method("eth_existing").is_none());
2779 assert!(ipc.method("eth_new").is_none());
2780 }
2781
2782 #[test]
2783 fn test_merge_if_module_configured_with_lazy_evaluation() {
2784 let config = TransportRpcModuleConfig::default().with_http([RethRpcModule::Eth]);
2786
2787 let mut modules =
2788 TransportRpcModules { config, http: Some(RpcModule::new(())), ws: None, ipc: None };
2789
2790 let mut closure_called = false;
2792
2793 let result = modules.merge_if_module_configured_with(RethRpcModule::Eth, || {
2795 closure_called = true;
2796 let mut methods = RpcModule::new(());
2797 methods.register_method("eth_test", |_, _, _| "test").unwrap();
2798 methods.into()
2799 });
2800
2801 assert!(result.is_ok());
2802 assert!(closure_called, "Closure should be called when module is configured");
2803 assert!(modules.http.as_ref().unwrap().method("eth_test").is_some());
2804
2805 closure_called = false;
2807 let result = modules.merge_if_module_configured_with(RethRpcModule::Debug, || {
2808 closure_called = true;
2809 RpcModule::new(()).into()
2810 });
2811
2812 assert!(result.is_ok());
2813 assert!(!closure_called, "Closure should NOT be called when module is not configured");
2814 }
2815}