Skip to content

Commit

Permalink
Use jsonrpsee subscriptions (#1533)
Browse files Browse the repository at this point in the history 
* splitted Substrate RPC trait

* introduce subscription methods

* removed commented code

* removed commented code
  • Loading branch information
@svyatonik @bkchr
svyatonik authored and bkchr committed Apr 10, 2024
1 parent 77af92b commit e7a7396
Show file tree
Hide file tree
Showing 2 changed files with 119 additions and 80 deletions.
83 changes: 37 additions & 46 deletions bridges/relays/client-substrate/src/client.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};

Expand All @@ -29,8 +33,7 @@ use codec::{Decode, Encode};
use frame_system::AccountInfo;
use futures::{SinkExt, StreamExt};
use jsonrpsee::{
core::{client::SubscriptionClientT, DeserializeOwned},
types::params::ParamsSer,
core::DeserializeOwned,
ws_client::{WsClient as RpcClient, WsClientBuilder as RpcClientBuilder},
};
use num_traits::{Bounded, Zero};
Expand Down Expand Up @@ -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))
.await
SubstrateChainClient::<C>::block_hash(&*genesis_hash_client, Some(number)).await
})
.await??;

Expand Down Expand Up @@ -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(())
Expand All @@ -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
}
Expand All @@ -252,15 +254,15 @@ 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
}

/// 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
}
Expand All @@ -271,15 +273,15 @@ 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
}

/// 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
}
Expand All @@ -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
}
Expand Down Expand Up @@ -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
}
Expand All @@ -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(
Expand All @@ -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
}
Expand All @@ -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);
Expand Down Expand Up @@ -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);
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(tx_hash)
})
Expand All @@ -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
.subscribe(
"author_submitAndWatchExtrinsic",
Some(ParamsSer::Array(vec![jsonrpsee::core::to_json_value(extrinsic)
.map_err(|e| Error::RpcError(e.into()))?])),
"author_unwatchExtrinsic",
)
.await
.map_err(|e| {
log::error!(target: "bridge", "Failed to send transaction to {} node: {:?}", C::NAME, e);
e
})?;
let subscription =
SubstrateAuthorClient::<C>::submit_and_watch_extrinsic(&*client, extrinsic)
.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)
})
Expand All @@ -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
}
Expand All @@ -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)?;

Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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)
})
Expand All @@ -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<_>>())
Expand All @@ -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
Expand All @@ -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
.subscribe(
"grandpa_subscribeJustifications",
None,
"grandpa_unsubscribeJustifications",
)
.await?)
Ok(SubstrateGrandpaClient::<C>::subscribe_justifications(&*client).await?)
})
.await?;
let (sender, receiver) = futures::channel::mpsc::channel(MAX_SUBSCRIPTION_CAPACITY);
Expand Down
116 changes: 82 additions & 34 deletions bridges/relays/client-substrate/src/rpc.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -28,52 +28,100 @@ use sp_core::{
use sp_rpc::number::NumberOrHex;
use sp_version::RuntimeVersion;

#[rpc(client, client_bounds(C: Chain))]
pub(crate) trait Substrate<C> {
#[method(name = "system_health", param_kind = array)]
async fn system_health(&self) -> RpcResult<Health>;
#[method(name = "system_properties", param_kind = array)]
async fn system_properties(&self) -> RpcResult<sc_chain_spec::Properties>;
#[method(name = "chain_getHeader", param_kind = array)]
async fn chain_get_header(&self, block_hash: Option<C::Hash>) -> RpcResult<C::Header>;
#[method(name = "chain_getFinalizedHead", param_kind = array)]
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>;
#[method(name = "chain_getBlockHash", param_kind = array)]
async fn chain_get_block_hash(
&self,
block_number: Option<C::BlockNumber>,
) -> RpcResult<C::Hash>;
#[method(name = "system_accountNextIndex", param_kind = array)]
async fn system_account_next_index(&self, account_id: C::AccountId) -> RpcResult<C::Index>;
#[method(name = "author_submitExtrinsic", param_kind = array)]
async fn author_submit_extrinsic(&self, extrinsic: Bytes) -> RpcResult<C::Hash>;
#[method(name = "author_pendingExtrinsics", param_kind = array)]
async fn author_pending_extrinsics(&self) -> RpcResult<Vec<Bytes>>;
#[method(name = "state_call", param_kind = array)]
async fn state_call(
/// RPC methods of Substrate `system` namespace, that we are using.
#[rpc(client, client_bounds(C: Chain), namespace = "system")]
pub(crate) trait SubstrateSystem<C> {
/// Return node health.
#[method(name = "health")]
async fn health(&self) -> RpcResult<Health>;
/// Return system properties.
#[method(name = "properties")]
async fn properties(&self) -> RpcResult<sc_chain_spec::Properties>;
}

/// RPC methods of Substrate `chain` namespace, that we are using.
#[rpc(client, client_bounds(C: Chain), namespace = "chain")]
pub(crate) trait SubstrateChain<C> {
/// Get block hash by its number.
#[method(name = "getBlockHash")]
async fn block_hash(&self, block_number: Option<C::BlockNumber>) -> RpcResult<C::Hash>;
/// Return block header by its hash.
#[method(name = "getHeader")]
async fn header(&self, block_hash: Option<C::Hash>) -> RpcResult<C::Header>;
/// Return best finalized block hash.
#[method(name = "getFinalizedHead")]
async fn finalized_head(&self) -> RpcResult<C::Hash>;
/// Return signed block (with justifications) by its hash.
#[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)]
async fn state_get_storage(
/// Get value of the runtime 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)]
async fn state_prove_storage(
/// Get proof of the runtime storage value.
#[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)]
async fn payment_query_fee_details(
}

/// RPC methods of Substrate `grandpa` namespace, that we are using.
#[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>,
Expand Down

0 comments on commit e7a7396

Please sign in to comment.