pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-20 05:51:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into lexnv/codegen-config

Browse Source
This commit is contained in:
Alexandru Vasile
2024-09-13 18:59:58 +03:00
parent 609b701d4d 52252242b0
commit b13fcd66b8
111 changed files with 24479 additions and 7527 deletions
Download Patch File Download Diff File Expand all files Collapse all files
subxt/Cargo.toml
+27 -10
View File
@@ -26,9 +26,11 @@ default = ["jsonrpsee", "native"]
# Exactly 1 of "web" and "native" is expected.
native = [
"jsonrpsee?/async-client",
"jsonrpsee?/client-ws-transport-native-tls",
"jsonrpsee?/client-ws-transport-tls",
"jsonrpsee?/ws-client",
"subxt-lightclient?/native",
"tokio-util"
"tokio-util",
"tokio?/sync",
]
# Enable this for web/wasm builds.
@@ -36,14 +38,17 @@ native = [
web = [
"jsonrpsee?/async-wasm-client",
"jsonrpsee?/client-web-transport",
"jsonrpsee?/wasm-client",
"getrandom/js",
"subxt-lightclient?/web",
"subxt-macro/web",
"instant/wasm-bindgen"
"instant/wasm-bindgen",
"tokio?/sync",
"finito?/wasm-bindgen",
]
# Enable this to use the reconnecting rpc client
unstable-reconnecting-rpc-client = ["dep:reconnecting-jsonrpsee-ws-client"]
unstable-reconnecting-rpc-client = ["dep:finito", "dep:tokio", "jsonrpsee", "wasm-bindgen-futures"]
# Enable this to use jsonrpsee (allowing for example `OnlineClient::from_url`).
jsonrpsee = [
@@ -65,6 +70,9 @@ unstable-metadata = []
# Note that this feature is experimental and things may break or not work as expected.
unstable-light-client = ["subxt-lightclient"]
# Activate this to expose the ability to generate metadata from Wasm runtime files.
runtime-path = ["subxt-macro/runtime-path"]
[dependencies]
async-trait = { workspace = true }
codec = { package = "parity-scale-codec", workspace = true, features = ["derive"] }
@@ -98,9 +106,6 @@ subxt-core = { workspace = true, features = ["std"] }
subxt-metadata = { workspace = true, features = ["std"] }
subxt-lightclient = { workspace = true, optional = true, default-features = false }
# Reconnecting jsonrpc ws client
reconnecting-jsonrpsee-ws-client = { version = "0.4", optional = true }
# For parsing urls to disallow insecure schemes
url = { workspace = true }
@@ -110,11 +115,18 @@ getrandom = { workspace = true, optional = true }
# Included if "native" feature is enabled
tokio-util = { workspace = true, features = ["compat"], optional = true }
# Included if the reconnecting rpc client feature is enabled
# Only the `tokio/sync` is used in the reconnecting rpc client
# and that compiles both for native and web.
tokio = { workspace = true, optional = true }
finito = { workspace = true, optional = true }
wasm-bindgen-futures = { workspace = true, optional = true }
[dev-dependencies]
bitvec = { workspace = true }
codec = { workspace = true, features = ["derive", "bit-vec"] }
scale-info = { workspace = true, features = ["bit-vec"] }
tokio = { workspace = true, features = ["macros", "time", "rt-multi-thread"] }
tokio = { workspace = true, features = ["macros", "time", "rt-multi-thread", "sync"] }
sp-core = { workspace = true }
sp-keyring = { workspace = true }
sp-runtime = { workspace = true }
@@ -125,6 +137,11 @@ subxt-signer = { path = "../signer", features = ["unstable-eth"] }
# the light-client wlll emit INFO logs with
# `GrandPa warp sync finished` and `Finalized block runtime ready.`
tracing-subscriber = { workspace = true }
# These deps are needed to test the reconnecting rpc client
jsonrpsee = { workspace = true, features = ["server"] }
tower = "0.4"
hyper = "1"
http-body = "1"
[[example]]
name = "light_client_basic"
@@ -137,8 +154,8 @@ path = "examples/light_client_local_node.rs"
required-features = ["unstable-light-client", "jsonrpsee", "native"]
[[example]]
name = "reconnecting_rpc_client"
path = "examples/reconnecting_rpc_client.rs"
name = "setup_reconnecting_rpc_client"
path = "examples/setup_reconnecting_rpc_client.rs"
required-features = ["unstable-reconnecting-rpc-client"]
[package.metadata.docs.rs]
subxt/examples/reconnecting_rpc_client.rs
-73
View File
@@ -1,73 +0,0 @@
//! Example to utilize the `reconnecting rpc client` in subxt
//! which hidden behind behind `--feature unstable-reconnecting-rpc-client`
//!
//! To utilize full logs from the RPC client use:
//! `RUST_LOG="jsonrpsee=trace,reconnecting_jsonrpsee_ws_client=trace"`
#![allow(missing_docs)]
use std::time::Duration;
use subxt::backend::rpc::reconnecting_rpc_client::{Client, ExponentialBackoff, PingConfig};
use subxt::backend::rpc::RpcClient;
use subxt::error::{Error, RpcError};
use subxt::{OnlineClient, PolkadotConfig};
// Generate an interface that we can use from the node's metadata.
#[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata_small.scale")]
pub mod polkadot {}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
tracing_subscriber::fmt::init();
// Create a new client with with a reconnecting RPC client.
let rpc = Client::builder()
// Reconnect with exponential backoff
//
// This API is "iterator-like" so one could limit it to only
// reconnect x times and then quit.
.retry_policy(ExponentialBackoff::from_millis(100).max_delay(Duration::from_secs(10)))
// Send period WebSocket pings/pongs every 6th second and if it's not ACK:ed in 30 seconds
// then disconnect.
//
// This is just a way to ensure that the connection isn't idle if no message is sent that often
.enable_ws_ping(
PingConfig::new()
.ping_interval(Duration::from_secs(6))
.inactive_limit(Duration::from_secs(30)),
)
// There are other configurations as well that can be found here:
// <https://docs.rs/reconnecting-jsonrpsee-ws-client/latest/reconnecting_jsonrpsee_ws_client/struct.ClientBuilder.html>
.build("ws://localhost:9944".to_string())
.await?;
let api: OnlineClient<PolkadotConfig> =
OnlineClient::from_rpc_client(RpcClient::new(rpc.clone())).await?;
// Subscribe to all finalized blocks:
let mut blocks_sub = api.blocks().subscribe_finalized().await?;
// For each block, print a bunch of information about it:
while let Some(block) = blocks_sub.next().await {
let block = match block {
Ok(b) => b,
Err(Error::Rpc(RpcError::DisconnectedWillReconnect(err))) => {
println!("{err}");
continue;
}
Err(e) => {
return Err(e.into());
}
};
let block_number = block.header().number;
let block_hash = block.hash();
println!("Block #{block_number} ({block_hash})");
}
println!("RPC client reconnected `{}` times", rpc.reconnect_count());
Ok(())
}
subxt/examples/setup_config_signed_extension.rs
+3 -2
View File
@@ -37,6 +37,7 @@ impl Config for CustomConfig {
signed_extensions::CheckMortality<Self>,
signed_extensions::ChargeAssetTxPayment<Self>,
signed_extensions::ChargeTransactionPayment,
signed_extensions::CheckMetadataHash,
// And add a new one of our own:
CustomSignedExtension,
),
@@ -83,8 +84,8 @@ impl ExtrinsicParamsEncoder for CustomSignedExtension {
pub fn custom(
params: DefaultExtrinsicParamsBuilder<CustomConfig>,
) -> <<CustomConfig as Config>::ExtrinsicParams as ExtrinsicParams<CustomConfig>>::Params {
let (a, b, c, d, e, f, g) = params.build();
(a, b, c, d, e, f, g, ())
let (a, b, c, d, e, f, g, h) = params.build();
(a, b, c, d, e, f, g, h, ())
}
#[tokio::main]
subxt/examples/setup_reconnecting_rpc_client.rs
+84
View File
@@ -0,0 +1,84 @@
//! Example to utilize the `reconnecting rpc client` in subxt
//! which hidden behind behind `--feature unstable-reconnecting-rpc-client`
//!
//! To utilize full logs from the RPC client use:
//! `RUST_LOG="jsonrpsee=trace,subxt-reconnecting-rpc-client=trace"`
#![allow(missing_docs)]
use std::time::Duration;
use futures::StreamExt;
use subxt::backend::rpc::reconnecting_rpc_client::{ExponentialBackoff, RpcClient};
use subxt::{OnlineClient, PolkadotConfig};
// Generate an interface that we can use from the node's metadata.
#[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata_small.scale")]
pub mod polkadot {}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
tracing_subscriber::fmt::init();
// Create a new client with with a reconnecting RPC client.
let rpc = RpcClient::builder()
// Reconnect with exponential backoff
//
// This API is "iterator-like" and we use `take` to limit the number of retries.
.retry_policy(
ExponentialBackoff::from_millis(100)
.max_delay(Duration::from_secs(10))
.take(3),
)
// There are other configurations as well that can be found at [`reconnecting_rpc_client::ClientBuilder`].
.build("ws://localhost:9944".to_string())
.await?;
// If you want to use the unstable backend with the reconnecting RPC client, you can do so like this:
//
// ```
// use subxt::backend::unstable::UnstableBackend;
// use subxt::OnlineClient;
//
// let (backend, mut driver) = UnstableBackend::builder().build(RpcClient::new(rpc.clone()));
// tokio::spawn(async move {
// while let Some(val) = driver.next().await {
// if let Err(e) = val {
// eprintln!("Error driving unstable backend: {e}; terminating client");
// }
// }
// });
// let api: OnlineClient<PolkadotConfig> = OnlineClient::from_backend(Arc::new(backend)).await?;
// ```
let api: OnlineClient<PolkadotConfig> = OnlineClient::from_rpc_client(rpc.clone()).await?;
// Run for at most 100 blocks and print a bunch of information about it.
//
// The subscription is automatically re-started when the RPC client has reconnected.
// You can test that by stopping the polkadot node and restarting it.
let mut blocks_sub = api.blocks().subscribe_finalized().await?.take(100);
while let Some(block) = blocks_sub.next().await {
let block = match block {
Ok(b) => b,
Err(e) => {
// This can only happen on the legacy backend and the unstable backend
// will handle this internally.
if e.is_disconnected_will_reconnect() {
println!("The RPC connection was lost and we may have missed a few blocks");
continue;
}
return Err(e.into());
}
};
let block_number = block.number();
let block_hash = block.hash();
println!("Block #{block_number} ({block_hash})");
}
Ok(())
}
subxt/examples/storage_fetch.rs
+2 -2
View File
@@ -22,10 +22,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.storage()
.at_latest()
.await?
.fetch_raw(subxt_core::storage::get_address_bytes(&storage_query, &api.metadata()).unwrap())
.fetch(&storage_query)
.await?;
let v = hex::encode(result.unwrap());
let v = result.unwrap().data.free;
println!("Alice: {v}");
Ok(())
}
subxt/examples/tx_basic_frontier.rs
+4 -11
View File
@@ -6,7 +6,8 @@
#![allow(missing_docs)]
use subxt::OnlineClient;
use subxt_signer::eth::{dev, AccountId20, Signature};
use subxt_core::utils::AccountId20;
use subxt_signer::eth::{dev, Signature};
#[subxt::subxt(runtime_metadata_path = "../artifacts/frontier_metadata_small.scale")]
mod eth_runtime {}
@@ -25,28 +26,20 @@ impl subxt::Config for EthRuntimeConfig {
type AssetId = u32;
}
// This helper makes it easy to use our `AccountId20`'s with generated
// code that expects a generated `eth_runtime::runtime_types::fp_account:AccountId20` type.
impl From<AccountId20> for eth_runtime::runtime_types::fp_account::AccountId20 {
fn from(a: AccountId20) -> Self {
eth_runtime::runtime_types::fp_account::AccountId20(a.0)
}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let api = OnlineClient::<EthRuntimeConfig>::from_insecure_url("ws://127.0.0.1:9944").await?;
let alith = dev::alith();
let baltathar = dev::baltathar();
let dest = baltathar.account_id();
let dest = baltathar.public_key().to_account_id();
println!("baltathar pub: {}", hex::encode(baltathar.public_key().0));
println!("baltathar addr: {}", hex::encode(dest));
let balance_transfer_tx = eth_runtime::tx()
.balances()
.transfer_allow_death(dest.into(), 10_001);
.transfer_allow_death(dest, 10_001);
let events = api
.tx()
subxt/examples/tx_boxed.rs
+43
View File
@@ -0,0 +1,43 @@
#![allow(missing_docs)]
use subxt::{OnlineClient, PolkadotConfig};
use subxt_signer::sr25519::dev;
#[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata_small.scale")]
pub mod polkadot {}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let api = OnlineClient::<PolkadotConfig>::new().await?;
// Prepare some extrinsics. These are boxed so that they can live alongside each other.
let txs = [dynamic_remark(), balance_transfer(), remark()];
for tx in txs {
let from = dev::alice();
api.tx()
.sign_and_submit_then_watch_default(&tx, &from)
.await?
.wait_for_finalized_success()
.await?;
println!("Submitted tx");
}
Ok(())
}
fn balance_transfer() -> Box<dyn subxt::tx::Payload> {
let dest = dev::bob().public_key().into();
Box::new(polkadot::tx().balances().transfer_allow_death(dest, 10_000))
}
fn remark() -> Box<dyn subxt::tx::Payload> {
Box::new(polkadot::tx().system().remark(vec![1, 2, 3, 4, 5]))
}
fn dynamic_remark() -> Box<dyn subxt::tx::Payload> {
use subxt::dynamic::{tx, Value};
let tx_payload = tx("System", "remark", vec![Value::from_bytes("Hello")]);
Box::new(tx_payload)
}
subxt/examples/tx_partial.rs
+53
View File
@@ -0,0 +1,53 @@
#![allow(missing_docs)]
use subxt::{OnlineClient, PolkadotConfig};
use subxt_signer::sr25519::dev;
type BoxedError = Box<dyn std::error::Error + Send + Sync + 'static>;
#[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata_small.scale")]
pub mod polkadot {}
#[tokio::main]
async fn main() -> Result<(), BoxedError> {
// Spawned tasks require things held across await points to impl Send,
// so we use one to demonstrate that this is possible with `PartialExtrinsic`
tokio::spawn(signing_example()).await??;
Ok(())
}
async fn signing_example() -> Result<(), BoxedError> {
let api = OnlineClient::<PolkadotConfig>::new().await?;
// Build a balance transfer extrinsic.
let dest = dev::bob().public_key().into();
let balance_transfer_tx = polkadot::tx().balances().transfer_allow_death(dest, 10_000);
let alice = dev::alice();
// Create partial tx, ready to be signed.
let partial_tx = api
.tx()
.create_partial_signed(
&balance_transfer_tx,
&alice.public_key().to_account_id(),
Default::default(),
)
.await?;
// Simulate taking some time to get a signature back, in part to
// show that the `PartialExtrinsic` can be held across await points.
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
let signature = alice.sign(&partial_tx.signer_payload());
// Sign the transaction.
let tx = partial_tx
.sign_with_address_and_signature(&alice.public_key().to_address(), &signature.into());
// Submit it.
tx.submit_and_watch()
.await?
.wait_for_finalized_success()
.await?;
Ok(())
}
subxt/src/backend/legacy/mod.rs
+158 -68
View File
@@ -8,10 +8,12 @@
pub mod rpc_methods;
use self::rpc_methods::TransactionStatus as RpcTransactionStatus;
use crate::backend::utils::{retry, retry_stream};
use crate::backend::{
rpc::RpcClient, Backend, BlockRef, RuntimeVersion, StorageResponse, StreamOf, StreamOfResults,
TransactionStatus,
};
use crate::error::RpcError;
use crate::{config::Header, Config, Error};
use async_trait::async_trait;
use futures::{future, future::Either, stream, Future, FutureExt, Stream, StreamExt};
@@ -62,12 +64,21 @@ impl<T: Config> LegacyBackendBuilder<T> {
}
/// The legacy backend.
#[derive(Debug, Clone)]
#[derive(Debug)]
pub struct LegacyBackend<T> {
storage_page_size: u32,
methods: LegacyRpcMethods<T>,
}
impl<T> Clone for LegacyBackend<T> {
fn clone(&self) -> LegacyBackend<T> {
LegacyBackend {
storage_page_size: self.storage_page_size,
methods: self.methods.clone(),
}
}
}
impl<T: Config> LegacyBackend<T> {
/// Configure and construct an [`LegacyBackend`].
pub fn builder() -> LegacyBackendBuilder<T> {
@@ -84,16 +95,28 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for LegacyBackend<T> {
keys: Vec<Vec<u8>>,
at: T::Hash,
) -> Result<StreamOfResults<StorageResponse>, Error> {
fn get_entry<T: Config>(
key: Vec<u8>,
at: T::Hash,
methods: LegacyRpcMethods<T>,
) -> impl Future<Output = Result<Option<StorageResponse>, Error>> {
retry(move || {
let methods = methods.clone();
let key = key.clone();
async move {
let res = methods.state_get_storage(&key, Some(at)).await?;
Ok(res.map(move |value| StorageResponse { key, value }))
}
})
}
let keys = keys.clone();
let methods = self.methods.clone();
// For each key, return it + a future to get the result.
let iter = keys.into_iter().map(move |key| {
let methods = methods.clone();
async move {
let res = methods.state_get_storage(&key, Some(at)).await?;
Ok(res.map(|value| StorageResponse { key, value }))
}
});
let iter = keys
.into_iter()
.map(move |key| get_entry(key, at, methods.clone()));
let s = stream::iter(iter)
// Resolve the future
@@ -158,99 +181,159 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for LegacyBackend<T> {
}
async fn genesis_hash(&self) -> Result<T::Hash, Error> {
self.methods.genesis_hash().await
retry(|| self.methods.genesis_hash()).await
}
async fn block_header(&self, at: T::Hash) -> Result<Option<T::Header>, Error> {
self.methods.chain_get_header(Some(at)).await
retry(|| self.methods.chain_get_header(Some(at))).await
}
async fn block_body(&self, at: T::Hash) -> Result<Option<Vec<Vec<u8>>>, Error> {
let Some(details) = self.methods.chain_get_block(Some(at)).await? else {
return Ok(None);
};
Ok(Some(
details.block.extrinsics.into_iter().map(|b| b.0).collect(),
))
retry(|| async {
let Some(details) = self.methods.chain_get_block(Some(at)).await? else {
return Ok(None);
};
Ok(Some(
details.block.extrinsics.into_iter().map(|b| b.0).collect(),
))
})
.await
}
async fn latest_finalized_block_ref(&self) -> Result<BlockRef<T::Hash>, Error> {
let hash = self.methods.chain_get_finalized_head().await?;
Ok(BlockRef::from_hash(hash))
retry(|| async {
let hash = self.methods.chain_get_finalized_head().await?;
Ok(BlockRef::from_hash(hash))
})
.await
}
async fn current_runtime_version(&self) -> Result<RuntimeVersion, Error> {
let details = self.methods.state_get_runtime_version(None).await?;
Ok(RuntimeVersion {
spec_version: details.spec_version,
transaction_version: details.transaction_version,
retry(|| async {
let details = self.methods.state_get_runtime_version(None).await?;
Ok(RuntimeVersion {
spec_version: details.spec_version,
transaction_version: details.transaction_version,
})
})
.await
}
async fn stream_runtime_version(&self) -> Result<StreamOfResults<RuntimeVersion>, Error> {
let sub = self.methods.state_subscribe_runtime_version().await?;
let sub = sub.map(|r| {
r.map(|v| RuntimeVersion {
spec_version: v.spec_version,
transaction_version: v.transaction_version,
let methods = self.methods.clone();
let retry_sub = retry_stream(move || {
let methods = methods.clone();
Box::pin(async move {
let sub = methods.state_subscribe_runtime_version().await?;
let sub = sub.map(|r| {
r.map(|v| RuntimeVersion {
spec_version: v.spec_version,
transaction_version: v.transaction_version,
})
});
Ok(StreamOf(Box::pin(sub)))
})
})
.await?;
// For runtime version subscriptions we omit the `DisconnectedWillReconnect` error
// because the once it resubscribes it will emit the latest runtime version.
//
// Thus, it's technically possible that a runtime version can be missed if
// two runtime upgrades happen in quick succession, but this is very unlikely.
let stream = retry_sub.filter(|r| {
let forward = !matches!(r, Err(Error::Rpc(RpcError::DisconnectedWillReconnect(_))));
async move { forward }
});
Ok(StreamOf(Box::pin(sub)))
Ok(StreamOf(Box::pin(stream)))
}
async fn stream_all_block_headers(
&self,
) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
let sub = self.methods.chain_subscribe_all_heads().await?;
let sub = sub.map(|r| {
r.map(|h| {
let hash = h.hash();
(h, BlockRef::from_hash(hash))
let methods = self.methods.clone();
let retry_sub = retry_stream(move || {
let methods = methods.clone();
Box::pin(async move {
let sub = methods.chain_subscribe_all_heads().await?;
let sub = sub.map(|r| {
r.map(|h| {
let hash = h.hash();
(h, BlockRef::from_hash(hash))
})
});
Ok(StreamOf(Box::pin(sub)))
})
});
Ok(StreamOf(Box::pin(sub)))
})
.await?;
Ok(retry_sub)
}
async fn stream_best_block_headers(
&self,
) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
let sub = self.methods.chain_subscribe_new_heads().await?;
let sub = sub.map(|r| {
r.map(|h| {
let hash = h.hash();
(h, BlockRef::from_hash(hash))
let methods = self.methods.clone();
let retry_sub = retry_stream(move || {
let methods = methods.clone();
Box::pin(async move {
let sub = methods.chain_subscribe_new_heads().await?;
let sub = sub.map(|r| {
r.map(|h| {
let hash = h.hash();
(h, BlockRef::from_hash(hash))
})
});
Ok(StreamOf(Box::pin(sub)))
})
});
Ok(StreamOf(Box::pin(sub)))
})
.await?;
Ok(retry_sub)
}
async fn stream_finalized_block_headers(
&self,
) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
let sub: super::rpc::RpcSubscription<<T as Config>::Header> =
self.methods.chain_subscribe_finalized_heads().await?;
let this = self.clone();
// Get the last finalized block immediately so that the stream will emit every finalized block after this.
let last_finalized_block_ref = self.latest_finalized_block_ref().await?;
let last_finalized_block_num = self
.block_header(last_finalized_block_ref.hash())
.await?
.map(|h| h.number().into());
let retry_sub = retry_stream(move || {
let this = this.clone();
Box::pin(async move {
let sub = this.methods.chain_subscribe_finalized_heads().await?;
// Fill in any missing blocks, because the backend may not emit every finalized block; just the latest ones which
// are finalized each time.
let sub = subscribe_to_block_headers_filling_in_gaps(
self.methods.clone(),
sub,
last_finalized_block_num,
);
let sub = sub.map(|r| {
r.map(|h| {
let hash = h.hash();
(h, BlockRef::from_hash(hash))
// Get the last finalized block immediately so that the stream will emit every finalized block after this.
let last_finalized_block_ref = this.latest_finalized_block_ref().await?;
let last_finalized_block_num = this
.block_header(last_finalized_block_ref.hash())
.await?
.map(|h| h.number().into());
// Fill in any missing blocks, because the backend may not emit every finalized block; just the latest ones which
// are finalized each time.
let sub = subscribe_to_block_headers_filling_in_gaps(
this.methods.clone(),
sub,
last_finalized_block_num,
);
let sub = sub.map(|r| {
r.map(|h| {
let hash = h.hash();
(h, BlockRef::from_hash(hash))
})
});
Ok(StreamOf(Box::pin(sub)))
})
});
Ok(StreamOf(Box::pin(sub)))
})
.await?;
Ok(retry_sub)
}
async fn submit_transaction(
@@ -261,6 +344,7 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for LegacyBackend<T> {
.methods
.author_submit_and_watch_extrinsic(extrinsic)
.await?;
let sub = sub.filter_map(|r| {
let mapped = r
.map(|tx| {
@@ -309,7 +393,8 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for LegacyBackend<T> {
future::ready(mapped)
});
Ok(StreamOf(Box::pin(sub)))
Ok(StreamOf::new(Box::pin(sub)))
}
async fn call(
@@ -318,9 +403,7 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for LegacyBackend<T> {
call_parameters: Option<&[u8]>,
at: T::Hash,
) -> Result<Vec<u8>, Error> {
self.methods
.state_call(method, call_parameters, Some(at))
.await
retry(|| self.methods.state_call(method, call_parameters, Some(at))).await
}
}
@@ -431,6 +514,11 @@ impl<T: Config> Stream for StorageFetchDescendantKeysStream<T> {
return Poll::Ready(Some(Ok(keys)));
}
Err(e) => {
if e.is_disconnected_will_reconnect() {
this.keys_fut = Some(keys_fut);
continue;
}
// Error getting keys? Return it.
return Poll::Ready(Some(Err(e)));
}
@@ -513,7 +601,9 @@ impl<T: Config> Stream for StorageFetchDescendantValuesStream<T> {
let at = this.keys.at;
let results_fut = async move {
let keys = keys.iter().map(|k| &**k);
let values = methods.state_query_storage_at(keys, Some(at)).await?;
let values =
retry(|| methods.state_query_storage_at(keys.clone(), Some(at)))
.await?;
let values: VecDeque<_> = values
.into_iter()
.flat_map(|v| {
subxt/src/backend/legacy/rpc_methods.rs
+2 -2
View File
@@ -332,8 +332,7 @@ impl<T: Config> LegacyRpcMethods<T> {
public: Vec<u8>,
) -> Result<(), Error> {
let params = rpc_params![key_type, suri, Bytes(public)];
self.client.request("author_insertKey", params).await?;
Ok(())
self.client.request("author_insertKey", params).await
}
/// Generate new session keys and returns the corresponding public keys.
@@ -455,6 +454,7 @@ pub type EncodedJustification = Vec<u8>;
/// the RPC call `state_getRuntimeVersion`,
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(test, derive(serde::Serialize))]
pub struct RuntimeVersion {
/// Version of the runtime specification. A full-node will not attempt to use its native
/// runtime in substitute for the on-chain Wasm runtime unless all of `spec_name`,
subxt/src/backend/mod.rs
+405
View File
@@ -9,6 +9,7 @@
pub mod legacy;
pub mod rpc;
pub mod unstable;
pub mod utils;
use subxt_core::client::RuntimeVersion;
@@ -324,9 +325,413 @@ pub enum TransactionStatus<Hash> {
/// A response from calls like [`Backend::storage_fetch_values`] or
/// [`Backend::storage_fetch_descendant_values`].
#[cfg_attr(test, derive(serde::Serialize, Clone, PartialEq, Debug))]
pub struct StorageResponse {
/// The key.
pub key: Vec<u8>,
/// The associated value.
pub value: Vec<u8>,
}
#[cfg(test)]
mod test {
use super::*;
mod legacy {
use super::rpc::{RpcClient, RpcClientT};
use crate::backend::rpc::RawRpcSubscription;
use crate::backend::BackendExt;
use crate::{
backend::{
legacy::rpc_methods::Bytes, legacy::rpc_methods::RuntimeVersion,
legacy::LegacyBackend, StorageResponse,
},
error::RpcError,
};
use futures::StreamExt;
use serde::Serialize;
use serde_json::value::RawValue;
use std::{
collections::{HashMap, VecDeque},
sync::Arc,
};
use subxt_core::{config::DefaultExtrinsicParams, Config};
use tokio::sync::{mpsc, Mutex};
type RpcResult<T> = Result<T, RpcError>;
type Item = RpcResult<String>;
struct MockDataTable {
items: HashMap<Vec<u8>, VecDeque<Item>>,
}
impl MockDataTable {
fn new() -> Self {
MockDataTable {
items: HashMap::new(),
}
}
fn from_iter<'a, T: Serialize, I: IntoIterator<Item = (&'a str, RpcResult<T>)>>(
item: I,
) -> Self {
let mut data = Self::new();
for (key, item) in item.into_iter() {
data.push(key.into(), item);
}
data
}
fn push<I: Serialize>(&mut self, key: Vec<u8>, item: RpcResult<I>) {
let item = item.map(|x| serde_json::to_string(&x).unwrap());
match self.items.entry(key) {
std::collections::hash_map::Entry::Occupied(v) => v.into_mut().push_back(item),
std::collections::hash_map::Entry::Vacant(e) => {
e.insert(VecDeque::from([item]));
}
}
}
fn pop(&mut self, key: Vec<u8>) -> Item {
self.items.get_mut(&key).unwrap().pop_front().unwrap()
}
}
struct Subscription {
sender: mpsc::Sender<RpcResult<Vec<Item>>>,
receiver: mpsc::Receiver<RpcResult<Vec<Item>>>,
}
impl Subscription {
fn new() -> Self {
let (sender, receiver) = mpsc::channel(32);
Self { sender, receiver }
}
async fn from_iter<
T: Serialize,
S: IntoIterator<Item = RpcResult<Vec<RpcResult<T>>>>,
>(
items: S,
) -> Self {
let sub = Self::new();
for i in items {
let i: RpcResult<Vec<Item>> = i.map(|items| {
items
.into_iter()
.map(|item| item.map(|i| serde_json::to_string(&i).unwrap()))
.collect()
});
sub.write(i).await
}
sub
}
async fn read(&mut self) -> RpcResult<Vec<Item>> {
self.receiver.recv().await.unwrap()
}
async fn write(&self, items: RpcResult<Vec<Item>>) {
self.sender.send(items).await.unwrap()
}
}
struct Data {
request: MockDataTable,
subscription: Subscription,
}
struct MockRpcClientStorage {
data: Arc<Mutex<Data>>,
}
impl RpcClientT for MockRpcClientStorage {
fn request_raw<'a>(
&'a self,
method: &'a str,
params: Option<Box<serde_json::value::RawValue>>,
) -> super::rpc::RawRpcFuture<'a, Box<serde_json::value::RawValue>> {
Box::pin(async move {
match method {
"state_getStorage" => {
let mut data = self.data.lock().await;
let params = params.map(|p| p.get().to_string());
let rpc_params = jsonrpsee::types::Params::new(params.as_deref());
let key: sp_core::Bytes = rpc_params.sequence().next().unwrap();
let value = data.request.pop(key.0);
value.map(|v| serde_json::value::RawValue::from_string(v).unwrap())
}
"chain_getBlockHash" => {
let mut data = self.data.lock().await;
let value = data.request.pop("chain_getBlockHash".into());
value.map(|v| serde_json::value::RawValue::from_string(v).unwrap())
}
_ => todo!(),
}
})
}
fn subscribe_raw<'a>(
&'a self,
_sub: &'a str,
_params: Option<Box<serde_json::value::RawValue>>,
_unsub: &'a str,
) -> super::rpc::RawRpcFuture<'a, super::rpc::RawRpcSubscription> {
Box::pin(async {
let mut data = self.data.lock().await;
let values: RpcResult<Vec<RpcResult<Box<RawValue>>>> =
data.subscription.read().await.map(|v| {
v.into_iter()
.map(|v| {
v.map(|v| serde_json::value::RawValue::from_string(v).unwrap())
})
.collect::<Vec<RpcResult<Box<RawValue>>>>()
});
values.map(|v| RawRpcSubscription {
stream: futures::stream::iter(v).boxed(),
id: Some("ID".to_string()),
})
})
}
}
// Define dummy config
enum Conf {}
impl Config for Conf {
type Hash = crate::utils::H256;
type AccountId = crate::utils::AccountId32;
type Address = crate::utils::MultiAddress<Self::AccountId, ()>;
type Signature = crate::utils::MultiSignature;
type Hasher = crate::config::substrate::BlakeTwo256;
type Header = crate::config::substrate::SubstrateHeader<u32, Self::Hasher>;
type ExtrinsicParams = DefaultExtrinsicParams<Self>;
type AssetId = u32;
}
use crate::backend::Backend;
fn client_runtime_version(num: u32) -> crate::client::RuntimeVersion {
crate::client::RuntimeVersion {
spec_version: num,
transaction_version: num,
}
}
fn runtime_version(num: u32) -> RuntimeVersion {
RuntimeVersion {
spec_version: num,
transaction_version: num,
other: HashMap::new(),
}
}
fn bytes(str: &str) -> RpcResult<Option<Bytes>> {
Ok(Some(Bytes(str.into())))
}
fn storage_response<K: Into<Vec<u8>>, V: Into<Vec<u8>>>(key: K, value: V) -> StorageResponse
where
Vec<u8>: From<K>,
{
StorageResponse {
key: key.into(),
value: value.into(),
}
}
async fn build_mock_client<
'a,
T: Serialize,
D: IntoIterator<Item = (&'a str, RpcResult<T>)>,
S: IntoIterator<Item = RpcResult<Vec<RpcResult<T>>>>,
>(
table_data: D,
subscription_data: S,
) -> RpcClient {
let data = Data {
request: MockDataTable::from_iter(table_data),
subscription: Subscription::from_iter(subscription_data).await,
};
RpcClient::new(MockRpcClientStorage {
data: Arc::new(Mutex::new(data)),
})
}
#[tokio::test]
async fn storage_fetch_values() {
let mock_data = vec![
("ID1", bytes("Data1")),
(
"ID2",
Err(RpcError::DisconnectedWillReconnect(
"Reconnecting".to_string(),
)),
),
("ID2", bytes("Data2")),
(
"ID3",
Err(RpcError::DisconnectedWillReconnect(
"Reconnecting".to_string(),
)),
),
("ID3", bytes("Data3")),
];
let rpc_client = build_mock_client(mock_data, vec![]).await;
let backend: LegacyBackend<Conf> = LegacyBackend::builder().build(rpc_client);
// Test
let response = backend
.storage_fetch_values(
["ID1".into(), "ID2".into(), "ID3".into()].into(),
crate::utils::H256::random(),
)
.await
.unwrap();
let response = response
.map(|x| x.unwrap())
.collect::<Vec<StorageResponse>>()
.await;
let expected = vec![
storage_response("ID1", "Data1"),
storage_response("ID2", "Data2"),
storage_response("ID3", "Data3"),
];
assert_eq!(expected, response)
}
#[tokio::test]
async fn storage_fetch_value() {
// Setup
let mock_data = [
(
"ID1",
Err(RpcError::DisconnectedWillReconnect(
"Reconnecting".to_string(),
)),
),
("ID1", bytes("Data1")),
];
let rpc_client = build_mock_client(mock_data, vec![]).await;
// Test
let backend: LegacyBackend<Conf> = LegacyBackend::builder().build(rpc_client);
let response = backend
.storage_fetch_value("ID1".into(), crate::utils::H256::random())
.await
.unwrap();
let response = response.unwrap();
assert_eq!("Data1".to_owned(), String::from_utf8(response).unwrap())
}
#[tokio::test]
/// This test should cover the logic of the following methods:
/// - `genesis_hash`
/// - `block_header`
/// - `block_body`
/// - `latest_finalized_block`
/// - `current_runtime_version`
/// - `current_runtime_version`
/// - `call`
/// The test covers them because they follow the simple pattern of:
/// ```no_run
/// async fn THE_THING(&self) -> Result<T::Hash, Error> {
/// retry(|| <DO THE THING> ).await
/// }
/// ```
async fn simple_fetch() {
let hash = crate::utils::H256::random();
let mock_data = vec![
(
"chain_getBlockHash",
Err(RpcError::DisconnectedWillReconnect(
"Reconnecting".to_string(),
)),
),
("chain_getBlockHash", Ok(Some(hash))),
];
let rpc_client = build_mock_client(mock_data, vec![]).await;
// Test
let backend: LegacyBackend<Conf> = LegacyBackend::builder().build(rpc_client);
let response = backend.genesis_hash().await.unwrap();
assert_eq!(hash, response)
}
#[tokio::test]
/// This test should cover the logic of the following methods:
/// - `stream_runtime_version`
/// - `stream_all_block_headers`
/// - `stream_best_block_headers`
/// The test covers them because they follow the simple pattern of:
/// ```no_run
/// async fn stream_the_thing(
/// &self,
/// ) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
/// let methods = self.methods.clone();
/// let retry_sub = retry_stream(move || {
/// let methods = methods.clone();
/// Box::pin(async move {
/// methods.do_the_thing().await?
/// });
/// Ok(StreamOf(Box::pin(sub)))
/// })
/// })
/// .await?;
/// Ok(retry_sub)
/// }
/// ```
async fn stream_simple() {
let mock_subscription_data = vec![
Ok(vec![
Ok(runtime_version(0)),
Err(RpcError::DisconnectedWillReconnect(
"Reconnecting".to_string(),
)),
Ok(runtime_version(1)),
]),
Ok(vec![
Err(RpcError::DisconnectedWillReconnect(
"Reconnecting".to_string(),
)),
Ok(runtime_version(2)),
Ok(runtime_version(3)),
]),
Ok(vec![
Ok(runtime_version(4)),
Ok(runtime_version(5)),
Err(RpcError::RequestRejected("Reconnecting".to_string())),
]),
];
let rpc_client = build_mock_client(vec![], mock_subscription_data).await;
// Test
let backend: LegacyBackend<Conf> = LegacyBackend::builder().build(rpc_client);
let mut results = backend.stream_runtime_version().await.unwrap();
let mut expected = VecDeque::from(vec![
Ok::<crate::client::RuntimeVersion, crate::Error>(client_runtime_version(0)),
Ok(client_runtime_version(4)),
Ok(client_runtime_version(5)),
]);
while let Some(res) = results.next().await {
if res.is_ok() {
assert_eq!(expected.pop_front().unwrap().unwrap(), res.unwrap())
} else {
assert!(matches!(
res,
Err(crate::Error::Rpc(RpcError::RequestRejected(_)))
))
}
}
assert!(expected.is_empty());
assert!(results.next().await.is_none())
}
}
}
subxt/src/backend/rpc/mod.rs
+2 -2
View File
@@ -65,8 +65,8 @@ crate::macros::cfg_unstable_light_client! {
}
crate::macros::cfg_reconnecting_rpc_client! {
mod reconnecting_jsonrpsee_impl;
pub use reconnecting_jsonrpsee_ws_client as reconnecting_rpc_client;
/// reconnecting rpc client.
pub mod reconnecting_rpc_client;
}
mod rpc_client;
subxt/src/backend/rpc/reconnecting_jsonrpsee_impl.rs
-52
View File
@@ -1,52 +0,0 @@
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
// see LICENSE for license details.
use super::{RawRpcFuture, RawRpcSubscription, RpcClientT};
use crate::error::RpcError;
use futures::{FutureExt, StreamExt, TryStreamExt};
use reconnecting_jsonrpsee_ws_client::SubscriptionId;
use serde_json::value::RawValue;
impl RpcClientT for reconnecting_jsonrpsee_ws_client::Client {
fn request_raw<'a>(
&'a self,
method: &'a str,
params: Option<Box<RawValue>>,
) -> RawRpcFuture<'a, Box<RawValue>> {
async {
self.request_raw(method.to_string(), params)
.await
.map_err(|e| RpcError::ClientError(Box::new(e)))
}
.boxed()
}
fn subscribe_raw<'a>(
&'a self,
sub: &'a str,
params: Option<Box<RawValue>>,
unsub: &'a str,
) -> RawRpcFuture<'a, RawRpcSubscription> {
async {
let sub = self
.subscribe_raw(sub.to_string(), params, unsub.to_string())
.await
.map_err(|e| RpcError::ClientError(Box::new(e)))?;
let id = match sub.id() {
SubscriptionId::Num(n) => n.to_string(),
SubscriptionId::Str(s) => s.to_string(),
};
let stream = sub
.map_err(|e| RpcError::DisconnectedWillReconnect(e.to_string()))
.boxed();
Ok(RawRpcSubscription {
stream,
id: Some(id),
})
}
.boxed()
}
}
subxt/src/backend/rpc/reconnecting_rpc_client/mod.rs
+640
View File
@@ -0,0 +1,640 @@
// Copyright 2019-2024 Parity Technologies (UK) Ltd.
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
// see LICENSE for license details.
//! # reconnecting-jsonrpsee-ws-client
//!
//! A simple reconnecting JSON-RPC WebSocket client for subxt which
//! automatically reconnects when the connection is lost but
//! it doesn't retain subscriptions and pending method calls when it reconnects.
//!
//! The logic which action to take for individual calls and subscriptions are
//! handled by the subxt backend implementations.
//!
//! # Example
//!
//! ```no_run
//! use std::time::Duration;
//! use futures::StreamExt;
//! use subxt::backend::rpc::reconnecting_rpc_client::{RpcClient, ExponentialBackoff};
//! use subxt::{OnlineClient, PolkadotConfig};
//!
//! #[tokio::main]
//! async fn main() {
//! let rpc = RpcClient::builder()
//! .retry_policy(ExponentialBackoff::from_millis(100).max_delay(Duration::from_secs(10)))
//! .build("ws://localhost:9944".to_string())
//! .await
//! .unwrap();
//!
//! let subxt_client: OnlineClient<PolkadotConfig> = OnlineClient::from_rpc_client(rpc.clone()).await.unwrap();
//! let mut blocks_sub = subxt_client.blocks().subscribe_finalized().await.unwrap();
//!
//! while let Some(block) = blocks_sub.next().await {
//! let block = match block {
//! Ok(b) => b,
//! Err(e) => {
//! if e.is_disconnected_will_reconnect() {
//! println!("The RPC connection was lost and we may have missed a few blocks");
//! continue;
//! } else {
//! panic!("Error: {}", e);
//! }
//! }
//! };
//! println!("Block #{} ({})", block.number(), block.hash());
//! }
//! }
//! ```
mod platform;
#[cfg(test)]
mod tests;
mod utils;
use std::{
pin::Pin,
sync::Arc,
task::{self, Poll},
time::Duration,
};
use super::{RawRpcFuture, RawRpcSubscription, RpcClientT};
use crate::error::RpcError as SubxtRpcError;
use finito::Retry;
use futures::{FutureExt, Stream, StreamExt, TryStreamExt};
use jsonrpsee::core::{
client::{
Client as WsClient, ClientT, Subscription as RpcSubscription, SubscriptionClientT,
SubscriptionKind,
},
traits::ToRpcParams,
};
use platform::spawn;
use serde_json::value::RawValue;
use tokio::sync::{
mpsc::{self, UnboundedReceiver, UnboundedSender},
oneshot, Notify,
};
use utils::display_close_reason;
// re-exports
pub use finito::{ExponentialBackoff, FibonacciBackoff, FixedInterval};
pub use jsonrpsee::core::client::IdKind;
pub use jsonrpsee::{core::client::error::Error as RpcError, rpc_params, types::SubscriptionId};
#[cfg(feature = "native")]
pub use jsonrpsee::ws_client::{HeaderMap, PingConfig};
const LOG_TARGET: &str = "subxt-reconnecting-rpc-client";
/// Method result.
pub type MethodResult = Result<Box<RawValue>, Error>;
/// Subscription result.
pub type SubscriptionResult = Result<Box<RawValue>, DisconnectedWillReconnect>;
/// The connection was closed, reconnect initiated and the subscription was dropped.
#[derive(Debug, thiserror::Error)]
#[error("The connection was closed because of `{0:?}` and reconnect initiated")]
pub struct DisconnectedWillReconnect(String);
/// New-type pattern which implements [`ToRpcParams`] that is required by jsonrpsee.
#[derive(Debug, Clone)]
struct RpcParams(Option<Box<RawValue>>);
impl ToRpcParams for RpcParams {
fn to_rpc_params(self) -> Result<Option<Box<RawValue>>, serde_json::Error> {
Ok(self.0)
}
}
#[derive(Debug)]
enum Op {
Call {
method: String,
params: RpcParams,
send_back: oneshot::Sender<MethodResult>,
},
Subscription {
subscribe_method: String,
params: RpcParams,
unsubscribe_method: String,
send_back: oneshot::Sender<Result<Subscription, Error>>,
},
}
/// Error that can occur when for a RPC call or subscription.
#[derive(Debug, thiserror::Error)]
pub enum Error {
/// The client was dropped by the user.
#[error("The client was dropped")]
Dropped,
/// The connection was closed and reconnect initiated.
#[error(transparent)]
DisconnectedWillReconnect(#[from] DisconnectedWillReconnect),
/// Other rpc error.
#[error("{0}")]
RpcError(RpcError),
}
/// Represent a single subscription.
pub struct Subscription {
id: SubscriptionId<'static>,
stream: mpsc::UnboundedReceiver<SubscriptionResult>,
}
impl Subscription {
/// Returns the next notification from the stream.
/// This may return `None` if the subscription has been terminated,
/// which may happen if the channel becomes full or is dropped.
///
/// **Note:** This has an identical signature to the [`StreamExt::next`]
/// method (and delegates to that). Import [`StreamExt`] if you'd like
/// access to other stream combinator methods.
#[allow(clippy::should_implement_trait)]
pub async fn next(&mut self) -> Option<SubscriptionResult> {
StreamExt::next(self).await
}
/// Get the subscription ID.
pub fn id(&self) -> SubscriptionId<'static> {
self.id.clone()
}
}
impl Stream for Subscription {
type Item = SubscriptionResult;
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut task::Context<'_>,
) -> task::Poll<Option<Self::Item>> {
match self.stream.poll_recv(cx) {
Poll::Ready(Some(msg)) => Poll::Ready(Some(msg)),
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
}
}
}
impl std::fmt::Debug for Subscription {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Subscription")
.field("id", &self.id)
.finish()
}
}
/// JSON-RPC client that reconnects automatically and may loose
/// subscription notifications when it reconnects.
#[derive(Clone, Debug)]
pub struct RpcClient {
tx: mpsc::UnboundedSender<Op>,
}
/// Builder for [`Client`].
#[derive(Clone, Debug)]
pub struct RpcClientBuilder<P> {
max_request_size: u32,
max_response_size: u32,
retry_policy: P,
#[cfg(feature = "native")]
ping_config: Option<PingConfig>,
#[cfg(feature = "native")]
// web doesn't support custom headers
// https://stackoverflow.com/a/4361358/6394734
headers: HeaderMap,
max_redirections: u32,
id_kind: IdKind,
max_log_len: u32,
max_concurrent_requests: u32,
request_timeout: Duration,
connection_timeout: Duration,
}
impl Default for RpcClientBuilder<ExponentialBackoff> {
fn default() -> Self {
Self {
max_request_size: 10 * 1024 * 1024,
max_response_size: 10 * 1024 * 1024,
retry_policy: ExponentialBackoff::from_millis(10).max_delay(Duration::from_secs(60)),
#[cfg(feature = "native")]
ping_config: Some(PingConfig::new()),
#[cfg(feature = "native")]
headers: HeaderMap::new(),
max_redirections: 5,
id_kind: IdKind::Number,
max_log_len: 1024,
max_concurrent_requests: 1024,
request_timeout: Duration::from_secs(60),
connection_timeout: Duration::from_secs(10),
}
}
}
impl RpcClientBuilder<ExponentialBackoff> {
/// Create a new builder.
pub fn new() -> Self {
Self::default()
}
}
impl<P> RpcClientBuilder<P>
where
P: Iterator<Item = Duration> + Send + Sync + 'static + Clone,
{
/// Configure the min response size a for websocket message.
///
/// Default: 10MB
pub fn max_request_size(mut self, max: u32) -> Self {
self.max_request_size = max;
self
}
/// Configure the max response size a for websocket message.
///
/// Default: 10MB
pub fn max_response_size(mut self, max: u32) -> Self {
self.max_response_size = max;
self
}
/// Set the max number of redirections to perform until a connection is regarded as failed.
///
/// Default: 5
pub fn max_redirections(mut self, redirect: u32) -> Self {
self.max_redirections = redirect;
self
}
/// Configure how many concurrent method calls are allowed.
///
/// Default: 1024
pub fn max_concurrent_requests(mut self, max: u32) -> Self {
self.max_concurrent_requests = max;
self
}
/// Configure how long until a method call is regarded as failed.
///
/// Default: 1 minute
pub fn request_timeout(mut self, timeout: Duration) -> Self {
self.request_timeout = timeout;
self
}
/// Set connection timeout for the WebSocket handshake
///
/// Default: 10 seconds
pub fn connection_timeout(mut self, timeout: Duration) -> Self {
self.connection_timeout = timeout;
self
}
/// Configure the data type of the request object ID
///
/// Default: number
pub fn id_format(mut self, kind: IdKind) -> Self {
self.id_kind = kind;
self
}
/// Set maximum length for logging calls and responses.
/// Logs bigger than this limit will be truncated.
///
/// Default: 1024
pub fn set_max_logging_length(mut self, max: u32) -> Self {
self.max_log_len = max;
self
}
#[cfg(feature = "native")]
#[cfg_attr(docsrs, doc(cfg(feature = "native")))]
/// Configure custom headers to use in the WebSocket handshake.
pub fn set_headers(mut self, headers: HeaderMap) -> Self {
self.headers = headers;
self
}
/// Configure which retry policy to use when a connection is lost.
///
/// Default: Exponential backoff 10ms
pub fn retry_policy<T>(self, retry_policy: T) -> RpcClientBuilder<T> {
RpcClientBuilder {
max_request_size: self.max_request_size,
max_response_size: self.max_response_size,
retry_policy,
#[cfg(feature = "native")]
ping_config: self.ping_config,
#[cfg(feature = "native")]
headers: self.headers,
max_redirections: self.max_redirections,
max_log_len: self.max_log_len,
id_kind: self.id_kind,
max_concurrent_requests: self.max_concurrent_requests,
request_timeout: self.request_timeout,
connection_timeout: self.connection_timeout,
}
}
#[cfg(feature = "native")]
#[cfg_attr(docsrs, doc(cfg(feature = "native")))]
/// Configure the WebSocket ping/pong interval.
///
/// Default: 30 seconds.
pub fn enable_ws_ping(mut self, ping_config: PingConfig) -> Self {
self.ping_config = Some(ping_config);
self
}
#[cfg(feature = "native")]
#[cfg_attr(docsrs, doc(cfg(feature = "native")))]
/// Disable WebSocket ping/pongs.
///
/// Default: 30 seconds.
pub fn disable_ws_ping(mut self) -> Self {
self.ping_config = None;
self
}
/// Build and connect to the target.
pub async fn build(self, url: String) -> Result<RpcClient, RpcError> {
let (tx, rx) = mpsc::unbounded_channel();
let client = Retry::new(self.retry_policy.clone(), || {
platform::ws_client(url.as_ref(), &self)
})
.await?;
platform::spawn(background_task(client, rx, url, self));
Ok(RpcClient { tx })
}
}
impl RpcClient {
/// Create a builder.
pub fn builder() -> RpcClientBuilder<ExponentialBackoff> {
RpcClientBuilder::new()
}
/// Perform a JSON-RPC method call.
pub async fn request(
&self,
method: String,
params: Option<Box<RawValue>>,
) -> Result<Box<RawValue>, Error> {
let (tx, rx) = oneshot::channel();
self.tx
.send(Op::Call {
method,
params: RpcParams(params),
send_back: tx,
})
.map_err(|_| Error::Dropped)?;
rx.await.map_err(|_| Error::Dropped)?
}
/// Perform a JSON-RPC subscription.
pub async fn subscribe(
&self,
subscribe_method: String,
params: Option<Box<RawValue>>,
unsubscribe_method: String,
) -> Result<Subscription, Error> {
let (tx, rx) = oneshot::channel();
self.tx
.send(Op::Subscription {
subscribe_method,
params: RpcParams(params),
unsubscribe_method,
send_back: tx,
})
.map_err(|_| Error::Dropped)?;
rx.await.map_err(|_| Error::Dropped)?
}
}
impl RpcClientT for RpcClient {
fn request_raw<'a>(
&'a self,
method: &'a str,
params: Option<Box<RawValue>>,
) -> RawRpcFuture<'a, Box<RawValue>> {
async {
self.request(method.to_string(), params)
.await
.map_err(|e| SubxtRpcError::DisconnectedWillReconnect(e.to_string()))
}
.boxed()
}
fn subscribe_raw<'a>(
&'a self,
sub: &'a str,
params: Option<Box<RawValue>>,
unsub: &'a str,
) -> RawRpcFuture<'a, RawRpcSubscription> {
async {
let sub = self
.subscribe(sub.to_string(), params, unsub.to_string())
.await
.map_err(|e| SubxtRpcError::ClientError(Box::new(e)))?;
let id = match sub.id() {
SubscriptionId::Num(n) => n.to_string(),
SubscriptionId::Str(s) => s.to_string(),
};
let stream = sub
.map_err(|e| SubxtRpcError::DisconnectedWillReconnect(e.to_string()))
.boxed();
Ok(RawRpcSubscription {
stream,
id: Some(id),
})
}
.boxed()
}
}
async fn background_task<P>(
mut client: Arc<WsClient>,
mut rx: UnboundedReceiver<Op>,
url: String,
client_builder: RpcClientBuilder<P>,
) where
P: Iterator<Item = Duration> + Send + 'static + Clone,
{
let disconnect = Arc::new(tokio::sync::Notify::new());
loop {
tokio::select! {
// An incoming JSON-RPC call to dispatch.
next_message = rx.recv() => {
match next_message {
None => break,
Some(op) => {
spawn(dispatch_call(client.clone(), op, disconnect.clone()));
}
};
}
// The connection was terminated and try to reconnect.
_ = client.on_disconnect() => {
let params = ReconnectParams {
url: &url,
client_builder: &client_builder,
close_reason: client.disconnect_reason().await,
};
client = match reconnect(params).await {
Ok(client) => client,
Err(e) => {
tracing::debug!(target: LOG_TARGET, "Failed to reconnect: {e}; terminating the connection");
break;
}
};
}
}
}
disconnect.notify_waiters();
}
async fn dispatch_call(client: Arc<WsClient>, op: Op, on_disconnect: Arc<tokio::sync::Notify>) {
match op {
Op::Call {
method,
params,
send_back,
} => {
match client.request::<Box<RawValue>, _>(&method, params).await {
Ok(rp) => {
// Fails only if the request is dropped by the client.
let _ = send_back.send(Ok(rp));
}
Err(RpcError::RestartNeeded(e)) => {
// Fails only if the request is dropped by the client.
let _ = send_back.send(Err(DisconnectedWillReconnect(e.to_string()).into()));
}
Err(e) => {
// Fails only if the request is dropped by the client.
let _ = send_back.send(Err(Error::RpcError(e)));
}
}
}
Op::Subscription {
subscribe_method,
params,
unsubscribe_method,
send_back,
} => {
match client
.subscribe::<Box<RawValue>, _>(
&subscribe_method,
params.clone(),
&unsubscribe_method,
)
.await
{
Ok(sub) => {
let (tx, rx) = mpsc::unbounded_channel();
let sub_id = match sub.kind() {
SubscriptionKind::Subscription(id) => id.clone().into_owned(),
_ => unreachable!("No method subscriptions possible in this crate; qed"),
};
platform::spawn(subscription_handler(
tx.clone(),
sub,
on_disconnect.clone(),
client.clone(),
));
let stream = Subscription {
id: sub_id,
stream: rx,
};
// Fails only if the request is dropped by the client.
let _ = send_back.send(Ok(stream));
}
Err(RpcError::RestartNeeded(e)) => {
// Fails only if the request is dropped by the client.
let _ = send_back.send(Err(DisconnectedWillReconnect(e.to_string()).into()));
}
Err(e) => {
// Fails only if the request is dropped.
let _ = send_back.send(Err(Error::RpcError(e)));
}
}
}
}
}
/// Handler for each individual subscription.
async fn subscription_handler(
sub_tx: UnboundedSender<SubscriptionResult>,
mut rpc_sub: RpcSubscription<Box<RawValue>>,
client_closed: Arc<Notify>,
client: Arc<WsClient>,
) {
loop {
tokio::select! {
next_msg = rpc_sub.next() => {
let Some(notif) = next_msg else {
let close = client.disconnect_reason().await;
_ = sub_tx.send(Err(DisconnectedWillReconnect(close.to_string())));
break;
};
let msg = notif.expect("RawValue is valid JSON; qed");
// Fails only if subscription was closed by the user.
if sub_tx.send(Ok(msg)).is_err() {
break;
}
}
// This channel indices whether the subscription was closed by user.
_ = sub_tx.closed() => {
break;
}
// This channel indicates whether the main task has been closed.
// at this point no further messages are processed.
_ = client_closed.notified() => {
break;
}
}
}
}
struct ReconnectParams<'a, P> {
url: &'a str,
client_builder: &'a RpcClientBuilder<P>,
close_reason: RpcError,
}
async fn reconnect<P>(params: ReconnectParams<'_, P>) -> Result<Arc<WsClient>, RpcError>
where
P: Iterator<Item = Duration> + Send + 'static + Clone,
{
let ReconnectParams {
url,
client_builder,
close_reason,
} = params;
let retry_policy = client_builder.retry_policy.clone();
tracing::debug!(target: LOG_TARGET, "Connection to {url} was closed: `{}`; starting to reconnect", display_close_reason(&close_reason));
let client = Retry::new(retry_policy.clone(), || {
platform::ws_client(url, client_builder)
})
.await?;
tracing::debug!(target: LOG_TARGET, "Connection to {url} was successfully re-established");
Ok(client)
}
subxt/src/backend/rpc/reconnecting_rpc_client/platform.rs
+83
View File
@@ -0,0 +1,83 @@
// Copyright 2019-2024 Parity Technologies (UK) Ltd.
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
// see LICENSE for license details.
use crate::backend::rpc::reconnecting_rpc_client::{RpcClientBuilder, RpcError};
use jsonrpsee::core::client::Client;
use std::sync::Arc;
#[cfg(feature = "native")]
pub use tokio::spawn;
#[cfg(feature = "web")]
pub use wasm_bindgen_futures::spawn_local as spawn;
#[cfg(feature = "native")]
pub async fn ws_client<P>(
url: &str,
builder: &RpcClientBuilder<P>,
) -> Result<Arc<Client>, RpcError> {
use jsonrpsee::ws_client::WsClientBuilder;
let RpcClientBuilder {
max_request_size,
max_response_size,
ping_config,
headers,
max_redirections,
id_kind,
max_concurrent_requests,
max_log_len,
request_timeout,
connection_timeout,
..
} = builder;
let mut ws_client_builder = WsClientBuilder::new()
.max_request_size(*max_request_size)
.max_response_size(*max_response_size)
.set_headers(headers.clone())
.max_redirections(*max_redirections as usize)
.max_buffer_capacity_per_subscription(tokio::sync::Semaphore::MAX_PERMITS)
.max_concurrent_requests(*max_concurrent_requests as usize)
.set_max_logging_length(*max_log_len)
.set_tcp_no_delay(true)
.request_timeout(*request_timeout)
.connection_timeout(*connection_timeout)
.id_format(*id_kind);
if let Some(ping) = ping_config {
ws_client_builder = ws_client_builder.enable_ws_ping(*ping);
}
let client = ws_client_builder.build(url).await?;
Ok(Arc::new(client))
}
#[cfg(feature = "web")]
pub async fn ws_client<P>(
url: &str,
builder: &RpcClientBuilder<P>,
) -> Result<Arc<Client>, RpcError> {
use jsonrpsee::wasm_client::WasmClientBuilder;
let RpcClientBuilder {
id_kind,
max_concurrent_requests,
max_log_len,
request_timeout,
..
} = builder;
let ws_client_builder = WasmClientBuilder::new()
.max_buffer_capacity_per_subscription(tokio::sync::Semaphore::MAX_PERMITS)
.max_concurrent_requests(*max_concurrent_requests as usize)
.set_max_logging_length(*max_log_len)
.request_timeout(*request_timeout)
.id_format(*id_kind);
let client = ws_client_builder.build(url).await?;
Ok(Arc::new(client))
}
subxt/src/backend/rpc/reconnecting_rpc_client/tests.rs
+270
View File
@@ -0,0 +1,270 @@
// Copyright 2019-2024 Parity Technologies (UK) Ltd.
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
// see LICENSE for license details.
use super::*;
use futures::{future::Either, FutureExt};
use jsonrpsee::core::BoxError;
use jsonrpsee::server::{
http, stop_channel, ws, ConnectionGuard, ConnectionState, HttpRequest, HttpResponse, RpcModule,
RpcServiceBuilder, ServerConfig, SubscriptionMessage,
};
#[tokio::test]
async fn call_works() {
let (_handle, addr) = run_server().await.unwrap();
let client = RpcClient::builder().build(addr).await.unwrap();
assert!(client.request("say_hello".to_string(), None).await.is_ok(),)
}
#[tokio::test]
async fn sub_works() {
let (_handle, addr) = run_server().await.unwrap();
let client = RpcClient::builder()
.retry_policy(ExponentialBackoff::from_millis(50))
.build(addr)
.await
.unwrap();
let mut sub = client
.subscribe(
"subscribe_lo".to_string(),
None,
"unsubscribe_lo".to_string(),
)
.await
.unwrap();
assert!(sub.next().await.is_some());
}
#[tokio::test]
async fn sub_with_reconnect() {
let (handle, addr) = run_server().await.unwrap();
let client = RpcClient::builder().build(addr.clone()).await.unwrap();
let mut sub = client
.subscribe(
"subscribe_lo".to_string(),
None,
"unsubscribe_lo".to_string(),
)
.await
.unwrap();
let _ = handle.send(());
// Hack to wait for the server to restart.
tokio::time::sleep(Duration::from_millis(100)).await;
assert!(matches!(sub.next().await, Some(Ok(_))));
assert!(matches!(
sub.next().await,
Some(Err(DisconnectedWillReconnect(_)))
));
// Restart the server.
let (_handle, _) = run_server_with_settings(Some(&addr), false).await.unwrap();
// Hack to wait for the server to restart.
tokio::time::sleep(Duration::from_millis(100)).await;
// Subscription should work after reconnect.
let mut sub = client
.subscribe(
"subscribe_lo".to_string(),
None,
"unsubscribe_lo".to_string(),
)
.await
.unwrap();
assert!(matches!(sub.next().await, Some(Ok(_))));
}
#[tokio::test]
async fn call_with_reconnect() {
let (handle, addr) = run_server_with_settings(None, true).await.unwrap();
let client = Arc::new(RpcClient::builder().build(addr.clone()).await.unwrap());
let req_fut = client.request("say_hello".to_string(), None).boxed();
let timeout_fut = tokio::time::sleep(Duration::from_secs(5));
// If the call isn't replied in 5 secs then it's regarded as it's still pending.
let req_fut = match futures::future::select(Box::pin(timeout_fut), req_fut).await {
Either::Left((_, f)) => f,
Either::Right(_) => panic!("RPC call finished"),
};
// Close the connection with a pending call.
let _ = handle.send(());
// Restart the server
let (_handle, _) = run_server_with_settings(Some(&addr), false).await.unwrap();
// Hack to wait for the server to restart.
tokio::time::sleep(Duration::from_millis(100)).await;
// This call should fail because reconnect.
assert!(req_fut.await.is_err());
// Future call should work after reconnect.
assert!(client.request("say_hello".to_string(), None).await.is_ok());
}
async fn run_server() -> Result<(tokio::sync::broadcast::Sender<()>, String), BoxError> {
run_server_with_settings(None, false).await
}
async fn run_server_with_settings(
url: Option<&str>,
dont_respond_to_method_calls: bool,
) -> Result<(tokio::sync::broadcast::Sender<()>, String), BoxError> {
use jsonrpsee::server::HttpRequest;
let sockaddr = match url {
Some(url) => url.strip_prefix("ws://").unwrap(),
None => "127.0.0.1:0",
};
let mut i = 0;
let listener = loop {
if let Ok(l) = tokio::net::TcpListener::bind(sockaddr).await {
break l;
}
tokio::time::sleep(Duration::from_millis(100)).await;
if i >= 10 {
panic!("Addr already in use");
}
i += 1;
};
let mut module = RpcModule::new(());
if dont_respond_to_method_calls {
module.register_async_method("say_hello", |_, _, _| async {
futures::future::pending::<()>().await;
"timeout"
})?;
} else {
module.register_async_method("say_hello", |_, _, _| async { "lo" })?;
}
module.register_subscription(
"subscribe_lo",
"subscribe_lo",
"unsubscribe_lo",
|_params, pending, _ctx, _| async move {
let sink = pending.accept().await.unwrap();
let i = 0;
loop {
if sink
.send(SubscriptionMessage::from_json(&i).unwrap())
.await
.is_err()
{
break;
}
tokio::time::sleep(std::time::Duration::from_secs(6)).await;
}
},
)?;
let (tx, mut rx) = tokio::sync::broadcast::channel(4);
let tx2 = tx.clone();
let (stop_handle, server_handle) = stop_channel();
let addr = listener.local_addr().expect("Could not find local addr");
tokio::spawn(async move {
loop {
let sock = tokio::select! {
res = listener.accept() => {
match res {
Ok((stream, _remote_addr)) => stream,
Err(e) => {
tracing::error!("Failed to accept connection: {:?}", e);
continue;
}
}
}
_ = rx.recv() => {
break
}
};
let module = module.clone();
let rx2 = tx2.subscribe();
let tx2 = tx2.clone();
let stop_handle2 = stop_handle.clone();
let svc = tower::service_fn(move |req: HttpRequest<hyper::body::Incoming>| {
let module = module.clone();
let tx = tx2.clone();
let stop_handle = stop_handle2.clone();
let conn_permit = ConnectionGuard::new(1).try_acquire().unwrap();
if ws::is_upgrade_request(&req) {
let rpc_service = RpcServiceBuilder::new();
let conn = ConnectionState::new(stop_handle, 1, conn_permit);
async move {
let mut rx = tx.subscribe();
let (rp, conn_fut) =
ws::connect(req, ServerConfig::default(), module, conn, rpc_service)
.await
.unwrap();
tokio::spawn(async move {
tokio::select! {
_ = conn_fut => (),
_ = rx.recv() => {},
}
});
Ok::<_, BoxError>(rp)
}
.boxed()
} else {
async { Ok(http::response::denied()) }.boxed()
}
});
tokio::spawn(serve_with_graceful_shutdown(sock, svc, rx2));
}
drop(server_handle);
});
Ok((tx, format!("ws://{}", addr)))
}
async fn serve_with_graceful_shutdown<S, B, I>(
io: I,
service: S,
mut rx: tokio::sync::broadcast::Receiver<()>,
) where
S: tower::Service<HttpRequest<hyper::body::Incoming>, Response = HttpResponse<B>>
+ Clone
+ Send
+ 'static,
S::Future: Send,
S::Response: Send,
S::Error: Into<BoxError>,
B: http_body::Body<Data = hyper::body::Bytes> + Send + 'static,
B::Error: Into<BoxError>,
I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
{
if let Err(e) =
jsonrpsee::server::serve_with_graceful_shutdown(io, service, rx.recv().map(|_| ())).await
{
tracing::error!("Error while serving: {:?}", e);
}
}
subxt/src/backend/rpc/reconnecting_rpc_client/utils.rs
+14
View File
@@ -0,0 +1,14 @@
// Copyright 2019-2024 Parity Technologies (UK) Ltd.
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
// see LICENSE for license details.
//! Utils.
use crate::backend::rpc::reconnecting_rpc_client::RpcError;
pub fn display_close_reason(err: &RpcError) -> String {
match err {
RpcError::RestartNeeded(e) => e.to_string(),
other => other.to_string(),
}
}
subxt/src/backend/unstable/follow_stream.rs
+12
View File
@@ -148,6 +148,12 @@ impl<Hash> Stream for FollowStream<Hash> {
continue;
}
Poll::Ready(Err(e)) => {
// Re-start if a reconnecting backend was enabled.
if e.is_disconnected_will_reconnect() {
this.stream = InnerStreamState::Stopped;
continue;
}
// Finish forever if there's an error, passing it on.
this.stream = InnerStreamState::Finished;
return Poll::Ready(Some(Err(e)));
@@ -182,6 +188,12 @@ impl<Hash> Stream for FollowStream<Hash> {
return Poll::Ready(Some(Ok(FollowStreamMsg::Event(ev))));
}
Poll::Ready(Some(Err(e))) => {
// Re-start if a reconnecting backend was enabled.
if e.is_disconnected_will_reconnect() {
this.stream = InnerStreamState::Stopped;
continue;
}
// Finish forever if there's an error, passing it on.
this.stream = InnerStreamState::Finished;
return Poll::Ready(Some(Err(e)));
subxt/src/backend/unstable/follow_stream_driver.rs
+201 -3
View File
@@ -5,7 +5,7 @@
use super::follow_stream_unpin::{BlockRef, FollowStreamMsg, FollowStreamUnpin};
use crate::backend::unstable::rpc_methods::{FollowEvent, Initialized, RuntimeEvent};
use crate::config::BlockHash;
use crate::error::Error;
use crate::error::{Error, RpcError};
use futures::stream::{Stream, StreamExt};
use std::collections::{HashMap, HashSet, VecDeque};
use std::ops::DerefMut;
@@ -267,8 +267,9 @@ impl<Hash: BlockHash> Shared<Hash> {
shared.seen_runtime_events.clear();
init_message.finalized_block_hashes =
finalized_ev.finalized_block_hashes.clone();
init_message
.finalized_block_hashes
.clone_from(&finalized_ev.finalized_block_hashes);
if let Some(runtime_ev) = newest_runtime {
init_message.finalized_block_runtime = Some(runtime_ev);
@@ -379,6 +380,103 @@ struct SubscriberDetails<Hash: BlockHash> {
waker: Option<Waker>,
}
/// A stream that subscribes to finalized blocks
/// and indicates whether a block was missed if was restarted.
#[derive(Debug)]
pub struct FollowStreamFinalizedHeads<Hash: BlockHash, F> {
stream: FollowStreamDriverSubscription<Hash>,
sub_id: Option<String>,
last_seen_block: Option<BlockRef<Hash>>,
f: F,
is_done: bool,
}
impl<Hash: BlockHash, F> Unpin for FollowStreamFinalizedHeads<Hash, F> {}
impl<Hash, F> FollowStreamFinalizedHeads<Hash, F>
where
Hash: BlockHash,
F: Fn(FollowEvent<BlockRef<Hash>>) -> Vec<BlockRef<Hash>>,
{
pub fn new(stream: FollowStreamDriverSubscription<Hash>, f: F) -> Self {
Self {
stream,
sub_id: None,
last_seen_block: None,
f,
is_done: false,
}
}
}
impl<Hash, F> Stream for FollowStreamFinalizedHeads<Hash, F>
where
Hash: BlockHash,
F: Fn(FollowEvent<BlockRef<Hash>>) -> Vec<BlockRef<Hash>>,
{
type Item = Result<(String, Vec<BlockRef<Hash>>), Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if self.is_done {
return Poll::Ready(None);
}
loop {
let Some(ev) = futures::ready!(self.stream.poll_next_unpin(cx)) else {
self.is_done = true;
return Poll::Ready(None);
};
let block_refs = match ev {
FollowStreamMsg::Ready(sub_id) => {
self.sub_id = Some(sub_id);
continue;
}
FollowStreamMsg::Event(FollowEvent::Finalized(finalized)) => {
self.last_seen_block = finalized.finalized_block_hashes.last().cloned();
(self.f)(FollowEvent::Finalized(finalized))
}
FollowStreamMsg::Event(FollowEvent::Initialized(mut init)) => {
let prev = self.last_seen_block.take();
self.last_seen_block = init.finalized_block_hashes.last().cloned();
if let Some(p) = prev {
let Some(pos) = init
.finalized_block_hashes
.iter()
.position(|b| b.hash() == p.hash())
else {
return Poll::Ready(Some(Err(RpcError::DisconnectedWillReconnect(
"Missed at least one block when the connection was lost".to_owned(),
)
.into())));
};
// If we got older blocks than `prev`, we need to remove them
// because they should already have been sent at this point.
init.finalized_block_hashes.drain(0..=pos);
}
(self.f)(FollowEvent::Initialized(init))
}
FollowStreamMsg::Event(ev) => (self.f)(ev),
};
if block_refs.is_empty() {
continue;
}
let sub_id = self
.sub_id
.clone()
.expect("Ready is always emitted before any other event");
return Poll::Ready(Some(Ok((sub_id, block_refs))));
}
}
}
#[cfg(test)]
mod test_utils {
use super::super::follow_stream_unpin::test_utils::test_unpin_stream_getter;
@@ -401,6 +499,9 @@ mod test_utils {
#[cfg(test)]
mod test {
use futures::TryStreamExt;
use sp_core::H256;
use super::super::follow_stream::test_utils::{
ev_best_block, ev_finalized, ev_initialized, ev_new_block,
};
@@ -544,4 +645,101 @@ mod test {
];
assert_eq!(evs, expected);
}
#[tokio::test]
async fn subscribe_finalized_blocks_restart_works() {
let mut driver = test_follow_stream_driver_getter(
|| {
[
Ok(ev_initialized(0)),
Ok(ev_new_block(0, 1)),
Ok(ev_best_block(1)),
Ok(ev_finalized([1], [])),
Ok(FollowEvent::Stop),
Ok(ev_initialized(1)),
Ok(ev_finalized([2], [])),
Err(Error::Other("ended".to_owned())),
]
},
10,
);
let handle = driver.handle();
tokio::spawn(async move { while driver.next().await.is_some() {} });
let f = |ev| match ev {
FollowEvent::Finalized(ev) => ev.finalized_block_hashes,
FollowEvent::Initialized(ev) => ev.finalized_block_hashes,
_ => vec![],
};
let stream = FollowStreamFinalizedHeads::new(handle.subscribe(), f);
let evs: Vec<_> = stream.try_collect().await.unwrap();
let expected = vec![
(
"sub_id_0".to_string(),
vec![BlockRef::new(H256::from_low_u64_le(0))],
),
(
"sub_id_0".to_string(),
vec![BlockRef::new(H256::from_low_u64_le(1))],
),
(
"sub_id_5".to_string(),
vec![BlockRef::new(H256::from_low_u64_le(2))],
),
];
assert_eq!(evs, expected);
}
#[tokio::test]
async fn subscribe_finalized_blocks_restart_with_missed_blocks() {
let mut driver = test_follow_stream_driver_getter(
|| {
[
Ok(ev_initialized(0)),
Ok(FollowEvent::Stop),
// Emulate that we missed some blocks.
Ok(ev_initialized(13)),
Ok(ev_finalized([14], [])),
Err(Error::Other("ended".to_owned())),
]
},
10,
);
let handle = driver.handle();
tokio::spawn(async move { while driver.next().await.is_some() {} });
let f = |ev| match ev {
FollowEvent::Finalized(ev) => ev.finalized_block_hashes,
FollowEvent::Initialized(ev) => ev.finalized_block_hashes,
_ => vec![],
};
let evs: Vec<_> = FollowStreamFinalizedHeads::new(handle.subscribe(), f)
.collect()
.await;
assert_eq!(
evs[0].as_ref().unwrap(),
&(
"sub_id_0".to_string(),
vec![BlockRef::new(H256::from_low_u64_le(0))]
)
);
assert!(
matches!(&evs[1], Err(Error::Rpc(RpcError::DisconnectedWillReconnect(e))) if e.contains("Missed at least one block when the connection was lost"))
);
assert_eq!(
evs[2].as_ref().unwrap(),
&(
"sub_id_2".to_string(),
vec![BlockRef::new(H256::from_low_u64_le(14))]
)
);
}
}
subxt/src/backend/unstable/follow_stream_unpin.rs
+1 -1
View File
@@ -474,7 +474,7 @@ pub(super) mod test_utils {
pub type UnpinRx<Hash> = std::sync::mpsc::Receiver<(Hash, Arc<str>)>;
/// Get a `FolowStreamUnpin` from an iterator over events.
/// Get a [`FollowStreamUnpin`] from an iterator over events.
pub fn test_unpin_stream_getter<Hash, F, I>(
events: F,
max_life: usize,
subxt/src/backend/unstable/mod.rs
+178 -144
View File
@@ -18,21 +18,22 @@ mod storage_items;
pub mod rpc_methods;
use self::follow_stream_driver::FollowStreamFinalizedHeads;
use self::rpc_methods::{
FollowEvent, MethodResponse, RuntimeEvent, StorageQuery, StorageQueryType, StorageResultType,
};
use crate::backend::{
rpc::RpcClient, Backend, BlockRef, BlockRefT, RuntimeVersion, StorageResponse, StreamOf,
StreamOfResults, TransactionStatus,
rpc::RpcClient, utils::retry, Backend, BlockRef, BlockRefT, RuntimeVersion, StorageResponse,
StreamOf, StreamOfResults, TransactionStatus,
};
use crate::config::BlockHash;
use crate::error::{Error, RpcError};
use crate::Config;
use async_trait::async_trait;
use follow_stream_driver::{FollowStreamDriver, FollowStreamDriverHandle};
use futures::future::Either;
use futures::{Stream, StreamExt};
use std::collections::HashMap;
use std::sync::Arc;
use std::task::Poll;
use storage_items::StorageItems;
@@ -136,43 +137,50 @@ impl<T: Config> UnstableBackend<T> {
}
/// Stream block headers based on the provided filter fn
async fn stream_headers<F, I>(
async fn stream_headers<F>(
&self,
f: F,
) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error>
where
F: Fn(FollowEvent<follow_stream_unpin::BlockRef<T::Hash>>) -> I + Copy + Send + 'static,
I: IntoIterator<Item = follow_stream_unpin::BlockRef<T::Hash>> + Send + 'static,
<I as IntoIterator>::IntoIter: Send,
F: Fn(
FollowEvent<follow_stream_unpin::BlockRef<T::Hash>>,
) -> Vec<follow_stream_unpin::BlockRef<T::Hash>>
+ Send
+ Sync
+ 'static,
{
let sub_id = get_subscription_id(&self.follow_handle).await?;
let sub_id = Arc::new(sub_id);
let methods = self.methods.clone();
let headers = self.follow_handle.subscribe().events().flat_map(move |ev| {
let sub_id = sub_id.clone();
let methods = methods.clone();
let block_refs = f(ev).into_iter();
futures::stream::iter(block_refs).filter_map(move |block_ref| {
let sub_id = sub_id.clone();
let headers =
FollowStreamFinalizedHeads::new(self.follow_handle.subscribe(), f).flat_map(move |r| {
let methods = methods.clone();
async move {
let res = methods
.chainhead_v1_header(&sub_id, block_ref.hash())
.await
.transpose()?;
let (sub_id, block_refs) = match r {
Ok(ev) => ev,
Err(e) => return Either::Left(futures::stream::once(async { Err(e) })),
};
let header = match res {
Ok(header) => header,
Err(e) => return Some(Err(e)),
};
Either::Right(
futures::stream::iter(block_refs).filter_map(move |block_ref| {
let methods = methods.clone();
let sub_id = sub_id.clone();
Some(Ok((header, block_ref.into())))
}
})
});
async move {
let res = methods
.chainhead_v1_header(&sub_id, block_ref.hash())
.await
.transpose()?;
let header = match res {
Ok(header) => header,
Err(e) => return Some(Err(e)),
};
Some(Ok((header, block_ref.into())))
}
}),
)
});
Ok(StreamOf(Box::pin(headers)))
}
@@ -194,31 +202,34 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
keys: Vec<Vec<u8>>,
at: T::Hash,
) -> Result<StreamOfResults<StorageResponse>, Error> {
let queries = keys.iter().map(|key| StorageQuery {
key: &**key,
query_type: StorageQueryType::Value,
});
retry(|| async {
let queries = keys.iter().map(|key| StorageQuery {
key: &**key,
query_type: StorageQueryType::Value,
});
let storage_items =
StorageItems::from_methods(queries, at, &self.follow_handle, self.methods.clone())
.await?;
let storage_items =
StorageItems::from_methods(queries, at, &self.follow_handle, self.methods.clone())
.await?;
let storage_result_stream = storage_items.filter_map(|val| async move {
let val = match val {
Ok(val) => val,
Err(e) => return Some(Err(e)),
};
let stream = storage_items.filter_map(|val| async move {
let val = match val {
Ok(val) => val,
Err(e) => return Some(Err(e)),
};
let StorageResultType::Value(result) = val.result else {
return None;
};
Some(Ok(StorageResponse {
key: val.key.0,
value: result.0,
}))
});
let StorageResultType::Value(result) = val.result else {
return None;
};
Some(Ok(StorageResponse {
key: val.key.0,
value: result.0,
}))
});
Ok(StreamOf(Box::pin(storage_result_stream)))
Ok(StreamOf(Box::pin(stream)))
})
.await
}
async fn storage_fetch_descendant_keys(
@@ -226,22 +237,25 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
key: Vec<u8>,
at: T::Hash,
) -> Result<StreamOfResults<Vec<u8>>, Error> {
// Ask for hashes, and then just ignore them and return the keys that come back.
let query = StorageQuery {
key: &*key,
query_type: StorageQueryType::DescendantsHashes,
};
retry(|| async {
// Ask for hashes, and then just ignore them and return the keys that come back.
let query = StorageQuery {
key: &*key,
query_type: StorageQueryType::DescendantsHashes,
};
let storage_items = StorageItems::from_methods(
std::iter::once(query),
at,
&self.follow_handle,
self.methods.clone(),
)
.await?;
let storage_items = StorageItems::from_methods(
std::iter::once(query),
at,
&self.follow_handle,
self.methods.clone(),
)
.await?;
let storage_result_stream = storage_items.map(|val| val.map(|v| v.key.0));
Ok(StreamOf(Box::pin(storage_result_stream)))
let storage_result_stream = storage_items.map(|val| val.map(|v| v.key.0));
Ok(StreamOf(Box::pin(storage_result_stream)))
})
.await
}
async fn storage_fetch_descendant_values(
@@ -249,72 +263,81 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
key: Vec<u8>,
at: T::Hash,
) -> Result<StreamOfResults<StorageResponse>, Error> {
let query = StorageQuery {
key: &*key,
query_type: StorageQueryType::DescendantsValues,
};
let storage_items = StorageItems::from_methods(
std::iter::once(query),
at,
&self.follow_handle,
self.methods.clone(),
)
.await?;
let storage_result_stream = storage_items.filter_map(|val| async move {
let val = match val {
Ok(val) => val,
Err(e) => return Some(Err(e)),
retry(|| async {
let query = StorageQuery {
key: &*key,
query_type: StorageQueryType::DescendantsValues,
};
let StorageResultType::Value(result) = val.result else {
return None;
};
Some(Ok(StorageResponse {
key: val.key.0,
value: result.0,
}))
});
let storage_items = StorageItems::from_methods(
std::iter::once(query),
at,
&self.follow_handle,
self.methods.clone(),
)
.await?;
Ok(StreamOf(Box::pin(storage_result_stream)))
let storage_result_stream = storage_items.filter_map(|val| async move {
let val = match val {
Ok(val) => val,
Err(e) => return Some(Err(e)),
};
let StorageResultType::Value(result) = val.result else {
return None;
};
Some(Ok(StorageResponse {
key: val.key.0,
value: result.0,
}))
});
Ok(StreamOf(Box::pin(storage_result_stream)))
})
.await
}
async fn genesis_hash(&self) -> Result<T::Hash, Error> {
self.methods.chainspec_v1_genesis_hash().await
retry(|| self.methods.chainspec_v1_genesis_hash()).await
}
async fn block_header(&self, at: T::Hash) -> Result<Option<T::Header>, Error> {
let sub_id = get_subscription_id(&self.follow_handle).await?;
self.methods.chainhead_v1_header(&sub_id, at).await
retry(|| async {
let sub_id = get_subscription_id(&self.follow_handle).await?;
self.methods.chainhead_v1_header(&sub_id, at).await
})
.await
}
async fn block_body(&self, at: T::Hash) -> Result<Option<Vec<Vec<u8>>>, Error> {
let sub_id = get_subscription_id(&self.follow_handle).await?;
retry(|| async {
let sub_id = get_subscription_id(&self.follow_handle).await?;
// Subscribe to the body response and get our operationId back.
let follow_events = self.follow_handle.subscribe().events();
let status = self.methods.chainhead_v1_body(&sub_id, at).await?;
let operation_id = match status {
MethodResponse::LimitReached => {
return Err(RpcError::request_rejected("limit reached").into())
}
MethodResponse::Started(s) => s.operation_id,
};
// Wait for the response to come back with the correct operationId.
let mut exts_stream = follow_events.filter_map(|ev| {
let FollowEvent::OperationBodyDone(body) = ev else {
return std::future::ready(None);
// Subscribe to the body response and get our operationId back.
let follow_events = self.follow_handle.subscribe().events();
let status = self.methods.chainhead_v1_body(&sub_id, at).await?;
let operation_id = match status {
MethodResponse::LimitReached => {
return Err(RpcError::request_rejected("limit reached").into())
}
MethodResponse::Started(s) => s.operation_id,
};
if body.operation_id != operation_id {
return std::future::ready(None);
}
let exts: Vec<_> = body.value.into_iter().map(|ext| ext.0).collect();
std::future::ready(Some(exts))
});
Ok(exts_stream.next().await)
// Wait for the response to come back with the correct operationId.
let mut exts_stream = follow_events.filter_map(|ev| {
let FollowEvent::OperationBodyDone(body) = ev else {
return std::future::ready(None);
};
if body.operation_id != operation_id {
return std::future::ready(None);
}
let exts: Vec<_> = body.value.into_iter().map(|ext| ext.0).collect();
std::future::ready(Some(exts))
});
Ok(exts_stream.next().await)
})
.await
}
async fn latest_finalized_block_ref(&self) -> Result<BlockRef<T::Hash>, Error> {
@@ -423,12 +446,16 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
std::future::ready(Some(Ok(runtime_version)))
});
Ok(StreamOf(Box::pin(runtime_stream)))
Ok(StreamOf::new(Box::pin(runtime_stream)))
}
async fn stream_all_block_headers(
&self,
) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
// TODO: https://github.com/paritytech/subxt/issues/1568
//
// It's possible that blocks may be silently missed if
// a reconnection occurs because it's restarted by the unstable backend.
self.stream_headers(|ev| match ev {
FollowEvent::Initialized(init) => init.finalized_block_hashes,
FollowEvent::NewBlock(ev) => {
@@ -442,6 +469,10 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
async fn stream_best_block_headers(
&self,
) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
// TODO: https://github.com/paritytech/subxt/issues/1568
//
// It's possible that blocks may be silently missed if
// a reconnection occurs because it's restarted by the unstable backend.
self.stream_headers(|ev| match ev {
FollowEvent::Initialized(init) => init.finalized_block_hashes,
FollowEvent::BestBlockChanged(ev) => vec![ev.best_block_hash],
@@ -638,37 +669,40 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
call_parameters: Option<&[u8]>,
at: T::Hash,
) -> Result<Vec<u8>, Error> {
let sub_id = get_subscription_id(&self.follow_handle).await?;
retry(|| async {
let sub_id = get_subscription_id(&self.follow_handle).await?;
// Subscribe to the body response and get our operationId back.
let follow_events = self.follow_handle.subscribe().events();
let call_parameters = call_parameters.unwrap_or(&[]);
let status = self
.methods
.chainhead_v1_call(&sub_id, at, method, call_parameters)
.await?;
let operation_id = match status {
MethodResponse::LimitReached => {
return Err(RpcError::request_rejected("limit reached").into())
}
MethodResponse::Started(s) => s.operation_id,
};
// Wait for the response to come back with the correct operationId.
let mut call_data_stream = follow_events.filter_map(|ev| {
let FollowEvent::OperationCallDone(body) = ev else {
return std::future::ready(None);
// Subscribe to the body response and get our operationId back.
let follow_events = self.follow_handle.subscribe().events();
let call_parameters = call_parameters.unwrap_or(&[]);
let status = self
.methods
.chainhead_v1_call(&sub_id, at, method, call_parameters)
.await?;
let operation_id = match status {
MethodResponse::LimitReached => {
return Err(RpcError::request_rejected("limit reached").into())
}
MethodResponse::Started(s) => s.operation_id,
};
if body.operation_id != operation_id {
return std::future::ready(None);
}
std::future::ready(Some(body.output.0))
});
call_data_stream
.next()
.await
.ok_or_else(|| RpcError::SubscriptionDropped.into())
// Wait for the response to come back with the correct operationId.
let mut call_data_stream = follow_events.filter_map(|ev| {
let FollowEvent::OperationCallDone(body) = ev else {
return std::future::ready(None);
};
if body.operation_id != operation_id {
return std::future::ready(None);
}
std::future::ready(Some(body.output.0))
});
call_data_stream
.next()
.await
.ok_or_else(|| RpcError::SubscriptionDropped.into())
})
.await
}
}
subxt/src/backend/unstable/rpc_methods.rs
+3 -9
View File
@@ -77,9 +77,7 @@ impl<T: Config> UnstableRpcMethods<T> {
"chainHead_v1_continue",
rpc_params![follow_subscription, operation_id],
)
.await?;
Ok(())
.await
}
/// Stops an operation started with `chainHead_v1_body`, `chainHead_v1_call`, or
@@ -97,9 +95,7 @@ impl<T: Config> UnstableRpcMethods<T> {
"chainHead_v1_stopOperation",
rpc_params![follow_subscription, operation_id],
)
.await?;
Ok(())
.await
}
/// Call the `chainHead_v1_body` method and return an operation ID to obtain the block's body.
@@ -222,9 +218,7 @@ impl<T: Config> UnstableRpcMethods<T> {
) -> Result<(), Error> {
self.client
.request("chainHead_v1_unpin", rpc_params![subscription_id, hash])
.await?;
Ok(())
.await
}
/// Return the genesis hash.
subxt/src/backend/unstable/storage_items.rs
+5
View File
@@ -111,6 +111,11 @@ impl<T: Config> Stream for StorageItems<T> {
return Poll::Pending;
}
Poll::Ready(Err(e)) => {
if e.is_disconnected_will_reconnect() {
self.continue_fut = Some((self.continue_call)());
continue;
}
self.done = true;
return Poll::Ready(Some(Err(e)));
}
subxt/src/backend/utils.rs
+271
View File
@@ -0,0 +1,271 @@
//! RPC utils.
use super::{StreamOf, StreamOfResults};
use crate::error::Error;
use futures::future::BoxFuture;
use futures::{FutureExt, Stream, StreamExt};
use std::{future::Future, pin::Pin, task::Poll};
/// Resubscribe callback.
type ResubscribeGetter<T> = Box<dyn FnMut() -> ResubscribeFuture<T> + Send>;
/// Future that resolves to a subscription stream.
type ResubscribeFuture<T> = Pin<Box<dyn Future<Output = Result<StreamOfResults<T>, Error>> + Send>>;
pub(crate) enum PendingOrStream<T> {
Pending(BoxFuture<'static, Result<StreamOfResults<T>, Error>>),
Stream(StreamOfResults<T>),
}
impl<T> std::fmt::Debug for PendingOrStream<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PendingOrStream::Pending(_) => write!(f, "Pending"),
PendingOrStream::Stream(_) => write!(f, "Stream"),
}
}
}
/// Retry subscription.
struct RetrySubscription<T> {
resubscribe: ResubscribeGetter<T>,
state: Option<PendingOrStream<T>>,
}
impl<T> std::marker::Unpin for RetrySubscription<T> {}
impl<T> Stream for RetrySubscription<T> {
type Item = Result<T, Error>;
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> Poll<Option<Self::Item>> {
loop {
let Some(mut this) = self.state.take() else {
return Poll::Ready(None);
};
match this {
PendingOrStream::Stream(ref mut s) => match s.poll_next_unpin(cx) {
Poll::Ready(Some(Err(err))) => {
if err.is_disconnected_will_reconnect() {
self.state = Some(PendingOrStream::Pending((self.resubscribe)()));
}
return Poll::Ready(Some(Err(err)));
}
Poll::Ready(None) => return Poll::Ready(None),
Poll::Ready(Some(Ok(val))) => {
self.state = Some(this);
return Poll::Ready(Some(Ok(val)));
}
Poll::Pending => {
self.state = Some(this);
return Poll::Pending;
}
},
PendingOrStream::Pending(mut fut) => match fut.poll_unpin(cx) {
Poll::Ready(Ok(stream)) => {
self.state = Some(PendingOrStream::Stream(stream));
continue;
}
Poll::Ready(Err(err)) => {
if err.is_disconnected_will_reconnect() {
self.state = Some(PendingOrStream::Pending((self.resubscribe)()));
}
return Poll::Ready(Some(Err(err)));
}
Poll::Pending => {
self.state = Some(PendingOrStream::Pending(fut));
return Poll::Pending;
}
},
};
}
}
}
/// Retry a future until it doesn't return a disconnected error.
///
/// # Example
///
/// ```no_run
/// use subxt::backend::utils::retry;
///
/// async fn some_future() -> Result<(), subxt::error::Error> {
/// Ok(())
/// }
///
/// #[tokio::main]
/// async fn main() {
/// let result = retry(|| some_future()).await;
/// }
/// ```
pub async fn retry<T, F, R>(mut retry_future: F) -> Result<R, Error>
where
F: FnMut() -> T,
T: Future<Output = Result<R, Error>>,
{
const REJECTED_MAX_RETRIES: usize = 10;
let mut rejected_retries = 0;
loop {
match retry_future().await {
Ok(v) => return Ok(v),
Err(e) => {
if e.is_disconnected_will_reconnect() {
continue;
}
// TODO: https://github.com/paritytech/subxt/issues/1567
// This is a hack because if a reconnection occurs
// the order of pending calls is not guaranteed.
//
// Such that it's possible the a pending future completes
// before `chainHead_follow` is established with fresh
// subscription id.
//
if e.is_rejected() && rejected_retries < REJECTED_MAX_RETRIES {
rejected_retries += 1;
continue;
}
return Err(e);
}
}
}
}
/// Create a retry stream that will resubscribe on disconnect.
///
/// It's important to note that this function is intended to work only for stateless subscriptions.
/// If the subscription takes input or modifies state, this function should not be used.
///
/// # Example
///
/// ```no_run
/// use subxt::backend::{utils::retry_stream, StreamOf};
/// use futures::future::FutureExt;
///
/// #[tokio::main]
/// async fn main() {
/// retry_stream(|| {
/// // This needs to return a stream of results but if you are using
/// // the subxt backend already it will return StreamOf so you can just
/// // return it directly in the async block below.
/// async move { Ok(StreamOf::new(Box::pin(futures::stream::iter([Ok(2)])))) }.boxed()
/// }).await;
/// }
/// ```
pub async fn retry_stream<F, R>(sub_stream: F) -> Result<StreamOfResults<R>, Error>
where
F: FnMut() -> ResubscribeFuture<R> + Send + 'static + Clone,
R: Send + 'static,
{
let stream = retry(sub_stream.clone()).await?;
let resubscribe = Box::new(move || {
let sub_stream = sub_stream.clone();
async move { retry(sub_stream).await }.boxed()
});
// The extra Box is to encapsulate the retry subscription type
Ok(StreamOf::new(Box::pin(RetrySubscription {
state: Some(PendingOrStream::Stream(stream)),
resubscribe,
})))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::backend::StreamOf;
fn disconnect_err() -> Error {
Error::Rpc(crate::error::RpcError::DisconnectedWillReconnect(
String::new(),
))
}
fn custom_err() -> Error {
Error::Other(String::new())
}
#[tokio::test]
async fn retry_stream_works() {
let retry_stream = retry_stream(|| {
async {
Ok(StreamOf::new(Box::pin(futures::stream::iter([
Ok(1),
Ok(2),
Ok(3),
Err(disconnect_err()),
]))))
}
.boxed()
})
.await
.unwrap();
let result = retry_stream
.take(5)
.collect::<Vec<Result<usize, Error>>>()
.await;
assert!(matches!(result[0], Ok(r) if r == 1));
assert!(matches!(result[1], Ok(r) if r == 2));
assert!(matches!(result[2], Ok(r) if r == 3));
assert!(matches!(result[3], Err(ref e) if e.is_disconnected_will_reconnect()));
assert!(matches!(result[4], Ok(r) if r == 1));
}
#[tokio::test]
async fn retry_sub_works() {
let stream = futures::stream::iter([Ok(1), Err(disconnect_err())]);
let resubscribe = Box::new(move || {
async move { Ok(StreamOf::new(Box::pin(futures::stream::iter([Ok(2)])))) }.boxed()
});
let retry_stream = RetrySubscription {
state: Some(PendingOrStream::Stream(StreamOf::new(Box::pin(stream)))),
resubscribe,
};
let result: Vec<_> = retry_stream.collect().await;
assert!(matches!(result[0], Ok(r) if r == 1));
assert!(matches!(result[1], Err(ref e) if e.is_disconnected_will_reconnect()));
assert!(matches!(result[2], Ok(r) if r == 2));
}
#[tokio::test]
async fn retry_sub_err_terminates_stream() {
let stream = futures::stream::iter([Ok(1)]);
let resubscribe = Box::new(move || async move { Err(custom_err()) }.boxed());
let retry_stream = RetrySubscription {
state: Some(PendingOrStream::Stream(StreamOf::new(Box::pin(stream)))),
resubscribe,
};
assert_eq!(retry_stream.count().await, 1);
}
#[tokio::test]
async fn retry_sub_resubscribe_err() {
let stream = futures::stream::iter([Ok(1), Err(disconnect_err())]);
let resubscribe = Box::new(move || async move { Err(custom_err()) }.boxed());
let retry_stream = RetrySubscription {
state: Some(PendingOrStream::Stream(StreamOf::new(Box::pin(stream)))),
resubscribe,
};
let result: Vec<_> = retry_stream.collect().await;
assert!(matches!(result[0], Ok(r) if r == 1));
assert!(matches!(result[1], Err(ref e) if e.is_disconnected_will_reconnect()));
assert!(matches!(result[2], Err(ref e) if matches!(e, Error::Other(_))));
}
}
subxt/src/blocks/blocks_client.rs
+6 -6
View File
@@ -95,8 +95,8 @@ where
{
let client = self.client.clone();
header_sub_fut_to_block_sub(self.clone(), async move {
let sub = client.backend().stream_all_block_headers().await?;
BlockStreamRes::Ok(sub)
let stream = client.backend().stream_all_block_headers().await?;
BlockStreamRes::Ok(stream)
})
}
@@ -112,8 +112,8 @@ where
{
let client = self.client.clone();
header_sub_fut_to_block_sub(self.clone(), async move {
let sub = client.backend().stream_best_block_headers().await?;
BlockStreamRes::Ok(sub)
let stream = client.backend().stream_best_block_headers().await?;
BlockStreamRes::Ok(stream)
})
}
@@ -126,8 +126,8 @@ where
{
let client = self.client.clone();
header_sub_fut_to_block_sub(self.clone(), async move {
let sub = client.backend().stream_finalized_block_headers().await?;
BlockStreamRes::Ok(sub)
let stream = client.backend().stream_finalized_block_headers().await?;
BlockStreamRes::Ok(stream)
})
}
}
subxt/src/blocks/extrinsic_types.rs
+5
View File
@@ -159,6 +159,11 @@ where
}
}
/// See [`subxt_core::blocks::ExtrinsicDetails::hash()`].
pub fn hash(&self) -> T::Hash {
self.inner.hash()
}
/// See [`subxt_core::blocks::ExtrinsicDetails::is_signed()`].
pub fn is_signed(&self) -> bool {
self.inner.is_signed()
subxt/src/book/setup/codegen.rs
+1 -1
View File
@@ -46,7 +46,7 @@
//! Using the [`#[subxt]`](crate::subxt) macro carries some downsides:
//!
//! - Using it to generate an interface will have a small impact on compile times (though much less of
//! one if you only need a few pallets).
//! one if you only need a few pallets).
//! - IDE support for autocompletion and documentation when using the macro interface can be poor.
//! - It's impossible to manually look at the generated code to understand and debug things.
//!
subxt/src/book/usage/transactions.rs
+17
View File
@@ -196,6 +196,14 @@
//! This example doesn't wait for the transaction to be included in a block; it just submits it and
//! hopes for the best!
//!
//! ### Boxing transaction payloads
//!
//! Transaction payloads can be boxed so that they all share a common type and can be stored together.
//!
//! ```rust,ignore
#![doc = include_str!("../../../examples/tx_boxed.rs")]
//! ```
//!
//! ### Custom handling of transaction status updates
//!
//! If you'd like more control or visibility over exactly which status updates are being emitted for
@@ -205,6 +213,15 @@
#![doc = include_str!("../../../examples/tx_status_stream.rs")]
//! ```
//!
//! ### Signing transactions externally
//!
//! Subxt also allows you to get hold of the signer payload and hand that off to something else to be
//! signed. The signature can then be provided back to Subxt to build the final transaction to submit:
//!
//! ```rust,ignore
#![doc = include_str!("../../../examples/tx_partial.rs")]
//! ```
//!
//! Take a look at the API docs for [`crate::tx::TxProgress`], [`crate::tx::TxStatus`] and
//! [`crate::tx::TxInBlock`] for more options.
//!
subxt/src/client/online_client.rs
+1 -2
View File
@@ -432,9 +432,8 @@ impl<T: Config> ClientRuntimeUpdater<T> {
/// Instead that's up to the user of this API to decide when to update and
/// to perform the actual updating.
pub async fn runtime_updates(&self) -> Result<RuntimeUpdaterStream<T>, Error> {
let stream = self.0.backend().stream_runtime_version().await?;
Ok(RuntimeUpdaterStream {
stream,
stream: self.0.backend().stream_runtime_version().await?,
client: self.0.clone(),
})
}
subxt/src/error/mod.rs
+5
View File
@@ -125,6 +125,11 @@ impl Error {
pub fn is_disconnected_will_reconnect(&self) -> bool {
matches!(self, Error::Rpc(RpcError::DisconnectedWillReconnect(_)))
}
/// Checks whether the error was caused by a RPC request being rejected.
pub fn is_rejected(&self) -> bool {
matches!(self, Error::Rpc(RpcError::RequestRejected(_)))
}
}
/// An RPC error. Since we are generic over the RPC client that is used,
subxt/src/macros.rs
+1 -1
View File
@@ -56,7 +56,7 @@ macro_rules! cfg_jsonrpsee_web {
macro_rules! cfg_reconnecting_rpc_client {
($($item:item)*) => {
$(
#[cfg(feature = "unstable-reconnecting-rpc-client")]
#[cfg(all(feature = "unstable-reconnecting-rpc-client", any(feature = "native", feature = "web")))]
#[cfg_attr(docsrs, doc(cfg(feature = "unstable-reconnecting-rpc-client")))]
$item
)*
subxt/src/tx/tx_client.rs
+3 -4
View File
@@ -599,11 +599,10 @@ pub enum TransactionInvalid {
///
/// # Possible causes
///
/// For `FRAME`-based runtimes this would be caused by `current block number
/// For `FRAME`-based runtimes this would be caused by `current block number`
/// - Era::birth block number > BlockHashCount`. (e.g. in Polkadot `BlockHashCount` = 2400, so
/// a
/// transaction with birth block number 1337 would be valid up until block number 1337 + 2400,
/// after which point the transaction would be considered to have an ancient birth block.)
/// a transaction with birth block number 1337 would be valid up until block number 1337 + 2400,
/// after which point the transaction would be considered to have an ancient birth block.)
AncientBirthBlock,
/// The transaction would exhaust the resources of current block.
///
subxt/src/tx/tx_progress.rs
+1 -1
View File
@@ -11,11 +11,11 @@ use crate::{
client::OnlineClientT,
error::{DispatchError, Error, RpcError, TransactionError},
events::EventsClient,
utils::strip_compact_prefix,
Config,
};
use derive_where::derive_where;
use futures::{Stream, StreamExt};
use subxt_core::utils::strip_compact_prefix;
/// This struct represents a subscription to the progress of some transaction.
pub struct TxProgress<T: Config, C> {