🏗️ start websocket support
This commit is contained in:
parent
21c73c1649
commit
55839e5479
|
@ -19,41 +19,22 @@ use crate::database::{Database, FileDB};
|
|||
use crate::node::Node;
|
||||
use crate::rpc::Rpc;
|
||||
|
||||
pub struct Client<DB: Database> {
|
||||
node: Arc<RwLock<Node>>,
|
||||
rpc: Option<Rpc>,
|
||||
db: Option<DB>,
|
||||
}
|
||||
|
||||
impl Client<FileDB> {
|
||||
fn new(config: Config) -> Result<Self> {
|
||||
let config = Arc::new(config);
|
||||
let node = Node::new(config.clone())?;
|
||||
let node = Arc::new(RwLock::new(node));
|
||||
|
||||
let rpc = config.rpc_port.map(|port| Rpc::new(node.clone(), port));
|
||||
|
||||
let data_dir = config.data_dir.clone();
|
||||
let db = data_dir.map(FileDB::new);
|
||||
|
||||
Ok(Client { node, rpc, db })
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct ClientBuilder {
|
||||
network: Option<Network>,
|
||||
consensus_rpc: Option<String>,
|
||||
execution_rpc: Option<String>,
|
||||
checkpoint: Option<Vec<u8>>,
|
||||
rpc_port: Option<u16>,
|
||||
data_dir: Option<PathBuf>,
|
||||
config: Option<Config>,
|
||||
pub network: Option<Network>,
|
||||
pub consensus_rpc: Option<String>,
|
||||
pub execution_rpc: Option<String>,
|
||||
pub checkpoint: Option<Vec<u8>>,
|
||||
pub rpc_port: Option<u16>,
|
||||
pub data_dir: Option<PathBuf>,
|
||||
pub config: Option<Config>,
|
||||
pub with_ws: bool,
|
||||
pub with_http: bool,
|
||||
}
|
||||
|
||||
impl ClientBuilder {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
Self::default().with_http(true)
|
||||
}
|
||||
|
||||
pub fn network(mut self, network: Network) -> Self {
|
||||
|
@ -77,6 +58,34 @@ impl ClientBuilder {
|
|||
self
|
||||
}
|
||||
|
||||
/// Enables the client to serve a websocket connection.
|
||||
///
|
||||
/// # Example
|
||||
/// ```rust
|
||||
/// let mut client_builder = client::ClientBuilder::new().with_ws(true);
|
||||
/// assert_eq!(client_builder.with_ws, true);
|
||||
/// client_builder = client_builder.with_ws(false);
|
||||
/// assert_eq!(client_builder.with_ws, false);
|
||||
/// ```
|
||||
pub fn with_ws(mut self, option: bool) -> Self {
|
||||
self.with_ws = option;
|
||||
self
|
||||
}
|
||||
|
||||
/// Enables the client to serve an http connection (enabled by default).
|
||||
///
|
||||
/// # Example
|
||||
/// ```rust
|
||||
/// let mut client_builder = client::ClientBuilder::new();
|
||||
/// assert_eq!(client_builder.with_http, true);
|
||||
/// client_builder = client_builder.with_http(false);
|
||||
/// assert_eq!(client_builder.with_http, false);
|
||||
/// ```
|
||||
pub fn with_http(mut self, option: bool) -> Self {
|
||||
self.with_http = option;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn rpc_port(mut self, port: u16) -> Self {
|
||||
self.rpc_port = Some(port);
|
||||
self
|
||||
|
@ -152,16 +161,43 @@ impl ClientBuilder {
|
|||
chain: base_config.chain,
|
||||
forks: base_config.forks,
|
||||
max_checkpoint_age: base_config.max_checkpoint_age,
|
||||
with_ws: self.with_ws,
|
||||
with_http: self.with_http,
|
||||
};
|
||||
|
||||
Client::new(config)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct Client<DB: Database> {
|
||||
node: Arc<RwLock<Node>>,
|
||||
rpc: Option<Rpc>,
|
||||
db: Option<DB>,
|
||||
}
|
||||
|
||||
impl Client<FileDB> {
|
||||
fn new(config: Config) -> Result<Self> {
|
||||
let config = Arc::new(config);
|
||||
let node = Node::new(config.clone())?;
|
||||
let node = Arc::new(RwLock::new(node));
|
||||
|
||||
let rpc = config
|
||||
.rpc_port
|
||||
.map(|port| Rpc::new(node.clone(), config.with_http, config.with_ws, port));
|
||||
|
||||
let data_dir = config.data_dir.clone();
|
||||
let db = data_dir.map(FileDB::new);
|
||||
|
||||
Ok(Client { node, rpc, db })
|
||||
}
|
||||
}
|
||||
|
||||
impl<DB: Database> Client<DB> {
|
||||
pub async fn start(&mut self) -> Result<()> {
|
||||
if let Some(rpc) = &mut self.rpc {
|
||||
rpc.start().await?;
|
||||
// We can start both ws and http servers since they only run if enabled in the config.
|
||||
rpc.start_ws().await?;
|
||||
rpc.start_http().await?;
|
||||
}
|
||||
|
||||
let res = self.node.write().await.sync().await;
|
||||
|
|
|
@ -1,9 +1,5 @@
|
|||
use ethers::{
|
||||
abi::AbiEncode,
|
||||
types::{Address, Filter, Log, Transaction, TransactionReceipt, H256, U256},
|
||||
};
|
||||
use ethers::types::{Address, Filter, Log, Transaction, TransactionReceipt, H256};
|
||||
use eyre::Result;
|
||||
use log::info;
|
||||
use std::{fmt::Display, net::SocketAddr, str::FromStr, sync::Arc};
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
|
@ -11,6 +7,7 @@ use jsonrpsee::{
|
|||
core::{async_trait, server::rpc_module::Methods, Error},
|
||||
http_server::{HttpServerBuilder, HttpServerHandle},
|
||||
proc_macros::rpc,
|
||||
ws_server::{WsServerBuilder, WsServerHandle},
|
||||
};
|
||||
|
||||
use crate::{errors::NodeError, node::Node};
|
||||
|
@ -23,31 +20,43 @@ use execution::types::{CallOpts, ExecutionBlock};
|
|||
|
||||
pub struct Rpc {
|
||||
node: Arc<RwLock<Node>>,
|
||||
handle: Option<HttpServerHandle>,
|
||||
http_handle: Option<HttpServerHandle>,
|
||||
ws_handle: Option<WsServerHandle>,
|
||||
pub with_http: bool,
|
||||
pub with_ws: bool,
|
||||
port: u16,
|
||||
}
|
||||
|
||||
impl Rpc {
|
||||
pub fn new(node: Arc<RwLock<Node>>, port: u16) -> Self {
|
||||
pub fn new(node: Arc<RwLock<Node>>, with_http: bool, with_ws: bool, port: u16) -> Self {
|
||||
Rpc {
|
||||
node,
|
||||
handle: None,
|
||||
http_handle: None,
|
||||
ws_handle: None,
|
||||
with_http,
|
||||
with_ws,
|
||||
port,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn start(&mut self) -> Result<SocketAddr> {
|
||||
let rpc_inner = RpcInner {
|
||||
node: self.node.clone(),
|
||||
port: self.port,
|
||||
};
|
||||
pub async fn start_http(&mut self) -> Result<Option<SocketAddr>> {
|
||||
if self.with_http {
|
||||
let (handle, addr) = RpcInner::from(&*self).start_http().await?;
|
||||
self.http_handle = Some(handle);
|
||||
log::info!("http rpc server started at {}", addr);
|
||||
return Ok(Some(addr));
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
let (handle, addr) = start(rpc_inner).await?;
|
||||
self.handle = Some(handle);
|
||||
|
||||
info!("rpc server started at {}", addr);
|
||||
|
||||
Ok(addr)
|
||||
pub async fn start_ws(&mut self) -> Result<Option<SocketAddr>> {
|
||||
if self.with_ws {
|
||||
let (handle, addr) = RpcInner::from(&*self).start_ws().await?;
|
||||
self.ws_handle = Some(handle);
|
||||
log::info!("http rpc server started at {}", addr);
|
||||
return Ok(Some(addr));
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -108,6 +117,53 @@ struct RpcInner {
|
|||
port: u16,
|
||||
}
|
||||
|
||||
impl From<&Rpc> for RpcInner {
|
||||
fn from(rpc: &Rpc) -> Self {
|
||||
RpcInner {
|
||||
node: Arc::clone(&rpc.node),
|
||||
port: rpc.port,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl RpcInner {
|
||||
pub async fn start_http(&self) -> Result<(HttpServerHandle, SocketAddr)> {
|
||||
let addr = format!("127.0.0.1:{}", self.port);
|
||||
let server = HttpServerBuilder::default().build(addr).await?;
|
||||
|
||||
let addr = server.local_addr()?;
|
||||
|
||||
let mut methods = Methods::new();
|
||||
let eth_methods: Methods = EthRpcServer::into_rpc(self.clone()).into();
|
||||
let net_methods: Methods = NetRpcServer::into_rpc(self.clone()).into();
|
||||
|
||||
methods.merge(eth_methods)?;
|
||||
methods.merge(net_methods)?;
|
||||
|
||||
let handle = server.start(methods)?;
|
||||
|
||||
Ok((handle, addr))
|
||||
}
|
||||
|
||||
pub async fn start_ws(&self) -> Result<(WsServerHandle, SocketAddr)> {
|
||||
let addr = format!("127.0.0.1:{}", self.port);
|
||||
let server = WsServerBuilder::default().build(addr).await?;
|
||||
|
||||
let addr = server.local_addr()?;
|
||||
|
||||
let mut methods = Methods::new();
|
||||
let eth_methods: Methods = EthRpcServer::into_rpc(self.clone()).into();
|
||||
let net_methods: Methods = NetRpcServer::into_rpc(self.clone()).into();
|
||||
|
||||
methods.merge(eth_methods)?;
|
||||
methods.merge(net_methods)?;
|
||||
|
||||
let handle = server.start(methods)?;
|
||||
|
||||
Ok((handle, addr))
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EthRpcServer for RpcInner {
|
||||
async fn get_balance(&self, address: &str, block: BlockTag) -> Result<String, Error> {
|
||||
|
@ -115,7 +171,7 @@ impl EthRpcServer for RpcInner {
|
|||
let node = self.node.read().await;
|
||||
let balance = convert_err(node.get_balance(&address, block).await)?;
|
||||
|
||||
Ok(format_hex(&balance))
|
||||
Ok(common::utils::format_hex(&balance))
|
||||
}
|
||||
|
||||
async fn get_transaction_count(&self, address: &str, block: BlockTag) -> Result<String, Error> {
|
||||
|
@ -164,13 +220,13 @@ impl EthRpcServer for RpcInner {
|
|||
async fn gas_price(&self) -> Result<String, Error> {
|
||||
let node = self.node.read().await;
|
||||
let gas_price = convert_err(node.get_gas_price())?;
|
||||
Ok(format_hex(&gas_price))
|
||||
Ok(common::utils::format_hex(&gas_price))
|
||||
}
|
||||
|
||||
async fn max_priority_fee_per_gas(&self) -> Result<String, Error> {
|
||||
let node = self.node.read().await;
|
||||
let tip = convert_err(node.get_priority_fee())?;
|
||||
Ok(format_hex(&tip))
|
||||
Ok(common::utils::format_hex(&tip))
|
||||
}
|
||||
|
||||
async fn block_number(&self) -> Result<String, Error> {
|
||||
|
@ -237,35 +293,6 @@ impl NetRpcServer for RpcInner {
|
|||
}
|
||||
}
|
||||
|
||||
async fn start(rpc: RpcInner) -> Result<(HttpServerHandle, SocketAddr)> {
|
||||
let addr = format!("127.0.0.1:{}", rpc.port);
|
||||
let server = HttpServerBuilder::default().build(addr).await?;
|
||||
|
||||
let addr = server.local_addr()?;
|
||||
|
||||
let mut methods = Methods::new();
|
||||
let eth_methods: Methods = EthRpcServer::into_rpc(rpc.clone()).into();
|
||||
let net_methods: Methods = NetRpcServer::into_rpc(rpc).into();
|
||||
|
||||
methods.merge(eth_methods)?;
|
||||
methods.merge(net_methods)?;
|
||||
|
||||
let handle = server.start(methods)?;
|
||||
|
||||
Ok((handle, addr))
|
||||
}
|
||||
|
||||
fn convert_err<T, E: Display>(res: Result<T, E>) -> Result<T, Error> {
|
||||
res.map_err(|err| Error::Custom(err.to_string()))
|
||||
}
|
||||
|
||||
fn format_hex(num: &U256) -> String {
|
||||
let stripped = num
|
||||
.encode_hex()
|
||||
.strip_prefix("0x")
|
||||
.unwrap()
|
||||
.trim_start_matches('0')
|
||||
.to_string();
|
||||
|
||||
format!("0x{}", stripped)
|
||||
}
|
||||
|
|
|
@ -1,9 +1,22 @@
|
|||
use ethers::prelude::Address;
|
||||
use ethers::{
|
||||
abi::AbiEncode,
|
||||
types::{Address, U256},
|
||||
};
|
||||
use eyre::Result;
|
||||
use ssz_rs::{Node, Vector};
|
||||
|
||||
use super::types::Bytes32;
|
||||
|
||||
pub fn format_hex(num: &U256) -> String {
|
||||
let stripped = num
|
||||
.encode_hex()
|
||||
.strip_prefix("0x")
|
||||
.unwrap()
|
||||
.trim_start_matches('0')
|
||||
.to_string();
|
||||
format!("0x{}", stripped)
|
||||
}
|
||||
|
||||
pub fn hex_str_to_bytes(s: &str) -> Result<Vec<u8>> {
|
||||
let stripped = s.strip_prefix("0x").unwrap_or(s);
|
||||
Ok(hex::decode(stripped)?)
|
||||
|
|
|
@ -27,6 +27,8 @@ pub struct Config {
|
|||
pub chain: ChainConfig,
|
||||
pub forks: Forks,
|
||||
pub max_checkpoint_age: u64,
|
||||
pub with_ws: bool,
|
||||
pub with_http: bool,
|
||||
}
|
||||
|
||||
impl Config {
|
||||
|
|
Loading…
Reference in New Issue