ethrpc_rs/
rpc.rs

1//! The [`Rpc`] client and the [`Handler`] trait.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use base64::Engine;
9use rsurl::aio;
10use serde::de::DeserializeOwned;
11use serde_json::{Map, Value};
12
13use crate::error::{Error, Result};
14use crate::jsonrpc::{Request, Response};
15
16/// A locally-handled RPC method. Receives the positional parameters and returns
17/// a JSON value (or an error). Registered with [`Rpc::set_override`].
18pub type OverrideFn = Arc<dyn Fn(&[Value]) -> Result<Value> + Send + Sync>;
19
20/// Any backend capable of executing JSON-RPC calls. Implemented by [`Rpc`] and
21/// [`RpcList`](crate::RpcList).
22#[async_trait]
23pub trait Handler: Send + Sync {
24    /// Performs a JSON-RPC call with positional parameters.
25    async fn call(&self, method: &str, params: Vec<Value>) -> Result<Value>;
26}
27
28/// A connection to an Ethereum JSON-RPC endpoint over HTTP, with optional basic
29/// authentication and local method overrides.
30#[derive(Clone, Default)]
31pub struct Rpc {
32    host: String,
33    lag: Duration,
34    block: u64,
35    username: String,
36    password: String,
37    overrides: HashMap<String, OverrideFn>,
38}
39
40impl Rpc {
41    /// Returns a new RPC client targeting the given endpoint. Pass an empty host
42    /// to build an override-only handler.
43    pub fn new(host: impl Into<String>) -> Rpc {
44        Rpc {
45            host: host.into(),
46            ..Default::default()
47        }
48    }
49
50    /// Redirects calls to `method` to a local function instead of the remote
51    /// node. The function receives the call's positional parameters.
52    pub fn set_override<F>(&mut self, method: impl Into<String>, f: F)
53    where
54        F: Fn(&[Value]) -> Result<Value> + Send + Sync + 'static,
55    {
56        self.overrides.insert(method.into(), Arc::new(f));
57    }
58
59    /// Sets the host for subsequent requests.
60    pub fn set_host(&mut self, host: impl Into<String>) {
61        self.host = host.into();
62    }
63
64    /// Returns the configured host.
65    pub fn host(&self) -> &str {
66        &self.host
67    }
68
69    /// Sets HTTP basic auth credentials for subsequent requests.
70    pub fn set_basic_auth(&mut self, username: impl Into<String>, password: impl Into<String>) {
71        self.username = username.into();
72        self.password = password.into();
73    }
74
75    /// Returns how long this endpoint took to answer `eth_blockNumber` during
76    /// [`evaluate`](crate::evaluate), or zero if it was never probed.
77    pub fn lag(&self) -> Duration {
78        self.lag
79    }
80
81    /// Returns the latest block number observed during
82    /// [`evaluate`](crate::evaluate), or zero if it was never probed.
83    pub fn block(&self) -> u64 {
84        self.block
85    }
86
87    pub(crate) fn set_probe(&mut self, lag: Duration, block: u64) {
88        self.lag = lag;
89        self.block = block;
90    }
91
92    /// Performs a request using positional arguments.
93    pub async fn call(&self, method: &str, params: Vec<Value>) -> Result<Value> {
94        self.send(&Request::new(method, params)).await
95    }
96
97    /// Performs a request using named arguments.
98    pub async fn call_named(&self, method: &str, params: Map<String, Value>) -> Result<Value> {
99        self.send(&Request::with_map(method, params)).await
100    }
101
102    /// Performs a request and deserializes the result into `T` via serde.
103    pub async fn call_as<T: DeserializeOwned>(
104        &self,
105        method: &str,
106        params: Vec<Value>,
107    ) -> Result<T> {
108        Ok(serde_json::from_value(self.call(method, params).await?)?)
109    }
110
111    /// Sends a raw [`Request`] to the endpoint and returns the raw result value.
112    pub async fn send(&self, req: &Request) -> Result<Value> {
113        // Local override: run the function instead of hitting the node.
114        if let Some(f) = self.overrides.get(&req.method) {
115            return match &req.params {
116                Value::Array(arr) => f(arr),
117                _ => Err(Error::Other(
118                    "function requires positional arguments instead of named arguments".to_string(),
119                )),
120            };
121        }
122
123        if self.host.is_empty() {
124            // Override-only handler: anything else is "not found".
125            return Err(Error::NotFound);
126        }
127
128        let body = serde_json::to_vec(req)?;
129        let mut hreq =
130            aio::Request::post(self.host.clone(), body).header("Content-Type", "application/json");
131
132        if !self.username.is_empty() || !self.password.is_empty() {
133            let token = base64::engine::general_purpose::STANDARD
134                .encode(format!("{}:{}", self.username, self.password));
135            hreq = hreq.header("Authorization", format!("Basic {token}"));
136        }
137
138        let rt = aio::TokioRuntime;
139        let resp = aio::request(&rt, &hreq).await?;
140        let status = resp.status;
141        let body = resp.body;
142
143        // Some servers return JSON-RPC errors over HTTP 4xx/5xx. Try to decode
144        // either way; if decoding fails on a non-2xx, surface the HTTP status.
145        let decoded: std::result::Result<Response, _> = serde_json::from_slice(&body);
146
147        if !(200..300).contains(&status) {
148            if let Ok(r) = &decoded {
149                if let Some(eo) = &r.error {
150                    return Err(Error::Rpc(eo.clone()));
151                }
152            }
153            return Err(Error::Http {
154                status,
155                method: req.method.clone(),
156                body: snippet(&body),
157            });
158        }
159
160        let res = decoded?;
161        if let Some(eo) = res.error {
162            return Err(Error::Rpc(eo));
163        }
164        Ok(res.result)
165    }
166}
167
168#[async_trait]
169impl Handler for Rpc {
170    async fn call(&self, method: &str, params: Vec<Value>) -> Result<Value> {
171        Rpc::call(self, method, params).await
172    }
173}
174
175/// Returns a trimmed, lossy snippet of up to 200 bytes of `body`.
176fn snippet(body: &[u8]) -> String {
177    let end = body.len().min(200);
178    String::from_utf8_lossy(&body[..end]).trim().to_string()
179}
180
181/// Options controlling how [`Rpc::forward`] builds its response.
182#[derive(Debug, Clone, Default)]
183pub struct ForwardOptions {
184    /// Pretty-print the JSON body.
185    pub pretty: bool,
186    /// If set, emit `Cache-Control: public, max-age=<seconds>`.
187    pub cache: Option<Duration>,
188}
189
190/// A response produced by [`Rpc::forward`], ready to be written to whatever HTTP
191/// framework the caller uses.
192#[derive(Debug, Clone)]
193pub struct ForwardResponse {
194    /// The HTTP status code to send.
195    pub status: u16,
196    /// The response headers, in order.
197    pub headers: Vec<(String, String)>,
198    /// The response body.
199    pub body: Vec<u8>,
200}
201
202/// Hop-by-hop headers per RFC 7230 §6.1 — must not be forwarded by
203/// intermediaries. Compared case-insensitively.
204const HOP_BY_HOP: &[&str] = &[
205    "connection",
206    "keep-alive",
207    "proxy-authenticate",
208    "proxy-authorization",
209    "te",
210    "trailer",
211    "transfer-encoding",
212    "upgrade",
213];
214
215fn is_hop_by_hop(name: &str) -> bool {
216    HOP_BY_HOP.iter().any(|h| name.eq_ignore_ascii_case(h))
217}
218
219impl Rpc {
220    /// Builds the response for a JSON-RPC request suitable for proxying back to
221    /// an HTTP client. Overridden methods are executed locally; otherwise the
222    /// request is forwarded to the node and its response relayed (with
223    /// hop-by-hop headers stripped). The returned [`ForwardResponse`] is
224    /// framework-agnostic — write its status, headers, and body to your
225    /// response object.
226    pub async fn forward(&self, req: &Request, opts: &ForwardOptions) -> ForwardResponse {
227        // Local override: run it and encode a JSON-RPC response body.
228        if let Some(f) = self.overrides.get(&req.method) {
229            let mut headers = vec![
230                ("Content-Type".to_string(), "application/json".to_string()),
231                (
232                    "Access-Control-Allow-Methods".to_string(),
233                    "GET, POST, OPTIONS".to_string(),
234                ),
235            ];
236            cache_header(&mut headers, opts);
237
238            let body = match &req.params {
239                Value::Array(arr) => match f(arr) {
240                    Ok(res) => encode_json(
241                        &crate::jsonrpc::ResponseIntf {
242                            jsonrpc: "2.0".to_string(),
243                            result: Some(res),
244                            error: None,
245                            id: req.id.clone(),
246                        },
247                        opts.pretty,
248                    ),
249                    // JSON-RPC convention: transport stays 200, error in body.
250                    Err(e) => encode_json(&req.make_error(&e), opts.pretty),
251                },
252                _ => encode_json(
253                    &req.make_error(&Error::Other(
254                        "function only supports positional arguments".to_string(),
255                    )),
256                    opts.pretty,
257                ),
258            };
259            return ForwardResponse {
260                status: 200,
261                headers,
262                body,
263            };
264        }
265
266        if self.host.is_empty() {
267            return ForwardResponse {
268                status: 404,
269                headers: vec![("Content-Type".to_string(), "text/plain".to_string())],
270                body: b"404 page not found\n".to_vec(),
271            };
272        }
273
274        // Forward to the node.
275        let enc = match serde_json::to_vec(req) {
276            Ok(b) => b,
277            Err(e) => return internal_error(&e.to_string()),
278        };
279        let mut hreq =
280            aio::Request::post(self.host.clone(), enc).header("Content-Type", "application/json");
281        if !self.username.is_empty() || !self.password.is_empty() {
282            let token = base64::engine::general_purpose::STANDARD
283                .encode(format!("{}:{}", self.username, self.password));
284            hreq = hreq.header("Authorization", format!("Basic {token}"));
285        }
286
287        let rt = aio::TokioRuntime;
288        let resp = match aio::request(&rt, &hreq).await {
289            Ok(r) => r,
290            Err(e) => return internal_error(&e.to_string()),
291        };
292
293        let mut headers: Vec<(String, String)> = resp
294            .headers
295            .iter()
296            .filter(|(k, _)| !is_hop_by_hop(k))
297            .filter(|(k, _)| !(opts.pretty && k.eq_ignore_ascii_case("content-length")))
298            .cloned()
299            .collect();
300        headers.push((
301            "Access-Control-Allow-Methods".to_string(),
302            "GET, POST, OPTIONS".to_string(),
303        ));
304        cache_header(&mut headers, opts);
305
306        let body = if opts.pretty {
307            match serde_json::from_slice::<Value>(&resp.body) {
308                // Go re-indents the proxied body with two spaces.
309                Ok(v) => encode_json_indent(&v, b"  "),
310                Err(_) => resp.body.clone(),
311            }
312        } else {
313            resp.body.clone()
314        };
315
316        ForwardResponse {
317            status: resp.status,
318            headers,
319            body,
320        }
321    }
322}
323
324fn cache_header(headers: &mut Vec<(String, String)>, opts: &ForwardOptions) {
325    if let Some(d) = opts.cache {
326        if d > Duration::ZERO {
327            headers.push((
328                "Cache-Control".to_string(),
329                format!("public, max-age={}", d.as_secs()),
330            ));
331        }
332    }
333}
334
335/// Encodes a serializable value to JSON bytes; when `pretty`, uses a 4-space
336/// indent to match the Go override encoder (`SetIndent("", "    ")`).
337fn encode_json<T: serde::Serialize>(v: &T, pretty: bool) -> Vec<u8> {
338    if pretty {
339        return encode_json_indent(v, b"    ");
340    }
341    serde_json::to_vec(v).unwrap_or_default()
342}
343
344/// Encodes `v` as pretty JSON using the given indent unit.
345fn encode_json_indent<T: serde::Serialize>(v: &T, indent: &[u8]) -> Vec<u8> {
346    let mut buf = Vec::new();
347    let fmt = serde_json::ser::PrettyFormatter::with_indent(indent);
348    let mut ser = serde_json::Serializer::with_formatter(&mut buf, fmt);
349    if v.serialize(&mut ser).is_ok() {
350        buf
351    } else {
352        serde_json::to_vec(v).unwrap_or_default()
353    }
354}
355
356fn internal_error(msg: &str) -> ForwardResponse {
357    ForwardResponse {
358        status: 500,
359        headers: vec![("Content-Type".to_string(), "text/plain".to_string())],
360        body: format!("{msg}\n").into_bytes(),
361    }
362}
ethrpc_rs/rpc.rs

ethrpc_rs/
rpc.rs
