refactor(client): use native async fn in traits instead async_trait crate (#1551)

* refactor(client): ClientT/SubscribeClienT use native async fns

* fix ui tests

* refactor(client): use async native traits in transports

* macros(client): use native async fn in traits

* fix ui tests

Author: niklasad1

client/http-client/Cargo.toml

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

client/http-client/src/client.rs

@@ -32,7 +32,6 @@ use std::time::Duration;
 use crate::rpc_service::RpcService;
 use crate::transport::{self, Error as TransportError, HttpBackend, HttpTransportClientBuilder};
 use crate::{HttpRequest, HttpResponse};
-use async_trait::async_trait;
 use hyper::body::Bytes;
 use hyper::http::{Extensions, HeaderMap};
 use jsonrpsee_core::client::{
@@ -350,150 +349,157 @@ impl HttpClient<HttpBackend> {
 	}
 }
 
-#[async_trait]
 impl<S> ClientT for HttpClient<S>
 where
 	S: RpcServiceT<Error = Error, Response = MethodResponse> + Send + Sync,
 {
-	async fn notification<Params>(&self, method: &str, params: Params) -> Result<(), Error>
+	fn notification<Params>(&self, method: &str, params: Params) -> impl Future<Output = Result<(), Error>> + Send
 	where
 		Params: ToRpcParams + Send,
 	{
-		let _permit = match self.request_guard.as_ref() {
-			Some(permit) => permit.acquire().await.ok(),
-			None => None,
-		};
-		let params = params.to_rpc_params()?.map(StdCow::Owned);
-
-		run_future_until_timeout(
-			self.service.notification(Notification::new(method.into(), params)),
-			self.request_timeout,
-		)
-		.await
-		.map_err(|e| Error::Transport(e.into()))?;
-		Ok(())
+		async {
+			let _permit = match self.request_guard.as_ref() {
+				Some(permit) => permit.acquire().await.ok(),
+				None => None,
+			};
+			let params = params.to_rpc_params()?.map(StdCow::Owned);
+
+			run_future_until_timeout(
+				self.service.notification(Notification::new(method.into(), params)),
+				self.request_timeout,
+			)
+			.await
+			.map_err(|e| Error::Transport(e.into()))?;
+			Ok(())
+		}
 	}
 
-	async fn request<R, Params>(&self, method: &str, params: Params) -> Result<R, Error>
+	fn request<R, Params>(&self, method: &str, params: Params) -> impl Future<Output = Result<R, Error>> + Send
 	where
 		R: DeserializeOwned,
 		Params: ToRpcParams + Send,
 	{
-		let _permit = match self.request_guard.as_ref() {
-			Some(permit) => permit.acquire().await.ok(),
-			None => None,
-		};
-		let id = self.id_manager.next_request_id();
-		let params = params.to_rpc_params()?;
-
-		let method_response = run_future_until_timeout(
-			self.service.call(Request::borrowed(method, params.as_deref(), id.clone())),
-			self.request_timeout,
-		)
-		.await?
-		.into_method_call()
-		.expect("Method call must return a method call reponse; qed");
-
-		let rp = ResponseSuccess::try_from(method_response.into_inner())?;
-
-		let result = serde_json::from_str(rp.result.get()).map_err(Error::ParseError)?;
-		if rp.id == id { Ok(result) } else { Err(InvalidRequestId::NotPendingRequest(rp.id.to_string()).into()) }
+		async {
+			let _permit = match self.request_guard.as_ref() {
+				Some(permit) => permit.acquire().await.ok(),
+				None => None,
+			};
+			let id = self.id_manager.next_request_id();
+			let params = params.to_rpc_params()?;
+
+			let method_response = run_future_until_timeout(
+				self.service.call(Request::borrowed(method, params.as_deref(), id.clone())),
+				self.request_timeout,
+			)
+			.await?
+			.into_method_call()
+			.expect("Method call must return a method call reponse; qed");
+
+			let rp = ResponseSuccess::try_from(method_response.into_inner())?;
+
+			let result = serde_json::from_str(rp.result.get()).map_err(Error::ParseError)?;
+			if rp.id == id { Ok(result) } else { Err(InvalidRequestId::NotPendingRequest(rp.id.to_string()).into()) }
+		}
 	}
 
-	async fn batch_request<'a, R>(&self, batch: BatchRequestBuilder<'a>) -> Result<BatchResponse<'a, R>, Error>
+	fn batch_request<'a, R>(
+		&self,
+		batch: BatchRequestBuilder<'a>,
+	) -> impl Future<Output = Result<BatchResponse<'a, R>, Error>> + Send
 	where
 		R: DeserializeOwned + fmt::Debug + 'a,
 	{
-		let _permit = match self.request_guard.as_ref() {
-			Some(permit) => permit.acquire().await.ok(),
-			None => None,
-		};
-		let batch = batch.build()?;
-		let id = self.id_manager.next_request_id();
-		let id_range = generate_batch_id_range(id, batch.len() as u64)?;
-
-		let mut batch_request = Batch::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);
-			let req = Request {
-				jsonrpc: TwoPointZero,
-				method: method.into(),
-				params: params.map(StdCow::Owned),
-				id,
-				extensions: Extensions::new(),
+		async {
+			let _permit = match self.request_guard.as_ref() {
+				Some(permit) => permit.acquire().await.ok(),
+				None => None,
 			};
-			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 batch = batch.build()?;
+			let id = self.id_manager.next_request_id();
+			let id_range = generate_batch_id_range(id, batch.len() as u64)?;
+
+			let mut batch_request = Batch::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);
+				let req = Request {
+					jsonrpc: TwoPointZero,
+					method: method.into(),
+					params: params.map(StdCow::Owned),
+					id,
+					extensions: Extensions::new(),
+				};
+				batch_request.push(req);
+			}
 
