server: bounded channels and backpressure

benches/helpers.rs

@@ -132,7 +132,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;
+	use jsonrpsee::{core::server::rpc_module::SubscriptionMessage, server::ServerBuilder};
 
 	let server = ServerBuilder::default()
 		.max_request_body_size(u32::MAX)
@@ -146,11 +146,17 @@ pub async fn ws_server(handle: tokio::runtime::Handle) -> (String, jsonrpsee::se
 	let mut module = gen_rpc_module();
 
 	module
-		.register_subscription(SUB_METHOD_NAME, SUB_METHOD_NAME, UNSUB_METHOD_NAME, |_params, mut sink, _ctx| {
-			let x = "Hello";
-			tokio::spawn(async move { sink.send(&x) });
-			Ok(())
-		})
+		.register_subscription(
+			SUB_METHOD_NAME,
+			SUB_METHOD_NAME,
+			UNSUB_METHOD_NAME,
+			|_params, pending, _ctx| async move {
+				let sink = pending.accept().await?;
+				let msg = SubscriptionMessage::from_json(&"Hello")?;
+				sink.send(msg).await?;
+				Ok(())
+			},
+		)
 		.unwrap();
 
 	let addr = format!("ws://{}", server.local_addr().unwrap());

core/Cargo.toml

@@ -27,7 +27,6 @@ soketto = { version = "0.7.1", optional = true }
 parking_lot = { version = "0.12", optional = true }
 tokio = { version = "1.16", optional = true }
 wasm-bindgen-futures = { version = "0.4.19", optional = true }
-futures-channel = { version = "0.3.14", optional = true }
 futures-timer = { version = "3", optional = true }
 globset = { version = "0.4", optional = true }
 http = { version = "0.2.7", optional = true }
@@ -45,7 +44,8 @@ server = [
 	"rand",
 	"tokio/rt",
 	"tokio/sync",
-	"futures-channel",
+	"tokio/macros",
+	"tokio/time",
 ]
 client = ["futures-util/sink", "tokio/sync"]
 async-client = [

core/src/client/async_client/mod.rs

@@ -40,7 +40,7 @@ use tracing::instrument;
 
 use super::{generate_batch_id_range, FrontToBack, IdKind, RequestIdManager};
 
-/// Wrapper over a [`oneshot::Receiver`](futures_channel::oneshot::Receiver) that reads
+/// Wrapper over a [`oneshot::Receiver`](tokio::sync::oneshot::Receiver) that reads
 /// the underlying channel once and then stores the result in String.
 /// It is possible that the error is read more than once if several calls are made
 /// when the background thread has been terminated.

core/src/error.rs

@@ -27,8 +27,7 @@
 use std::fmt;
 
 use jsonrpsee_types::error::{
-	CallError, ErrorObject, ErrorObjectOwned, CALL_EXECUTION_FAILED_CODE, INVALID_PARAMS_CODE, SUBSCRIPTION_CLOSED,
-	UNKNOWN_ERROR_CODE,
+	CallError, ErrorObject, ErrorObjectOwned, CALL_EXECUTION_FAILED_CODE, INVALID_PARAMS_CODE, UNKNOWN_ERROR_CODE,
 };
 
 /// Convenience type for displaying errors.
@@ -108,21 +107,6 @@ pub enum Error {
 	/// Access control verification of HTTP headers failed.
 	#[error("HTTP header: `{0}` value: `{1}` verification failed")]
 	HttpHeaderRejected(&'static str, String),
-	/// Failed to execute a method because a resource was already at capacity
-	#[error("Resource at capacity: {0}")]
-	ResourceAtCapacity(&'static str),
-	/// Failed to register a resource due to a name conflict
-	#[error("Resource name already taken: {0}")]
-	ResourceNameAlreadyTaken(&'static str),
-	/// Failed to initialize resources for a method at startup
-	#[error("Resource name `{0}` not found for method `{1}`")]
-	ResourceNameNotFoundForMethod(&'static str, &'static str),
-	/// Trying to claim resources for a method execution, but the method resources have not been initialized
-	#[error("Method `{0}` has uninitialized resources")]
-	UninitializedMethod(Box<str>),
-	/// Failed to register a resource due to a maximum number of resources already registered
-	#[error("Maximum number of resources reached")]
-	MaxResourcesReached,
 	/// Custom error.
 	#[error("Custom error: {0}")]
 	Custom(String),
@@ -161,34 +145,6 @@ impl From<Error> for ErrorObjectOwned {
 	}
 }
 
-/// A type to represent when a subscription gets closed
-/// by either the server or client side.
-#[derive(Clone, Debug)]
-pub enum SubscriptionClosed {
-	/// The remote peer closed the connection or called the unsubscribe method.
-	RemotePeerAborted,
-	/// The subscription was completed successfully by the server.
-	Success,
-	/// The subscription failed during execution by the server.
-	Failed(ErrorObject<'static>),
-}
-
-impl From<SubscriptionClosed> for ErrorObjectOwned {
-	fn from(err: SubscriptionClosed) -> Self {
-		match err {
-			SubscriptionClosed::RemotePeerAborted => {
-				ErrorObject::owned(SUBSCRIPTION_CLOSED, "Subscription was closed by the remote peer", None::<()>)
-			}
-			SubscriptionClosed::Success => ErrorObject::owned(
-				SUBSCRIPTION_CLOSED,
-				"Subscription was completed by the server successfully",
-				None::<()>,
-			),
-			SubscriptionClosed::Failed(err) => err,
-		}
-	}
-}
-
 /// Generic transport error.
 #[derive(Debug, thiserror::Error)]
 pub enum GenericTransportError {
@@ -229,3 +185,79 @@ impl From<hyper::Error> for Error {
 		Error::Transport(hyper_err.into())
 	}
 }
+
+/// The error returned by the subscription's method for the rpc server implementation.
+///
+/// It provides an abstraction to make the API more ergonomic while handling errors
+/// that may occur during the subscription callback.
+#[derive(Debug)]
+pub enum SubscriptionCallbackError {
+	/// Error cause is propagated by other code or connection related.
+	None,
+	/// Some error happened to be logged.
+	Some(String),
+}
+
+// User defined error.
+impl From<anyhow::Error> for SubscriptionCallbackError {
+	fn from(e: anyhow::Error) -> Self {
+		Self::Some(format!("Other: {e}"))
+	}
+}
+
+// User defined error.
+impl From<Box<dyn std::error::Error>> for SubscriptionCallbackError {
+	fn from(e: Box<dyn std::error::Error>) -> Self {
+		Self::Some(format!("Other: {e}"))
+	}
+}
+
+impl From<CallError> for SubscriptionCallbackError {
+	fn from(e: CallError) -> Self {
+		Self::Some(e.to_string())
+	}
+}
+
+impl From<SubscriptionAcceptRejectError> for SubscriptionCallbackError {
+	fn from(_: SubscriptionAcceptRejectError) -> Self {
+		Self::None
+	}
+}
+
+impl From<serde_json::Error> for SubscriptionCallbackError {
+	fn from(e: serde_json::Error) -> Self {
+		Self::Some(format!("Failed to parse SubscriptionMessage::from_json: {e}"))
+	}
+}
+
+#[cfg(feature = "server")]
+impl From<crate::server::rpc_module::TrySendError> for SubscriptionCallbackError {
+	fn from(e: crate::server::rpc_module::TrySendError) -> Self {
+		Self::Some(format!("SubscriptionSink::try_send failed: {e}"))
+	}
+}
+
+#[cfg(feature = "server")]
+impl From<crate::server::rpc_module::DisconnectError> for SubscriptionCallbackError {
+	fn from(e: crate::server::rpc_module::DisconnectError) -> Self {
+		Self::Some(format!("SubscriptionSink::send failed: {e}"))
+	}
+}
+
+#[cfg(feature = "server")]
+impl From<crate::server::rpc_module::SendTimeoutError> for SubscriptionCallbackError {
+	fn from(e: crate::server::rpc_module::SendTimeoutError) -> Self {
+		Self::Some(format!("SubscriptionSink::send_timeout failed: {e}"))
+	}
+}
+
+/// The error returned while accepting or rejecting a subscription.
+#[derive(Debug, Copy, Clone)]
+pub enum SubscriptionAcceptRejectError {
+	/// The method was already called.
+	AlreadyCalled,
+	/// The remote peer closed the connection or called the unsubscribe method.
+	RemotePeerAborted,
+	/// The subscription response message was too large.
+	MessageTooLarge,
+}

core/src/lib.rs

@@ -58,11 +58,19 @@ cfg_client! {
 /// Shared tracing helpers to trace RPC calls.
 pub mod tracing;
 pub use async_trait::async_trait;
-pub use error::Error;
+pub use error::{Error, SubscriptionAcceptRejectError, SubscriptionCallbackError};
 
 /// JSON-RPC result.
 pub type RpcResult<T> = std::result::Result<T, Error>;
 
+/// The return type of the subscription's method for the rpc server implementation.
+///
+/// **Note**: The error does not contain any data and is discarded on drop.
+pub type SubscriptionResult = Result<(), SubscriptionCallbackError>;
+
+/// Empty server `RpcParams` type to use while registering modules.
+pub type EmptyServerParams = Vec<()>;
+
 /// Re-exports for proc-macro library to not require any additional
 /// dependencies to be explicitly added on the client side.
 #[doc(hidden)]