feat: client rpc middleware

client/http-client/Cargo.toml

@@ -19,6 +19,7 @@ workspace = true
 [dependencies]
 async-trait = { workspace = true }
 base64 = { workspace = true }
+futures-util = { workspace = true }
 hyper = { workspace = true, features = ["client", "http1", "http2"] }
 hyper-rustls = { workspace = true, features = ["http1", "http2", "tls12", "logging", "ring"], optional = true }
 hyper-util = { workspace = true, features = ["client", "client-legacy", "tokio", "http1", "http2"] }

client/http-client/src/client.rs

@@ -29,19 +29,21 @@
 use std::sync::Arc;
 use std::time::Duration;
 
-use crate::transport::{self, Error as TransportError, HttpBackend, HttpTransportClient, HttpTransportClientBuilder};
-use crate::types::{NotificationSer, RequestSer, Response};
+use crate::rpc_service::RpcService;
+use crate::transport::{self, Error as TransportError, HttpBackend, HttpTransportClientBuilder};
+use crate::types::Response;
 use crate::{HttpRequest, HttpResponse};
 use async_trait::async_trait;
 use hyper::body::Bytes;
 use hyper::http::HeaderMap;
 use jsonrpsee_core::client::{
 	BatchResponse, ClientT, Error, IdKind, RequestIdManager, Subscription, SubscriptionClientT, generate_batch_id_range,
 };
+use jsonrpsee_core::middleware::{RpcServiceBuilder, RpcServiceT};
 use jsonrpsee_core::params::BatchRequestBuilder;
 use jsonrpsee_core::traits::ToRpcParams;
 use jsonrpsee_core::{BoxError, JsonRawValue, TEN_MB_SIZE_BYTES};
-use jsonrpsee_types::{ErrorObject, InvalidRequestId, ResponseSuccess, TwoPointZero};
+use jsonrpsee_types::{ErrorObject, InvalidRequestId, Notification, Request, ResponseSuccess, TwoPointZero};
 use serde::de::DeserializeOwned;
 use tokio::sync::Semaphore;
 use tower::layer::util::Identity;
@@ -75,7 +77,7 @@
 /// }
 /// ```
 #[derive(Clone, Debug)]
