1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
use futures::StreamExt;
use log::info;
use subxt::events::StaticEvent;
use crate::{
aleph_zero,
api::session::events::NewSession,
connections::AsConnection,
pallets::{session::SessionApi, staking::StakingApi},
EraIndex, SessionIndex,
};
/// When using waiting API, what kind of block status we should wait for.
pub enum BlockStatus {
/// Wait for event or block to be in the best chain.
Best,
/// Wait for the event or block to be in the finalized chain.
Finalized,
}
/// Waiting _for_ various events API
#[async_trait::async_trait]
pub trait AlephWaiting {
/// Wait for a particular block to be in a [`BlockStatus`].
/// Block number must match given predicate.
/// * `predicate` - a `u32` -> `bool` functor, first argument is a block number
/// * `status` - a [`BlockStatus`] of the block we wait for
///
/// # Examples
/// ```ignore
/// let finalized = connection.connection.get_finalized_block_hash().await;
/// let finalized_number = connection
/// .connection
/// .get_block_number(finalized)
/// .await
/// .unwrap();
/// connection
/// .connection
/// .wait_for_block(|n| n > finalized_number, BlockStatus::Finalized)
/// .await;
/// ```
async fn wait_for_block<P: Fn(u32) -> bool + Send>(&self, predicate: P, status: BlockStatus);
/// Wait for a particular event to be emitted on chain.
/// * `predicate` - a predicate that has one argument (ref to an emitted event)
/// * `status` - a [`BlockStatus`] of the event we wait for
///
/// # Examples
/// ```ignore
/// let event = connection
/// .wait_for_event(
/// |event: &BanValidators| {
/// info!("Received BanValidators event: {:?}", event.0);
/// true
/// },
/// BlockStatus::Best,
/// )
/// .await;
/// ```
async fn wait_for_event<T: StaticEvent, P: Fn(&T) -> bool + Send>(
&self,
predicate: P,
status: BlockStatus,
) -> T;
/// Wait for given era to happen.
/// * `era` - number of the era to wait for
/// * `status` - a [`BlockStatus`] of the era we wait for
async fn wait_for_era(&self, era: EraIndex, status: BlockStatus);
/// Wait for given session to happen.
/// * `session` - number of the session to wait for
/// * `status` - a [`BlockStatus`] of the session we wait for
async fn wait_for_session(&self, session: SessionIndex, status: BlockStatus);
}
/// nWaiting _from_ the current moment of time API
#[async_trait::async_trait]
pub trait WaitingExt {
/// Wait for a given number of sessions to wait from a current session.
/// `n` - number of sessions to wait from now
/// * `status` - a [`BlockStatus`] of the session we wait for
async fn wait_for_n_sessions(&self, n: SessionIndex, status: BlockStatus);
/// Wait for a given number of eras to wait from a current era.
/// `n` - number of eras to wait from now
/// * `status` - a [`BlockStatus`] of the era we wait for
async fn wait_for_n_eras(&self, n: EraIndex, status: BlockStatus);
}
#[async_trait::async_trait]
impl<C: AsConnection + Sync> AlephWaiting for C {
async fn wait_for_block<P: Fn(u32) -> bool + Send>(&self, predicate: P, status: BlockStatus) {
let mut block_sub = match status {
BlockStatus::Best => self
.as_connection()
.as_client()
.blocks()
.subscribe_best()
.await
.expect("Failed to subscribe to the best block stream!"),
BlockStatus::Finalized => self
.as_connection()
.as_client()
.blocks()
.subscribe_finalized()
.await
.expect("Failed to subscribe to the finalized block stream!"),
};
while let Some(Ok(block)) = block_sub.next().await {
if predicate(block.number()) {
return;
}
}
}
async fn wait_for_event<T: StaticEvent, P: Fn(&T) -> bool + Send>(
&self,
predicate: P,
status: BlockStatus,
) -> T {
let mut block_sub = match status {
BlockStatus::Best => self
.as_connection()
.as_client()
.blocks()
.subscribe_best()
.await
.expect("Failed to subscribe to the best block stream!"),
BlockStatus::Finalized => self
.as_connection()
.as_client()
.blocks()
.subscribe_finalized()
.await
.expect("Failed to subscribe to the finalized block stream!"),
};
info!(target: "aleph-client", "waiting for event {}.{}", T::PALLET, T::EVENT);
while let Some(Ok(block)) = block_sub.next().await {
let events = match block.events().await {
Ok(events) => events,
_ => continue,
};
for event in events.iter() {
let event = event.expect("Failed to obtain event from the block!");
if let Ok(Some(ev)) = event.as_event::<T>() {
if predicate(&ev) {
return ev;
}
}
}
}
panic!("No more blocks");
}
async fn wait_for_era(&self, era: EraIndex, status: BlockStatus) {
let addrs = aleph_zero::api::constants().staking().sessions_per_era();
let sessions_per_era = self
.as_connection()
.as_client()
.constants()
.at(&addrs)
.expect("Failed to obtain sessions_per_era const!");
let first_session_in_era = era * sessions_per_era;
self.wait_for_session(first_session_in_era, status).await;
}
async fn wait_for_session(&self, session: SessionIndex, status: BlockStatus) {
self.wait_for_event(|event: &NewSession| {
info!(target: "aleph-client", "waiting for session {:?}, current session {:?}", session, event.session_index);
event.session_index >= session
}, status)
.await;
}
}
#[async_trait::async_trait]
impl<C: AsConnection + Sync> WaitingExt for C {
async fn wait_for_n_sessions(&self, n: SessionIndex, status: BlockStatus) {
let current_session = self.get_session(None).await;
self.wait_for_session(current_session + n, status).await;
}
async fn wait_for_n_eras(&self, n: EraIndex, status: BlockStatus) {
let current_era = self.get_current_era(None).await;
self.wait_for_era(current_era + n, status).await;
}
}