Use jsonrpsee subscriptions (#1533)

* splitted Substrate RPC trait

* introduce subscription methods

* removed commented code

* removed commented code

bridges/relays/client-substrate/src/client.rs

@@ -18,7 +18,11 @@
 
 use crate::{
 	chain::{Chain, ChainWithBalances, TransactionStatusOf},
-	rpc::SubstrateClient,
+	rpc::{
+		SubstrateAuthorClient, SubstrateChainClient, SubstrateFrameSystemClient,
+		SubstrateGrandpaClient, SubstrateStateClient, SubstrateSystemClient,
+		SubstrateTransactionPaymentClient,
+	},
 	ConnectionParams, Error, HashOf, HeaderIdOf, Result,
 };
 
@@ -29,8 +33,7 @@ use codec::{Decode, Encode};
 use frame_system::AccountInfo;
 use futures::{SinkExt, StreamExt};
 use jsonrpsee::{
-	core::{client::SubscriptionClientT, DeserializeOwned},
+	core::DeserializeOwned,
-	types::params::ParamsSer,
 	ws_client::{WsClient as RpcClient, WsClientBuilder as RpcClientBuilder},
 };
 use num_traits::{Bounded, Zero};
@@ -153,8 +156,7 @@ impl<C: Chain> Client<C> {
 		let genesis_hash_client = client.clone();
 		let genesis_hash = tokio
 			.spawn(async move {
-				SubstrateClient::<C>::chain_get_block_hash(&*genesis_hash_client, Some(number))
+				SubstrateChainClient::<C>::block_hash(&*genesis_hash_client, Some(number)).await
-					.await
 			})
 			.await??;
 
@@ -212,7 +214,7 @@ impl<C: Chain> Client<C> {
 	/// Returns true if client is connected to at least one peer and is in synced state.
 	pub async fn ensure_synced(&self) -> Result<()> {
 		self.jsonrpsee_execute(|client| async move {
-			let health = SubstrateClient::<C>::system_health(&*client).await?;
+			let health = SubstrateSystemClient::<C>::health(&*client).await?;
 			let is_synced = !health.is_syncing && (!health.should_have_peers || health.peers > 0);
 			if is_synced {
 				Ok(())
@@ -231,7 +233,7 @@ impl<C: Chain> Client<C> {
 	/// Return hash of the best finalized block.
 	pub async fn best_finalized_header_hash(&self) -> Result<C::Hash> {
 		self.jsonrpsee_execute(|client| async move {
-			Ok(SubstrateClient::<C>::chain_get_finalized_head(&*client).await?)
+			Ok(SubstrateChainClient::<C>::finalized_head(&*client).await?)
 		})
 		.await
 	}
@@ -252,7 +254,7 @@ impl<C: Chain> Client<C> {
 		C::Header: DeserializeOwned,
 	{
 		self.jsonrpsee_execute(|client| async move {
-			Ok(SubstrateClient::<C>::chain_get_header(&*client, None).await?)
+			Ok(SubstrateChainClient::<C>::header(&*client, None).await?)
 		})
 		.await
 	}
@@ -260,7 +262,7 @@ impl<C: Chain> Client<C> {
 	/// Get a Substrate block from its hash.
 	pub async fn get_block(&self, block_hash: Option<C::Hash>) -> Result<C::SignedBlock> {
 		self.jsonrpsee_execute(move |client| async move {
-			Ok(SubstrateClient::<C>::chain_get_block(&*client, block_hash).await?)
+			Ok(SubstrateChainClient::<C>::block(&*client, block_hash).await?)
 		})
 		.await
 	}
@@ -271,7 +273,7 @@ impl<C: Chain> Client<C> {
 		C::Header: DeserializeOwned,
 	{
 		self.jsonrpsee_execute(move |client| async move {
-			Ok(SubstrateClient::<C>::chain_get_header(&*client, Some(block_hash)).await?)
+			Ok(SubstrateChainClient::<C>::header(&*client, Some(block_hash)).await?)
 		})
 		.await
 	}
@@ -279,7 +281,7 @@ impl<C: Chain> Client<C> {
 	/// Get a Substrate block hash by its number.
 	pub async fn block_hash_by_number(&self, number: C::BlockNumber) -> Result<C::Hash> {
 		self.jsonrpsee_execute(move |client| async move {
-			Ok(SubstrateClient::<C>::chain_get_block_hash(&*client, Some(number)).await?)
+			Ok(SubstrateChainClient::<C>::block_hash(&*client, Some(number)).await?)
 		})
 		.await
 	}
@@ -297,7 +299,7 @@ impl<C: Chain> Client<C> {
 	/// Return runtime version.
 	pub async fn runtime_version(&self) -> Result<RuntimeVersion> {
 		self.jsonrpsee_execute(move |client| async move {
-			Ok(SubstrateClient::<C>::state_runtime_version(&*client).await?)
+			Ok(SubstrateStateClient::<C>::runtime_version(&*client).await?)
 		})
 		.await
 	}
@@ -341,7 +343,7 @@ impl<C: Chain> Client<C> {
 		block_hash: Option<C::Hash>,
 	) -> Result<Option<StorageData>> {
 		self.jsonrpsee_execute(move |client| async move {
-			Ok(SubstrateClient::<C>::state_get_storage(&*client, storage_key, block_hash).await?)
+			Ok(SubstrateStateClient::<C>::storage(&*client, storage_key, block_hash).await?)
 		})
 		.await
 	}
@@ -354,7 +356,7 @@ impl<C: Chain> Client<C> {
 		self.jsonrpsee_execute(move |client| async move {
 			let storage_key = C::account_info_storage_key(&account);
 			let encoded_account_data =
-				SubstrateClient::<C>::state_get_storage(&*client, storage_key, None)
+				SubstrateStateClient::<C>::storage(&*client, storage_key, None)
 					.await?
 					.ok_or(Error::AccountDoesNotExist)?;
 			let decoded_account_data = AccountInfo::<C::Index, AccountData<C::Balance>>::decode(
@@ -371,7 +373,7 @@ impl<C: Chain> Client<C> {
 	/// Note: It's the caller's responsibility to make sure `account` is a valid SS58 address.
 	pub async fn next_account_index(&self, account: C::AccountId) -> Result<C::Index> {
 		self.jsonrpsee_execute(move |client| async move {
-			Ok(SubstrateClient::<C>::system_account_next_index(&*client, account).await?)
+			Ok(SubstrateFrameSystemClient::<C>::account_next_index(&*client, account).await?)
 		})
 		.await
 	}
@@ -381,7 +383,7 @@ impl<C: Chain> Client<C> {
 	/// Note: The given transaction needs to be SCALE encoded beforehand.
 	pub async fn submit_unsigned_extrinsic(&self, transaction: Bytes) -> Result<C::Hash> {
 		self.jsonrpsee_execute(move |client| async move {
-			let tx_hash = SubstrateClient::<C>::author_submit_extrinsic(&*client, transaction)
+			let tx_hash = SubstrateAuthorClient::<C>::submit_extrinsic(&*client, transaction)
 				.await
 				.map_err(|e| {
 					log::error!(target: "bridge", "Failed to send transaction to {} node: {:?}", C::NAME, e);
@@ -418,12 +420,12 @@ impl<C: Chain> Client<C> {
 
 		self.jsonrpsee_execute(move |client| async move {
 			let extrinsic = prepare_extrinsic(best_header_id, transaction_nonce)?;
-			let tx_hash = SubstrateClient::<C>::author_submit_extrinsic(&*client, extrinsic)
+			let tx_hash = SubstrateAuthorClient::<C>::submit_extrinsic(&*client, extrinsic)
 				.await
 				.map_err(|e| {
-					log::error!(target: "bridge", "Failed to send transaction to {} node: {:?}", C::NAME, e);
+				log::error!(target: "bridge", "Failed to send transaction to {} node: {:?}", C::NAME, e);
-					e
+				e
-				})?;
+			})?;
 			log::trace!(target: "bridge", "Sent transaction to {} node: {:?}", C::NAME, tx_hash);
 			Ok(tx_hash)
 		})
@@ -445,18 +447,13 @@ impl<C: Chain> Client<C> {
 			.jsonrpsee_execute(move |client| async move {
 				let extrinsic = prepare_extrinsic(best_header_id, transaction_nonce)?;
 				let tx_hash = C::Hasher::hash(&extrinsic.0);
-				let subscription = client
+				let subscription =
-					.subscribe(
+					SubstrateAuthorClient::<C>::submit_and_watch_extrinsic(&*client, extrinsic)
-						"author_submitAndWatchExtrinsic",
+						.await
-						Some(ParamsSer::Array(vec![jsonrpsee::core::to_json_value(extrinsic)
+						.map_err(|e| {
-							.map_err(|e| Error::RpcError(e.into()))?])),
+							log::error!(target: "bridge", "Failed to send transaction to {} node: {:?}", C::NAME, e);
-						"author_unwatchExtrinsic",
+							e
-					)
+						})?;
-					.await
-					.map_err(|e| {
-						log::error!(target: "bridge", "Failed to send transaction to {} node: {:?}", C::NAME, e);
-						e
-					})?;
 				log::trace!(target: "bridge", "Sent transaction to {} node: {:?}", C::NAME, tx_hash);
 				Ok(subscription)
 			})
@@ -474,7 +471,7 @@ impl<C: Chain> Client<C> {
 	/// Returns pending extrinsics from transaction pool.
 	pub async fn pending_extrinsics(&self) -> Result<Vec<Bytes>> {
 		self.jsonrpsee_execute(move |client| async move {
-			Ok(SubstrateClient::<C>::author_pending_extrinsics(&*client).await?)
+			Ok(SubstrateAuthorClient::<C>::pending_extrinsics(&*client).await?)
 		})
 		.await
 	}
@@ -490,7 +487,7 @@ impl<C: Chain> Client<C> {
 			let data = Bytes((TransactionSource::External, transaction, at_block).encode());
 
 			let encoded_response =
-				SubstrateClient::<C>::state_call(&*client, call, data, Some(at_block)).await?;
+				SubstrateStateClient::<C>::call(&*client, call, data, Some(at_block)).await?;
 			let validity = TransactionValidity::decode(&mut &encoded_response.0[..])
 				.map_err(Error::ResponseParseFailed)?;
 
@@ -506,7 +503,7 @@ impl<C: Chain> Client<C> {
 	) -> Result<InclusionFee<C::Balance>> {
 		self.jsonrpsee_execute(move |client| async move {
 			let fee_details =
-				SubstrateClient::<C>::payment_query_fee_details(&*client, transaction, None)
+				SubstrateTransactionPaymentClient::<C>::fee_details(&*client, transaction, None)
 					.await?;
 			let inclusion_fee = fee_details
 				.inclusion_fee
@@ -540,7 +537,7 @@ impl<C: Chain> Client<C> {
 			let data = Bytes(Vec::new());
 
 			let encoded_response =
-				SubstrateClient::<C>::state_call(&*client, call, data, Some(block)).await?;
+				SubstrateStateClient::<C>::call(&*client, call, data, Some(block)).await?;
 			let authority_list = encoded_response.0;
 
 			Ok(authority_list)
@@ -556,7 +553,7 @@ impl<C: Chain> Client<C> {
 		at_block: Option<C::Hash>,
 	) -> Result<Bytes> {
 		self.jsonrpsee_execute(move |client| async move {
-			SubstrateClient::<C>::state_call(&*client, method, data, at_block)
+			SubstrateStateClient::<C>::call(&*client, method, data, at_block)
 				.await
 				.map_err(Into::into)
 		})
@@ -570,7 +567,7 @@ impl<C: Chain> Client<C> {
 		at_block: C::Hash,
 	) -> Result<StorageProof> {
 		self.jsonrpsee_execute(move |client| async move {
-			SubstrateClient::<C>::state_prove_storage(&*client, keys, Some(at_block))
+			SubstrateStateClient::<C>::prove_storage(&*client, keys, Some(at_block))
 				.await
 				.map(|proof| {
 					StorageProof::new(proof.proof.into_iter().map(|b| b.0).collect::<Vec<_>>())
@@ -583,7 +580,7 @@ impl<C: Chain> Client<C> {
 	/// Return `tokenDecimals` property from the set of chain properties.
 	pub async fn token_decimals(&self) -> Result<Option<u64>> {
 		self.jsonrpsee_execute(move |client| async move {
-			let system_properties = SubstrateClient::<C>::system_properties(&*client).await?;
+			let system_properties = SubstrateSystemClient::<C>::properties(&*client).await?;
 			Ok(system_properties.get("tokenDecimals").and_then(|v| v.as_u64()))
 		})
 		.await
@@ -593,13 +590,7 @@ impl<C: Chain> Client<C> {
 	pub async fn subscribe_grandpa_justifications(&self) -> Result<Subscription<Bytes>> {
 		let subscription = self
 			.jsonrpsee_execute(move |client| async move {
-				Ok(client
+				Ok(SubstrateGrandpaClient::<C>::subscribe_justifications(&*client).await?)
-					.subscribe(
-						"grandpa_subscribeJustifications",
-						None,
-						"grandpa_unsubscribeJustifications",
-					)
-					.await?)
 			})
 			.await?;
 		let (sender, receiver) = futures::channel::mpsc::channel(MAX_SUBSCRIPTION_CAPACITY);

bridges/relays/client-substrate/src/rpc.rs

@@ -16,7 +16,7 @@
 
 //! The most generic Substrate node RPC interface.
 
-use crate::Chain;
+use crate::{Chain, TransactionStatusOf};
 
 use jsonrpsee::{core::RpcResult, proc_macros::rpc};
 use pallet_transaction_payment_rpc_runtime_api::FeeDetails;
@@ -28,52 +28,100 @@ use sp_core::{
 use sp_rpc::number::NumberOrHex;
 use sp_version::RuntimeVersion;
 
-#[rpc(client, client_bounds(C: Chain))]
+/// RPC methods of Substrate `system` namespace, that we are using.
-pub(crate) trait Substrate<C> {
+#[rpc(client, client_bounds(C: Chain), namespace = "system")]
-	#[method(name = "system_health", param_kind = array)]
+pub(crate) trait SubstrateSystem<C> {
-	async fn system_health(&self) -> RpcResult<Health>;
+	/// Return node health.
-	#[method(name = "system_properties", param_kind = array)]
+	#[method(name = "health")]
-	async fn system_properties(&self) -> RpcResult<sc_chain_spec::Properties>;
+	async fn health(&self) -> RpcResult<Health>;
-	#[method(name = "chain_getHeader", param_kind = array)]
+	/// Return system properties.
-	async fn chain_get_header(&self, block_hash: Option<C::Hash>) -> RpcResult<C::Header>;
+	#[method(name = "properties")]
-	#[method(name = "chain_getFinalizedHead", param_kind = array)]
+	async fn properties(&self) -> RpcResult<sc_chain_spec::Properties>;
-	async fn chain_get_finalized_head(&self) -> RpcResult<C::Hash>;
+}
-	#[method(name = "chain_getBlock", param_kind = array)]
+
-	async fn chain_get_block(&self, block_hash: Option<C::Hash>) -> RpcResult<C::SignedBlock>;
+/// RPC methods of Substrate `chain` namespace, that we are using.
-	#[method(name = "chain_getBlockHash", param_kind = array)]
+#[rpc(client, client_bounds(C: Chain), namespace = "chain")]
-	async fn chain_get_block_hash(
+pub(crate) trait SubstrateChain<C> {
-		&self,
+	/// Get block hash by its number.
-		block_number: Option<C::BlockNumber>,
+	#[method(name = "getBlockHash")]
-	) -> RpcResult<C::Hash>;
+	async fn block_hash(&self, block_number: Option<C::BlockNumber>) -> RpcResult<C::Hash>;
-	#[method(name = "system_accountNextIndex", param_kind = array)]
+	/// Return block header by its hash.
-	async fn system_account_next_index(&self, account_id: C::AccountId) -> RpcResult<C::Index>;
+	#[method(name = "getHeader")]
-	#[method(name = "author_submitExtrinsic", param_kind = array)]
+	async fn header(&self, block_hash: Option<C::Hash>) -> RpcResult<C::Header>;
-	async fn author_submit_extrinsic(&self, extrinsic: Bytes) -> RpcResult<C::Hash>;
+	/// Return best finalized block hash.
-	#[method(name = "author_pendingExtrinsics", param_kind = array)]
+	#[method(name = "getFinalizedHead")]
-	async fn author_pending_extrinsics(&self) -> RpcResult<Vec<Bytes>>;
+	async fn finalized_head(&self) -> RpcResult<C::Hash>;
-	#[method(name = "state_call", param_kind = array)]
+	/// Return signed block (with justifications) by its hash.
-	async fn state_call(
+	#[method(name = "getBlock")]
+	async fn block(&self, block_hash: Option<C::Hash>) -> RpcResult<C::SignedBlock>;
+}
+
+/// RPC methods of Substrate `author` namespace, that we are using.
+#[rpc(client, client_bounds(C: Chain), namespace = "author")]
+pub(crate) trait SubstrateAuthor<C> {
+	/// Submit extrinsic to the transaction pool.
+	#[method(name = "submitExtrinsic")]
+	async fn submit_extrinsic(&self, extrinsic: Bytes) -> RpcResult<C::Hash>;
+	/// Return vector of pending extrinsics from the transaction pool.
+	#[method(name = "pendingExtrinsics")]
+	async fn pending_extrinsics(&self) -> RpcResult<Vec<Bytes>>;
+	/// Submit and watch for extrinsic state.
+	#[subscription(name = "submitAndWatchExtrinsic", unsubscribe = "unwatchExtrinsic", item = TransactionStatusOf<C>)]
+	fn submit_and_watch_extrinsic(&self, extrinsic: Bytes);
+}
+
+/// RPC methods of Substrate `state` namespace, that we are using.
+#[rpc(client, client_bounds(C: Chain), namespace = "state")]
+pub(crate) trait SubstrateState<C> {
+	/// Get current runtime version.
+	#[method(name = "getRuntimeVersion")]
+	async fn runtime_version(&self) -> RpcResult<RuntimeVersion>;
+	/// Call given runtime method.
+	#[method(name = "call")]
+	async fn call(
 		&self,
 		method: String,
 		data: Bytes,
 		at_block: Option<C::Hash>,
 	) -> RpcResult<Bytes>;
-	#[method(name = "state_getStorage", param_kind = array)]
+	/// Get value of the runtime storage.
-	async fn state_get_storage(
+	#[method(name = "getStorage")]
+	async fn storage(
 		&self,
 		key: StorageKey,
 		at_block: Option<C::Hash>,
 	) -> RpcResult<Option<StorageData>>;
-	#[method(name = "state_getReadProof", param_kind = array)]
+	/// Get proof of the runtime storage value.
-	async fn state_prove_storage(
+	#[method(name = "getReadProof")]
+	async fn prove_storage(
 		&self,
 		keys: Vec<StorageKey>,
 		hash: Option<C::Hash>,
 	) -> RpcResult<ReadProof<C::Hash>>;
-	#[method(name = "state_getRuntimeVersion", param_kind = array)]
+}
-	async fn state_runtime_version(&self) -> RpcResult<RuntimeVersion>;
+
-	#[method(name = "payment_queryFeeDetails", param_kind = array)]
+/// RPC methods of Substrate `grandpa` namespace, that we are using.
-	async fn payment_query_fee_details(
+#[rpc(client, client_bounds(C: Chain), namespace = "grandpa")]
+pub(crate) trait SubstrateGrandpa<C> {
+	/// Subscribe to GRANDPA justifications.
+	#[subscription(name = "subscribeJustifications", unsubscribe = "unsubscribeJustifications", item = Bytes)]
+	fn subscribe_justifications(&self);
+}
+
+/// RPC methods of Substrate `system` frame pallet, that we are using.
+#[rpc(client, client_bounds(C: Chain), namespace = "system")]
+pub(crate) trait SubstrateFrameSystem<C> {
+	/// Return index of next account transaction.
+	#[method(name = "accountNextIndex")]
+	async fn account_next_index(&self, account_id: C::AccountId) -> RpcResult<C::Index>;
+}
+
+/// RPC methods of Substrate `pallet_transaction_payment` frame pallet, that we are using.
+#[rpc(client, client_bounds(C: Chain), namespace = "payment")]
+pub(crate) trait SubstrateTransactionPayment<C> {
+	/// Query transaction fee details.
+	#[method(name = "queryFeeDetails")]
+	async fn fee_details(
 		&self,
 		extrinsic: Bytes,
 		at_block: Option<C::Hash>,