diff --git a/Cargo.lock b/Cargo.lock index 61aeb4649..a1ee82ef3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4639,6 +4639,7 @@ dependencies = [ "jsonrpsee", "reth-ipc", "reth-network-api", + "reth-primitives", "reth-provider", "reth-rpc", "reth-rpc-api", diff --git a/crates/rpc/rpc-api/src/admin.rs b/crates/rpc/rpc-api/src/admin.rs index f8678d8a3..61a435931 100644 --- a/crates/rpc/rpc-api/src/admin.rs +++ b/crates/rpc/rpc-api/src/admin.rs @@ -1,31 +1,32 @@ -use jsonrpsee::{core::RpcResult as Result, proc_macros::rpc}; +use jsonrpsee::{core::RpcResult, proc_macros::rpc}; use reth_primitives::NodeRecord; use reth_rpc_types::NodeInfo; /// Admin namespace rpc interface that gives access to several non-standard RPC methods. -#[rpc(server)] +#[cfg_attr(not(feature = "client"), rpc(server))] +#[cfg_attr(feature = "client", rpc(server, client))] #[async_trait::async_trait] pub trait AdminApi { /// Adds the given node record to the peerset. #[method(name = "admin_addPeer")] - fn add_peer(&self, record: NodeRecord) -> Result; + fn add_peer(&self, record: NodeRecord) -> RpcResult; /// Disconnects from a remote node if the connection exists. /// /// Returns true if the peer was successfully removed. #[method(name = "admin_removePeer")] - fn remove_peer(&self, record: NodeRecord) -> Result; + fn remove_peer(&self, record: NodeRecord) -> RpcResult; /// Adds the given node record to the trusted peerset. #[method(name = "admin_addTrustedPeer")] - fn add_trusted_peer(&self, record: NodeRecord) -> Result; + fn add_trusted_peer(&self, record: NodeRecord) -> RpcResult; /// Removes a remote node from the trusted peer set, but it does not disconnect it /// automatically. /// /// Returns true if the peer was successfully removed. #[method(name = "admin_removeTrustedPeer")] - fn remove_trusted_peer(&self, record: NodeRecord) -> Result; + fn remove_trusted_peer(&self, record: NodeRecord) -> RpcResult; /// Creates an RPC subscription which serves events received from the network. #[subscription( @@ -33,9 +34,9 @@ pub trait AdminApi { unsubscribe = "admin_peerEvents_unsubscribe", item = String )] - fn subscribe(&self); + fn subscribe_peer_events(&self); /// Returns the ENR of the node. #[method(name = "admin_nodeInfo")] - async fn node_info(&self) -> Result; + async fn node_info(&self) -> RpcResult; } diff --git a/crates/rpc/rpc-api/src/lib.rs b/crates/rpc/rpc-api/src/lib.rs index ec573ede0..f0ca4688d 100644 --- a/crates/rpc/rpc-api/src/lib.rs +++ b/crates/rpc/rpc-api/src/lib.rs @@ -30,3 +30,16 @@ pub mod servers { trace::TraceApiServer, web3::Web3ApiServer, }; } + +/// re-export of all client traits +#[cfg(feature = "client")] +pub use clients::*; + +/// Aggregates all client traits. +#[cfg(feature = "client")] +pub mod clients { + pub use crate::{ + admin::AdminApiClient, debug::DebugApiClient, engine::EngineApiClient, eth::EthApiClient, + net::NetApiClient, trace::TraceApiClient, web3::Web3ApiClient, + }; +} diff --git a/crates/rpc/rpc-builder/Cargo.toml b/crates/rpc/rpc-builder/Cargo.toml index 983617a02..5c7a3f39b 100644 --- a/crates/rpc/rpc-builder/Cargo.toml +++ b/crates/rpc/rpc-builder/Cargo.toml @@ -22,8 +22,10 @@ serde = { version = "1.0", features = ["derive"] } [dev-dependencies] reth-tracing = { path = "../../tracing" } +reth-primitives = { path = "../../primitives" } reth-rpc-api = { path = "../rpc-api", features = ["client"] } reth-transaction-pool = { path = "../../transaction-pool", features = ["test-utils"] } reth-provider = { path = "../../storage/provider", features = ["test-utils"] } +reth-network-api = { path = "../../net/network-api", features = ["test-utils"] } -tokio = { version = "1", features = ["rt", "rt-multi-thread"] } +tokio = { version = "1", features = ["rt", "rt-multi-thread"] } \ No newline at end of file diff --git a/crates/rpc/rpc-builder/src/lib.rs b/crates/rpc/rpc-builder/src/lib.rs index af4ae908f..589cb9775 100644 --- a/crates/rpc/rpc-builder/src/lib.rs +++ b/crates/rpc/rpc-builder/src/lib.rs @@ -426,6 +426,21 @@ pub struct RpcServerConfig { /// === impl RpcServerConfig === impl RpcServerConfig { + /// Creates a new config with only http set + pub fn http(config: ServerBuilder) -> Self { + Self::default().with_http(config) + } + + /// Creates a new config with only ws set + pub fn ws(config: ServerBuilder) -> Self { + Self::default().with_ws(config) + } + + /// Creates a new config with only ipc set + pub fn ipc(config: IpcServerBuilder) -> Self { + Self::default().with_ipc(config) + } + /// Configures the http server pub fn with_http(mut self, config: ServerBuilder) -> Self { self.http_server_config = Some(config.http_only()); @@ -478,15 +493,23 @@ impl RpcServerConfig { .http_ws_addr .unwrap_or(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, DEFAULT_RPC_PORT))); + let mut local_addr = None; + let http_server = if let Some(builder) = self.http_server_config { let server = builder.build(socket_addr).await?; + if local_addr.is_none() { + local_addr = server.local_addr().ok(); + } Some(server) } else { None }; let ws_server = if let Some(builder) = self.ws_server_config { - let server = builder.build(socket_addr).await?; + let server = builder.build(socket_addr).await.unwrap(); + if local_addr.is_none() { + local_addr = server.local_addr().ok(); + } Some(server) } else { None @@ -502,7 +525,7 @@ impl RpcServerConfig { None }; - Ok(RpcServer { http: http_server, ws: ws_server, ipc: ipc_server }) + Ok(RpcServer { local_addr, http: http_server, ws: ws_server, ipc: ipc_server }) } } @@ -530,6 +553,21 @@ pub struct TransportRpcModuleConfig { // === impl TransportRpcModuleConfig === impl TransportRpcModuleConfig { + /// Creates a new config with only http set + pub fn http(http: impl Into) -> Self { + Self::default().with_http(http) + } + + /// Creates a new config with only ws set + pub fn ws(ws: impl Into) -> Self { + Self::default().with_ws(ws) + } + + /// Creates a new config with only ipc set + pub fn ipc(ipc: impl Into) -> Self { + Self::default().with_ipc(ipc) + } + /// Sets the [RpcModuleConfig] for the http transport. pub fn with_http(mut self, http: impl Into) -> Self { self.http = Some(http.into()); @@ -576,6 +614,8 @@ impl TransportRpcModules<()> { /// Container type for each transport ie. http, ws, and ipc server pub struct RpcServer { + /// The address of the http/ws server + local_addr: Option, /// http server http: Option, /// ws server @@ -587,6 +627,11 @@ pub struct RpcServer { // === impl RpcServer === impl RpcServer { + /// Returns the [`SocketAddr`] of the http/ws server if started. + fn local_addr(&self) -> Option { + self.local_addr + } + /// Starts the configured server by spawning the servers on the tokio runtime. /// /// This returns an [RpcServerHandle] that's connected to the server task(s) until the server is @@ -596,7 +641,8 @@ impl RpcServer { modules: TransportRpcModules<()>, ) -> Result { let TransportRpcModules { http, ws, ipc } = modules; - let mut handle = RpcServerHandle { http: None, ws: None, ipc: None }; + let mut handle = + RpcServerHandle { local_addr: self.local_addr, http: None, ws: None, ipc: None }; // Start all servers if let Some((server, module)) = @@ -636,6 +682,8 @@ impl std::fmt::Debug for RpcServer { #[derive(Clone)] #[must_use = "Server stop if dropped"] pub struct RpcServerHandle { + /// The address of the http/ws server + local_addr: Option, http: Option, ws: Option, ipc: Option, @@ -644,6 +692,11 @@ pub struct RpcServerHandle { // === impl RpcServerHandle === impl RpcServerHandle { + /// Returns the [`SocketAddr`] of the http/ws server if started. + fn local_addr(&self) -> Option { + self.local_addr + } + /// Tell the server to stop without waiting for the server to stop. pub fn stop(self) -> Result<(), RpcError> { if let Some(handle) = self.http { @@ -660,6 +713,35 @@ impl RpcServerHandle { Ok(()) } + + /// Returns the url to the http server + pub fn http_url(&self) -> Option { + self.local_addr.map(|addr| format!("http://{addr}")) + } + + /// Returns the url to the ws server + pub fn ws_url(&self) -> Option { + self.local_addr.map(|addr| format!("ws://{addr}")) + } + + /// Returns a http client connected to the server. + pub fn http_client(&self) -> Option { + let url = self.http_url()?; + let client = jsonrpsee::http_client::HttpClientBuilder::default() + .build(url) + .expect("Failed to create http client"); + Some(client) + } + + /// Returns a ws client connected to the server. + pub async fn ws_client(&self) -> Option { + let url = self.ws_url()?; + let client = jsonrpsee::ws_client::WsClientBuilder::default() + .build(url) + .await + .expect("Failed to create ws client"); + Some(client) + } } impl std::fmt::Debug for RpcServerHandle { diff --git a/crates/rpc/rpc-builder/tests/it/http.rs b/crates/rpc/rpc-builder/tests/it/http.rs index ccf0adc40..e00324732 100644 --- a/crates/rpc/rpc-builder/tests/it/http.rs +++ b/crates/rpc/rpc-builder/tests/it/http.rs @@ -1,8 +1,48 @@ //! Standalone http tests -use crate::utils::test_rpc_builder; +use crate::utils::{launch_http, launch_http_ws, launch_ws}; +use jsonrpsee::core::client::{ClientT, SubscriptionClientT}; +use reth_primitives::NodeRecord; +use reth_rpc_api::clients::AdminApiClient; +use reth_rpc_builder::RethRpcModule; + +async fn test_basic_admin_calls(client: &C) +where + C: ClientT + SubscriptionClientT + Sync, +{ + let url = "enode://6f8a80d14311c39f35f516fa664deaaaa13e85b2f7493f37f6144d86991ec012937307647bd3b9a82abe2974e1407241d54947bbb39763a4cac9f77166ad92a0@10.3.58.6:30303?discport=30301"; + let node: NodeRecord = url.parse().unwrap(); + + AdminApiClient::add_peer(client, node).await.unwrap(); + AdminApiClient::remove_peer(client, node).await.unwrap(); + AdminApiClient::add_trusted_peer(client, node).await.unwrap(); + AdminApiClient::remove_trusted_peer(client, node).await.unwrap(); + AdminApiClient::node_info(client).await.unwrap(); +} #[tokio::test(flavor = "multi_thread")] -async fn test_launch_http() { - let _builder = test_rpc_builder(); +async fn test_call_admin_functions_http() { + reth_tracing::init_test_tracing(); + + let handle = launch_http(vec![RethRpcModule::Admin]).await; + let client = handle.http_client().unwrap(); + test_basic_admin_calls(&client).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_call_admin_functions_ws() { + reth_tracing::init_test_tracing(); + + let handle = launch_ws(vec![RethRpcModule::Admin]).await; + let client = handle.ws_client().await.unwrap(); + test_basic_admin_calls(&client).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_call_admin_functions_http_and_ws() { + reth_tracing::init_test_tracing(); + + let handle = launch_http_ws(vec![RethRpcModule::Admin]).await; + let client = handle.http_client().unwrap(); + test_basic_admin_calls(&client).await; } diff --git a/crates/rpc/rpc-builder/tests/it/utils.rs b/crates/rpc/rpc-builder/tests/it/utils.rs index 8dc623c94..11afed706 100644 --- a/crates/rpc/rpc-builder/tests/it/utils.rs +++ b/crates/rpc/rpc-builder/tests/it/utils.rs @@ -1,8 +1,55 @@ +use reth_network_api::test_utils::NoopNetwork; use reth_provider::test_utils::NoopProvider; -use reth_rpc_builder::RpcModuleBuilder; +use reth_rpc_builder::{ + RpcModuleBuilder, RpcModuleConfig, RpcServerConfig, RpcServerHandle, TransportRpcModuleConfig, +}; use reth_transaction_pool::test_utils::{testing_pool, TestPool}; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; + +/// Localhost with port 0 so a free port is used. +pub fn test_address() -> SocketAddr { + SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0)) +} + +/// Launches a new server with http only with the given modules +pub async fn launch_http(modules: impl Into) -> RpcServerHandle { + let builder = test_rpc_builder(); + let server = builder.build(TransportRpcModuleConfig::http(modules)); + server + .start_server(RpcServerConfig::http(Default::default()).with_address(test_address())) + .await + .unwrap() +} + +/// Launches a new server with ws only with the given modules +pub async fn launch_ws(modules: impl Into) -> RpcServerHandle { + let builder = test_rpc_builder(); + let server = builder.build(TransportRpcModuleConfig::ws(modules)); + server + .start_server(RpcServerConfig::ws(Default::default()).with_address(test_address())) + .await + .unwrap() +} + +/// Launches a new server with http and ws and with the given modules +pub async fn launch_http_ws(modules: impl Into) -> RpcServerHandle { + let builder = test_rpc_builder(); + let modules = modules.into(); + let server = builder.build(TransportRpcModuleConfig::ws(modules.clone()).with_http(modules)); + server + .start_server( + RpcServerConfig::ws(Default::default()) + .with_http(Default::default()) + .with_address(test_address()), + ) + .await + .unwrap() +} /// Returns an [RpcModuleBuilder] with testing components. -pub fn test_rpc_builder() -> RpcModuleBuilder { - RpcModuleBuilder::default().with_client(NoopProvider::default()).with_pool(testing_pool()) +pub fn test_rpc_builder() -> RpcModuleBuilder { + RpcModuleBuilder::default() + .with_client(NoopProvider::default()) + .with_pool(testing_pool()) + .with_network(NoopNetwork::default()) } diff --git a/crates/rpc/rpc/src/admin.rs b/crates/rpc/rpc/src/admin.rs index 90e192ecd..eb0bbe8df 100644 --- a/crates/rpc/rpc/src/admin.rs +++ b/crates/rpc/rpc/src/admin.rs @@ -46,7 +46,7 @@ where Ok(true) } - fn subscribe( + fn subscribe_peer_events( &self, _subscription_sink: jsonrpsee::SubscriptionSink, ) -> jsonrpsee::types::SubscriptionResult {