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::{pending_block::PendingEnvBuilder, Call, EthApiSpec, EthTransactions, TraceExt},
48 node::RpcNodeCoreAdapter,
49 EthApiServer, EthApiTypes, FullEthApiServer, FullEthApiTypes, RpcBlock, RpcConvert,
50 RpcConverter, RpcHeader, RpcNodeCore, RpcReceipt, RpcTransaction, RpcTxReq,
51};
52use reth_rpc_eth_types::{receipt::EthReceiptConverter, EthConfig, EthSubscriptionIdProvider};
53use reth_rpc_layer::{
54 AuthLayer, Claims, CompressionLayer, DecompressionLayer, JwtAuthValidator, JwtSecret,
55};
56pub use reth_rpc_server_types::RethRpcModule;
57use reth_storage_api::{
58 BlockReader, ChangeSetReader, FullRpcProvider, NodePrimitivesProvider, StateProviderFactory,
59};
60use reth_tasks::{pool::BlockingTaskGuard, Runtime};
61use reth_tokio_util::EventSender;
62use reth_transaction_pool::{noop::NoopTransactionPool, TransactionPool};
63use serde::{Deserialize, Serialize};
64use std::{
65 collections::HashMap,
66 fmt::Debug,
67 net::{Ipv4Addr, SocketAddr, SocketAddrV4},
68 time::{Duration, SystemTime, UNIX_EPOCH},
69};
70use tower_http::{cors::CorsLayer, validate_request::ValidateRequestHeaderLayer};
71
72pub use cors::CorsDomainError;
73
74pub use jsonrpsee::server::ServerBuilder;
76use jsonrpsee::server::ServerConfigBuilder;
77pub use reth_ipc::server::{
78 Builder as IpcServerBuilder, RpcServiceBuilder as IpcRpcServiceBuilder,
79};
80pub use reth_rpc_server_types::{constants, RpcModuleSelection};
81pub use tower::layer::util::{Identity, Stack};
82
83pub mod auth;
85
86pub mod config;
88
89pub mod middleware;
91
92mod cors;
94
95pub mod error;
97
98pub mod eth;
100pub use eth::EthHandlers;
101
102mod metrics;
104use crate::middleware::RethRpcMiddleware;
105pub use metrics::{MeteredBatchRequestsFuture, MeteredRequestFuture, RpcRequestMetricsService};
106use reth_chain_state::{
107 CanonStateSubscriptions, ForkChoiceSubscriptions, PersistedBlockSubscriptions,
108};
109use reth_rpc::eth::sim_bundle::EthSimBundle;
110
111pub mod rate_limiter;
113
114#[derive(Debug, Clone)]
118pub struct RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, Consensus> {
119 provider: Provider,
121 pool: Pool,
123 network: Network,
125 executor: Option<Runtime>,
127 evm_config: EvmConfig,
129 consensus: Consensus,
131 _primitives: PhantomData<N>,
133}
134
135impl<N, Provider, Pool, Network, EvmConfig, Consensus>
138 RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, Consensus>
139{
140 pub const fn new(
142 provider: Provider,
143 pool: Pool,
144 network: Network,
145 executor: Runtime,
146 evm_config: EvmConfig,
147 consensus: Consensus,
148 ) -> Self {
149 Self {
150 provider,
151 pool,
152 network,
153 executor: Some(executor),
154 evm_config,
155 consensus,
156 _primitives: PhantomData,
157 }
158 }
159
160 pub fn with_provider<P>(
162 self,
163 provider: P,
164 ) -> RpcModuleBuilder<N, P, Pool, Network, EvmConfig, Consensus> {
165 let Self { pool, network, executor, evm_config, consensus, _primitives, .. } = self;
166 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
167 }
168
169 pub fn with_pool<P>(
171 self,
172 pool: P,
173 ) -> RpcModuleBuilder<N, Provider, P, Network, EvmConfig, Consensus> {
174 let Self { provider, network, executor, evm_config, consensus, _primitives, .. } = self;
175 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
176 }
177
178 pub fn with_noop_pool(
184 self,
185 ) -> RpcModuleBuilder<N, Provider, NoopTransactionPool, Network, EvmConfig, Consensus> {
186 let Self { provider, executor, network, evm_config, consensus, _primitives, .. } = self;
187 RpcModuleBuilder {
188 provider,
189 executor,
190 network,
191 evm_config,
192 pool: NoopTransactionPool::default(),
193 consensus,
194 _primitives,
195 }
196 }
197
198 pub fn with_network<Net>(
200 self,
201 network: Net,
202 ) -> RpcModuleBuilder<N, Provider, Pool, Net, EvmConfig, Consensus> {
203 let Self { provider, pool, executor, evm_config, consensus, _primitives, .. } = self;
204 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
205 }
206
207 pub fn with_noop_network(
213 self,
214 ) -> RpcModuleBuilder<N, Provider, Pool, NoopNetwork, EvmConfig, Consensus> {
215 let Self { provider, pool, executor, evm_config, consensus, _primitives, .. } = self;
216 RpcModuleBuilder {
217 provider,
218 pool,
219 executor,
220 network: NoopNetwork::default(),
221 evm_config,
222 consensus,
223 _primitives,
224 }
225 }
226
227 pub fn with_executor(self, executor: Runtime) -> Self {
229 let Self { pool, network, provider, evm_config, consensus, _primitives, .. } = self;
230 Self {
231 provider,
232 network,
233 pool,
234 executor: Some(executor),
235 evm_config,
236 consensus,
237 _primitives,
238 }
239 }
240
241 pub fn with_evm_config<E>(
243 self,
244 evm_config: E,
245 ) -> RpcModuleBuilder<N, Provider, Pool, Network, E, Consensus> {
246 let Self { provider, pool, executor, network, consensus, _primitives, .. } = self;
247 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
248 }
249
250 pub fn with_consensus<C>(
252 self,
253 consensus: C,
254 ) -> RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, C> {
255 let Self { provider, network, pool, executor, evm_config, _primitives, .. } = self;
256 RpcModuleBuilder { provider, network, pool, executor, evm_config, consensus, _primitives }
257 }
258
259 #[expect(clippy::type_complexity)]
261 pub fn eth_api_builder<ChainSpec>(
262 &self,
263 ) -> EthApiBuilder<
264 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>,
265 RpcConverter<Ethereum, EvmConfig, EthReceiptConverter<ChainSpec>>,
266 >
267 where
268 Provider: Clone,
269 Pool: Clone,
270 Network: Clone,
271 EvmConfig: Clone,
272 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>:
273 RpcNodeCore<Provider: ChainSpecProvider<ChainSpec = ChainSpec>, Evm = EvmConfig>,
274 {
275 EthApiBuilder::new(
276 self.provider.clone(),
277 self.pool.clone(),
278 self.network.clone(),
279 self.evm_config.clone(),
280 )
281 }
282
283 #[expect(clippy::type_complexity)]
289 pub fn bootstrap_eth_api<ChainSpec>(
290 &self,
291 ) -> EthApi<
292 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>,
293 RpcConverter<Ethereum, EvmConfig, EthReceiptConverter<ChainSpec>>,
294 >
295 where
296 Provider: Clone,
297 Pool: Clone,
298 Network: Clone,
299 EvmConfig: ConfigureEvm,
300 RpcNodeCoreAdapter<Provider, Pool, Network, EvmConfig>:
301 RpcNodeCore<Provider: ChainSpecProvider<ChainSpec = ChainSpec>, Evm = EvmConfig>,
302 RpcConverter<Ethereum, EvmConfig, EthReceiptConverter<ChainSpec>>: RpcConvert,
303 (): PendingEnvBuilder<EvmConfig>,
304 {
305 self.eth_api_builder().build()
306 }
307}
308
309impl<N, Provider, Pool, Network, EvmConfig, Consensus>
310 RpcModuleBuilder<N, Provider, Pool, Network, EvmConfig, Consensus>
311where
312 N: NodePrimitives,
313 Provider: FullRpcProvider<Block = N::Block, Receipt = N::Receipt, Header = N::BlockHeader>
314 + CanonStateSubscriptions<Primitives = N>
315 + ForkChoiceSubscriptions<Header = N::BlockHeader>
316 + PersistedBlockSubscriptions
317 + ChangeSetReader
318 + NodePrimitivesProvider<Primitives = N>,
319 Pool: TransactionPool + 'static,
320 Network: NetworkInfo + Peers + Clone + 'static,
321 EvmConfig: ConfigureEvm<Primitives = N> + 'static,
322 Consensus: FullConsensus<N> + Clone + 'static,
323{
324 pub fn build_with_auth_server<EthApi, Payload>(
331 self,
332 module_config: TransportRpcModuleConfig,
333 engine: impl IntoEngineApiRpcModule,
334 eth: EthApi,
335 engine_events: EventSender<ConsensusEngineEvent<N>>,
336 beacon_engine_handle: ConsensusEngineHandle<Payload>,
337 ) -> (
338 TransportRpcModules,
339 AuthRpcModule,
340 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>,
341 )
342 where
343 EthApi: FullEthApiServer<Provider = Provider, Pool = Pool>,
344 Payload: PayloadTypes,
345 {
346 let config = module_config.config.clone().unwrap_or_default();
347
348 let mut registry = self.into_registry(config, eth, engine_events);
349 let modules = registry.create_transport_rpc_modules(module_config);
350 let auth_module = registry.create_auth_module(engine, beacon_engine_handle);
351
352 (modules, auth_module, registry)
353 }
354
355 pub fn into_registry<EthApi>(
360 self,
361 config: RpcModuleConfig,
362 eth: EthApi,
363 engine_events: EventSender<ConsensusEngineEvent<N>>,
364 ) -> RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
365 where
366 EthApi: FullEthApiServer<Provider = Provider, Pool = Pool>,
367 {
368 let Self { provider, pool, network, executor, consensus, evm_config, .. } = self;
369 let executor =
370 executor.expect("RpcModuleBuilder requires a Runtime to be set via `with_executor`");
371 RpcRegistryInner::new(
372 provider,
373 pool,
374 network,
375 executor,
376 consensus,
377 config,
378 evm_config,
379 eth,
380 engine_events,
381 )
382 }
383
384 pub fn build<EthApi>(
387 self,
388 module_config: TransportRpcModuleConfig,
389 eth: EthApi,
390 engine_events: EventSender<ConsensusEngineEvent<N>>,
391 ) -> TransportRpcModules<()>
392 where
393 EthApi: FullEthApiServer<Provider = Provider, Pool = Pool>,
394 {
395 if module_config.is_empty() {
396 TransportRpcModules::default()
397 } else {
398 let config = module_config.config.clone().unwrap_or_default();
399 let mut registry = self.into_registry(config, eth, engine_events);
400 registry.create_transport_rpc_modules(module_config)
401 }
402 }
403}
404
405impl<N: NodePrimitives> Default for RpcModuleBuilder<N, (), (), (), (), ()> {
406 fn default() -> Self {
407 Self {
408 provider: (),
409 pool: (),
410 network: (),
411 executor: None,
412 evm_config: (),
413 consensus: (),
414 _primitives: PhantomData,
415 }
416 }
417}
418
419#[derive(Debug, Default, Clone, Eq, PartialEq, Serialize, Deserialize)]
421pub struct RpcModuleConfig {
422 eth: EthConfig,
424}
425
426impl RpcModuleConfig {
429 pub fn builder() -> RpcModuleConfigBuilder {
431 RpcModuleConfigBuilder::default()
432 }
433
434 pub const fn new(eth: EthConfig) -> Self {
436 Self { eth }
437 }
438
439 pub const fn eth(&self) -> &EthConfig {
441 &self.eth
442 }
443
444 pub const fn eth_mut(&mut self) -> &mut EthConfig {
446 &mut self.eth
447 }
448}
449
450#[derive(Clone, Debug, Default)]
452pub struct RpcModuleConfigBuilder {
453 eth: Option<EthConfig>,
454}
455
456impl RpcModuleConfigBuilder {
459 pub fn eth(mut self, eth: EthConfig) -> Self {
461 self.eth = Some(eth);
462 self
463 }
464
465 pub fn build(self) -> RpcModuleConfig {
467 let Self { eth } = self;
468 RpcModuleConfig { eth: eth.unwrap_or_default() }
469 }
470
471 pub const fn get_eth(&self) -> Option<&EthConfig> {
473 self.eth.as_ref()
474 }
475
476 pub const fn eth_mut(&mut self) -> &mut Option<EthConfig> {
478 &mut self.eth
479 }
480
481 pub fn eth_mut_or_default(&mut self) -> &mut EthConfig {
483 self.eth.get_or_insert_with(EthConfig::default)
484 }
485}
486
487#[derive(Debug)]
489pub struct RpcRegistryInner<Provider, Pool, Network, EthApi: EthApiTypes, EvmConfig, Consensus> {
490 provider: Provider,
491 pool: Pool,
492 network: Network,
493 executor: Runtime,
494 evm_config: EvmConfig,
495 consensus: Consensus,
496 eth: EthHandlers<EthApi>,
498 blocking_pool_guard: BlockingTaskGuard,
500 modules: HashMap<RethRpcModule, Methods>,
502 eth_config: EthConfig,
504 engine_events:
506 EventSender<ConsensusEngineEvent<<EthApi::RpcConvert as RpcConvert>::Primitives>>,
507}
508
509impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
512 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
513where
514 N: NodePrimitives,
515 Provider: StateProviderFactory
516 + CanonStateSubscriptions<Primitives = N>
517 + BlockReader<Block = N::Block, Receipt = N::Receipt>
518 + Clone
519 + Unpin
520 + 'static,
521 Pool: Send + Sync + Clone + 'static,
522 Network: Clone + 'static,
523 EthApi: FullEthApiTypes,
524 EvmConfig: ConfigureEvm<Primitives = N>,
525{
526 #[expect(clippy::too_many_arguments)]
528 pub fn new(
529 provider: Provider,
530 pool: Pool,
531 network: Network,
532 executor: Runtime,
533 consensus: Consensus,
534 config: RpcModuleConfig,
535 evm_config: EvmConfig,
536 eth_api: EthApi,
537 engine_events: EventSender<
538 ConsensusEngineEvent<<EthApi::Provider as NodePrimitivesProvider>::Primitives>,
539 >,
540 ) -> Self
541 where
542 EvmConfig: ConfigureEvm<Primitives = N>,
543 {
544 let blocking_pool_guard = BlockingTaskGuard::new(config.eth.max_tracing_requests);
545
546 let eth = EthHandlers::bootstrap(config.eth.clone(), executor.clone(), eth_api);
547
548 Self {
549 provider,
550 pool,
551 network,
552 eth,
553 executor,
554 consensus,
555 modules: Default::default(),
556 blocking_pool_guard,
557 eth_config: config.eth,
558 evm_config,
559 engine_events,
560 }
561 }
562}
563
564impl<Provider, Pool, Network, EthApi, Evm, Consensus>
565 RpcRegistryInner<Provider, Pool, Network, EthApi, Evm, Consensus>
566where
567 EthApi: EthApiTypes,
568{
569 pub const fn eth_api(&self) -> &EthApi {
571 &self.eth.api
572 }
573
574 pub const fn eth_handlers(&self) -> &EthHandlers<EthApi> {
576 &self.eth
577 }
578
579 pub const fn pool(&self) -> &Pool {
581 &self.pool
582 }
583
584 pub const fn tasks(&self) -> &Runtime {
586 &self.executor
587 }
588
589 pub const fn provider(&self) -> &Provider {
591 &self.provider
592 }
593
594 pub const fn evm_config(&self) -> &Evm {
596 &self.evm_config
597 }
598
599 pub fn methods(&self) -> Vec<Methods> {
601 self.modules.values().cloned().collect()
602 }
603
604 pub fn module(&self) -> RpcModule<()> {
606 let mut module = RpcModule::new(());
607 for methods in self.modules.values().cloned() {
608 module.merge(methods).expect("No conflicts");
609 }
610 module
611 }
612}
613
614impl<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
615 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
616where
617 Network: NetworkInfo + Clone + 'static,
618 EthApi: EthApiTypes,
619 Provider: BlockReader + ChainSpecProvider<ChainSpec: EthereumHardforks>,
620 EvmConfig: ConfigureEvm,
621{
622 pub fn admin_api(&self) -> AdminApi<Network, Provider::ChainSpec, Pool>
624 where
625 Network: Peers,
626 Pool: TransactionPool + 'static,
627 {
628 AdminApi::new(self.network.clone(), self.provider.chain_spec(), self.pool.clone())
629 }
630
631 pub fn web3_api(&self) -> Web3Api<Network> {
633 Web3Api::new(self.network.clone())
634 }
635
636 pub fn register_admin(&mut self) -> &mut Self
638 where
639 Network: Peers,
640 Pool: TransactionPool + 'static,
641 {
642 let adminapi = self.admin_api();
643 self.modules.insert(RethRpcModule::Admin, adminapi.into_rpc().into());
644 self
645 }
646
647 pub fn register_web3(&mut self) -> &mut Self {
649 let web3api = self.web3_api();
650 self.modules.insert(RethRpcModule::Web3, web3api.into_rpc().into());
651 self
652 }
653}
654
655impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
656 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
657where
658 N: NodePrimitives,
659 Provider: FullRpcProvider<
660 Header = N::BlockHeader,
661 Block = N::Block,
662 Receipt = N::Receipt,
663 Transaction = N::SignedTx,
664 > + ChangeSetReader
665 + CanonStateSubscriptions<Primitives = N>
666 + ForkChoiceSubscriptions<Header = N::BlockHeader>
667 + PersistedBlockSubscriptions,
668 Network: NetworkInfo + Peers + Clone + 'static,
669 EthApi: EthApiServer<
670 RpcTxReq<EthApi::NetworkTypes>,
671 RpcTransaction<EthApi::NetworkTypes>,
672 RpcBlock<EthApi::NetworkTypes>,
673 RpcReceipt<EthApi::NetworkTypes>,
674 RpcHeader<EthApi::NetworkTypes>,
675 TxTy<N>,
676 > + EthApiTypes,
677 EvmConfig: ConfigureEvm<Primitives = N> + 'static,
678{
679 pub fn register_eth(&mut self) -> &mut Self {
685 let eth_api = self.eth_api().clone();
686 self.modules.insert(RethRpcModule::Eth, eth_api.into_rpc().into());
687 self
688 }
689
690 pub fn register_ots(&mut self) -> &mut Self
696 where
697 EthApi: TraceExt + EthTransactions<Primitives = N>,
698 {
699 let otterscan_api = self.otterscan_api();
700 self.modules.insert(RethRpcModule::Ots, otterscan_api.into_rpc().into());
701 self
702 }
703
704 pub fn register_debug(&mut self) -> &mut Self
710 where
711 EthApi: EthTransactions + TraceExt,
712 {
713 let debug_api = self.debug_api();
714 self.modules.insert(RethRpcModule::Debug, debug_api.into_rpc().into());
715 self
716 }
717
718 pub fn register_trace(&mut self) -> &mut Self
724 where
725 EthApi: TraceExt,
726 {
727 let trace_api = self.trace_api();
728 self.modules.insert(RethRpcModule::Trace, trace_api.into_rpc().into());
729 self
730 }
731
732 pub fn register_net(&mut self) -> &mut Self
740 where
741 EthApi: EthApiSpec,
742 {
743 let netapi = self.net_api();
744 self.modules.insert(RethRpcModule::Net, netapi.into_rpc().into());
745 self
746 }
747
748 pub fn register_reth(&mut self) -> &mut Self {
756 let rethapi = self.reth_api();
757 self.modules.insert(RethRpcModule::Reth, rethapi.into_rpc().into());
758 self
759 }
760
761 pub fn otterscan_api(&self) -> OtterscanApi<EthApi> {
767 let eth_api = self.eth_api().clone();
768 OtterscanApi::new(eth_api)
769 }
770}
771
772impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
773 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
774where
775 N: NodePrimitives,
776 Provider: FullRpcProvider<
777 Block = N::Block,
778 Header = N::BlockHeader,
779 Transaction = N::SignedTx,
780 Receipt = N::Receipt,
781 > + ChangeSetReader,
782 Network: NetworkInfo + Peers + Clone + 'static,
783 EthApi: EthApiTypes,
784 EvmConfig: ConfigureEvm<Primitives = N>,
785{
786 pub fn trace_api(&self) -> TraceApi<EthApi> {
792 TraceApi::new(
793 self.eth_api().clone(),
794 self.blocking_pool_guard.clone(),
795 self.eth_config.clone(),
796 )
797 }
798
799 pub fn bundle_api(&self) -> EthBundle<EthApi>
805 where
806 EthApi: EthTransactions + Call,
807 {
808 let eth_api = self.eth_api().clone();
809 EthBundle::new(eth_api, self.blocking_pool_guard.clone())
810 }
811
812 pub fn debug_api(&self) -> DebugApi<EthApi>
818 where
819 EthApi: FullEthApiTypes,
820 {
821 DebugApi::new(
822 self.eth_api().clone(),
823 self.blocking_pool_guard.clone(),
824 self.tasks(),
825 self.engine_events.new_listener(),
826 )
827 }
828
829 pub fn net_api(&self) -> NetApi<Network, EthApi>
835 where
836 EthApi: EthApiSpec,
837 {
838 let eth_api = self.eth_api().clone();
839 NetApi::new(self.network.clone(), eth_api)
840 }
841
842 pub fn reth_api(&self) -> RethApi<Provider, EvmConfig> {
844 RethApi::new(
845 self.provider.clone(),
846 self.evm_config.clone(),
847 self.blocking_pool_guard.clone(),
848 self.executor.clone(),
849 )
850 }
851}
852
853impl<N, Provider, Pool, Network, EthApi, EvmConfig, Consensus>
854 RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
855where
856 N: NodePrimitives,
857 Provider: FullRpcProvider<Block = N::Block>
858 + CanonStateSubscriptions<Primitives = N>
859 + ForkChoiceSubscriptions<Header = N::BlockHeader>
860 + PersistedBlockSubscriptions
861 + ChangeSetReader,
862 Pool: TransactionPool + 'static,
863 Network: NetworkInfo + Peers + Clone + 'static,
864 EthApi: FullEthApiServer,
865 EvmConfig: ConfigureEvm<Primitives = N> + 'static,
866 Consensus: FullConsensus<N> + Clone + 'static,
867{
868 pub fn create_auth_module<Payload>(
875 &self,
876 engine_api: impl IntoEngineApiRpcModule,
877 beacon_engine_handle: ConsensusEngineHandle<Payload>,
878 ) -> AuthRpcModule
879 where
880 Payload: PayloadTypes,
881 {
882 let mut module = engine_api.into_rpc_module();
883
884 let reth_engine_api = RethEngineApi::new(beacon_engine_handle);
886 module
887 .merge(RethEngineApiServer::into_rpc(reth_engine_api).remove_context())
888 .expect("No conflicting methods");
889
890 let eth_handlers = self.eth_handlers();
892 let engine_eth = EngineEthApi::new(eth_handlers.api.clone(), eth_handlers.filter.clone());
893
894 module.merge(engine_eth.into_rpc()).expect("No conflicting methods");
895
896 AuthRpcModule { inner: module }
897 }
898
899 fn maybe_module(&mut self, config: Option<&RpcModuleSelection>) -> Option<RpcModule<()>> {
901 config.map(|config| self.module_for(config))
902 }
903
904 pub fn create_transport_rpc_modules(
908 &mut self,
909 config: TransportRpcModuleConfig,
910 ) -> TransportRpcModules<()> {
911 let mut modules = TransportRpcModules::default();
912 let http = self.maybe_module(config.http.as_ref());
913 let ws = self.maybe_module(config.ws.as_ref());
914 let ipc = self.maybe_module(config.ipc.as_ref());
915
916 modules.config = config;
917 modules.http = http;
918 modules.ws = ws;
919 modules.ipc = ipc;
920 modules
921 }
922
923 pub fn module_for(&mut self, config: &RpcModuleSelection) -> RpcModule<()> {
926 let mut module = RpcModule::new(());
927 let all_methods = self.reth_methods(config.iter_selection());
928 for methods in all_methods {
929 module.merge(methods).expect("No conflicts");
930 }
931 module
932 }
933
934 pub fn reth_methods(
943 &mut self,
944 namespaces: impl Iterator<Item = RethRpcModule>,
945 ) -> Vec<Methods> {
946 let EthHandlers { api: eth_api, filter: eth_filter, pubsub: eth_pubsub, .. } =
947 self.eth_handlers().clone();
948
949 let namespaces: Vec<_> = namespaces.collect();
951 namespaces
952 .iter()
953 .map(|namespace| {
954 self.modules
955 .entry(namespace.clone())
956 .or_insert_with(|| match namespace.clone() {
957 RethRpcModule::Admin => AdminApi::new(
958 self.network.clone(),
959 self.provider.chain_spec(),
960 self.pool.clone(),
961 )
962 .into_rpc()
963 .into(),
964 RethRpcModule::Debug => DebugApi::new(
965 eth_api.clone(),
966 self.blocking_pool_guard.clone(),
967 &self.executor,
968 self.engine_events.new_listener(),
969 )
970 .into_rpc()
971 .into(),
972 RethRpcModule::Eth => {
973 let mut module = eth_api.clone().into_rpc();
975 module.merge(eth_filter.clone().into_rpc()).expect("No conflicts");
976 module.merge(eth_pubsub.clone().into_rpc()).expect("No conflicts");
977 module
978 .merge(
979 EthBundle::new(
980 eth_api.clone(),
981 self.blocking_pool_guard.clone(),
982 )
983 .into_rpc(),
984 )
985 .expect("No conflicts");
986
987 module.into()
988 }
989 RethRpcModule::Net => {
990 NetApi::new(self.network.clone(), eth_api.clone()).into_rpc().into()
991 }
992 RethRpcModule::Trace => TraceApi::new(
993 eth_api.clone(),
994 self.blocking_pool_guard.clone(),
995 self.eth_config.clone(),
996 )
997 .into_rpc()
998 .into(),
999 RethRpcModule::Web3 => Web3Api::new(self.network.clone()).into_rpc().into(),
1000 RethRpcModule::Txpool => TxPoolApi::new(
1001 self.eth.api.pool().clone(),
1002 dyn_clone::clone(self.eth.api.converter()),
1003 )
1004 .into_rpc()
1005 .into(),
1006 RethRpcModule::Rpc => RPCApi::new(
1007 namespaces
1008 .iter()
1009 .map(|module| (module.to_string(), "1.0".to_string()))
1010 .collect(),
1011 )
1012 .into_rpc()
1013 .into(),
1014 RethRpcModule::Ots => OtterscanApi::new(eth_api.clone()).into_rpc().into(),
1015 RethRpcModule::Reth => RethApi::new(
1016 self.provider.clone(),
1017 self.evm_config.clone(),
1018 self.blocking_pool_guard.clone(),
1019 self.executor.clone(),
1020 )
1021 .into_rpc()
1022 .into(),
1023 RethRpcModule::Miner => MinerApi::default().into_rpc().into(),
1024 RethRpcModule::Mev => {
1025 EthSimBundle::new(eth_api.clone(), self.blocking_pool_guard.clone())
1026 .into_rpc()
1027 .into()
1028 }
1029 RethRpcModule::Flashbots |
1033 RethRpcModule::Testing |
1034 RethRpcModule::Other(_) => Default::default(),
1035 })
1036 .clone()
1037 })
1038 .collect::<Vec<_>>()
1039 }
1040}
1041
1042impl<Provider, Pool, Network, EthApi, EvmConfig, Consensus> Clone
1043 for RpcRegistryInner<Provider, Pool, Network, EthApi, EvmConfig, Consensus>
1044where
1045 EthApi: EthApiTypes,
1046 Provider: Clone,
1047 Pool: Clone,
1048 Network: Clone,
1049 EvmConfig: Clone,
1050 Consensus: Clone,
1051{
1052 fn clone(&self) -> Self {
1053 Self {
1054 provider: self.provider.clone(),
1055 pool: self.pool.clone(),
1056 network: self.network.clone(),
1057 executor: self.executor.clone(),
1058 evm_config: self.evm_config.clone(),
1059 consensus: self.consensus.clone(),
1060 eth: self.eth.clone(),
1061 blocking_pool_guard: self.blocking_pool_guard.clone(),
1062 modules: self.modules.clone(),
1063 eth_config: self.eth_config.clone(),
1064 engine_events: self.engine_events.clone(),
1065 }
1066 }
1067}
1068
1069#[derive(Debug)]
1081pub struct RpcServerConfig<RpcMiddleware = Identity> {
1082 http_server_config: Option<ServerConfigBuilder>,
1084 http_cors_domains: Option<String>,
1086 http_addr: Option<SocketAddr>,
1088 http_disable_compression: bool,
1090 http_compression_algorithms: Option<Vec<String>>,
1094 http_decompression_algorithms: Option<Vec<String>>,
1098 http_max_request_body_size: Option<u32>,
1100 ws_server_config: Option<ServerConfigBuilder>,
1102 ws_cors_domains: Option<String>,
1104 ws_addr: Option<SocketAddr>,
1106 ipc_server_config: Option<IpcServerBuilder<Identity, Identity>>,
1108 ipc_endpoint: Option<String>,
1110 jwt_secret: Option<JwtSecret>,
1112 rpc_metrics_enabled: bool,
1114 rpc_middleware: RpcMiddleware,
1116}
1117
1118impl Default for RpcServerConfig<Identity> {
1121 fn default() -> Self {
1123 Self {
1124 http_server_config: None,
1125 http_cors_domains: None,
1126 http_addr: None,
1127 http_disable_compression: false,
1128 http_compression_algorithms: None,
1129 http_decompression_algorithms: None,
1130 http_max_request_body_size: None,
1131 ws_server_config: None,
1132 ws_cors_domains: None,
1133 ws_addr: None,
1134 ipc_server_config: None,
1135 ipc_endpoint: None,
1136 jwt_secret: None,
1137 rpc_metrics_enabled: true,
1138 rpc_middleware: Default::default(),
1139 }
1140 }
1141}
1142
1143impl RpcServerConfig {
1144 pub fn http(config: ServerConfigBuilder) -> Self {
1146 Self::default().with_http(config)
1147 }
1148
1149 pub fn ws(config: ServerConfigBuilder) -> Self {
1151 Self::default().with_ws(config)
1152 }
1153
1154 pub fn ipc(config: IpcServerBuilder<Identity, Identity>) -> Self {
1156 Self::default().with_ipc(config)
1157 }
1158
1159 pub fn with_http(mut self, config: ServerConfigBuilder) -> Self {
1164 self.http_server_config =
1165 Some(config.set_id_provider(EthSubscriptionIdProvider::default()));
1166 self
1167 }
1168
1169 pub fn with_ws(mut self, config: ServerConfigBuilder) -> Self {
1174 self.ws_server_config = Some(config.set_id_provider(EthSubscriptionIdProvider::default()));
1175 self
1176 }
1177
1178 pub fn with_ipc(mut self, config: IpcServerBuilder<Identity, Identity>) -> Self {
1183 self.ipc_server_config = Some(config.set_id_provider(EthSubscriptionIdProvider::default()));
1184 self
1185 }
1186}
1187
1188impl<RpcMiddleware> RpcServerConfig<RpcMiddleware> {
1189 pub fn set_rpc_middleware<T>(self, rpc_middleware: T) -> RpcServerConfig<T> {
1191 RpcServerConfig {
1192 http_server_config: self.http_server_config,
1193 http_cors_domains: self.http_cors_domains,
1194 http_addr: self.http_addr,
1195 http_disable_compression: self.http_disable_compression,
1196 http_compression_algorithms: self.http_compression_algorithms,
1197 http_decompression_algorithms: self.http_decompression_algorithms,
1198 http_max_request_body_size: self.http_max_request_body_size,
1199 ws_server_config: self.ws_server_config,
1200 ws_cors_domains: self.ws_cors_domains,
1201 ws_addr: self.ws_addr,
1202 ipc_server_config: self.ipc_server_config,
1203 ipc_endpoint: self.ipc_endpoint,
1204 jwt_secret: self.jwt_secret,
1205 rpc_metrics_enabled: self.rpc_metrics_enabled,
1206 rpc_middleware,
1207 }
1208 }
1209
1210 pub const fn with_rpc_metrics_enabled(mut self, enabled: bool) -> Self {
1212 self.rpc_metrics_enabled = enabled;
1213 self
1214 }
1215
1216 pub fn with_cors(self, cors_domain: Option<String>) -> Self {
1218 self.with_http_cors(cors_domain.clone()).with_ws_cors(cors_domain)
1219 }
1220
1221 pub fn with_ws_cors(mut self, cors_domain: Option<String>) -> Self {
1223 self.ws_cors_domains = cors_domain;
1224 self
1225 }
1226
1227 pub fn with_http_cors(mut self, cors_domain: Option<String>) -> Self {
1229 self.http_cors_domains = cors_domain;
1230 self
1231 }
1232
1233 pub const fn with_http_disable_compression(mut self, http_disable_compression: bool) -> Self {
1235 self.http_disable_compression = http_disable_compression;
1236 self
1237 }
1238
1239 pub fn with_http_compression_algorithms(mut self, algos: Option<Vec<String>>) -> Self {
1244 self.http_compression_algorithms = algos;
1245 self
1246 }
1247
1248 pub fn with_http_decompression(mut self, algos: Option<Vec<String>>, max_size: u32) -> Self {
1252 self.http_decompression_algorithms = algos;
1253 self.http_max_request_body_size = Some(max_size);
1254 self
1255 }
1256
1257 pub const fn with_http_address(mut self, addr: SocketAddr) -> Self {
1262 self.http_addr = Some(addr);
1263 self
1264 }
1265
1266 pub const fn with_ws_address(mut self, addr: SocketAddr) -> Self {
1271 self.ws_addr = Some(addr);
1272 self
1273 }
1274
1275 pub fn with_id_provider<I>(mut self, id_provider: I) -> Self
1279 where
1280 I: IdProvider + Clone + 'static,
1281 {
1282 if let Some(config) = self.http_server_config {
1283 self.http_server_config = Some(config.set_id_provider(id_provider.clone()));
1284 }
1285 if let Some(config) = self.ws_server_config {
1286 self.ws_server_config = Some(config.set_id_provider(id_provider.clone()));
1287 }
1288 if let Some(ipc) = self.ipc_server_config {
1289 self.ipc_server_config = Some(ipc.set_id_provider(id_provider));
1290 }
1291
1292 self
1293 }
1294
1295 pub fn with_ipc_endpoint(mut self, path: impl Into<String>) -> Self {
1299 self.ipc_endpoint = Some(path.into());
1300 self
1301 }
1302
1303 pub const fn with_jwt_secret(mut self, secret: Option<JwtSecret>) -> Self {
1305 self.jwt_secret = secret;
1306 self
1307 }
1308
1309 pub fn with_tokio_runtime(mut self, tokio_runtime: Option<tokio::runtime::Handle>) -> Self {
1311 let Some(tokio_runtime) = tokio_runtime else { return self };
1312 if let Some(http_server_config) = self.http_server_config {
1313 self.http_server_config =
1314 Some(http_server_config.custom_tokio_runtime(tokio_runtime.clone()));
1315 }
1316 if let Some(ws_server_config) = self.ws_server_config {
1317 self.ws_server_config =
1318 Some(ws_server_config.custom_tokio_runtime(tokio_runtime.clone()));
1319 }
1320 if let Some(ipc_server_config) = self.ipc_server_config {
1321 self.ipc_server_config = Some(ipc_server_config.custom_tokio_runtime(tokio_runtime));
1322 }
1323 self
1324 }
1325
1326 pub const fn has_server(&self) -> bool {
1330 self.http_server_config.is_some() ||
1331 self.ws_server_config.is_some() ||
1332 self.ipc_server_config.is_some()
1333 }
1334
1335 pub const fn http_address(&self) -> Option<SocketAddr> {
1337 self.http_addr
1338 }
1339
1340 pub const fn ws_address(&self) -> Option<SocketAddr> {
1342 self.ws_addr
1343 }
1344
1345 pub fn ipc_endpoint(&self) -> Option<String> {
1347 self.ipc_endpoint.clone()
1348 }
1349
1350 pub const fn rpc_metrics_enabled(&self) -> bool {
1352 self.rpc_metrics_enabled
1353 }
1354
1355 fn maybe_cors_layer(cors: Option<String>) -> Result<Option<CorsLayer>, CorsDomainError> {
1357 cors.as_deref().map(cors::create_cors_layer).transpose()
1358 }
1359
1360 fn maybe_jwt_layer(jwt_secret: Option<JwtSecret>) -> Option<AuthLayer<JwtAuthValidator>> {
1362 jwt_secret.map(|secret| AuthLayer::new(JwtAuthValidator::new(secret)))
1363 }
1364
1365 fn maybe_compression_layer(
1371 addr: SocketAddr,
1372 disable_compression: bool,
1373 algos: Option<&[String]>,
1374 ) -> Option<CompressionLayer> {
1375 if disable_compression || addr.ip().to_canonical().is_loopback() {
1376 None
1377 } else {
1378 match algos {
1379 None => Some(CompressionLayer::new()),
1381 Some(algos) => Some(CompressionLayer::with_algorithms(algos)),
1382 }
1383 }
1384 }
1385
1386 fn maybe_decompression_layer(
1392 max_request_body_size: Option<u32>,
1393 algos: Option<&[String]>,
1394 ) -> Option<DecompressionLayer> {
1395 let algos = algos?;
1396 if algos.is_empty() {
1397 return None;
1398 }
1399
1400 let max = max_request_body_size?;
1401 Some(DecompressionLayer::new(algos, max as usize))
1402 }
1403
1404 pub async fn start(self, modules: &TransportRpcModules) -> Result<RpcServerHandle, RpcError>
1410 where
1411 RpcMiddleware: RethRpcMiddleware,
1412 {
1413 let mut http_handle = None;
1414 let mut ws_handle = None;
1415 let mut ipc_handle = None;
1416
1417 let http_socket_addr = self.http_addr.unwrap_or(SocketAddr::V4(SocketAddrV4::new(
1418 Ipv4Addr::LOCALHOST,
1419 constants::DEFAULT_HTTP_RPC_PORT,
1420 )));
1421
1422 let ws_socket_addr = self.ws_addr.unwrap_or(SocketAddr::V4(SocketAddrV4::new(
1423 Ipv4Addr::LOCALHOST,
1424 constants::DEFAULT_WS_RPC_PORT,
1425 )));
1426
1427 let rpc_metrics_enabled = self.rpc_metrics_enabled;
1428 let ipc_path =
1429 self.ipc_endpoint.clone().unwrap_or_else(|| constants::DEFAULT_IPC_ENDPOINT.into());
1430
1431 if let Some(builder) = self.ipc_server_config {
1432 let ipc = builder
1433 .set_rpc_middleware(
1434 IpcRpcServiceBuilder::new().option_layer(
1435 rpc_metrics_enabled
1436 .then(|| modules.ipc.as_ref().map(RpcRequestMetrics::ipc))
1437 .flatten(),
1438 ),
1439 )
1440 .build(ipc_path);
1441 ipc_handle = Some(ipc.start(modules.ipc.clone().expect("ipc server error")).await?);
1442 }
1443
1444 if self.http_addr == self.ws_addr &&
1446 self.http_server_config.is_some() &&
1447 self.ws_server_config.is_some()
1448 {
1449 let cors = match (self.ws_cors_domains.as_ref(), self.http_cors_domains.as_ref()) {
1450 (Some(ws_cors), Some(http_cors)) => {
1451 if ws_cors.trim() != http_cors.trim() {
1452 return Err(WsHttpSamePortError::ConflictingCorsDomains {
1453 http_cors_domains: Some(http_cors.clone()),
1454 ws_cors_domains: Some(ws_cors.clone()),
1455 }
1456 .into());
1457 }
1458 Some(ws_cors)
1459 }
1460 (a, b) => a.or(b),
1461 }
1462 .cloned();
1463
1464 modules.config.ensure_ws_http_identical()?;
1466
1467 if let Some(config) = self.http_server_config {
1468 let server = ServerBuilder::new()
1469 .set_http_middleware(
1470 tower::ServiceBuilder::new()
1471 .option_layer(Self::maybe_cors_layer(cors.clone())?)
1472 .option_layer(Self::maybe_jwt_layer(self.jwt_secret))
1473 .option_layer(Self::maybe_decompression_layer(
1474 self.http_max_request_body_size,
1475 self.http_decompression_algorithms.as_deref(),
1476 ))
1477 .option_layer(Self::maybe_compression_layer(
1478 http_socket_addr,
1479 self.http_disable_compression,
1480 self.http_compression_algorithms.as_deref(),
1481 ))
1482 .layer(ValidateRequestHeaderLayer::custom(
1483 cors::WsOriginValidator::new(cors.as_deref())?,
1484 )),
1485 )
1486 .set_rpc_middleware(
1487 RpcServiceBuilder::default()
1488 .option_layer(
1489 rpc_metrics_enabled
1490 .then(|| {
1491 modules
1492 .http
1493 .as_ref()
1494 .or(modules.ws.as_ref())
1495 .map(RpcRequestMetrics::same_port)
1496 })
1497 .flatten(),
1498 )
1499 .layer(self.rpc_middleware.clone()),
1500 )
1501 .set_config(config.build())
1502 .build(http_socket_addr)
1503 .await
1504 .map_err(|err| {
1505 RpcError::server_error(err, ServerKind::WsHttp(http_socket_addr))
1506 })?;
1507 let addr = server.local_addr().map_err(|err| {
1508 RpcError::server_error(err, ServerKind::WsHttp(http_socket_addr))
1509 })?;
1510 if let Some(module) = modules.http.as_ref().or(modules.ws.as_ref()) {
1511 let handle = server.start(module.clone());
1512 http_handle = Some(handle.clone());
1513 ws_handle = Some(handle);
1514 }
1515 return Ok(RpcServerHandle {
1516 http_local_addr: Some(addr),
1517 ws_local_addr: Some(addr),
1518 http: http_handle,
1519 ws: ws_handle,
1520 ipc_endpoint: self.ipc_endpoint.clone(),
1521 ipc: ipc_handle,
1522 jwt_secret: self.jwt_secret,
1523 });
1524 }
1525 }
1526
1527 let mut ws_local_addr = None;
1528 let mut ws_server = None;
1529 let mut http_local_addr = None;
1530 let mut http_server = None;
1531
1532 if let Some(config) = self.ws_server_config {
1533 let server = ServerBuilder::new()
1534 .set_config(config.ws_only().build())
1535 .set_http_middleware(
1536 tower::ServiceBuilder::new()
1537 .option_layer(Self::maybe_cors_layer(self.ws_cors_domains.clone())?)
1538 .option_layer(Self::maybe_jwt_layer(self.jwt_secret))
1539 .layer(ValidateRequestHeaderLayer::custom(cors::WsOriginValidator::new(
1540 self.ws_cors_domains.as_deref(),
1541 )?)),
1542 )
1543 .set_rpc_middleware(
1544 RpcServiceBuilder::default()
1545 .option_layer(
1546 rpc_metrics_enabled
1547 .then(|| modules.ws.as_ref().map(RpcRequestMetrics::ws))
1548 .flatten(),
1549 )
1550 .layer(self.rpc_middleware.clone()),
1551 )
1552 .build(ws_socket_addr)
1553 .await
1554 .map_err(|err| RpcError::server_error(err, ServerKind::WS(ws_socket_addr)))?;
1555
1556 let addr = server
1557 .local_addr()
1558 .map_err(|err| RpcError::server_error(err, ServerKind::WS(ws_socket_addr)))?;
1559
1560 ws_local_addr = Some(addr);
1561 ws_server = Some(server);
1562 }
1563
1564 if let Some(config) = self.http_server_config {
1565 let server = ServerBuilder::new()
1566 .set_config(config.http_only().build())
1567 .set_http_middleware(
1568 tower::ServiceBuilder::new()
1569 .option_layer(Self::maybe_cors_layer(self.http_cors_domains.clone())?)
1570 .option_layer(Self::maybe_jwt_layer(self.jwt_secret))
1571 .option_layer(Self::maybe_decompression_layer(
1572 self.http_max_request_body_size,
1573 self.http_decompression_algorithms.as_deref(),
1574 ))
1575 .option_layer(Self::maybe_compression_layer(
1576 http_socket_addr,
1577 self.http_disable_compression,
1578 self.http_compression_algorithms.as_deref(),
1579 )),
1580 )
1581 .set_rpc_middleware(
1582 RpcServiceBuilder::default()
1583 .option_layer(
1584 rpc_metrics_enabled
1585 .then(|| modules.http.as_ref().map(RpcRequestMetrics::http))
1586 .flatten(),
1587 )
1588 .layer(self.rpc_middleware.clone()),
1589 )
1590 .build(http_socket_addr)
1591 .await
1592 .map_err(|err| RpcError::server_error(err, ServerKind::Http(http_socket_addr)))?;
1593 let local_addr = server
1594 .local_addr()
1595 .map_err(|err| RpcError::server_error(err, ServerKind::Http(http_socket_addr)))?;
1596 http_local_addr = Some(local_addr);
1597 http_server = Some(server);
1598 }
1599
1600 http_handle = http_server
1601 .map(|http_server| http_server.start(modules.http.clone().expect("http server error")));
1602 ws_handle = ws_server
1603 .map(|ws_server| ws_server.start(modules.ws.clone().expect("ws server error")));
1604 Ok(RpcServerHandle {
1605 http_local_addr,
1606 ws_local_addr,
1607 http: http_handle,
1608 ws: ws_handle,
1609 ipc_endpoint: self.ipc_endpoint.clone(),
1610 ipc: ipc_handle,
1611 jwt_secret: self.jwt_secret,
1612 })
1613 }
1614}
1615
1616#[derive(Debug, Clone, Default, Eq, PartialEq)]
1628pub struct TransportRpcModuleConfig {
1629 http: Option<RpcModuleSelection>,
1631 ws: Option<RpcModuleSelection>,
1633 ipc: Option<RpcModuleSelection>,
1635 config: Option<RpcModuleConfig>,
1637}
1638
1639impl TransportRpcModuleConfig {
1642 pub fn set_http(http: impl Into<RpcModuleSelection>) -> Self {
1644 Self::default().with_http(http)
1645 }
1646
1647 pub fn set_ws(ws: impl Into<RpcModuleSelection>) -> Self {
1649 Self::default().with_ws(ws)
1650 }
1651
1652 pub fn set_ipc(ipc: impl Into<RpcModuleSelection>) -> Self {
1654 Self::default().with_ipc(ipc)
1655 }
1656
1657 pub fn with_http(mut self, http: impl Into<RpcModuleSelection>) -> Self {
1659 self.http = Some(http.into());
1660 self
1661 }
1662
1663 pub fn with_ws(mut self, ws: impl Into<RpcModuleSelection>) -> Self {
1665 self.ws = Some(ws.into());
1666 self
1667 }
1668
1669 pub fn with_ipc(mut self, ipc: impl Into<RpcModuleSelection>) -> Self {
1671 self.ipc = Some(ipc.into());
1672 self
1673 }
1674
1675 pub fn with_config(mut self, config: RpcModuleConfig) -> Self {
1677 self.config = Some(config);
1678 self
1679 }
1680
1681 pub const fn http_mut(&mut self) -> &mut Option<RpcModuleSelection> {
1683 &mut self.http
1684 }
1685
1686 pub const fn ws_mut(&mut self) -> &mut Option<RpcModuleSelection> {
1688 &mut self.ws
1689 }
1690
1691 pub const fn ipc_mut(&mut self) -> &mut Option<RpcModuleSelection> {
1693 &mut self.ipc
1694 }
1695
1696 pub const fn config_mut(&mut self) -> &mut Option<RpcModuleConfig> {
1698 &mut self.config
1699 }
1700
1701 pub const fn is_empty(&self) -> bool {
1703 self.http.is_none() && self.ws.is_none() && self.ipc.is_none()
1704 }
1705
1706 pub const fn http(&self) -> Option<&RpcModuleSelection> {
1708 self.http.as_ref()
1709 }
1710
1711 pub const fn ws(&self) -> Option<&RpcModuleSelection> {
1713 self.ws.as_ref()
1714 }
1715
1716 pub const fn ipc(&self) -> Option<&RpcModuleSelection> {
1718 self.ipc.as_ref()
1719 }
1720
1721 pub const fn config(&self) -> Option<&RpcModuleConfig> {
1723 self.config.as_ref()
1724 }
1725
1726 pub fn contains_any(&self, module: &RethRpcModule) -> bool {
1728 self.contains_http(module) || self.contains_ws(module) || self.contains_ipc(module)
1729 }
1730
1731 pub fn contains_http(&self, module: &RethRpcModule) -> bool {
1733 self.http.as_ref().is_some_and(|http| http.contains(module))
1734 }
1735
1736 pub fn contains_ws(&self, module: &RethRpcModule) -> bool {
1738 self.ws.as_ref().is_some_and(|ws| ws.contains(module))
1739 }
1740
1741 pub fn contains_ipc(&self, module: &RethRpcModule) -> bool {
1743 self.ipc.as_ref().is_some_and(|ipc| ipc.contains(module))
1744 }
1745
1746 fn ensure_ws_http_identical(&self) -> Result<(), WsHttpSamePortError> {
1749 if RpcModuleSelection::are_identical(self.http.as_ref(), self.ws.as_ref()) {
1750 Ok(())
1751 } else {
1752 let http_modules =
1753 self.http.as_ref().map(RpcModuleSelection::to_selection).unwrap_or_default();
1754 let ws_modules =
1755 self.ws.as_ref().map(RpcModuleSelection::to_selection).unwrap_or_default();
1756
1757 let http_not_ws = http_modules.difference(&ws_modules).cloned().collect();
1758 let ws_not_http = ws_modules.difference(&http_modules).cloned().collect();
1759 let overlap = http_modules.intersection(&ws_modules).cloned().collect();
1760
1761 Err(WsHttpSamePortError::ConflictingModules(Box::new(ConflictingModules {
1762 overlap,
1763 http_not_ws,
1764 ws_not_http,
1765 })))
1766 }
1767 }
1768}
1769
1770#[derive(Debug, Clone, Default)]
1772pub struct TransportRpcModules<Context = ()> {
1773 config: TransportRpcModuleConfig,
1775 http: Option<RpcModule<Context>>,
1777 ws: Option<RpcModule<Context>>,
1779 ipc: Option<RpcModule<Context>>,
1781}
1782
1783impl TransportRpcModules {
1786 pub fn with_config(mut self, config: TransportRpcModuleConfig) -> Self {
1789 self.config = config;
1790 self
1791 }
1792
1793 pub fn with_http(mut self, http: RpcModule<()>) -> Self {
1796 self.http = Some(http);
1797 self
1798 }
1799
1800 pub fn with_ws(mut self, ws: RpcModule<()>) -> Self {
1803 self.ws = Some(ws);
1804 self
1805 }
1806
1807 pub fn with_ipc(mut self, ipc: RpcModule<()>) -> Self {
1810 self.ipc = Some(ipc);
1811 self
1812 }
1813
1814 pub const fn module_config(&self) -> &TransportRpcModuleConfig {
1816 &self.config
1817 }
1818
1819 pub fn merge_if_module_configured(
1824 &mut self,
1825 module: RethRpcModule,
1826 other: impl Into<Methods>,
1827 ) -> Result<(), RegisterMethodError> {
1828 let other = other.into();
1829 if self.module_config().contains_http(&module) {
1830 self.merge_http(other.clone())?;
1831 }
1832 if self.module_config().contains_ws(&module) {
1833 self.merge_ws(other.clone())?;
1834 }
1835 if self.module_config().contains_ipc(&module) {
1836 self.merge_ipc(other)?;
1837 }
1838
1839 Ok(())
1840 }
1841
1842 pub fn merge_if_module_configured_with<F>(
1849 &mut self,
1850 module: RethRpcModule,
1851 f: F,
1852 ) -> Result<(), RegisterMethodError>
1853 where
1854 F: FnOnce() -> Methods,
1855 {
1856 if !self.module_config().contains_any(&module) {
1858 return Ok(());
1859 }
1860 self.merge_if_module_configured(module, f())
1861 }
1862
1863 pub fn merge_http(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
1869 if let Some(ref mut http) = self.http {
1870 return http.merge(other.into()).map(|_| true)
1871 }
1872 Ok(false)
1873 }
1874
1875 pub fn merge_ws(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
1881 if let Some(ref mut ws) = self.ws {
1882 return ws.merge(other.into()).map(|_| true)
1883 }
1884 Ok(false)
1885 }
1886
1887 pub fn merge_ipc(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
1893 if let Some(ref mut ipc) = self.ipc {
1894 return ipc.merge(other.into()).map(|_| true)
1895 }
1896 Ok(false)
1897 }
1898
1899 pub fn merge_configured(
1903 &mut self,
1904 other: impl Into<Methods>,
1905 ) -> Result<(), RegisterMethodError> {
1906 let other = other.into();
1907 self.merge_http(other.clone())?;
1908 self.merge_ws(other.clone())?;
1909 self.merge_ipc(other)?;
1910 Ok(())
1911 }
1912
1913 pub fn methods_by_module(&self, module: RethRpcModule) -> Methods {
1917 self.methods_by(|name| name.starts_with(module.as_str()))
1918 }
1919
1920 pub fn methods_by<F>(&self, mut filter: F) -> Methods
1924 where
1925 F: FnMut(&str) -> bool,
1926 {
1927 let mut methods = Methods::new();
1928
1929 let mut f =
1931 |name: &str, mm: &Methods| filter(name) && !mm.method_names().any(|m| m == name);
1932
1933 if let Some(m) = self.http_methods(|name| f(name, &methods)) {
1934 let _ = methods.merge(m);
1935 }
1936 if let Some(m) = self.ws_methods(|name| f(name, &methods)) {
1937 let _ = methods.merge(m);
1938 }
1939 if let Some(m) = self.ipc_methods(|name| f(name, &methods)) {
1940 let _ = methods.merge(m);
1941 }
1942 methods
1943 }
1944
1945 pub fn http_methods<F>(&self, filter: F) -> Option<Methods>
1949 where
1950 F: FnMut(&str) -> bool,
1951 {
1952 self.http.as_ref().map(|module| methods_by(module, filter))
1953 }
1954
1955 pub fn ws_methods<F>(&self, filter: F) -> Option<Methods>
1959 where
1960 F: FnMut(&str) -> bool,
1961 {
1962 self.ws.as_ref().map(|module| methods_by(module, filter))
1963 }
1964
1965 pub fn ipc_methods<F>(&self, filter: F) -> Option<Methods>
1969 where
1970 F: FnMut(&str) -> bool,
1971 {
1972 self.ipc.as_ref().map(|module| methods_by(module, filter))
1973 }
1974
1975 pub fn remove_http_method(&mut self, method_name: &'static str) -> bool {
1983 if let Some(http_module) = &mut self.http {
1984 http_module.remove_method(method_name).is_some()
1985 } else {
1986 false
1987 }
1988 }
1989
1990 pub fn remove_http_methods(&mut self, methods: impl IntoIterator<Item = &'static str>) {
1992 for name in methods {
1993 self.remove_http_method(name);
1994 }
1995 }
1996
1997 pub fn remove_ws_method(&mut self, method_name: &'static str) -> bool {
2005 if let Some(ws_module) = &mut self.ws {
2006 ws_module.remove_method(method_name).is_some()
2007 } else {
2008 false
2009 }
2010 }
2011
2012 pub fn remove_ws_methods(&mut self, methods: impl IntoIterator<Item = &'static str>) {
2014 for name in methods {
2015 self.remove_ws_method(name);
2016 }
2017 }
2018
2019 pub fn remove_ipc_method(&mut self, method_name: &'static str) -> bool {
2027 if let Some(ipc_module) = &mut self.ipc {
2028 ipc_module.remove_method(method_name).is_some()
2029 } else {
2030 false
2031 }
2032 }
2033
2034 pub fn remove_ipc_methods(&mut self, methods: impl IntoIterator<Item = &'static str>) {
2036 for name in methods {
2037 self.remove_ipc_method(name);
2038 }
2039 }
2040
2041 pub fn remove_method_from_configured(&mut self, method_name: &'static str) -> bool {
2045 let http_removed = self.remove_http_method(method_name);
2046 let ws_removed = self.remove_ws_method(method_name);
2047 let ipc_removed = self.remove_ipc_method(method_name);
2048
2049 http_removed || ws_removed || ipc_removed
2050 }
2051
2052 pub fn rename(
2056 &mut self,
2057 old_name: &'static str,
2058 new_method: impl Into<Methods>,
2059 ) -> Result<(), RegisterMethodError> {
2060 self.remove_method_from_configured(old_name);
2062
2063 self.merge_configured(new_method)
2065 }
2066
2067 pub fn replace_http(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
2074 let other = other.into();
2075 self.remove_http_methods(other.method_names());
2076 self.merge_http(other)
2077 }
2078
2079 pub fn replace_ipc(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
2086 let other = other.into();
2087 self.remove_ipc_methods(other.method_names());
2088 self.merge_ipc(other)
2089 }
2090
2091 pub fn replace_ws(&mut self, other: impl Into<Methods>) -> Result<bool, RegisterMethodError> {
2098 let other = other.into();
2099 self.remove_ws_methods(other.method_names());
2100 self.merge_ws(other)
2101 }
2102
2103 pub fn replace_configured(
2107 &mut self,
2108 other: impl Into<Methods>,
2109 ) -> Result<bool, RegisterMethodError> {
2110 let other = other.into();
2111 self.replace_http(other.clone())?;
2112 self.replace_ws(other.clone())?;
2113 self.replace_ipc(other)?;
2114 Ok(true)
2115 }
2116
2117 pub fn add_or_replace_http(
2121 &mut self,
2122 other: impl Into<Methods>,
2123 ) -> Result<bool, RegisterMethodError> {
2124 let other = other.into();
2125 self.remove_http_methods(other.method_names());
2126 self.merge_http(other)
2127 }
2128
2129 pub fn add_or_replace_ws(
2133 &mut self,
2134 other: impl Into<Methods>,
2135 ) -> Result<bool, RegisterMethodError> {
2136 let other = other.into();
2137 self.remove_ws_methods(other.method_names());
2138 self.merge_ws(other)
2139 }
2140
2141 pub fn add_or_replace_ipc(
2145 &mut self,
2146 other: impl Into<Methods>,
2147 ) -> Result<bool, RegisterMethodError> {
2148 let other = other.into();
2149 self.remove_ipc_methods(other.method_names());
2150 self.merge_ipc(other)
2151 }
2152
2153 pub fn add_or_replace_configured(
2155 &mut self,
2156 other: impl Into<Methods>,
2157 ) -> Result<(), RegisterMethodError> {
2158 let other = other.into();
2159 self.add_or_replace_http(other.clone())?;
2160 self.add_or_replace_ws(other.clone())?;
2161 self.add_or_replace_ipc(other)?;
2162 Ok(())
2163 }
2164 pub fn add_or_replace_if_module_configured(
2167 &mut self,
2168 module: RethRpcModule,
2169 other: impl Into<Methods>,
2170 ) -> Result<(), RegisterMethodError> {
2171 let other = other.into();
2172 if self.module_config().contains_http(&module) {
2173 self.add_or_replace_http(other.clone())?;
2174 }
2175 if self.module_config().contains_ws(&module) {
2176 self.add_or_replace_ws(other.clone())?;
2177 }
2178 if self.module_config().contains_ipc(&module) {
2179 self.add_or_replace_ipc(other)?;
2180 }
2181 Ok(())
2182 }
2183}
2184
2185fn methods_by<T, F>(module: &RpcModule<T>, mut filter: F) -> Methods
2187where
2188 F: FnMut(&str) -> bool,
2189{
2190 let mut methods = Methods::new();
2191 let method_names = module.method_names().filter(|name| filter(name));
2192
2193 for name in method_names {
2194 if let Some(matched_method) = module.method(name).cloned() {
2195 let _ = methods.verify_and_insert(name, matched_method);
2196 }
2197 }
2198
2199 methods
2200}
2201
2202#[derive(Clone, Debug)]
2207#[must_use = "Server stops if dropped"]
2208pub struct RpcServerHandle {
2209 http_local_addr: Option<SocketAddr>,
2211 ws_local_addr: Option<SocketAddr>,
2212 http: Option<ServerHandle>,
2213 ws: Option<ServerHandle>,
2214 ipc_endpoint: Option<String>,
2215 ipc: Option<jsonrpsee::server::ServerHandle>,
2216 jwt_secret: Option<JwtSecret>,
2217}
2218
2219impl RpcServerHandle {
2222 fn bearer_token(&self) -> Option<String> {
2224 self.jwt_secret.as_ref().map(|secret| {
2225 format!(
2226 "Bearer {}",
2227 secret
2228 .encode(&Claims {
2229 iat: (SystemTime::now().duration_since(UNIX_EPOCH).unwrap() +
2230 Duration::from_secs(60))
2231 .as_secs(),
2232 exp: None,
2233 })
2234 .unwrap()
2235 )
2236 })
2237 }
2238 pub const fn http_local_addr(&self) -> Option<SocketAddr> {
2240 self.http_local_addr
2241 }
2242
2243 pub const fn ws_local_addr(&self) -> Option<SocketAddr> {
2245 self.ws_local_addr
2246 }
2247
2248 pub fn stop(self) -> Result<(), AlreadyStoppedError> {
2250 if let Some(handle) = self.http {
2251 handle.stop()?
2252 }
2253
2254 if let Some(handle) = self.ws {
2255 handle.stop()?
2256 }
2257
2258 if let Some(handle) = self.ipc {
2259 handle.stop()?
2260 }
2261
2262 Ok(())
2263 }
2264
2265 pub fn ipc_endpoint(&self) -> Option<String> {
2267 self.ipc_endpoint.clone()
2268 }
2269
2270 pub fn http_url(&self) -> Option<String> {
2272 self.http_local_addr.map(|addr| format!("http://{addr}"))
2273 }
2274
2275 pub fn ws_url(&self) -> Option<String> {
2277 self.ws_local_addr.map(|addr| format!("ws://{addr}"))
2278 }
2279
2280 pub fn http_client(&self) -> Option<jsonrpsee::http_client::HttpClient> {
2282 let url = self.http_url()?;
2283
2284 let client = if let Some(token) = self.bearer_token() {
2285 jsonrpsee::http_client::HttpClientBuilder::default()
2286 .set_headers(HeaderMap::from_iter([(AUTHORIZATION, token.parse().unwrap())]))
2287 .build(url)
2288 } else {
2289 jsonrpsee::http_client::HttpClientBuilder::default().build(url)
2290 };
2291
2292 client.expect("failed to create http client").into()
2293 }
2294
2295 pub async fn ws_client(&self) -> Option<jsonrpsee::ws_client::WsClient> {
2297 let url = self.ws_url()?;
2298 let mut builder = jsonrpsee::ws_client::WsClientBuilder::default();
2299
2300 if let Some(token) = self.bearer_token() {
2301 let headers = HeaderMap::from_iter([(AUTHORIZATION, token.parse().unwrap())]);
2302 builder = builder.set_headers(headers);
2303 }
2304
2305 let client = builder.build(url).await.expect("failed to create ws client");
2306 Some(client)
2307 }
2308
2309 pub fn eth_http_provider(
2311 &self,
2312 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static> {
2313 self.new_http_provider_for()
2314 }
2315
2316 pub fn eth_http_provider_with_wallet<W>(
2319 &self,
2320 wallet: W,
2321 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static>
2322 where
2323 W: IntoWallet<alloy_network::Ethereum, NetworkWallet: Clone + Unpin + 'static>,
2324 {
2325 let rpc_url = self.http_url()?;
2326 let provider =
2327 ProviderBuilder::new().wallet(wallet).connect_http(rpc_url.parse().expect("valid url"));
2328 Some(provider)
2329 }
2330
2331 pub fn new_http_provider_for<N>(&self) -> Option<impl Provider<N> + Clone + Unpin + 'static>
2336 where
2337 N: RecommendedFillers<RecommendedFillers: Unpin>,
2338 {
2339 let rpc_url = self.http_url()?;
2340 let provider = ProviderBuilder::default()
2341 .with_recommended_fillers()
2342 .connect_http(rpc_url.parse().expect("valid url"));
2343 Some(provider)
2344 }
2345
2346 pub async fn eth_ws_provider(
2348 &self,
2349 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static> {
2350 self.new_ws_provider_for().await
2351 }
2352
2353 pub async fn eth_ws_provider_with_wallet<W>(
2356 &self,
2357 wallet: W,
2358 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static>
2359 where
2360 W: IntoWallet<alloy_network::Ethereum, NetworkWallet: Clone + Unpin + 'static>,
2361 {
2362 let rpc_url = self.ws_url()?;
2363 let provider = ProviderBuilder::new()
2364 .wallet(wallet)
2365 .connect(&rpc_url)
2366 .await
2367 .expect("failed to create ws client");
2368 Some(provider)
2369 }
2370
2371 pub async fn new_ws_provider_for<N>(&self) -> Option<impl Provider<N> + Clone + Unpin + 'static>
2376 where
2377 N: RecommendedFillers<RecommendedFillers: Unpin>,
2378 {
2379 let rpc_url = self.ws_url()?;
2380 let provider = ProviderBuilder::default()
2381 .with_recommended_fillers()
2382 .connect(&rpc_url)
2383 .await
2384 .expect("failed to create ws client");
2385 Some(provider)
2386 }
2387
2388 pub async fn eth_ipc_provider(
2390 &self,
2391 ) -> Option<impl Provider<alloy_network::Ethereum> + Clone + Unpin + 'static> {
2392 self.new_ipc_provider_for().await
2393 }
2394
2395 pub async fn new_ipc_provider_for<N>(
2400 &self,
2401 ) -> Option<impl Provider<N> + Clone + Unpin + 'static>
2402 where
2403 N: RecommendedFillers<RecommendedFillers: Unpin>,
2404 {
2405 let rpc_url = self.ipc_endpoint()?;
2406 let provider = ProviderBuilder::default()
2407 .with_recommended_fillers()
2408 .connect(&rpc_url)
2409 .await
2410 .expect("failed to create ipc client");
2411 Some(provider)
2412 }
2413}
2414
2415#[cfg(test)]
2416mod tests {
2417 use super::*;
2418
2419 #[test]
2420 fn parse_eth_call_bundle_selection() {
2421 let selection = "eth,admin,debug".parse::<RpcModuleSelection>().unwrap();
2422 assert_eq!(
2423 selection,
2424 RpcModuleSelection::Selection(
2425 [RethRpcModule::Eth, RethRpcModule::Admin, RethRpcModule::Debug,].into()
2426 )
2427 );
2428 }
2429
2430 #[test]
2431 fn parse_rpc_module_selection() {
2432 let selection = "all".parse::<RpcModuleSelection>().unwrap();
2433 assert_eq!(selection, RpcModuleSelection::All);
2434 }
2435
2436 #[test]
2437 fn parse_rpc_module_selection_none() {
2438 let selection = "none".parse::<RpcModuleSelection>().unwrap();
2439 assert_eq!(selection, RpcModuleSelection::Selection(Default::default()));
2440 }
2441
2442 #[test]
2443 fn parse_rpc_unique_module_selection() {
2444 let selection = "eth,admin,eth,net".parse::<RpcModuleSelection>().unwrap();
2445 assert_eq!(
2446 selection,
2447 RpcModuleSelection::Selection(
2448 [RethRpcModule::Eth, RethRpcModule::Admin, RethRpcModule::Net,].into()
2449 )
2450 );
2451 }
2452
2453 #[test]
2454 fn identical_selection() {
2455 assert!(RpcModuleSelection::are_identical(
2456 Some(&RpcModuleSelection::All),
2457 Some(&RpcModuleSelection::All),
2458 ));
2459 assert!(!RpcModuleSelection::are_identical(
2460 Some(&RpcModuleSelection::All),
2461 Some(&RpcModuleSelection::Standard),
2462 ));
2463 assert!(RpcModuleSelection::are_identical(
2464 Some(&RpcModuleSelection::Selection(RpcModuleSelection::Standard.to_selection())),
2465 Some(&RpcModuleSelection::Standard),
2466 ));
2467 assert!(RpcModuleSelection::are_identical(
2468 Some(&RpcModuleSelection::Selection([RethRpcModule::Eth].into())),
2469 Some(&RpcModuleSelection::Selection([RethRpcModule::Eth].into())),
2470 ));
2471 assert!(RpcModuleSelection::are_identical(
2472 None,
2473 Some(&RpcModuleSelection::Selection(Default::default())),
2474 ));
2475 assert!(RpcModuleSelection::are_identical(
2476 Some(&RpcModuleSelection::Selection(Default::default())),
2477 None,
2478 ));
2479 assert!(RpcModuleSelection::are_identical(None, None));
2480 }
2481
2482 #[test]
2483 fn test_rpc_module_str() {
2484 macro_rules! assert_rpc_module {
2485 ($($s:expr => $v:expr,)*) => {
2486 $(
2487 let val: RethRpcModule = $s.parse().unwrap();
2488 assert_eq!(val, $v);
2489 assert_eq!(val.to_string(), $s);
2490 )*
2491 };
2492 }
2493 assert_rpc_module!
2494 (
2495 "admin" => RethRpcModule::Admin,
2496 "debug" => RethRpcModule::Debug,
2497 "eth" => RethRpcModule::Eth,
2498 "net" => RethRpcModule::Net,
2499 "trace" => RethRpcModule::Trace,
2500 "web3" => RethRpcModule::Web3,
2501 "rpc" => RethRpcModule::Rpc,
2502 "ots" => RethRpcModule::Ots,
2503 "reth" => RethRpcModule::Reth,
2504 );
2505 }
2506
2507 #[test]
2508 fn test_default_selection() {
2509 let selection = RpcModuleSelection::Standard.to_selection();
2510 assert_eq!(selection, [RethRpcModule::Eth, RethRpcModule::Net, RethRpcModule::Web3].into())
2511 }
2512
2513 #[test]
2514 fn test_create_rpc_module_config() {
2515 let selection = vec!["eth", "admin"];
2516 let config = RpcModuleSelection::try_from_selection(selection).unwrap();
2517 assert_eq!(
2518 config,
2519 RpcModuleSelection::Selection([RethRpcModule::Eth, RethRpcModule::Admin].into())
2520 );
2521 }
2522
2523 #[test]
2524 fn test_configure_transport_config() {
2525 let config = TransportRpcModuleConfig::default()
2526 .with_http([RethRpcModule::Eth, RethRpcModule::Admin]);
2527 assert_eq!(
2528 config,
2529 TransportRpcModuleConfig {
2530 http: Some(RpcModuleSelection::Selection(
2531 [RethRpcModule::Eth, RethRpcModule::Admin].into()
2532 )),
2533 ws: None,
2534 ipc: None,
2535 config: None,
2536 }
2537 )
2538 }
2539
2540 #[test]
2541 fn test_configure_transport_config_none() {
2542 let config = TransportRpcModuleConfig::default().with_http(Vec::<RethRpcModule>::new());
2543 assert_eq!(
2544 config,
2545 TransportRpcModuleConfig {
2546 http: Some(RpcModuleSelection::Selection(Default::default())),
2547 ws: None,
2548 ipc: None,
2549 config: None,
2550 }
2551 )
2552 }
2553
2554 fn create_test_module() -> RpcModule<()> {
2555 let mut module = RpcModule::new(());
2556 module.register_method("anything", |_, _, _| "succeed").unwrap();
2557 module
2558 }
2559
2560 #[test]
2561 fn test_remove_http_method() {
2562 let mut modules =
2563 TransportRpcModules { http: Some(create_test_module()), ..Default::default() };
2564 assert!(modules.remove_http_method("anything"));
2566
2567 assert!(!modules.remove_http_method("non_existent_method"));
2569
2570 assert!(modules.http.as_ref().unwrap().method("anything").is_none());
2572 }
2573
2574 #[test]
2575 fn test_remove_ws_method() {
2576 let mut modules =
2577 TransportRpcModules { ws: Some(create_test_module()), ..Default::default() };
2578
2579 assert!(modules.remove_ws_method("anything"));
2581
2582 assert!(!modules.remove_ws_method("non_existent_method"));
2584
2585 assert!(modules.ws.as_ref().unwrap().method("anything").is_none());
2587 }
2588
2589 #[test]
2590 fn test_remove_ipc_method() {
2591 let mut modules =
2592 TransportRpcModules { ipc: Some(create_test_module()), ..Default::default() };
2593
2594 assert!(modules.remove_ipc_method("anything"));
2596
2597 assert!(!modules.remove_ipc_method("non_existent_method"));
2599
2600 assert!(modules.ipc.as_ref().unwrap().method("anything").is_none());
2602 }
2603
2604 #[test]
2605 fn test_remove_method_from_configured() {
2606 let mut modules = TransportRpcModules {
2607 http: Some(create_test_module()),
2608 ws: Some(create_test_module()),
2609 ipc: Some(create_test_module()),
2610 ..Default::default()
2611 };
2612
2613 assert!(modules.remove_method_from_configured("anything"));
2615
2616 assert!(!modules.remove_method_from_configured("anything"));
2618
2619 assert!(!modules.remove_method_from_configured("non_existent_method"));
2621
2622 assert!(modules.http.as_ref().unwrap().method("anything").is_none());
2624 assert!(modules.ws.as_ref().unwrap().method("anything").is_none());
2625 assert!(modules.ipc.as_ref().unwrap().method("anything").is_none());
2626 }
2627
2628 #[test]
2629 fn test_transport_rpc_module_rename() {
2630 let mut modules = TransportRpcModules {
2631 http: Some(create_test_module()),
2632 ws: Some(create_test_module()),
2633 ipc: Some(create_test_module()),
2634 ..Default::default()
2635 };
2636
2637 assert!(modules.http.as_ref().unwrap().method("anything").is_some());
2639 assert!(modules.ws.as_ref().unwrap().method("anything").is_some());
2640 assert!(modules.ipc.as_ref().unwrap().method("anything").is_some());
2641
2642 assert!(modules.http.as_ref().unwrap().method("something").is_none());
2644 assert!(modules.ws.as_ref().unwrap().method("something").is_none());
2645 assert!(modules.ipc.as_ref().unwrap().method("something").is_none());
2646
2647 let mut other_module = RpcModule::new(());
2649 other_module.register_method("something", |_, _, _| "fails").unwrap();
2650
2651 modules.rename("anything", other_module).expect("rename failed");
2653
2654 assert!(modules.http.as_ref().unwrap().method("anything").is_none());
2656 assert!(modules.ws.as_ref().unwrap().method("anything").is_none());
2657 assert!(modules.ipc.as_ref().unwrap().method("anything").is_none());
2658
2659 assert!(modules.http.as_ref().unwrap().method("something").is_some());
2661 assert!(modules.ws.as_ref().unwrap().method("something").is_some());
2662 assert!(modules.ipc.as_ref().unwrap().method("something").is_some());
2663 }
2664
2665 #[test]
2666 fn test_replace_http_method() {
2667 let mut modules =
2668 TransportRpcModules { http: Some(create_test_module()), ..Default::default() };
2669
2670 let mut other_module = RpcModule::new(());
2671 other_module.register_method("something", |_, _, _| "fails").unwrap();
2672
2673 assert!(modules.replace_http(other_module.clone()).unwrap());
2674
2675 assert!(modules.http.as_ref().unwrap().method("something").is_some());
2676
2677 other_module.register_method("anything", |_, _, _| "fails").unwrap();
2678 assert!(modules.replace_http(other_module.clone()).unwrap());
2679
2680 assert!(modules.http.as_ref().unwrap().method("anything").is_some());
2681 }
2682 #[test]
2683 fn test_replace_ipc_method() {
2684 let mut modules =
2685 TransportRpcModules { ipc: Some(create_test_module()), ..Default::default() };
2686
2687 let mut other_module = RpcModule::new(());
2688 other_module.register_method("something", |_, _, _| "fails").unwrap();
2689
2690 assert!(modules.replace_ipc(other_module.clone()).unwrap());
2691
2692 assert!(modules.ipc.as_ref().unwrap().method("something").is_some());
2693
2694 other_module.register_method("anything", |_, _, _| "fails").unwrap();
2695 assert!(modules.replace_ipc(other_module.clone()).unwrap());
2696
2697 assert!(modules.ipc.as_ref().unwrap().method("anything").is_some());
2698 }
2699 #[test]
2700 fn test_replace_ws_method() {
2701 let mut modules =
2702 TransportRpcModules { ws: Some(create_test_module()), ..Default::default() };
2703
2704 let mut other_module = RpcModule::new(());
2705 other_module.register_method("something", |_, _, _| "fails").unwrap();
2706
2707 assert!(modules.replace_ws(other_module.clone()).unwrap());
2708
2709 assert!(modules.ws.as_ref().unwrap().method("something").is_some());
2710
2711 other_module.register_method("anything", |_, _, _| "fails").unwrap();
2712 assert!(modules.replace_ws(other_module.clone()).unwrap());
2713
2714 assert!(modules.ws.as_ref().unwrap().method("anything").is_some());
2715 }
2716
2717 #[test]
2718 fn test_replace_configured() {
2719 let mut modules = TransportRpcModules {
2720 http: Some(create_test_module()),
2721 ws: Some(create_test_module()),
2722 ipc: Some(create_test_module()),
2723 ..Default::default()
2724 };
2725 let mut other_module = RpcModule::new(());
2726 other_module.register_method("something", |_, _, _| "fails").unwrap();
2727
2728 assert!(modules.replace_configured(other_module).unwrap());
2729
2730 assert!(modules.http.as_ref().unwrap().method("something").is_some());
2732 assert!(modules.ipc.as_ref().unwrap().method("something").is_some());
2733 assert!(modules.ws.as_ref().unwrap().method("something").is_some());
2734
2735 assert!(modules.http.as_ref().unwrap().method("anything").is_some());
2736 assert!(modules.ipc.as_ref().unwrap().method("anything").is_some());
2737 assert!(modules.ws.as_ref().unwrap().method("anything").is_some());
2738 }
2739
2740 #[test]
2741 fn test_add_or_replace_if_module_configured() {
2742 let config = TransportRpcModuleConfig::default()
2744 .with_http([RethRpcModule::Eth])
2745 .with_ws([RethRpcModule::Eth]);
2746
2747 let mut http_module = RpcModule::new(());
2749 http_module.register_method("eth_existing", |_, _, _| "original").unwrap();
2750
2751 let mut ws_module = RpcModule::new(());
2753 ws_module.register_method("eth_existing", |_, _, _| "original").unwrap();
2754
2755 let ipc_module = RpcModule::new(());
2757
2758 let mut modules = TransportRpcModules {
2760 config,
2761 http: Some(http_module),
2762 ws: Some(ws_module),
2763 ipc: Some(ipc_module),
2764 };
2765
2766 let mut new_module = RpcModule::new(());
2768 new_module.register_method("eth_existing", |_, _, _| "replaced").unwrap(); new_module.register_method("eth_new", |_, _, _| "added").unwrap(); let new_methods: Methods = new_module.into();
2771
2772 let result = modules.add_or_replace_if_module_configured(RethRpcModule::Eth, new_methods);
2774 assert!(result.is_ok(), "Function should succeed");
2775
2776 let http = modules.http.as_ref().unwrap();
2778 assert!(http.method("eth_existing").is_some());
2779 assert!(http.method("eth_new").is_some());
2780
2781 let ws = modules.ws.as_ref().unwrap();
2783 assert!(ws.method("eth_existing").is_some());
2784 assert!(ws.method("eth_new").is_some());
2785
2786 let ipc = modules.ipc.as_ref().unwrap();
2788 assert!(ipc.method("eth_existing").is_none());
2789 assert!(ipc.method("eth_new").is_none());
2790 }
2791
2792 #[test]
2793 fn test_merge_if_module_configured_with_lazy_evaluation() {
2794 let config = TransportRpcModuleConfig::default().with_http([RethRpcModule::Eth]);
2796
2797 let mut modules =
2798 TransportRpcModules { config, http: Some(RpcModule::new(())), ws: None, ipc: None };
2799
2800 let mut closure_called = false;
2802
2803 let result = modules.merge_if_module_configured_with(RethRpcModule::Eth, || {
2805 closure_called = true;
2806 let mut methods = RpcModule::new(());
2807 methods.register_method("eth_test", |_, _, _| "test").unwrap();
2808 methods.into()
2809 });
2810
2811 assert!(result.is_ok());
2812 assert!(closure_called, "Closure should be called when module is configured");
2813 assert!(modules.http.as_ref().unwrap().method("eth_test").is_some());
2814
2815 closure_called = false;
2817 let result = modules.merge_if_module_configured_with(RethRpcModule::Debug, || {
2818 closure_called = true;
2819 RpcModule::new(()).into()
2820 });
2821
2822 assert!(result.is_ok());
2823 assert!(!closure_called, "Closure should NOT be called when module is not configured");
2824 }
2825}