middleware: RpcServiceT distinct return types for notif, batch, call (#1564)

* middleware: fix option layer

Close #1560

* middleware: RpcServiceT unique ret tys for futures

* remove ToJson trait bound from RpcServiceT

* fix nit in error docs

* fix tests

* move ToJson to core::traits

* remove future.boxed() in RpcService impl

* remove needless clone

* simplify code in example

Author: niklasad1

client/http-client/src/client.rs

@@ -35,8 +35,8 @@ use crate::{HttpRequest, HttpResponse};
 use hyper::body::Bytes;
 use hyper::http::{Extensions, HeaderMap};
 use jsonrpsee_core::client::{
-	BatchResponse, ClientT, Error, IdKind, MethodResponse, RequestIdManager, Subscription, SubscriptionClientT,
-	generate_batch_id_range,
+	BatchResponse, ClientT, Error, IdKind, MiddlewareBatchResponse, MiddlewareMethodResponse, MiddlewareNotifResponse,
+	RequestIdManager, Subscription, SubscriptionClientT, generate_batch_id_range,
 };
 use jsonrpsee_core::middleware::layer::{RpcLogger, RpcLoggerLayer};
 use jsonrpsee_core::middleware::{Batch, RpcServiceBuilder, RpcServiceT};
@@ -351,7 +351,12 @@ impl HttpClient<HttpBackend> {
 
 impl<S> ClientT for HttpClient<S>
 where
-	S: RpcServiceT<Error = Error, Response = MethodResponse> + Send + Sync,
+	S: RpcServiceT<
+			MethodResponse = Result<MiddlewareMethodResponse, Error>,
+			BatchResponse = Result<MiddlewareBatchResponse, Error>,
+			NotificationResponse = Result<MiddlewareNotifResponse, Error>,
+		> + Send
+		+ Sync,
 {
 	fn notification<Params>(&self, method: &str, params: Params) -> impl Future<Output = Result<(), Error>> + Send
 	where
@@ -363,13 +368,9 @@ where
 				None => None,
 			};
 			let params = params.to_rpc_params()?.map(StdCow::Owned);
+			let fut = self.service.notification(Notification::new(method.into(), params));
 
-			run_future_until_timeout(
-				self.service.notification(Notification::new(method.into(), params)),
-				self.request_timeout,
-			)
-			.await
-			.map_err(|e| Error::Transport(e.into()))?;
+			run_future_until_timeout(fut, self.request_timeout).await.map_err(|e| Error::Transport(e.into()))?;
 			Ok(())
 		}
 	}
@@ -392,8 +393,7 @@ where
 				self.request_timeout,
 			)
 			.await?
-			.into_method_call()
-			.expect("Method call must return a method call reponse; qed");
+			.into_response();
 
 			let rp = ResponseSuccess::try_from(method_response.into_inner())?;
 
@@ -405,7 +405,7 @@ where
 	fn batch_request<'a, R>(
 		&self,
 		batch: BatchRequestBuilder<'a>,
-	) -> impl Future<Output = Result<BatchResponse<'a, R>, Error>> + Send
+	) -> impl Future<Output = Result<jsonrpsee_core::client::BatchResponse<'a, R>, Error>> + Send
 	where
 		R: DeserializeOwned + fmt::Debug + 'a,
 	{
@@ -431,19 +431,18 @@ where
 				batch_request.push(req);
 			}
 
-			let rp = run_future_until_timeout(self.service.batch(batch_request), self.request_timeout).await?;
-			let json_rps = rp.into_batch().expect("Batch must return a batch reponse; qed");
+			let rps = run_future_until_timeout(self.service.batch(batch_request), self.request_timeout).await?;
 
 			let mut batch_response = Vec::new();
 			let mut success = 0;
 			let mut failed = 0;
 
 			// Fill the batch response with placeholder values.