-pub struct HttpClientBuilder<L = Identity> {
+pub struct HttpClientBuilder<HttpMiddleware = Identity, RpcMiddleware = Identity> {
 	max_request_size: u32,
 	max_response_size: u32,
 	request_timeout: Duration,
@@ -84,12 +86,13 @@
 	id_kind: IdKind,
 	max_log_length: u32,
 	headers: HeaderMap,
-	service_builder: tower::ServiceBuilder<L>,
+	service_builder: tower::ServiceBuilder<HttpMiddleware>,
+	rpc_middleware: RpcServiceBuilder<RpcMiddleware>,
 	tcp_no_delay: bool,
 	max_concurrent_requests: Option<usize>,
 }
 
-impl<L> HttpClientBuilder<L> {
+impl<HttpMiddleware, RpcMiddleware> HttpClientBuilder<HttpMiddleware, RpcMiddleware> {
 	/// Set the maximum size of a request body in bytes. Default is 10 MiB.
 	pub fn max_request_size(mut self, size: u32) -> Self {
 		self.max_request_size = size;
@@ -215,8 +218,29 @@
 		self
 	}
 
+	/// Set the RPC middleware.
+	pub fn set_rpc_middleware<T>(self, rpc_builder: RpcServiceBuilder<T>) -> HttpClientBuilder<HttpMiddleware, T> {
+		HttpClientBuilder {
+			#[cfg(feature = "tls")]
+			certificate_store: self.certificate_store,
+			id_kind: self.id_kind,
+			headers: self.headers,
+			max_log_length: self.max_log_length,
+			max_request_size: self.max_request_size,
+			max_response_size: self.max_response_size,
+			service_builder: self.service_builder,
+			rpc_middleware: rpc_builder,
+			request_timeout: self.request_timeout,
+			tcp_no_delay: self.tcp_no_delay,
+			max_concurrent_requests: self.max_concurrent_requests,
+		}
+	}
+
 	/// Set custom tower middleware.
-	pub fn set_http_middleware<T>(self, service_builder: tower::ServiceBuilder<T>) -> HttpClientBuilder<T> {
+	pub fn set_http_middleware<T>(
+		self,
+		service_builder: tower::ServiceBuilder<T>,
+	) -> HttpClientBuilder<T, RpcMiddleware> {
 		HttpClientBuilder {
 			#[cfg(feature = "tls")]
 			certificate_store: self.certificate_store,
@@ -226,23 +250,26 @@
 			max_request_size: self.max_request_size,
 			max_response_size: self.max_response_size,
 			service_builder,
+			rpc_middleware: self.rpc_middleware,
 			request_timeout: self.request_timeout,
 			tcp_no_delay: self.tcp_no_delay,
 			max_concurrent_requests: self.max_concurrent_requests,
 		}
 	}
 }
 
-impl<B, S, L> HttpClientBuilder<L>
+impl<B, S, S2, HttpMiddleware, RpcMiddleware> HttpClientBuilder<HttpMiddleware, RpcMiddleware>
 where
-	L: Layer<transport::HttpBackend, Service = S>,
+	RpcMiddleware: Layer<RpcService<S>, Service = S2>,
+	for<'a> <RpcMiddleware as Layer<RpcService<S>>>::Service: RpcServiceT<'a>,
+	HttpMiddleware: Layer<transport::HttpBackend, Service = S>,
 	S: Service<HttpRequest, Response = HttpResponse<B>, Error = TransportError> + Clone,
 	B: http_body::Body<Data = Bytes> + Send + Unpin + 'static,
 	B::Data: Send,
 	B::Error: Into<BoxError>,
 {
 	/// Build the HTTP client with target to connect to.
-	pub fn build(self, target: impl AsRef<str>) -> Result<HttpClient<S>, Error> {
+	pub fn build(self, target: impl AsRef<str>) -> Result<HttpClient<S2>, Error> {
 		let Self {
 			max_request_size,
 			max_response_size,
@@ -254,10 +281,11 @@
 			max_log_length,
 			service_builder,
 			tcp_no_delay,
+			rpc_middleware,
 			..
 		} = self;
 
-		let transport = HttpTransportClientBuilder {
+		let http = HttpTransportClientBuilder {
 			max_request_size,
 			max_response_size,
 			headers,
@@ -266,6 +294,7 @@
 			service_builder,
 			#[cfg(feature = "tls")]
 			certificate_store,
+			request_timeout,
 		}
 		.build(target)
 		.map_err(|e| Error::Transport(e.into()))?;
@@ -275,9 +304,8 @@
 			.map(|max_concurrent_requests| Arc::new(Semaphore::new(max_concurrent_requests)));
 
 		Ok(HttpClient {
-			transport,
+			transport: rpc_middleware.service(RpcService::new(http, max_response_size)),
 			id_manager: Arc::new(RequestIdManager::new(id_kind)),
-			request_timeout,
 			request_guard,
 		})
 	}
@@ -295,6 +323,7 @@
 			max_log_length: 4096,
 			headers: HeaderMap::new(),
 			service_builder: tower::ServiceBuilder::new(),
+			rpc_middleware: RpcServiceBuilder::default(),
 			tcp_no_delay: true,
 			max_concurrent_requests: None,
 		}
@@ -310,11 +339,9 @@
 
 /// JSON-RPC HTTP Client that provides functionality to perform method calls and notifications.
 #[derive(Debug, Clone)]
-pub struct HttpClient<S = HttpBackend> {
+pub struct HttpClient<S> {
 	/// HTTP transport client.
-	transport: HttpTransportClient<S>,
-	/// Request timeout. Defaults to 60sec.
-	request_timeout: Duration,
+	transport: S,
 	/// Request ID manager.
 	id_manager: Arc<RequestIdManager>,
 	/// Concurrent requests limit guard.
@@ -329,18 +356,14 @@
 
 	/// Returns configured request timeout.
 	pub fn request_timeout(&self) -> Duration {
-		self.request_timeout
+		todo!();
 	}
 }
 
 #[async_trait]
-impl<B, S> ClientT for HttpClient<S>
+impl<S> ClientT for HttpClient<S>
 where
-	S: Service<HttpRequest, Response = HttpResponse<B>, Error = TransportError> + Send + Sync + Clone,
-	<S as Service<HttpRequest>>::Future: Send,
-	B: http_body::Body<Data = Bytes> + Send + Unpin + 'static,
-	B::Error: Into<BoxError>,
-	B::Data: Send,
+	for<'a> S: RpcServiceT<'a, Error = Error> + Send + Sync,
 {
 	#[instrument(name = "notification", skip(self, params), level = "trace")]
 	async fn notification<Params>(&self, method: &str, params: Params) -> Result<(), Error>
@@ -351,17 +374,10 @@
 			Some(permit) => permit.acquire().await.ok(),
 			None => None,
 		};
-		let params = params.to_rpc_params()?;
-		let notif =
-			serde_json::to_string(&NotificationSer::borrowed(&method, params.as_deref())).map_err(Error::ParseError)?;
-
-		let fut = self.transport.send(notif);
-
-		match tokio::time::timeout(self.request_timeout, fut).await {
-			Ok(Ok(ok)) => Ok(ok),
-			Err(_) => Err(Error::RequestTimeout),
-			Ok(Err(e)) => Err(Error::Transport(e.into())),
-		}
+		let params = params.to_rpc_params()?.map(StdCow::Owned);
+		let n = Notification { jsonrpc: TwoPointZero, method: method.into(), params };
+		self.transport.notification(n).await?;
+		Ok(())
 	}
 
 	#[instrument(name = "method_call", skip(self, params), level = "trace")]
@@ -377,23 +393,13 @@
 		let id = self.id_manager.next_request_id();
 		let params = params.to_rpc_params()?;
 
-		let request = RequestSer::borrowed(&id, &method, params.as_deref());
-		let raw = serde_json::to_string(&request).map_err(Error::ParseError)?;
-
-		let fut = self.transport.send_and_read_body(raw);
-		let body = match tokio::time::timeout(self.request_timeout, fut).await {
-			Ok(Ok(body)) => body,
-			Err(_e) => {
-				return Err(Error::RequestTimeout);
-			}
-			Ok(Err(e)) => {
-				return Err(Error::Transport(e.into()));
-			}
-		};
+		let request = Request::new(method.into(), params.as_deref(), id.clone());
+		let rp = self.transport.call(request).await?;
 
 		// NOTE: it's decoded first to `JsonRawValue` and then to `R` below to get
 		// a better error message if `R` couldn't be decoded.
-		let response = ResponseSuccess::try_from(serde_json::from_slice::<Response<&JsonRawValue>>(&body)?)?;
+		// TODO: this is inefficient, we should avoid double parsing.
+		let response = ResponseSuccess::try_from(serde_json::from_str::<Response<&JsonRawValue>>(&rp.as_result())?)?;
 
 		let result = serde_json::from_str(response.result.get()).map_err(Error::ParseError)?;
 
@@ -420,51 +426,47 @@
 		let mut batch_request = Vec::with_capacity(batch.len());
 		for ((method, params), id) in batch.into_iter().zip(id_range.clone()) {
 			let id = self.id_manager.as_id_kind().into_id(id);
-			batch_request.push(RequestSer {
+			batch_request.push(Request {
 				jsonrpc: TwoPointZero,
-				id,
 				method: method.into(),
 				params: params.map(StdCow::Owned),
+				id,
+				extensions: Default::default(),
 			});
 		}
 
-		let fut = self.transport.send_and_read_body(serde_json::to_string(&batch_request).map_err(Error::ParseError)?);
-
-		let body = match tokio::time::timeout(self.request_timeout, fut).await {
-			Ok(Ok(body)) => body,
-			Err(_e) => return Err(Error::RequestTimeout),
-			Ok(Err(e)) => return Err(Error::Transport(e.into())),
-		};
+		let batch = self.transport.batch(batch_request).await?;
+		// TODO: this is inefficient, we should avoid double parsing.
+		let json_responses: Vec<Response<&JsonRawValue>> = serde_json::from_str(&batch.as_result())?;
 
-		let json_rps: Vec<Response<&JsonRawValue>> = serde_json::from_slice(&body).map_err(Error::ParseError)?;
+		let mut batch_response = Vec::new();
+		let mut success = 0;
+		let mut failed = 0;
 
-		let mut responses = Vec::with_capacity(json_rps.len());
-		let mut successful_calls = 0;
-		let mut failed_calls = 0;
-
-		for _ in 0..json_rps.len() {
-			responses.push(Err(ErrorObject::borrowed(0, "", None)));
+		// Fill the batch response with placeholder values.
+		for _ in 0..json_responses.len() {
+			batch_response.push(Err(ErrorObject::borrowed(0, "", None)));
 		}
 
-		for rp in json_rps {
+		for rp in json_responses.into_iter() {
 			let id = rp.id.try_parse_inner_as_number()?;
 
 			let res = match ResponseSuccess::try_from(rp) {
 				Ok(r) => {
-					let result = serde_json::from_str(r.result.get())?;
-					successful_calls += 1;
-					Ok(result)
+					let v = serde_json::from_str(r.result.get()).map_err(Error::ParseError)?;
+					success += 1;
+					Ok(v)
 				}
 				Err(err) => {
-					failed_calls += 1;
+					failed += 1;
 					Err(err)
 				}
 			};
 
 			let maybe_elem = id
 				.checked_sub(id_range.start)
 				.and_then(|p| p.try_into().ok())
-				.and_then(|p: usize| responses.get_mut(p));
+				.and_then(|p: usize| batch_response.get_mut(p));
 
 			if let Some(elem) = maybe_elem {
 				*elem = res;
@@ -473,18 +475,14 @@
 			}
 		}
 
-		Ok(BatchResponse::new(successful_calls, responses, failed_calls))
+		Ok(BatchResponse::new(success, batch_response, failed))
 	}
 }
 
 #[async_trait]
-impl<B, S> SubscriptionClientT for HttpClient<S>
+impl<S> SubscriptionClientT for HttpClient<S>
 where
-	S: Service<HttpRequest, Response = HttpResponse<B>, Error = TransportError> + Send + Sync + Clone,
-	<S as Service<HttpRequest>>::Future: Send,
-	B: http_body::Body<Data = Bytes> + Send + Unpin + 'static,
-	B::Data: Send,
-	B::Error: Into<BoxError>,
+	for<'a> S: RpcServiceT<'a, Error = Error> + Send + Sync,
 {
 	/// Send a subscription request to the server. Not implemented for HTTP; will always return
 	/// [`Error::HttpNotImplemented`].

client/http-client/src/lib.rs

@@ -36,6 +36,7 @@
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
 mod client;
+mod rpc_service;
 
 /// HTTP transport.
 pub mod transport;