anvil_server/
handler.rs

1use crate::RpcHandler;
2use anvil_rpc::{
3    error::RpcError,
4    request::{Id, Request, RpcCall, RpcMethodCall, Version},
5    response::{Response, RpcResponse},
6};
7use axum::{
8    Json,
9    extract::{State, rejection::JsonRejection},
10    http::StatusCode,
11    response::{IntoResponse, Response as AxumResponse},
12};
13use futures::{FutureExt, future};
14
15/// Handles incoming JSON-RPC Request.
16// NOTE: `handler` must come first because the `request` extractor consumes the request body.
17pub async fn handle<Http: RpcHandler, Ws>(
18    State((handler, _)): State<(Http, Ws)>,
19    request: Result<Json<Request>, JsonRejection>,
20) -> AxumResponse {
21    match request {
22        Ok(Json(req)) => handle_request(req, handler)
23            .await
24            .map(Json)
25            .map(IntoResponse::into_response)
26            .unwrap_or_else(|| StatusCode::NO_CONTENT.into_response()),
27        Err(JsonRejection::JsonSyntaxError(err)) => {
28            warn!(target: "rpc", ?err, "invalid request");
29            Json(Response::error(RpcError::parse_error())).into_response()
30        }
31        Err(err) => {
32            warn!(target: "rpc", ?err, "invalid request");
33            Json(Response::error(RpcError::invalid_request())).into_response()
34        }
35    }
36}
37
38/// Handle the JSON-RPC [Request]
39///
40/// This will try to deserialize the payload into the request type of the handler and if successful
41/// invoke the handler
42pub async fn handle_request<Handler: RpcHandler>(
43    req: Request,
44    handler: Handler,
45) -> Option<Response> {
46    /// processes batch calls
47    fn responses_as_batch(outs: Vec<Option<RpcResponse>>) -> Option<Response> {
48        let batch: Vec<_> = outs.into_iter().flatten().collect();
49        (!batch.is_empty()).then_some(Response::Batch(batch))
50    }
51
52    match req {
53        Request::Single(call) => handle_call(call, handler).await.map(Response::Single),
54        Request::Batch(calls) => {
55            if calls.is_empty() {
56                return Some(Response::error(RpcError::invalid_request()));
57            }
58            future::join_all(calls.into_iter().map(move |call| handle_call(call, handler.clone())))
59                .map(responses_as_batch)
60                .await
61        }
62    }
63}
64
65/// handle a single RPC method call
66async fn handle_call<Handler: RpcHandler>(call: RpcCall, handler: Handler) -> Option<RpcResponse> {
67    match call {
68        RpcCall::MethodCall(call) => {
69            trace!(target: "rpc", id = ?call.id , method = ?call.method,  "handling call");
70            Some(handler.on_call(call).await)
71        }
72        RpcCall::Notification(notification) => {
73            let call = RpcMethodCall {
74                jsonrpc: notification.jsonrpc.unwrap_or(Version::V2),
75                method: notification.method,
76                params: notification.params,
77                id: Id::Null,
78            };
79            trace!(target: "rpc", method = ?call.method, "handling rpc notification");
80            drop(handler.on_call(call).await);
81            None
82        }
83        RpcCall::Invalid { id } => {
84            warn!(target: "rpc", ?id,  "invalid rpc call");
85            Some(RpcResponse::invalid_request(id))
86        }
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93    use crate::{ServerConfig, http_router};
94    use anvil_rpc::{
95        request::{RequestParams, RpcNotification},
96        response::ResponseResult,
97    };
98    use axum::{
99        body::{Body, to_bytes},
100        http::{Request as HttpRequest, header},
101    };
102    use std::{
103        pin::pin,
104        sync::{
105            Arc,
106            atomic::{AtomicUsize, Ordering},
107        },
108        task::{Context, Poll, Waker},
109    };
110    use tower::ServiceExt;
111
112    #[derive(Clone, Default)]
113    struct TestHandler {
114        requests: Arc<AtomicUsize>,
115    }
116
117    #[derive(Clone, Debug, serde::Deserialize)]
118    #[serde(tag = "method", content = "params")]
119    enum TypedRequest {
120        #[serde(rename = "known")]
121        Known(Vec<TestParam>),
122    }
123
124    #[derive(Clone, Debug, serde::Deserialize)]
125    enum TestParam {
126        #[serde(rename = "allowed")]
127        Allowed,
128    }
129
130    #[derive(Clone)]
131    struct TypedHandler;
132
133    impl RpcHandler for TestHandler {
134        type Request = serde_json::Value;
135
136        async fn on_request(&self, _request: Self::Request) -> ResponseResult {
137            self.requests.fetch_add(1, Ordering::Relaxed);
138            ResponseResult::success(())
139        }
140    }
141
142    impl RpcHandler for TypedHandler {
143        type Request = TypedRequest;
144
145        async fn on_request(&self, request: Self::Request) -> ResponseResult {
146            let TypedRequest::Known(params) = request;
147            drop(params);
148            ResponseResult::success(())
149        }
150    }
151
152    fn notification() -> RpcCall {
153        RpcCall::Notification(RpcNotification {
154            jsonrpc: Some(Version::V2),
155            method: "record".to_owned(),
156            params: RequestParams::None,
157        })
158    }
159
160    fn method_call(id: i64) -> RpcCall {
161        RpcCall::MethodCall(RpcMethodCall {
162            jsonrpc: Version::V2,
163            method: "record".to_owned(),
164            params: RequestParams::None,
165            id: Id::Number(id),
166        })
167    }
168
169    fn typed_call(method: &str) -> RpcMethodCall {
170        RpcMethodCall {
171            jsonrpc: Version::V2,
172            method: method.to_owned(),
173            params: RequestParams::Array(vec![serde_json::json!("bogus")]),
174            id: Id::Number(1),
175        }
176    }
177
178    fn run_ready<F: Future>(future: F) -> F::Output {
179        let waker = Waker::noop();
180        let mut cx = Context::from_waker(waker);
181        let mut future = pin!(future);
182        match future.as_mut().poll(&mut cx) {
183            Poll::Ready(output) => output,
184            Poll::Pending => panic!("future unexpectedly pending"),
185        }
186    }
187
188    #[test]
189    fn empty_batch_returns_invalid_request() {
190        let response = run_ready(handle_request(Request::Batch(vec![]), TestHandler::default()));
191
192        assert_eq!(response, Some(Response::error(RpcError::invalid_request())));
193    }
194
195    #[test]
196    fn distinguishes_unknown_methods_from_unknown_parameter_variants() {
197        let unknown_method = run_ready(TypedHandler.on_call(typed_call("unknown")));
198        let invalid_params = run_ready(TypedHandler.on_call(typed_call("known")));
199
200        assert_eq!(
201            serde_json::to_value(unknown_method).unwrap()["error"]["code"],
202            serde_json::json!(-32601)
203        );
204        assert_eq!(
205            serde_json::to_value(invalid_params).unwrap()["error"]["code"],
206            serde_json::json!(-32602)
207        );
208    }
209
210    #[test]
211    fn notification_only_batch_executes_without_response() {
212        let handler = TestHandler::default();
213        let response = run_ready(handle_request(
214            Request::Batch(vec![notification(), notification()]),
215            handler.clone(),
216        ));
217
218        assert_eq!(response, None);
219        assert_eq!(handler.requests.load(Ordering::Relaxed), 2);
220    }
221
222    #[test]
223    fn mixed_batch_executes_notification_without_including_response() {
224        let handler = TestHandler::default();
225        let response = run_ready(handle_request(
226            Request::Batch(vec![notification(), method_call(1)]),
227            handler.clone(),
228        ));
229
230        assert_eq!(
231            response,
232            Some(Response::Batch(vec![RpcResponse::new(
233                Id::Number(1),
234                ResponseResult::success(())
235            )]))
236        );
237        assert_eq!(handler.requests.load(Ordering::Relaxed), 2);
238    }
239
240    #[test]
241    fn http_notification_returns_no_content_after_execution() {
242        let handler = TestHandler::default();
243        let response = run_ready(handle(
244            State((handler.clone(), ())),
245            Ok(Json(Request::Single(notification()))),
246        ));
247
248        assert_eq!(response.status(), StatusCode::NO_CONTENT);
249        assert!(run_ready(to_bytes(response.into_body(), usize::MAX)).unwrap().is_empty());
250        assert_eq!(handler.requests.load(Ordering::Relaxed), 1);
251    }
252
253    #[test]
254    fn malformed_json_returns_parse_error() {
255        let request = HttpRequest::post("/")
256            .header(header::CONTENT_TYPE, "application/json")
257            .body(Body::from("{"))
258            .unwrap();
259        let response = run_ready(
260            http_router(ServerConfig::default(), TestHandler::default()).oneshot(request),
261        )
262        .unwrap();
263        let body = run_ready(to_bytes(response.into_body(), usize::MAX)).unwrap();
264
265        assert_eq!(
266            serde_json::from_slice::<Response>(&body).unwrap(),
267            Response::error(RpcError::parse_error())
268        );
269    }
270}
anvil_server/handler.rs

anvil_server/
handler.rs
