pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-04-28 02:48:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Stabilize chainHead methods (#1538)

Browse Source 
* Stabilize chainHead methods

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Rename fn snake case

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* docs: Fix documentation

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

---------

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
This commit is contained in:
Alexandru Vasile
2024-04-19 12:27:50 +03:00
committed by GitHub
parent ac606cf625
commit c124e17336
6 changed files with 62 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
subxt/src/backend/unstable/follow_stream.rs
+1 -1
View File
@@ -105,7 +105,7 @@ impl<Hash> FollowStream<Hash> {
let methods = methods.clone();
Box::pin(async move {
// Make the RPC call:
let stream = methods.chainhead_unstable_follow(true).await?;
let stream = methods.chainhead_v1_follow(true).await?;
// Extract the subscription ID:
let Some(sub_id) = stream.subscription_id().map(ToOwned::to_owned) else {
return Err(Error::Other(
subxt/src/backend/unstable/follow_stream_unpin.rs
+1 -1
View File
@@ -282,7 +282,7 @@ impl<Hash: BlockHash> FollowStreamUnpin<Hash> {
let methods = methods.clone();
let fut: UnpinFut = Box::pin(async move {
// We ignore any errors trying to unpin at the moment.
let _ = methods.chainhead_unstable_unpin(&sub_id, hash).await;
let _ = methods.chainhead_v1_unpin(&sub_id, hash).await;
});
fut
});
subxt/src/backend/unstable/mod.rs
+4 -4
View File
@@ -160,7 +160,7 @@ impl<T: Config> UnstableBackend<T> {
async move {
let res = methods
.chainhead_unstable_header(&sub_id, block_ref.hash())
.chainhead_v1_header(&sub_id, block_ref.hash())
.await
.transpose()?;
@@ -286,7 +286,7 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
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_unstable_header(&sub_id, at).await
self.methods.chainhead_v1_header(&sub_id, at).await
}
async fn block_body(&self, at: T::Hash) -> Result<Option<Vec<Vec<u8>>>, Error> {
@@ -294,7 +294,7 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
// Subscribe to the body response and get our operationId back.
let follow_events = self.follow_handle.subscribe().events();
let status = self.methods.chainhead_unstable_body(&sub_id, at).await?;
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())
@@ -645,7 +645,7 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
let call_parameters = call_parameters.unwrap_or(&[]);
let status = self
.methods
.chainhead_unstable_call(&sub_id, at, method, call_parameters)
.chainhead_v1_call(&sub_id, at, method, call_parameters)
.await?;
let operation_id = match status {
MethodResponse::LimitReached => {
subxt/src/backend/unstable/rpc_methods.rs
+35 -44
View File
@@ -33,48 +33,48 @@ impl<T: Config> UnstableRpcMethods<T> {
}
}
/// Subscribe to `chainHead_unstable_follow` to obtain all reported blocks by the chain.
/// Subscribe to `chainHead_v1_follow` to obtain all reported blocks by the chain.
///
/// The subscription ID can be used to make queries for the
/// block's body ([`chainhead_unstable_body`](UnstableRpcMethods::chainhead_unstable_follow)),
/// block's header ([`chainhead_unstable_header`](UnstableRpcMethods::chainhead_unstable_header)),
/// block's storage ([`chainhead_unstable_storage`](UnstableRpcMethods::chainhead_unstable_storage)) and submitting
/// runtime API calls at this block ([`chainhead_unstable_call`](UnstableRpcMethods::chainhead_unstable_call)).
/// block's body ([`chainHead_v1_body`](UnstableRpcMethods::chainhead_v1_follow)),
/// block's header ([`chainHead_v1_header`](UnstableRpcMethods::chainhead_v1_header)),
/// block's storage ([`chainHead_v1_storage`](UnstableRpcMethods::chainhead_v1_storage)) and submitting
/// runtime API calls at this block ([`chainHead_v1_call`](UnstableRpcMethods::chainhead_v1_call)).
///
/// # Note
///
/// When the user is no longer interested in a block, the user is responsible
/// for calling the [`chainhead_unstable_unpin`](UnstableRpcMethods::chainhead_unstable_unpin) method.
/// for calling the [`chainHead_v1_unpin`](UnstableRpcMethods::chainhead_v1_unpin) method.
/// Failure to do so will result in the subscription being stopped by generating the `Stop` event.
pub async fn chainhead_unstable_follow(
pub async fn chainhead_v1_follow(
&self,
with_runtime: bool,
) -> Result<FollowSubscription<T::Hash>, Error> {
let sub = self
.client
.subscribe(
"chainHead_unstable_follow",
"chainHead_v1_follow",
rpc_params![with_runtime],
"chainHead_unstable_unfollow",
"chainHead_v1_unfollow",
)
.await?;
Ok(FollowSubscription { sub, done: false })
}
/// Resumes a storage fetch started with chainHead_unstable_storage after it has generated an
/// Resumes a storage fetch started with chainHead_v1_storage after it has generated an
/// `operationWaitingForContinue` event.
///
/// Has no effect if the operationId is invalid or refers to an operation that has emitted a
/// `{"event": "operationInaccessible"` event, or if the followSubscription is invalid or stale.
pub async fn chainhead_unstable_continue(
pub async fn chainhead_v1_continue(
&self,
follow_subscription: &str,
operation_id: &str,
) -> Result<(), Error> {
self.client
.request(
"chainHead_unstable_continue",
"chainHead_v1_continue",
rpc_params![follow_subscription, operation_id],
)
.await?;
@@ -82,19 +82,19 @@ impl<T: Config> UnstableRpcMethods<T> {
Ok(())
}
/// Stops an operation started with `chainHead_unstable_body`, `chainHead_unstable_call`, or
/// `chainHead_unstable_storage¦. If the operation was still in progress, this interrupts it.
/// Stops an operation started with `chainHead_v1_body`, `chainHead_v1_call`, or
/// `chainHead_v1_storage¦. If the operation was still in progress, this interrupts it.
/// If the operation was already finished, this call has no effect.
///
/// Has no effect if the `followSubscription` is invalid or stale.
pub async fn chainhead_unstable_stop_operation(
pub async fn chainhead_v1_stop_operation(
&self,
follow_subscription: &str,
operation_id: &str,
) -> Result<(), Error> {
self.client
.request(
"chainHead_unstable_stopOperation",
"chainHead_v1_stopOperation",
rpc_params![follow_subscription, operation_id],
)
.await?;
@@ -102,7 +102,7 @@ impl<T: Config> UnstableRpcMethods<T> {
Ok(())
}
/// Call the `chainHead_unstable_body` method and return an operation ID to obtain the block's body.
/// Call the `chainHead_v1_body` method and return an operation ID to obtain the block's body.
///
/// The response events are provided on the `chainHead_follow` subscription and identified by
/// the returned operation ID.
@@ -110,30 +110,27 @@ impl<T: Config> UnstableRpcMethods<T> {
/// # Note
///
/// The subscription ID is obtained from an open subscription created by
/// [`chainhead_unstable_follow`](UnstableRpcMethods::chainhead_unstable_follow).
pub async fn chainhead_unstable_body(
/// [`chainHead_v1_follow`](UnstableRpcMethods::chainhead_v1_follow).
pub async fn chainhead_v1_body(
&self,
subscription_id: &str,
hash: T::Hash,
) -> Result<MethodResponse, Error> {
let response = self
.client
.request(
"chainHead_unstable_body",
rpc_params![subscription_id, hash],
)
.request("chainHead_v1_body", rpc_params![subscription_id, hash])
.await?;
Ok(response)
}
/// Get the block's header using the `chainHead_unstable_header` method.
/// Get the block's header using the `chainHead_v1_header` method.
///
/// # Note
///
/// The subscription ID is obtained from an open subscription created by
/// [`chainhead_unstable_follow`](UnstableRpcMethods::chainhead_unstable_follow).
pub async fn chainhead_unstable_header(
/// [`chainHead_v1_follow`](UnstableRpcMethods::chainhead_v1_follow).
pub async fn chainhead_v1_header(
&self,
subscription_id: &str,
hash: T::Hash,
@@ -141,10 +138,7 @@ impl<T: Config> UnstableRpcMethods<T> {
// header returned as hex encoded SCALE encoded bytes.
let header: Option<Bytes> = self
.client
.request(
"chainHead_unstable_header",
rpc_params![subscription_id, hash],
)
.request("chainHead_v1_header", rpc_params![subscription_id, hash])
.await?;
let header = header
@@ -153,7 +147,7 @@ impl<T: Config> UnstableRpcMethods<T> {
Ok(header)
}
/// Call the `chainhead_unstable_storage` method and return an operation ID to obtain the block's storage.
/// Call the `chainHead_v1_storage` method and return an operation ID to obtain the block's storage.
///
/// The response events are provided on the `chainHead_follow` subscription and identified by
/// the returned operation ID.
@@ -161,8 +155,8 @@ impl<T: Config> UnstableRpcMethods<T> {
/// # Note
///
/// The subscription ID is obtained from an open subscription created by
/// [`chainhead_unstable_follow`](UnstableRpcMethods::chainhead_unstable_follow).
pub async fn chainhead_unstable_storage(
/// [`chainHead_v1_follow`](UnstableRpcMethods::chainhead_v1_follow).
pub async fn chainhead_v1_storage(
&self,
subscription_id: &str,
hash: T::Hash,
@@ -180,7 +174,7 @@ impl<T: Config> UnstableRpcMethods<T> {
let response = self
.client
.request(
"chainHead_unstable_storage",
"chainHead_v1_storage",
rpc_params![subscription_id, hash, items, child_key.map(to_hex)],
)
.await?;
@@ -188,7 +182,7 @@ impl<T: Config> UnstableRpcMethods<T> {
Ok(response)
}
/// Call the `chainhead_unstable_storage` method and return an operation ID to obtain the runtime API result.
/// Call the `chainHead_v1_storage` method and return an operation ID to obtain the runtime API result.
///
/// The response events are provided on the `chainHead_follow` subscription and identified by
/// the returned operation ID.
@@ -196,8 +190,8 @@ impl<T: Config> UnstableRpcMethods<T> {
/// # Note
///
/// The subscription ID is obtained from an open subscription created by
/// [`chainhead_unstable_follow`](UnstableRpcMethods::chainhead_unstable_follow).
pub async fn chainhead_unstable_call(
/// [`chainHead_v1_follow`](UnstableRpcMethods::chainhead_v1_follow).
pub async fn chainhead_v1_call(
&self,
subscription_id: &str,
hash: T::Hash,
@@ -207,7 +201,7 @@ impl<T: Config> UnstableRpcMethods<T> {
let response = self
.client
.request(
"chainHead_unstable_call",
"chainHead_v1_call",
rpc_params![subscription_id, hash, function, to_hex(call_parameters)],
)
.await?;
@@ -220,17 +214,14 @@ impl<T: Config> UnstableRpcMethods<T> {
/// # Note
///
/// The subscription ID is obtained from an open subscription created by
/// [`chainhead_unstable_follow`](UnstableRpcMethods::chainhead_unstable_follow).
pub async fn chainhead_unstable_unpin(
/// [`chainHead_v1_follow`](UnstableRpcMethods::chainhead_v1_follow).
pub async fn chainhead_v1_unpin(
&self,
subscription_id: &str,
hash: T::Hash,
) -> Result<(), Error> {
self.client
.request(
"chainHead_unstable_unpin",
rpc_params![subscription_id, hash],
)
.request("chainHead_v1_unpin", rpc_params![subscription_id, hash])
.await?;
Ok(())
subxt/src/backend/unstable/storage_items.rs
+2 -6
View File
@@ -42,7 +42,7 @@ impl<T: Config> StorageItems<T> {
// Subscribe to events and make the initial request to get an operation ID.
let follow_events = follow_handle.subscribe().events();
let status = methods
.chainhead_unstable_storage(&sub_id, at, queries, None)
.chainhead_v1_storage(&sub_id, at, queries, None)
.await?;
let operation_id: Arc<str> = match status {
MethodResponse::LimitReached => {
@@ -59,11 +59,7 @@ impl<T: Config> StorageItems<T> {
let operation_id = operation_id.clone();
let methods = methods.clone();
Box::pin(async move {
methods
.chainhead_unstable_continue(&sub_id, &operation_id)
.await
})
Box::pin(async move { methods.chainhead_v1_continue(&sub_id, &operation_id).await })
})
};
testing/integration-tests/src/full_client/client/unstable_rpcs.rs
+19 -19
View File
@@ -22,13 +22,13 @@ use subxt::{
use subxt_signer::sr25519::dev;
#[subxt_test]
async fn chainhead_unstable_follow() {
async fn chainhead_v1_follow() {
let ctx = test_context().await;
let rpc = ctx.unstable_rpc_methods().await;
let legacy_rpc = ctx.legacy_rpc_methods().await;
// Check subscription with runtime updates set on false.
let mut blocks = rpc.chainhead_unstable_follow(false).await.unwrap();
let mut blocks = rpc.chainhead_v1_follow(false).await.unwrap();
let event = blocks.next().await.unwrap().unwrap();
// The initialized event should contain the finalized block hash.
let finalized_block_hash = legacy_rpc.chain_get_finalized_head().await.unwrap();
@@ -41,7 +41,7 @@ async fn chainhead_unstable_follow() {
);
// Expect subscription to produce runtime versions.
let mut blocks = rpc.chainhead_unstable_follow(true).await.unwrap();
let mut blocks = rpc.chainhead_v1_follow(true).await.unwrap();
let event = blocks.next().await.unwrap().unwrap();
// The initialized event should contain the finalized block hash.
let finalized_block_hash = legacy_rpc.chain_get_finalized_head().await.unwrap();
@@ -62,11 +62,11 @@ async fn chainhead_unstable_follow() {
}
#[subxt_test]
async fn chainhead_unstable_body() {
async fn chainhead_v1_body() {
let ctx = test_context().await;
let rpc = ctx.unstable_rpc_methods().await;
let mut blocks = rpc.chainhead_unstable_follow(false).await.unwrap();
let mut blocks = rpc.chainhead_v1_follow(false).await.unwrap();
let event = blocks.next().await.unwrap().unwrap();
let hash = match event {
FollowEvent::Initialized(init) => *init.finalized_block_hashes.last().unwrap(),
@@ -75,7 +75,7 @@ async fn chainhead_unstable_body() {
let sub_id = blocks.subscription_id().unwrap();
// Fetch the block's body.
let response = rpc.chainhead_unstable_body(sub_id, hash).await.unwrap();
let response = rpc.chainhead_v1_body(sub_id, hash).await.unwrap();
let operation_id = match response {
MethodResponse::Started(started) => started.operation_id,
MethodResponse::LimitReached => panic!("Expected started response"),
@@ -90,12 +90,12 @@ async fn chainhead_unstable_body() {
}
#[subxt_test]
async fn chainhead_unstable_header() {
async fn chainhead_v1_header() {
let ctx = test_context().await;
let rpc = ctx.unstable_rpc_methods().await;
let legacy_rpc = ctx.legacy_rpc_methods().await;
let mut blocks = rpc.chainhead_unstable_follow(false).await.unwrap();
let mut blocks = rpc.chainhead_v1_follow(false).await.unwrap();
let event = blocks.next().await.unwrap().unwrap();
let hash = match event {
FollowEvent::Initialized(init) => *init.finalized_block_hashes.last().unwrap(),
@@ -109,7 +109,7 @@ async fn chainhead_unstable_header() {
.unwrap()
.unwrap();
let old_header = rpc
.chainhead_unstable_header(sub_id, hash)
.chainhead_v1_header(sub_id, hash)
.await
.unwrap()
.unwrap();
@@ -118,12 +118,12 @@ async fn chainhead_unstable_header() {
}
#[subxt_test]
async fn chainhead_unstable_storage() {
async fn chainhead_v1_storage() {
let ctx = test_context().await;
let api = ctx.client();
let rpc = ctx.unstable_rpc_methods().await;
let mut blocks = rpc.chainhead_unstable_follow(false).await.unwrap();
let mut blocks = rpc.chainhead_v1_follow(false).await.unwrap();
let event = blocks.next().await.unwrap().unwrap();
let hash = match event {
FollowEvent::Initialized(init) => *init.finalized_block_hashes.last().unwrap(),
@@ -142,7 +142,7 @@ async fn chainhead_unstable_storage() {
// Fetch storage.
let response = rpc
.chainhead_unstable_storage(sub_id, hash, items, None)
.chainhead_v1_storage(sub_id, hash, items, None)
.await
.unwrap();
let operation_id = match response {
@@ -164,11 +164,11 @@ async fn chainhead_unstable_storage() {
}
#[subxt_test]
async fn chainhead_unstable_call() {
async fn chainhead_v1_call() {
let ctx = test_context().await;
let rpc = ctx.unstable_rpc_methods().await;
let mut blocks = rpc.chainhead_unstable_follow(true).await.unwrap();
let mut blocks = rpc.chainhead_v1_follow(true).await.unwrap();
let event = blocks.next().await.unwrap().unwrap();
let hash = match event {
FollowEvent::Initialized(init) => *init.finalized_block_hashes.last().unwrap(),
@@ -179,7 +179,7 @@ async fn chainhead_unstable_call() {
let alice_id = dev::alice().public_key().to_account_id();
// Runtime API call.
let response = rpc
.chainhead_unstable_call(
.chainhead_v1_call(
sub_id,
hash,
"AccountNonceApi_account_nonce",
@@ -201,11 +201,11 @@ async fn chainhead_unstable_call() {
}
#[subxt_test]
async fn chainhead_unstable_unpin() {
async fn chainhead_v1_unpin() {
let ctx = test_context().await;
let rpc = ctx.unstable_rpc_methods().await;
let mut blocks = rpc.chainhead_unstable_follow(true).await.unwrap();
let mut blocks = rpc.chainhead_v1_follow(true).await.unwrap();
let event = blocks.next().await.unwrap().unwrap();
let hash = match event {
FollowEvent::Initialized(init) => *init.finalized_block_hashes.last().unwrap(),
@@ -213,9 +213,9 @@ async fn chainhead_unstable_unpin() {
};
let sub_id = blocks.subscription_id().unwrap();
assert!(rpc.chainhead_unstable_unpin(sub_id, hash).await.is_ok());
assert!(rpc.chainhead_v1_unpin(sub_id, hash).await.is_ok());
// The block was already unpinned.
assert!(rpc.chainhead_unstable_unpin(sub_id, hash).await.is_err());
assert!(rpc.chainhead_v1_unpin(sub_id, hash).await.is_err());
}
#[cfg(fullclient)]