fix(tracing): use tracing instrument macro

client/http-client/Cargo.toml

@@ -21,7 +21,6 @@ serde_json = "1.0"
 thiserror = "1.0"
 tokio = { version = "1.16", features = ["time"] }
 tracing = "0.1.34"
-tracing-futures = "0.2.5"
 
 [dev-dependencies]
 jsonrpsee-test-utils = { path = "../../test-utils" }

client/http-client/src/client.rs

@@ -33,13 +33,12 @@ use async_trait::async_trait;
 use hyper::http::HeaderMap;
 use jsonrpsee_core::client::{CertificateStore, ClientT, IdKind, RequestIdManager, Subscription, SubscriptionClientT};
 use jsonrpsee_core::params::BatchRequestBuilder;
-use jsonrpsee_core::tracing::RpcTracing;
 use jsonrpsee_core::traits::ToRpcParams;
 use jsonrpsee_core::{Error, JsonRawValue, TEN_MB_SIZE_BYTES};
 use jsonrpsee_types::error::CallError;
 use rustc_hash::FxHashMap;
 use serde::de::DeserializeOwned;
-use tracing_futures::Instrument;
+use tracing::instrument;
 
 /// Http Client Builder.
 ///
@@ -168,28 +167,26 @@ pub struct HttpClient {
 
 #[async_trait]
 impl ClientT for HttpClient {
+	#[instrument(name = "notification", skip(self, params), level = "trace")]
 	async fn notification<Params>(&self, method: &str, params: Params) -> Result<(), Error>
 	where
 		Params: ToRpcParams + Send,
 	{
-		let trace = RpcTracing::notification(method);
-		async {
-			let params = params.to_rpc_params()?;
-			let notif = serde_json::to_string(&NotificationSer::new(method, params)).map_err(Error::ParseError)?;
+		let params = params.to_rpc_params()?;
+		let notif = serde_json::to_string(&NotificationSer::new(method, params)).map_err(Error::ParseError)?;
 
-			let fut = self.transport.send(notif);
+		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())),
-			}
+		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())),
 		}
-		.instrument(trace.into_span())
-		.await
 	}
 
 	/// Perform a request towards the server.
+
+	#[instrument(name = "method_call", skip(self, params), level = "trace")]
 	async fn request<R, Params>(&self, method: &str, params: Params) -> Result<R, Error>
 	where
 		R: DeserializeOwned,
@@ -200,102 +197,93 @@ impl ClientT for HttpClient {
 		let params = params.to_rpc_params()?;
 
 		let request = RequestSer::new(&id, method, params);
-		let trace = RpcTracing::method_call(method);
-
-		async {
-			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 raw = serde_json::to_string(&request).map_err(Error::ParseError)?;
 
-			// 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: Response<&JsonRawValue> = match serde_json::from_slice(&body) {
-				Ok(response) => response,
-				Err(_) => {
-					let err: ErrorResponse = serde_json::from_slice(&body).map_err(Error::ParseError)?;
-					return Err(Error::Call(CallError::Custom(err.error_object().clone().into_owned())));
-				}
-			};
+		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()));
+			}
+		};
+
+		// 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: Response<&JsonRawValue> = match serde_json::from_slice(&body) {
+			Ok(response) => response,
+			Err(_) => {
+				let err: ErrorResponse = serde_json::from_slice(&body).map_err(Error::ParseError)?;
+				return Err(Error::Call(CallError::Custom(err.error_object().clone().into_owned())));
+			}
+		};
 
-			let result = serde_json::from_str(response.result.get()).map_err(Error::ParseError)?;
+		let result = serde_json::from_str(response.result.get()).map_err(Error::ParseError)?;
 
-			if response.id == id {
-				Ok(result)
-			} else {
-				Err(Error::InvalidRequestId)
-			}
+		if response.id == id {
+			Ok(result)
+		} else {
+			Err(Error::InvalidRequestId)
 		}
-		.instrument(trace.into_span())
-		.await
 	}
 
