jsonrpsee_core/server/
helpers.rs

1// Copyright 2019-2021 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any
4// person obtaining a copy of this software and associated
5// documentation files (the "Software"), to deal in the
6// Software without restriction, including without
7// limitation the rights to use, copy, modify, merge,
8// publish, distribute, sublicense, and/or sell copies of
9// the Software, and to permit persons to whom the Software
10// is furnished to do so, subject to the following
11// conditions:
12//
13// The above copyright notice and this permission notice
14// shall be included in all copies or substantial portions
15// of the Software.
16//
17// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
18// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
19// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
20// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
21// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
22// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
23// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
24// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
25// DEALINGS IN THE SOFTWARE.
26
27use std::time::Duration;
28
29use jsonrpsee_types::{ErrorCode, ErrorObject, Id, InvalidRequest, Response, ResponsePayload};
30use serde_json::value::RawValue;
31use tokio::sync::mpsc;
32
33use super::{DisconnectError, SendTimeoutError, TrySendError};
34
35/// Sink that is used to send back the result to the server for a specific method.
36#[derive(Clone, Debug)]
37pub struct MethodSink {
38    /// Channel sender.
39    tx: mpsc::Sender<String>,
40    /// Max response size in bytes for a executed call.
41    max_response_size: u32,
42}
43
44impl MethodSink {
45    /// Create a new `MethodSink` with unlimited response size.
46    pub fn new(tx: mpsc::Sender<String>) -> Self {
47        MethodSink { tx, max_response_size: u32::MAX }
48    }
49
50    /// Create a new `MethodSink` with a limited response size.
51    pub fn new_with_limit(tx: mpsc::Sender<String>, max_response_size: u32) -> Self {
52        MethodSink { tx, max_response_size }
53    }
54
55    /// Returns whether this channel is closed without needing a context.
56    pub fn is_closed(&self) -> bool {
57        self.tx.is_closed()
58    }
59
60    /// Same as [`tokio::sync::mpsc::Sender::closed`].
61    ///
62    /// # Cancel safety
63    /// This method is cancel safe. Once the channel is closed,
64    /// it stays closed forever and all future calls to closed will return immediately.
65    pub async fn closed(&self) {
66        self.tx.closed().await
67    }
68
69    /// Get the max response size.
70    pub const fn max_response_size(&self) -> u32 {
71        self.max_response_size
72    }
73
74    /// Attempts to send out the message immediately and fails if the underlying
75    /// connection has been closed or if the message buffer is full.
76    ///
77    /// Returns the message if the send fails such that either can be thrown away or re-sent later.
78    pub fn try_send(&mut self, msg: String) -> Result<(), TrySendError> {
79        self.tx.try_send(msg).map_err(Into::into)
80    }
81
82    /// Async send which will wait until there is space in channel buffer or that the subscription
83    /// is disconnected.
84    pub async fn send(&self, msg: String) -> Result<(), DisconnectError> {
85        self.tx.send(msg).await.map_err(Into::into)
86    }
87
88    /// Send a JSON-RPC error to the client
89    pub async fn send_error<'a>(
90        &self,
91        id: Id<'a>,
92        err: ErrorObject<'a>,
93    ) -> Result<(), DisconnectError> {
94        let payload = ResponsePayload::<()>::error_borrowed(err);
95        let json = serde_json::to_string(&Response::new(payload, id)).expect("valid JSON; qed");
96
97        self.send(json).await
98    }
99
100    /// Similar to `MethodSink::send` but only waits for a limited time.
101    pub async fn send_timeout(
102        &self,
103        msg: String,
104        timeout: Duration,
105    ) -> Result<(), SendTimeoutError> {
106        self.tx.send_timeout(msg, timeout).await.map_err(Into::into)
107    }
108
109    /// Get the capacity of the channel.
110    pub fn capacity(&self) -> usize {
111        self.tx.capacity()
112    }
113
114    /// Get the max capacity of the channel.
115    pub fn max_capacity(&self) -> usize {
116        self.tx.max_capacity()
117    }
118}
119
120/// Distinguish malformed JSON from an invalid request and recover its [`Id`] when possible.
121pub fn prepare_error(data: &[u8]) -> (Id<'_>, ErrorCode) {
122    let Ok(value) = serde_json::from_slice::<&RawValue>(data) else {
123        return (Id::Null, ErrorCode::ParseError);
124    };
125    let id = if value.get().starts_with('{') {
126        serde_json::from_slice::<InvalidRequest>(data).map(|request| request.id).unwrap_or(Id::Null)
127    } else {
128        Id::Null
129    };
130    (id, ErrorCode::InvalidRequest)
131}
132
133#[cfg(test)]
134mod tests {
135    use super::prepare_error;
136    use jsonrpsee_types::{ErrorCode, Id};
137
138    #[test]
139    fn error_classification_invalid_envelopes_without_id() {
140        for data in
141            [r#"{"jsonrpc":"1.0","method":"example","params":[]}"#, r#"{"jsonrpc":"2.0"}"#, r#"{}"#]
142        {
143            assert_eq!(
144                prepare_error(data.as_bytes()),
145                (Id::Null, ErrorCode::InvalidRequest),
146                "{data}"
147            );
148        }
149    }
150
151    #[test]
152    fn error_classification_malformed_json() {
153        for data in ["", " ", "{", "[", r#"{"id":1,"method":}"#, r#"{"id":1} trailing"#] {
154            assert_eq!(prepare_error(data.as_bytes()), (Id::Null, ErrorCode::ParseError), "{data}");
155        }
156    }
157
158    #[test]
159    fn error_classification_recoverable_ids() {
160        for (data, id) in [
161            (r#"{"jsonrpc":"1.0","method":"example","id":0}"#, Id::Number(0)),
162            (r#"{"jsonrpc":"1.0","id":18446744073709551615}"#, Id::Number(u64::MAX)),
163            (" \n {\"id\":\"keep-id\"} \t ", Id::Str("keep-id".into())),
164            (r#"{"id":"escaped\u002did"}"#, Id::Str("escaped-id".into())),
165            (r#"{"id":null}"#, Id::Null),
166        ] {
167            assert_eq!(prepare_error(data.as_bytes()), (id, ErrorCode::InvalidRequest), "{data}");
168        }
169    }
170
171    #[test]
172    fn error_classification_unrecoverable_ids_and_nonobjects() {
173        for data in [
174            r#"{"id":true}"#,
175            r#"{"id":{}}"#,
176            r#"{"id":[]}"#,
177            r#"{"id":-1}"#,
178            r#"["not-an-object-id"]"#,
179            "[1]",
180            "[]",
181            "true",
182            "null",
183            "42",
184            r#""text""#,
185        ] {
186            assert_eq!(
187                prepare_error(data.as_bytes()),
188                (Id::Null, ErrorCode::InvalidRequest),
189                "{data}"
190            );
191        }
192    }
193}
jsonrpsee_core/server/helpers.rs

jsonrpsee_core/server/
helpers.rs