-		let mut batch_response = Vec::new();
-		let mut success = 0;
-		let mut failed = 0;
+			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");
 
-		// Fill the batch response with placeholder values.
-		for _ in 0..json_rps.len() {
-			batch_response.push(Err(ErrorObject::borrowed(0, "", None)));
-		}
+			let mut batch_response = Vec::new();
+			let mut success = 0;
+			let mut failed = 0;
 
-		for rp in json_rps.into_iter() {
-			let id = rp.id().try_parse_inner_as_number()?;
+			// Fill the batch response with placeholder values.
+			for _ in 0..json_rps.len() {
+				batch_response.push(Err(ErrorObject::borrowed(0, "", None)));
+			}
 
-			let res = match ResponseSuccess::try_from(rp.into_inner()) {
-				Ok(r) => {
-					let v = serde_json::from_str(r.result.get()).map_err(Error::ParseError)?;
-					success += 1;
-					Ok(v)
-				}
-				Err(err) => {
-					failed += 1;
-					Err(err)
+			for rp in json_rps.into_iter() {
+				let id = rp.id().try_parse_inner_as_number()?;
+
+				let res = match ResponseSuccess::try_from(rp.into_inner()) {
+					Ok(r) => {
+						let v = serde_json::from_str(r.result.get()).map_err(Error::ParseError)?;
+						success += 1;
+						Ok(v)
+					}
+					Err(err) => {
+						failed += 1;
+						Err(err)
+					}
+				};
+
+				let maybe_elem = id
+					.checked_sub(id_range.start)
+					.and_then(|p| p.try_into().ok())
+					.and_then(|p: usize| batch_response.get_mut(p));
+
+				if let Some(elem) = maybe_elem {
+					*elem = res;
+				} else {
+					return Err(InvalidRequestId::NotPendingRequest(id.to_string()).into());
 				}
-			};
-
-			let maybe_elem = id
-				.checked_sub(id_range.start)
-				.and_then(|p| p.try_into().ok())
-				.and_then(|p: usize| batch_response.get_mut(p));
-
-			if let Some(elem) = maybe_elem {
-				*elem = res;
-			} else {
-				return Err(InvalidRequestId::NotPendingRequest(id.to_string()).into());
 			}
-		}
 
-		Ok(BatchResponse::new(success, batch_response, failed))
+			Ok(BatchResponse::new(success, batch_response, failed))
+		}
 	}
 }
 
