#![warn(missing_docs)]
pub mod middleware;
use std::{error::Error as StdError, net::SocketAddr, time::Duration};
use http::header::HeaderValue;
use jsonrpsee::{
server::middleware::{HostFilterLayer, ProxyGetRequestLayer},
RpcModule,
};
use tokio::net::TcpListener;
use tower_http::cors::{AllowOrigin, CorsLayer};
pub use crate::middleware::RpcMetrics;
pub use jsonrpsee::core::{
id_providers::{RandomIntegerIdProvider, RandomStringIdProvider},
traits::IdProvider,
};
const MEGABYTE: u32 = 1024 * 1024;
pub type Server = jsonrpsee::server::ServerHandle;
#[derive(Debug)]
pub struct Config<'a, M: Send + Sync + 'static> {
pub addrs: [SocketAddr; 2],
pub cors: Option<&'a Vec<String>>,
pub max_connections: u32,
pub max_subs_per_conn: u32,
pub max_payload_in_mb: u32,
pub max_payload_out_mb: u32,
pub metrics: Option<RpcMetrics>,
pub message_buffer_capacity: u32,
pub rpc_api: RpcModule<M>,
pub id_provider: Option<Box<dyn IdProvider>>,
pub tokio_handle: tokio::runtime::Handle,
}
pub async fn start_server<M: Send + Sync + 'static>(
config: Config<'_, M>,
) -> Result<Server, Box<dyn StdError + Send + Sync>> {
let Config {
addrs,
cors,
max_payload_in_mb,
max_payload_out_mb,
max_connections,
max_subs_per_conn,
metrics,
message_buffer_capacity,
id_provider,
tokio_handle,
rpc_api,
} = config;
let std_listener = TcpListener::bind(addrs.as_slice()).await?.into_std()?;
let local_addr = std_listener.local_addr().ok();
let host_filter = hosts_filtering(cors.is_some(), local_addr);
let middleware = tower::ServiceBuilder::new()
.option_layer(host_filter)
.layer(ProxyGetRequestLayer::new("/health", "system_health")?)
.layer(try_into_cors(cors)?);
let mut builder = jsonrpsee::server::Server::builder()
.max_request_body_size(max_payload_in_mb.saturating_mul(MEGABYTE))
.max_response_body_size(max_payload_out_mb.saturating_mul(MEGABYTE))
.max_connections(max_connections)
.max_subscriptions_per_connection(max_subs_per_conn)
.ping_interval(Duration::from_secs(30))
.set_middleware(middleware)
.set_message_buffer_capacity(message_buffer_capacity)
.custom_tokio_runtime(tokio_handle);
if let Some(provider) = id_provider {
builder = builder.set_id_provider(provider);
} else {
builder = builder.set_id_provider(RandomStringIdProvider::new(16));
};
let rpc_api = build_rpc_api(rpc_api);
let handle = if let Some(metrics) = metrics {
let server = builder.set_logger(metrics).build_from_tcp(std_listener)?;
server.start(rpc_api)
} else {
let server = builder.build_from_tcp(std_listener)?;
server.start(rpc_api)
};
log::info!(
"Running JSON-RPC server: addr={}, allowed origins={}",
local_addr.map_or_else(|| "unknown".to_string(), |a| a.to_string()),
format_cors(cors)
);
Ok(handle)
}
fn hosts_filtering(enabled: bool, addr: Option<SocketAddr>) -> Option<HostFilterLayer> {
let port = addr.map_or("*".to_string(), |p| p.port().to_string());
if enabled {
let hosts =
[format!("localhost:{port}"), format!("127.0.0.1:{port}"), format!("[::1]:{port}")];
Some(HostFilterLayer::new(hosts).expect("Valid hosts; qed"))
} else {
None
}
}
fn build_rpc_api<M: Send + Sync + 'static>(mut rpc_api: RpcModule<M>) -> RpcModule<M> {
let mut available_methods = rpc_api.method_names().collect::<Vec<_>>();
available_methods.push("rpc_methods");
available_methods.sort();
rpc_api
.register_method("rpc_methods", move |_, _| {
serde_json::json!({
"methods": available_methods,
})
})
.expect("infallible all other methods have their own address space; qed");
rpc_api
}
fn try_into_cors(
maybe_cors: Option<&Vec<String>>,
) -> Result<CorsLayer, Box<dyn StdError + Send + Sync>> {
if let Some(cors) = maybe_cors {
let mut list = Vec::new();
for origin in cors {
list.push(HeaderValue::from_str(origin)?);
}
Ok(CorsLayer::new().allow_origin(AllowOrigin::list(list)))
} else {
Ok(CorsLayer::permissive())
}
}
fn format_cors(maybe_cors: Option<&Vec<String>>) -> String {
if let Some(cors) = maybe_cors {
format!("{:?}", cors)
} else {
format!("{:?}", ["*"])
}
}