-			for _ in 0..json_rps.len() {
+			for _ in 0..rps.len() {
 				batch_response.push(Err(ErrorObject::borrowed(0, "", None)));
 			}
 
-			for rp in json_rps.into_iter() {
+			for rp in rps.into_iter() {
 				let id = rp.id().try_parse_inner_as_number()?;
 
 				let res = match ResponseSuccess::try_from(rp.into_inner()) {
@@ -477,7 +476,12 @@ where
 
 impl<S> SubscriptionClientT for HttpClient<S>
 where
-	S: RpcServiceT<Error = Error, Response = MethodResponse> + Send + Sync,
+	S: RpcServiceT<
+			MethodResponse = Result<MiddlewareMethodResponse, Error>,
+			BatchResponse = Result<MiddlewareBatchResponse, Error>,
+			NotificationResponse = Result<MiddlewareNotifResponse, Error>,
+		> + Send
+		+ Sync,
 {
 	/// Send a subscription request to the server. Not implemented for HTTP; will always return
 	/// [`Error::HttpNotImplemented`].
@@ -503,9 +507,9 @@ where
 	}
 }
 
-async fn run_future_until_timeout<F>(fut: F, timeout: Duration) -> Result<MethodResponse, Error>
+async fn run_future_until_timeout<F, T>(fut: F, timeout: Duration) -> Result<T, Error>
 where
-	F: std::future::Future<Output = Result<MethodResponse, Error>>,
+	F: std::future::Future<Output = Result<T, Error>>,
 {
 	match tokio::time::timeout(timeout, fut).await {
 		Ok(Ok(r)) => Ok(r),

client/http-client/src/rpc_service.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use hyper::body::Bytes;
 use jsonrpsee_core::{
 	BoxError, JsonRawValue,
-	client::{Error, MethodResponse},
+	client::{Error, MiddlewareBatchResponse, MiddlewareMethodResponse, MiddlewareNotifResponse},
 	middleware::{Batch, Notification, Request, RpcServiceT},
 };
 use jsonrpsee_types::Response;
@@ -34,21 +34,24 @@ where
 	B::Data: Send,
 	B::Error: Into<BoxError>,
 {
-	type Error = Error;
-	type Response = MethodResponse;
+	type BatchResponse = Result<MiddlewareBatchResponse, Error>;
+	type MethodResponse = Result<MiddlewareMethodResponse, Error>;
+	type NotificationResponse = Result<MiddlewareNotifResponse, Error>;
 
-	fn call<'a>(&self, request: Request<'a>) -> impl Future<Output = Result<Self::Response, Self::Error>> + Send + 'a {
+	fn call<'a>(&self, request: Request<'a>) -> impl Future<Output = Self::MethodResponse> + Send + 'a {
 		let service = self.service.clone();
 
 		async move {
 			let raw = serde_json::to_string(&request)?;
 			let bytes = service.send_and_read_body(raw).await.map_err(|e| Error::Transport(e.into()))?;
-			let json_rp: Response<Box<JsonRawValue>> = serde_json::from_slice(&bytes)?;
-			Ok(MethodResponse::method_call(json_rp.into_owned().into(), request.extensions))
+			let mut rp: Response<Box<JsonRawValue>> = serde_json::from_slice(&bytes)?;
+			rp.extensions = request.extensions;
+
+			Ok(MiddlewareMethodResponse::response(rp.into_owned().into()))
 		}
 	}
 
-	fn batch<'a>(&self, batch: Batch<'a>) -> impl Future<Output = Result<Self::Response, Self::Error>> + Send + 'a {
+	fn batch<'a>(&self, batch: Batch<'a>) -> impl Future<Output = Self::BatchResponse> + Send + 'a {
 		let service = self.service.clone();
 
 		async move {
@@ -59,20 +62,20 @@ where
 				.map(|r| r.into_owned().into())
 				.collect();
 
-			Ok(MethodResponse::batch(rp, batch.into_extensions()))
+			Ok(rp)
 		}
 	}
 
 	fn notification<'a>(
 		&self,
 		notif: Notification<'a>,
-	) -> impl Future<Output = Result<Self::Response, Self::Error>> + Send + 'a {
+	) -> impl Future<Output = Self::NotificationResponse> + Send + 'a {
 		let service = self.service.clone();
 
 		async move {
 			let raw = serde_json::to_string(&notif)?;
 			service.send(raw).await.map_err(|e| Error::Transport(e.into()))?;
-			Ok(MethodResponse::notification(notif.extensions))
+			Ok(notif.extensions.into())
 		}
 	}
 }

core/src/client/async_client/mod.rs

@@ -31,7 +31,7 @@ mod manager;
 mod rpc_service;
 mod utils;
 
-pub use rpc_service::{Error as RpcServiceError, RpcService};
+pub use rpc_service::RpcService;
 
 use std::borrow::Cow as StdCow;
 use std::time::Duration;
@@ -66,7 +66,10 @@ use tokio::sync::{mpsc, oneshot};
 use tower::layer::util::Identity;
 
 use self::utils::{InactivityCheck, IntervalStream};
