rings_node/native/endpoint/
mod.rs

1//! rings-node service run with `Swarm` and chord stabilization.
2#![warn(missing_docs)]
3mod http_error;
4mod ws;
5
6use std::net::SocketAddr;
7use std::sync::Arc;
8
9use axum::extract::ConnectInfo;
10use axum::extract::State;
11use axum::extract::WebSocketUpgrade;
12use axum::response::IntoResponse;
13use axum::routing::get;
14use axum::routing::post;
15use axum::Router;
16use jsonrpc_core::MetaIoHandler;
17use rings_rpc::protos::rings_node::NodeInfoResponse;
18use tower_http::cors::CorsLayer;
19
20use self::http_error::HttpError;
21use crate::processor::Processor;
22
23/// JSON-RPC state
24#[derive(Clone)]
25pub struct JsonRpcState<M>
26where M: jsonrpc_core::Middleware<Arc<Processor>>
27{
28    processor: Arc<Processor>,
29    io_handler: MetaIoHandler<Arc<Processor>, M>,
30}
31
32/// websocket state
33#[derive(Clone)]
34#[allow(dead_code)]
35pub struct WsState {
36    processor: Arc<Processor>,
37}
38
39/// Status state
40#[derive(Clone)]
41pub struct StatusState {
42    processor: Arc<Processor>,
43}
44
45struct ExternalRpcMiddleware;
46struct InternalRpcMiddleware;
47
48/// Run a web server to handle jsonrpc request locally
49pub async fn run_internal_api(port: u16, processor: Arc<Processor>) -> anyhow::Result<()> {
50    let binding_addr = SocketAddr::from(([127, 0, 0, 1], port));
51
52    let jsonrpc_handler = MetaIoHandler::with_middleware(InternalRpcMiddleware);
53    let jsonrpc_state = Arc::new(JsonRpcState {
54        processor: processor.clone(),
55        io_handler: jsonrpc_handler,
56    });
57
58    let ws_state = Arc::new(WsState {
59        processor: processor.clone(),
60    });
61
62    let status_state = Arc::new(StatusState { processor });
63
64    let axum_make_service = Router::new()
65        .route(
66            "/",
67            post(jsonrpc_io_handler).with_state(jsonrpc_state.clone()),
68        )
69        .route("/ws", get(ws_handler).with_state(ws_state))
70        .route("/status", get(status_handler).with_state(status_state))
71        .layer(CorsLayer::permissive())
72        .layer(axum::middleware::from_fn(node_info_header))
73        .into_make_service_with_connect_info::<SocketAddr>();
74
75    println!("JSON-RPC endpoint: http://{binding_addr}");
76    println!("WebSocket endpoint: http://{binding_addr}/ws");
77    axum::Server::bind(&binding_addr)
78        .serve(axum_make_service)
79        .await?;
80    Ok(())
81}
82
83/// Run a web server to handle jsonrpc request from external
84pub async fn run_external_api(addr: String, processor: Arc<Processor>) -> anyhow::Result<()> {
85    let binding_addr: SocketAddr = addr.parse()?;
86
87    let jsonrpc_handler = MetaIoHandler::with_middleware(ExternalRpcMiddleware);
88    let jsonrpc_state = Arc::new(JsonRpcState {
89        processor: processor.clone(),
90        io_handler: jsonrpc_handler,
91    });
92
93    let status_state = Arc::new(StatusState { processor });
94
95    let axum_make_service = Router::new()
96        .route(
97            "/",
98            post(jsonrpc_io_handler).with_state(jsonrpc_state.clone()),
99        )
100        .route("/status", get(status_handler).with_state(status_state))
101        .layer(CorsLayer::permissive())
102        .layer(axum::middleware::from_fn(node_info_header))
103        .into_make_service_with_connect_info::<SocketAddr>();
104
105    println!("JSON-RPC endpoint: http://{addr}");
106    axum::Server::bind(&binding_addr)
107        .serve(axum_make_service)
108        .await?;
109    Ok(())
110}
111
112async fn jsonrpc_io_handler<M>(
113    State(state): State<Arc<JsonRpcState<M>>>,
114    body: String,
115) -> Result<JsonResponse, HttpError>
116where
117    M: jsonrpc_core::Middleware<Arc<Processor>>,
118{
119    let r = state
120        .io_handler
121        .handle_request(&body, state.processor.clone())
122        .await
123        .ok_or(HttpError::BadRequest)?;
124    Ok(JsonResponse(r))
125}
126
127async fn node_info_header<B>(
128    req: http::Request<B>,
129    next: axum::middleware::Next<B>,
130) -> axum::response::Response {
131    let mut res = next.run(req).await;
132    let headers = res.headers_mut();
133
134    if let Ok(version) = http::HeaderValue::from_str(crate::util::build_version().as_str()) {
135        headers.insert("X-NODE-VERSION", version);
136    }
137    res
138}
139
140async fn status_handler(
141    State(state): State<Arc<StatusState>>,
142) -> Result<axum::Json<NodeInfoResponse>, HttpError> {
143    let info = state
144        .processor
145        .get_node_info()
146        .await
147        .map_err(|_| HttpError::Internal)?;
148    Ok(axum::Json(info))
149}
150
151/// JSON response struct
152#[derive(Debug, Clone)]
153pub struct JsonResponse(String);
154
155impl IntoResponse for JsonResponse {
156    fn into_response(self) -> axum::response::Response {
157        ([("content-type", "application/json")], self.0).into_response()
158    }
159}
160
161async fn ws_handler(
162    State(state): State<Arc<WsState>>,
163    ws: WebSocketUpgrade,
164    ConnectInfo(addr): ConnectInfo<SocketAddr>,
165) -> impl IntoResponse {
166    tracing::info!("ws connected, remote: {}", addr);
167    ws.on_upgrade(move |socket| self::ws::handle_socket(state, socket))
168}
169
170mod jsonrpc_middleware_impl {
171    use std::future::Future;
172
173    use jsonrpc_core::futures_util::future;
174    use jsonrpc_core::futures_util::future::Either;
175    use jsonrpc_core::futures_util::FutureExt;
176    use jsonrpc_core::middleware::NoopCallFuture;
177    use jsonrpc_core::middleware::NoopFuture;
178    use jsonrpc_core::*;
179    use rings_rpc::protos::rings_node_handler::ExternalRpcHandler;
180    use rings_rpc::protos::rings_node_handler::InternalRpcHandler;
181
182    use super::*;
183
184    impl Middleware<Arc<Processor>> for InternalRpcMiddleware {
185        type Future = NoopFuture;
186        type CallFuture = NoopCallFuture;
187
188        fn on_call<F, X>(
189            &self,
190            call: Call,
191            meta: Arc<Processor>,
192            next: F,
193        ) -> Either<Self::CallFuture, X>
194        where
195            F: Fn(Call, Arc<Processor>) -> X + Send + Sync,
196            X: Future<Output = Option<Output>> + Send + 'static,
197        {
198            match call {
199                Call::MethodCall(req) => {
200                    let fut = InternalRpcHandler
201                        .handle_request(meta, req.method, req.params.into())
202                        .then(move |res| {
203                            future::ready(Some(Output::from(res, req.id, req.jsonrpc)))
204                        });
205                    Either::Left(Box::pin(fut))
206                }
207                _ => Either::Left(Box::pin(next(call, meta))),
208            }
209        }
210    }
211
212    impl Middleware<Arc<Processor>> for ExternalRpcMiddleware {
213        type Future = NoopFuture;
214        type CallFuture = NoopCallFuture;
215
216        fn on_call<F, X>(
217            &self,
218            call: Call,
219            meta: Arc<Processor>,
220            next: F,
221        ) -> Either<Self::CallFuture, X>
222        where
223            F: Fn(Call, Arc<Processor>) -> X + Send + Sync,
224            X: Future<Output = Option<Output>> + Send + 'static,
225        {
226            match call {
227                Call::MethodCall(req) => {
228                    let fut = ExternalRpcHandler
229                        .handle_request(meta, req.method, req.params.into())
230                        .then(move |res| {
231                            future::ready(Some(Output::from(res, req.id, req.jsonrpc)))
232                        });
233                    Either::Left(Box::pin(fut))
234                }
235                _ => Either::Left(Box::pin(next(call, meta))),
236            }
237        }
238    }
239}
rings_node/native/endpoint/mod.rs

rings_node/native/endpoint/
mod.rs