+	#[instrument(name = "batch", skip(self, batch), level = "trace")]
 	async fn batch_request<'a, R>(&self, batch: BatchRequestBuilder<'a>) -> Result<Vec<R>, Error>
 	where
 		R: DeserializeOwned + Default + Clone,
 	{
 		let batch = batch.build();
 		let guard = self.id_manager.next_request_ids(batch.len())?;
 		let ids: Vec<Id> = guard.inner();
-		let trace = RpcTracing::batch();
-
-		async {
-			let mut batch_request = Vec::with_capacity(batch.len());
-			// NOTE(niklasad1): `ID` is not necessarily monotonically increasing.
-			let mut ordered_requests = Vec::with_capacity(batch.len());
-			let mut request_set = FxHashMap::with_capacity_and_hasher(batch.len(), Default::default());
-
-			for (pos, (method, params)) in batch.into_iter().enumerate() {
-				batch_request.push(RequestSer::new(&ids[pos], method, params));
-				ordered_requests.push(&ids[pos]);
-				request_set.insert(&ids[pos], pos);
-			}
 
-			let fut =
-				self.transport.send_and_read_body(serde_json::to_string(&batch_request).map_err(Error::ParseError)?);
+		let mut batch_request = Vec::with_capacity(batch.len());
+		// NOTE(niklasad1): `ID` is not necessarily monotonically increasing.
+		let mut ordered_requests = Vec::with_capacity(batch.len());
+		let mut request_set = FxHashMap::with_capacity_and_hasher(batch.len(), Default::default());
 
-			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())),
-			};
+		for (pos, (method, params)) in batch.into_iter().enumerate() {
+			batch_request.push(RequestSer::new(&ids[pos], method, params));
+			ordered_requests.push(&ids[pos]);
+			request_set.insert(&ids[pos], pos);
+		}
 
-			// 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 rps: Vec<Response<&JsonRawValue>> =
-				serde_json::from_slice(&body).map_err(|_| match serde_json::from_slice::<ErrorResponse>(&body) {
-					Ok(e) => Error::Call(CallError::Custom(e.error_object().clone().into_owned())),
-					Err(e) => Error::ParseError(e),
-				})?;
-
-			// NOTE: `R::default` is placeholder and will be replaced in loop below.
-			let mut responses = vec![R::default(); ordered_requests.len()];
-			for rp in rps {
-				let pos = match request_set.get(&rp.id) {
-					Some(pos) => *pos,
-					None => return Err(Error::InvalidRequestId),
-				};
-				let result = serde_json::from_str(rp.result.get()).map_err(Error::ParseError)?;
-				responses[pos] = result;
-			}
-			Ok(responses)
+		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())),
+		};
+
+		// 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 rps: Vec<Response<&JsonRawValue>> =
+			serde_json::from_slice(&body).map_err(|_| match serde_json::from_slice::<ErrorResponse>(&body) {
+				Ok(e) => Error::Call(CallError::Custom(e.error_object().clone().into_owned())),
+				Err(e) => Error::ParseError(e),
+			})?;
+
+		// NOTE: `R::default` is placeholder and will be replaced in loop below.
+		let mut responses = vec![R::default(); ordered_requests.len()];
+		for rp in rps {
+			let pos = match request_set.get(&rp.id) {
+				Some(pos) => *pos,
+				None => return Err(Error::InvalidRequestId),
+			};
+			let result = serde_json::from_str(rp.result.get()).map_err(Error::ParseError)?;
+			responses[pos] = result;
 		}
-		.instrument(trace.into_span())
-		.await
+
+		Ok(responses)
 	}
 }
 
 #[async_trait]
 impl SubscriptionClientT for HttpClient {
 	/// Send a subscription request to the server. Not implemented for HTTP; will always return [`Error::HttpNotImplemented`].
+	#[instrument(name = "subscription", fields(method = _subscribe_method), skip(self, _params, _subscribe_method, _unsubscribe_method), level = "trace")]
 	async fn subscribe<'a, N, Params>(
 		&self,
 		_subscribe_method: &'a str,
@@ -310,6 +298,7 @@ impl SubscriptionClientT for HttpClient {
 	}
 
 	/// Subscribe to a specific method. Not implemented for HTTP; will always return [`Error::HttpNotImplemented`].
+	#[instrument(name = "subscribe_method", fields(method = _method), skip(self, _method), level = "trace")]
 	async fn subscribe_to_method<'a, N>(&self, _method: &'a str) -> Result<Subscription<N>, Error>
 	where
 		N: DeserializeOwned,

core/Cargo.toml

@@ -23,7 +23,6 @@ async-channel = { version = "1.6", optional = true }
 async-lock = { version = "2.4", optional = true }
 futures-util = { version = "0.3.14", default-features = false, optional = true }
 hyper = { version = "0.14.10", default-features = false, features = ["stream"], optional = true }
-tracing-futures = { version = "0.2", optional = true }
 rustc-hash = { version = "1", optional = true }
 rand = { version = "0.8", optional = true }
 soketto = { version = "0.7.1", optional = true }
@@ -55,15 +54,13 @@ async-client = [
 	"tokio/macros",
 	"tokio/rt",
 	"tokio/sync",
-	"tracing-futures",
 	"futures-timer",
 ]
 async-wasm-client = [
 	"async-lock",
 	"client",
 	"wasm-bindgen-futures",
 	"rustc-hash/std",
-	"tracing-futures",
 	"futures-timer/wasm-bindgen",
 ]