-use super::{FrontToBack, IdKind, MethodResponse, RequestIdManager, generate_batch_id_range, subscription_channel};
+use super::{
+	FrontToBack, IdKind, MiddlewareBatchResponse, MiddlewareMethodResponse, MiddlewareNotifResponse, RequestIdManager,
+	generate_batch_id_range, subscription_channel,
+};
 
 pub(crate) type Notification<'a> = jsonrpsee_types::Notification<'a, Option<Box<JsonRawValue>>>;
 
@@ -455,16 +458,13 @@ impl<L> Client<L> {
 		!self.to_back.is_closed()
 	}
 
-	async fn run_future_until_timeout(
-		&self,
-		fut: impl Future<Output = Result<MethodResponse, RpcServiceError>>,
-	) -> Result<MethodResponse, Error> {
+	async fn run_future_until_timeout<T>(&self, fut: impl Future<Output = Result<T, Error>>) -> Result<T, Error> {
 		tokio::pin!(fut);
 
 		match futures_util::future::select(fut, futures_timer::Delay::new(self.request_timeout)).await {
 			Either::Left((Ok(r), _)) => Ok(r),
-			Either::Left((Err(RpcServiceError::Client(e)), _)) => Err(e),
-			Either::Left((Err(RpcServiceError::FetchFromBackend), _)) => Err(self.on_disconnect().await),
+			Either::Left((Err(Error::ServiceDisconnect), _)) => Err(self.on_disconnect().await),
+			Either::Left((Err(e), _)) => Err(e),
 			Either::Right(_) => Err(Error::RequestTimeout),
 		}
 	}
@@ -495,7 +495,12 @@ impl<L> Drop for Client<L> {
 
 impl<L> ClientT for Client<L>
 where
-	L: RpcServiceT<Error = RpcServiceError, Response = MethodResponse> + Send + Sync,
+	L: RpcServiceT<
+			MethodResponse = Result<MiddlewareMethodResponse, Error>,
+			BatchResponse = Result<MiddlewareBatchResponse, Error>,
+			NotificationResponse = Result<MiddlewareNotifResponse, Error>,
+		> + Send
+		+ Sync,
 {
 	fn notification<Params>(&self, method: &str, params: Params) -> impl Future<Output = Result<(), Error>> + Send
 	where
@@ -520,8 +525,8 @@ where
 			let id = self.id_manager.next_request_id();
 			let params = params.to_rpc_params()?;
 			let fut = self.service.call(Request::borrowed(method, params.as_deref(), id.clone()));
-			let rp = self.run_future_until_timeout(fut).await?.into_method_call().expect("Method call response");
-			let success = ResponseSuccess::try_from(rp.into_inner())?;
+			let rp = self.run_future_until_timeout(fut).await?;
+			let success = ResponseSuccess::try_from(rp.into_response().into_inner())?;
 
 			serde_json::from_str(success.result.get()).map_err(Into::into)
 		}
@@ -554,7 +559,7 @@ where
 			b.extensions_mut().insert(IsBatch { id_range });
 
 			let fut = self.service.batch(b);
-			let json_values = self.run_future_until_timeout(fut).await?.into_batch().expect("Batch response");
+			let json_values = self.run_future_until_timeout(fut).await?;
 
 			let mut responses = Vec::with_capacity(json_values.len());
 			let mut successful_calls = 0;
@@ -580,7 +585,12 @@ where
 
 impl<L> SubscriptionClientT for Client<L>
 where
-	L: RpcServiceT<Error = RpcServiceError, Response = MethodResponse> + Send + Sync,
+	L: RpcServiceT<
+			MethodResponse = Result<MiddlewareMethodResponse, Error>,
+			BatchResponse = Result<MiddlewareBatchResponse, Error>,
+			NotificationResponse = Result<MiddlewareNotifResponse, Error>,
+		> + Send
+		+ Sync,
 {
 	/// Send a subscription request to the server.
 	///
@@ -617,10 +627,12 @@ where
 			};
 
 			let fut = self.service.call(req);
-			let (sub_id, notifs_rx) =
-				self.run_future_until_timeout(fut).await?.into_subscription().expect("Subscription response");
-
-			Ok(Subscription::new(self.to_back.clone(), notifs_rx, SubscriptionKind::Subscription(sub_id)))
+			let sub = self
+				.run_future_until_timeout(fut)
+				.await?
+				.into_subscription()
+				.expect("Extensions set to subscription, must return subscription; qed");
+			Ok(Subscription::new(self.to_back.clone(), sub.stream, SubscriptionKind::Subscription(sub.sub_id)))
 		}
 	}