-#[async_trait]
 impl<S> SubscriptionClientT for HttpClient<S>
 where
 	S: RpcServiceT<Error = Error, Response = MethodResponse> + Send + Sync,
 {
 	/// Send a subscription request to the server. Not implemented for HTTP; will always return
 	/// [`Error::HttpNotImplemented`].
-	async fn subscribe<'a, N, Params>(
+	fn subscribe<'a, N, Params>(
 		&self,
 		_subscribe_method: &'a str,
 		_params: Params,
 		_unsubscribe_method: &'a str,
-	) -> Result<Subscription<N>, Error>
+	) -> impl Future<Output = Result<Subscription<N>, Error>>
 	where
 		Params: ToRpcParams + Send,
 		N: DeserializeOwned,
 	{
-		Err(Error::HttpNotImplemented)
+		async { Err(Error::HttpNotImplemented) }
 	}
 
 	/// Subscribe to a specific method. Not implemented for HTTP; will always return [`Error::HttpNotImplemented`].
-	async fn subscribe_to_method<'a, N>(&self, _method: &'a str) -> Result<Subscription<N>, Error>
+	fn subscribe_to_method<N>(&self, _method: &str) -> impl Future<Output = Result<Subscription<N>, Error>>
 	where
 		N: DeserializeOwned,
 	{
-		Err(Error::HttpNotImplemented)
+		async { Err(Error::HttpNotImplemented) }
 	}
 }
 

client/transport/src/web.rs

@@ -4,7 +4,6 @@ use futures_channel::mpsc;
 use futures_util::sink::SinkExt;
 use futures_util::stream::{SplitSink, SplitStream, StreamExt};
 use gloo_net::websocket::{Message, WebSocketError, futures::WebSocket};
-use jsonrpsee_core::async_trait;
 use jsonrpsee_core::client::{ReceivedMessage, TransportReceiverT, TransportSenderT};
 
 /// Web-sys transport error that can occur.
@@ -45,28 +44,30 @@ impl fmt::Debug for Receiver {
 	}
 }
 
-#[async_trait(?Send)]
 impl TransportSenderT for Sender {
 	type Error = Error;
 
-	async fn send(&mut self, msg: String) -> Result<(), Self::Error> {
-		self.0.send(Message::Text(msg)).await.map_err(|e| Error::WebSocket(e))?;
-		Ok(())
+	fn send(&mut self, msg: String) -> impl Future<Output = Result<(), Self::Error>> {
+		async {
+			self.0.send(Message::Text(msg)).await.map_err(|e| Error::WebSocket(e))?;
+			Ok(())
+		}
 	}
 }
 
-#[async_trait(?Send)]
 impl TransportReceiverT for Receiver {
 	type Error = Error;
 
-	async fn receive(&mut self) -> Result<ReceivedMessage, Self::Error> {
-		match self.0.next().await {
-			Some(Ok(msg)) => match msg {
-				Message::Bytes(bytes) => Ok(ReceivedMessage::Bytes(bytes)),
-				Message::Text(txt) => Ok(ReceivedMessage::Text(txt)),
-			},
-			Some(Err(err)) => Err(Error::WebSocket(err)),
-			None => Err(Error::SenderDisconnected),
+	fn receive(&mut self) -> impl Future<Output = Result<ReceivedMessage, Self::Error>> {
+		async {
+			match self.0.next().await {
+				Some(Ok(msg)) => match msg {
+					Message::Bytes(bytes) => Ok(ReceivedMessage::Bytes(bytes)),
+					Message::Text(txt) => Ok(ReceivedMessage::Text(txt)),
+				},
+				Some(Err(err)) => Err(Error::WebSocket(err)),
+				None => Err(Error::SenderDisconnected),
+			}
 		}
 	}
 }