use async_trait::async_trait;
use ethers_providers::{
Authorization, ConnectionDetails, Http, HttpRateLimitRetryPolicy, JsonRpcClient, JsonRpcError,
JwtAuth, JwtKey, ProviderError, RetryClient, RetryClientBuilder, RpcError, Ws,
};
use reqwest::{
header::{HeaderName, HeaderValue},
Url,
};
use serde::{de::DeserializeOwned, Serialize};
use std::{fmt::Debug, str::FromStr, sync::Arc, time::Duration};
use thiserror::Error;
use tokio::sync::RwLock;
#[derive(Debug)]
enum InnerClient {
Http(RetryClient<Http>),
Ws(Ws),
}
#[derive(Error, Debug)]
pub enum RuntimeClientError {
#[error(transparent)]
ProviderError(ProviderError),
#[error("Failed to lock the client")]
LockError,
#[error("URL scheme is not supported: {0}")]
BadScheme(String),
#[error("Invalid HTTP header: {0}")]
BadHeader(String),
#[error("Invalid IPC file path: {0}")]
BadPath(String),
}
impl RpcError for RuntimeClientError {
fn as_error_response(&self) -> Option<&JsonRpcError> {
match self {
RuntimeClientError::ProviderError(err) => err.as_error_response(),
_ => None,
}
}
fn as_serde_error(&self) -> Option<&serde_json::Error> {
match self {
RuntimeClientError::ProviderError(e) => e.as_serde_error(),
_ => None,
}
}
}
impl From<RuntimeClientError> for ProviderError {
fn from(src: RuntimeClientError) -> Self {
match src {
RuntimeClientError::ProviderError(err) => err,
_ => ProviderError::JsonRpcClientError(Box::new(src)),
}
}
}
#[derive(Clone, Debug, Error)]
pub struct RuntimeClient {
client: Arc<RwLock<Option<InnerClient>>>,
url: Url,
max_retry: u32,
timeout_retry: u32,
initial_backoff: u64,
timeout: Duration,
compute_units_per_second: u64,
jwt: Option<String>,
headers: Vec<String>,
}
pub struct RuntimeClientBuilder {
url: Url,
max_retry: u32,
timeout_retry: u32,
initial_backoff: u64,
timeout: Duration,
compute_units_per_second: u64,
jwt: Option<String>,
headers: Vec<String>,
}
impl ::core::fmt::Display for RuntimeClient {
fn fmt(&self, f: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result {
write!(f, "RuntimeClient")
}
}
fn build_auth(jwt: String) -> eyre::Result<Authorization> {
let jwt = hex::decode(jwt)?;
let secret = JwtKey::from_slice(&jwt).map_err(|err| eyre::eyre!("Invalid JWT: {}", err))?;
let auth = JwtAuth::new(secret, None, None);
let token = auth.generate_token()?;
let auth = Authorization::Bearer(token);
Ok(auth)
}
impl RuntimeClient {
async fn connect(&self) -> Result<InnerClient, RuntimeClientError> {
match self.url.scheme() {
"http" | "https" => {
let mut client_builder = reqwest::Client::builder().timeout(self.timeout);
let mut headers = reqwest::header::HeaderMap::new();
if let Some(jwt) = self.jwt.as_ref() {
let auth = build_auth(jwt.clone()).map_err(|err| {
RuntimeClientError::ProviderError(ProviderError::CustomError(
err.to_string(),
))
})?;
let mut auth_value: HeaderValue = HeaderValue::from_str(&auth.to_string())
.expect("Header should be valid string");
auth_value.set_sensitive(true);
headers.insert(reqwest::header::AUTHORIZATION, auth_value);
};
for header in self.headers.iter() {
let make_err = || RuntimeClientError::BadHeader(header.to_string());
let (key, val) = header.split_once(':').ok_or_else(make_err)?;
headers.insert(
HeaderName::from_str(key.trim()).map_err(|_| make_err())?,
HeaderValue::from_str(val.trim()).map_err(|_| make_err())?,
);
}
client_builder = client_builder.default_headers(headers);
let client = client_builder
.build()
.map_err(|e| RuntimeClientError::ProviderError(e.into()))?;
let provider = Http::new_with_client(self.url.clone(), client);
#[allow(clippy::box_default)]
let provider = RetryClientBuilder::default()
.initial_backoff(Duration::from_millis(self.initial_backoff))
.rate_limit_retries(self.max_retry)
.timeout_retries(self.timeout_retry)
.compute_units_per_second(self.compute_units_per_second)
.build(provider, Box::new(HttpRateLimitRetryPolicy));
Ok(InnerClient::Http(provider))
}
"ws" | "wss" => {
let auth: Option<Authorization> = self
.jwt
.as_ref()
.and_then(|jwt| build_auth(jwt.clone()).ok());
let connection_details = ConnectionDetails::new(self.url.as_str(), auth);
let client =
Ws::connect_with_reconnects(connection_details, self.max_retry as usize)
.await
.map_err(|e| RuntimeClientError::ProviderError(e.into()))?;
Ok(InnerClient::Ws(client))
}
_ => Err(RuntimeClientError::BadScheme(self.url.to_string())),
}
}
}
impl RuntimeClientBuilder {
pub fn new(
url: Url,
max_retry: u32,
timeout_retry: u32,
initial_backoff: u64,
timeout: Duration,
compute_units_per_second: u64,
) -> Self {
Self {
url,
max_retry,
timeout,
timeout_retry,
initial_backoff,
compute_units_per_second,
jwt: None,
headers: vec![],
}
}
pub fn with_jwt(mut self, jwt: Option<String>) -> Self {
self.jwt = jwt;
self
}
pub fn with_headers(mut self, headers: Vec<String>) -> Self {
self.headers = headers;
self
}
pub fn build(self) -> RuntimeClient {
RuntimeClient {
client: Arc::new(RwLock::new(None)),
url: self.url,
max_retry: self.max_retry,
timeout_retry: self.timeout_retry,
initial_backoff: self.initial_backoff,
timeout: self.timeout,
compute_units_per_second: self.compute_units_per_second,
jwt: self.jwt,
headers: self.headers,
}
}
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl JsonRpcClient for RuntimeClient {
type Error = RuntimeClientError;
async fn request<T, R>(&self, method: &str, params: T) -> Result<R, Self::Error>
where
T: Debug + Serialize + Send + Sync,
R: DeserializeOwned + Send,
{
if self.client.read().await.is_none() {
let mut w = self.client.write().await;
*w = Some(
self.connect()
.await
.map_err(|e| RuntimeClientError::ProviderError(e.into()))?,
);
}
let res = match self.client.read().await.as_ref().unwrap() {
InnerClient::Http(http) => RetryClient::request(http, method, params)
.await
.map_err(|e| RuntimeClientError::ProviderError(e.into())),
InnerClient::Ws(ws) => JsonRpcClient::request(ws, method, params)
.await
.map_err(|e| RuntimeClientError::ProviderError(e.into())),
}?;
Ok(res)
}
}