unify usage of JSON via Box<RawValue> (#1545)

* feat: client rpc middleware

* PoC works

* cargo fmt

* more refactoring

* use Cow in Notification to avoid alloc

* cleanup some todos

* rpc trait: return Result<MethodResponse, Err>

* remove infallible err

* make it compile

* fix tests

* introduce client method response type

* fix faulty imports

* minor cleanup

* introduce Batch/BatchEntry for middleware

* remove ignore for batch test

* fix rustdocs

* add rpc middleware for the async client

* remove serialize specific types

* commit missing file

* no serde_json::Value

* more nit fixing

* more cleanup

* refactor method response client

* fix some nits

* add client middleware rpc

* fix wasm build

* add client middleware example

* Update examples/examples/rpc_middleware_client.rs

* ToJson -> RawValue

* replace Future type with impl Trait

* revert changelog

* remove logger response future

* some cleanup

* move request timeout from transport to client

* more nit fixing

* have pass over examples

* show proper batch middleware example

* middleware: clean up batch type

* fix wasm build

* Update Cargo.toml

* doc: fix typo

* core: remove tracing mod

* fix more clippy

* refactor: use json rawvalue

* replace StringError with SubscriptionErr

* cleanup json APIs with RawValue

* cleanup and fix tests

* remove from_json api

* revert faulty change in types

* revert faulty change in types v2

* fix rustdocs

* use explicit JSON null value instead of default

* cleanup Rawvalue::from_string usage

* fix tests

* Update core/src/traits.rs

* Update benches/bench.rs

* Update core/src/params.rs

* cleanup expects on RawValue

* improve docs

* replace manual serialization with to_raw_value

It was much faster

* fix clippy

* fix nit

Author: niklasad1

benches/bench.rs

@@ -113,7 +113,7 @@ pub fn jsonrpsee_types_v2(crit: &mut Criterion) {
 	// Construct the serialized array request using the `RawValue` directly.
 	crit.bench_function("jsonrpsee_types_array_params_baseline", |b| {
 		b.iter(|| {
-			let params = serde_json::value::RawValue::from_string("[1, 2]".to_string()).unwrap();
+			let params = serde_json::value::to_raw_value(&[1, 2]).unwrap();
 
 			let request = Request::borrowed("say_hello", Some(&params), Id::Number(0));
 			v2_serialize(request);
@@ -136,8 +136,7 @@ pub fn jsonrpsee_types_v2(crit: &mut Criterion) {
 	// Construct the serialized object request using the `RawValue` directly.
 	crit.bench_function("jsonrpsee_types_object_params_baseline", |b| {
 		b.iter(|| {
-			let params = serde_json::value::RawValue::from_string(r#"{"key": 1}"#.to_string()).unwrap();
-
+			let params = serde_json::value::to_raw_value(r#"{"key": 1}"#).unwrap();
 			let request = Request::borrowed("say_hello", Some(&params), Id::Number(0));
 			v2_serialize(request);
 		})

benches/helpers.rs

@@ -131,7 +131,7 @@ pub async fn http_server(handle: tokio::runtime::Handle) -> (String, jsonrpsee::
 /// Run jsonrpsee WebSocket server for benchmarks.
 #[cfg(not(feature = "jsonrpc-crate"))]
 pub async fn ws_server(handle: tokio::runtime::Handle) -> (String, jsonrpsee::server::ServerHandle) {
-	use jsonrpsee::server::{ServerBuilder, ServerConfig, SubscriptionMessage};
+	use jsonrpsee::server::{ServerBuilder, ServerConfig};
 
 	let config = ServerConfig::builder()
 		.max_request_body_size(u32::MAX)
@@ -150,8 +150,8 @@ pub async fn ws_server(handle: tokio::runtime::Handle) -> (String, jsonrpsee::se
 			UNSUB_METHOD_NAME,
 			|_params, pending, _ctx, _| async move {
 				let sink = pending.accept().await?;
-				let msg = SubscriptionMessage::from("Hello");
-				sink.send(msg).await?;
+				let json = serde_json::value::to_raw_value(&"Hello").unwrap();
+				sink.send(json).await?;
 
 				Ok(())
 			},

core/src/client/mod.rs

@@ -787,7 +787,7 @@ impl ToJson for MethodResponse {
 	fn to_json(&self) -> Result<Box<RawValue>, serde_json::Error> {
 		match &self.inner {
 			MethodResponseKind::MethodCall(call) => call.to_json(),
-			MethodResponseKind::Notification => Ok(Box::<RawValue>::default()),
+			MethodResponseKind::Notification => Ok(RawValue::NULL.to_owned()),
 			MethodResponseKind::Batch(json) => serde_json::value::to_raw_value(json),
 			MethodResponseKind::Subscription(s) => serde_json::value::to_raw_value(&s.rp),
 		}

core/src/error.rs

@@ -24,13 +24,53 @@
 // IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-/// A type that returns the error as a `String` from `SubscriptionCallback`.
-#[derive(Debug)]
-pub struct StringError(pub(crate) String);
+use serde::Serialize;
+use serde_json::value::RawValue;
 
-impl<T: ToString> From<T> for StringError {
+#[derive(Debug, Clone)]
+pub(crate) enum InnerSubscriptionErr {
+	String(String),
+	Json(Box<RawValue>),
+}
+
+/// Error returned when a subscription fails where the error is returned
+/// as special error notification with the following format:
+///
+/// ```json
+/// {"jsonrpc":"2.0", "method":"subscription_error", "params": {"subscription": "sub_id", "error": <error message from this type>}}
+/// ```
+///
+/// It's recommended to use [`SubscriptionError::from_json`] to create a new instance of this error
+/// if the underlying error is a JSON value. That will ensure that the error is serialized correctly.
+///
+/// SubscriptionError::from will serialize the error as a string, which is not
+/// recommended and should only by used in the value of a `String` type.
+/// It's mainly provided for convenience and to allow for easy conversion any type that implements StdError.
+#[derive(Debug, Clone)]
+pub struct SubscriptionError(pub(crate) InnerSubscriptionErr);
+
+impl Serialize for SubscriptionError {
+	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+	where
+		S: serde::Serializer,
+	{
+		match &self.0 {
+			InnerSubscriptionErr::String(s) => serializer.serialize_str(s),
+			InnerSubscriptionErr::Json(json) => json.serialize(serializer),
+		}
+	}
+}
+
+impl<T: ToString> From<T> for SubscriptionError {
 	fn from(val: T) -> Self {
-		StringError(val.to_string())
+		Self(InnerSubscriptionErr::String(val.to_string()))
+	}
+}
+
+impl SubscriptionError {
+	/// Create a new `SubscriptionError` from a JSON value.
+	pub fn from_json(json: Box<RawValue>) -> Self {
+		Self(InnerSubscriptionErr::Json(json))
 	}
 }
 

core/src/lib.rs

@@ -61,7 +61,7 @@ cfg_client_or_server! {
 }
 
 pub use async_trait::async_trait;
-pub use error::{RegisterMethodError, StringError};
+pub use error::{RegisterMethodError, SubscriptionError};
 
 /// JSON-RPC result.
 pub type RpcResult<T> = std::result::Result<T, jsonrpsee_types::ErrorObjectOwned>;
@@ -99,7 +99,7 @@ pub use std::borrow::Cow;
 pub const TEN_MB_SIZE_BYTES: u32 = 10 * 1024 * 1024;
 
 /// The return type if the subscription wants to return `Result`.
-pub type SubscriptionResult = Result<(), StringError>;
+pub type SubscriptionResult = Result<(), SubscriptionError>;
 
 /// Type erased error.
 pub type BoxError = Box<dyn std::error::Error + Send + Sync>;

core/src/params.rs

@@ -33,6 +33,7 @@ use serde_json::value::RawValue;
 /// Helper module for building parameters.
 mod params_builder {
 	use serde::Serialize;
+	use serde_json::value::RawValue;
 
 	/// Initial number of bytes for a parameter length.
 	const PARAM_BYTES_CAPACITY: usize = 128;
@@ -106,8 +107,8 @@ mod params_builder {
 			Ok(())
 		}
 
-		/// Finish the building process and return a JSON compatible string.
-		pub(crate) fn build(mut self) -> Option<String> {
+		/// Finish the building process and return a JSON object.
+		pub(crate) fn build(mut self) -> Option<Box<RawValue>> {
 			if self.bytes.is_empty() {
 				return None;
 			}
@@ -120,7 +121,8 @@ mod params_builder {
 			}
 
 			// Safety: This is safe because JSON does not emit invalid UTF-8.
-			Some(unsafe { String::from_utf8_unchecked(self.bytes) })
+			let json_str = unsafe { String::from_utf8_unchecked(self.bytes) };
+			Some(RawValue::from_string(json_str).expect("Valid JSON String; qed"))
 		}
 	}
 }
@@ -164,7 +166,7 @@ impl Default for ObjectParams {
 
 impl ToRpcParams for ObjectParams {
 	fn to_rpc_params(self) -> Result<Option<Box<RawValue>>, serde_json::Error> {
-		if let Some(json) = self.0.build() { RawValue::from_string(json).map(Some) } else { Ok(None) }
+		Ok(self.0.build())
 	}
 }
 
@@ -206,7 +208,7 @@ impl Default for ArrayParams {
 
 impl ToRpcParams for ArrayParams {
 	fn to_rpc_params(self) -> Result<Option<Box<RawValue>>, serde_json::Error> {
-		if let Some(json) = self.0.build() { RawValue::from_string(json).map(Some) } else { Ok(None) }
+		Ok(self.0.build())
 	}
 }
 

core/src/server/error.rs

@@ -25,6 +25,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::server::SubscriptionMessage;
+use serde_json::value::RawValue;
 use tokio::sync::mpsc;
 
 /// Error that may occur during [`crate::server::MethodSink::try_send`] or [`crate::server::SubscriptionSink::try_send`].
@@ -60,23 +61,23 @@ pub enum SendTimeoutError {
 #[error("The remote peer closed the connection")]
 pub struct PendingSubscriptionAcceptError;
 
-impl From<mpsc::error::SendError<String>> for DisconnectError {
-	fn from(e: mpsc::error::SendError<String>) -> Self {
+impl From<mpsc::error::SendError<Box<RawValue>>> for DisconnectError {
+	fn from(e: mpsc::error::SendError<Box<RawValue>>) -> Self {
 		DisconnectError(SubscriptionMessage::from_complete_message(e.0))
 	}
 }
 
-impl From<mpsc::error::TrySendError<String>> for TrySendError {
-	fn from(e: mpsc::error::TrySendError<String>) -> Self {
+impl From<mpsc::error::TrySendError<Box<RawValue>>> for TrySendError {
+	fn from(e: mpsc::error::TrySendError<Box<RawValue>>) -> Self {
 		match e {
 			mpsc::error::TrySendError::Closed(m) => Self::Closed(SubscriptionMessage::from_complete_message(m)),
 			mpsc::error::TrySendError::Full(m) => Self::Full(SubscriptionMessage::from_complete_message(m)),
 		}
 	}
 }
 
-impl From<mpsc::error::SendTimeoutError<String>> for SendTimeoutError {
-	fn from(e: mpsc::error::SendTimeoutError<String>) -> Self {
+impl From<mpsc::error::SendTimeoutError<Box<RawValue>>> for SendTimeoutError {
+	fn from(e: mpsc::error::SendTimeoutError<Box<RawValue>>) -> Self {
 		match e {
 			mpsc::error::SendTimeoutError::Closed(m) => Self::Closed(SubscriptionMessage::from_complete_message(m)),
 			mpsc::error::SendTimeoutError::Timeout(m) => Self::Timeout(SubscriptionMessage::from_complete_message(m)),

core/src/server/helpers.rs

@@ -27,27 +27,28 @@
 use std::time::Duration;
 
 use jsonrpsee_types::{ErrorCode, ErrorObject, Id, InvalidRequest, Response, ResponsePayload};
+use serde_json::value::RawValue;
 use tokio::sync::mpsc;
 
-use super::{DisconnectError, SendTimeoutError, SubscriptionMessage, TrySendError};
+use super::{DisconnectError, SendTimeoutError, TrySendError};
 
 /// Sink that is used to send back the result to the server for a specific method.
 #[derive(Clone, Debug)]
 pub struct MethodSink {
 	/// Channel sender.
-	tx: mpsc::Sender<String>,
+	tx: mpsc::Sender<Box<RawValue>>,
 	/// Max response size in bytes for a executed call.
 	max_response_size: u32,
 }
 
 impl MethodSink {
 	/// Create a new `MethodSink` with unlimited response size.
-	pub fn new(tx: mpsc::Sender<String>) -> Self {
+	pub fn new(tx: mpsc::Sender<Box<RawValue>>) -> Self {
 		MethodSink { tx, max_response_size: u32::MAX }
 	}
 
 	/// Create a new `MethodSink` with a limited response size.
-	pub fn new_with_limit(tx: mpsc::Sender<String>, max_response_size: u32) -> Self {
+	pub fn new_with_limit(tx: mpsc::Sender<Box<RawValue>>, max_response_size: u32) -> Self {
 		MethodSink { tx, max_response_size }
 	}
 
@@ -74,25 +75,25 @@ impl MethodSink {
 	/// connection has been closed or if the message buffer is full.
 	///
 	/// Returns the message if the send fails such that either can be thrown away or re-sent later.
-	pub fn try_send(&mut self, msg: String) -> Result<(), TrySendError> {
+	pub fn try_send(&mut self, msg: Box<RawValue>) -> Result<(), TrySendError> {
 		self.tx.try_send(msg).map_err(Into::into)
 	}
 
 	/// Async send which will wait until there is space in channel buffer or that the subscription is disconnected.
-	pub async fn send(&self, msg: String) -> Result<(), DisconnectError> {
+	pub async fn send(&self, msg: Box<RawValue>) -> Result<(), DisconnectError> {
 		self.tx.send(msg).await.map_err(Into::into)
 	}
 
 	/// Send a JSON-RPC error to the client
 	pub async fn send_error<'a>(&self, id: Id<'a>, err: ErrorObject<'a>) -> Result<(), DisconnectError> {
 		let payload = ResponsePayload::<()>::error_borrowed(err);
-		let json = serde_json::to_string(&Response::new(payload, id)).expect("valid JSON; qed");
+		let json = serde_json::value::to_raw_value(&Response::new(payload, id)).expect("valid JSON; qed");
 
 		self.send(json).await
 	}
 
 	/// Similar to `MethodSink::send` but only waits for a limited time.
-	pub async fn send_timeout(&self, msg: String, timeout: Duration) -> Result<(), SendTimeoutError> {
+	pub async fn send_timeout(&self, msg: Box<RawValue>, timeout: Duration) -> Result<(), SendTimeoutError> {
 		self.tx.send_timeout(msg, timeout).await.map_err(Into::into)
 	}
 
@@ -112,7 +113,7 @@ impl MethodSink {
 			// The permit is thrown away here because it's just
 			// a way to ensure that the return buffer has space.
 			Ok(_) => Ok(()),
-			Err(_) => Err(DisconnectError(SubscriptionMessage::empty())),
+			Err(_) => Err(DisconnectError(RawValue::NULL.to_owned().into())),
 		}
 	}
 }

core/src/server/method_response.rs

@@ -191,7 +191,7 @@ impl MethodResponse {
 			Ok(_) => {
 				// Safety - serde_json does not emit invalid UTF-8.
 				let result = unsafe { String::from_utf8_unchecked(writer.into_bytes()) };
-				let json = RawValue::from_string(result).expect("JSON serialization infallible; qed");
+				let json = RawValue::from_string(result).expect("Valid JSON String; qed");
 
 				Self { json, success_or_error, kind, on_close: rp.on_exit, extensions: Extensions::new() }
 			}
@@ -261,7 +261,7 @@ impl MethodResponse {
 	/// Create notification response which is a response that doesn't expect a reply.
 	pub fn notification() -> Self {
 		Self {
-			json: Box::<RawValue>::default(),
+			json: RawValue::NULL.to_owned(),
 			success_or_error: MethodResponseResult::Success,
 			kind: ResponseKind::Notification,
 			on_close: None,
@@ -370,7 +370,7 @@ impl BatchResponseBuilder {
 		} else {
 			self.result.pop();
 			self.result.push(']');
-			let json = RawValue::from_string(self.result).expect("JSON serialization infallible; qed");
+			let json = RawValue::from_string(self.result).expect("BatchResponse builds a valid JSON String; qed");
 			BatchResponse { json, extensions: self.extensions }
 		}
 	}