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
macro_rules! pages_ahead_unordered_state_def {
($($extra_bounds:tt)*) => {
struct PagesAheadUnorderedState<'p, P, R>
where
P: 'p + PageTurner<R>,
$($extra_bounds)*
{
page_turner: P,
numbered_requests: EnumerableRequestChunks<R>,
in_progress: FuturesUnordered<NumberedRequestFuture<'p, P, R>>,
first_error: Option<(usize, PageError<P, R>)>,
last_page: Option<usize>,
}
impl<'p, P, R> PagesAheadUnorderedState<'p, P, R>
where
P: 'p + PageTurner<R>,
R: 'p + RequestAhead,
$($extra_bounds)*
{
fn new(page_turner: P, request: R, chunk_size: usize, limit: Limit) -> Self {
let numbered_requests = RequestIter::new(request, limit)
.enumerate()
.chunks(chunk_size);
Self {
page_turner,
numbered_requests,
in_progress: FuturesUnordered::new(),
first_error: None,
last_page: None,
}
}
/// Updates the error so that an error with the least `new_err_num` remains while other ones
/// get discarded
fn update_err(&mut self, new_err_num: usize, new_err: PageError<P, R>) {
match &self.first_error {
Some((old_err_num, _)) if new_err_num < *old_err_num => {
self.first_error = Some((new_err_num, new_err));
}
Some(_) => {}
None => self.first_error = Some((new_err_num, new_err)),
}
}
}
};
}
macro_rules! request_pages_ahead_unordered_decl {
($($extra_bounds:tt)*) => {
async fn request_pages_ahead_unordered<'p, P, R>(
mut state: Box<PagesAheadUnorderedState<'p, P, R>>,
) -> Result<Option<(PageItems<P, R>, Box<PagesAheadUnorderedState<'p, P, R>>)>, PageError<P, R>>
where
P: 'p + Clone + PageTurner<R>,
R: 'p + RequestAhead,
$($extra_bounds)*
{
// This and nested loops are required to discard all errors except the error for the first failed request without yielding them to the user.
loop {
// Once we're in this branch no code below will be executed
if let Some(last_page_num) = state.last_page {
while let Some((num, result)) = state.in_progress.next().await {
match result {
Ok(turned_page) => return Ok(Some((turned_page.items, state))),
Err(new_err) => {
state.update_err(num, new_err);
}
}
}
match state.first_error.take() {
Some((err_num, err)) if err_num <= last_page_num => {
return Err(err);
}
// If an error occured past the last existing page it will be discarded at this
// point
_ => {
return Ok(None);
}
}
}
// Once we're in this branch no code below will be executed
while state.first_error.is_some() {
match state.in_progress.next().await {
Some((num, result)) => match result {
Ok(TurnedPage {
items,
next_request,
}) => {
if next_request.is_none() {
state.last_page = Some(num);
}
return Ok(Some((items, state)));
}
Err(new_err) => state.update_err(num, new_err),
},
// If at least one of `requests_ahead_count` futures returned an error and
// we haven't found the last page in other responses - return the first error
None => return Err(state.first_error.unwrap().1),
}
}
// Schedule
if state.in_progress.is_empty() {
// Initial schedule of the first futures chunk
match state.numbered_requests.next_chunk() {
// If chunk is some then there is at least 1 request inside
Some(chunk) => {
for req in chunk {
let local_page_turner = state.page_turner.clone();
state.in_progress.push(Box::pin(async move {
(req.0, local_page_turner.turn_page(req.1).await)
}));
}
}
None => {
return Ok(None);
}
}
} else {
// At this point one of the first requests succeeded. Lets push the next one from the next_chunk to proceed in
// a sliding window maner.
if let Some(req) = state.numbered_requests.next_item() {
let local_page_turner = state.page_turner.clone();
state.in_progress.push(Box::pin(async move {
(req.0, local_page_turner.turn_page(req.1).await)
}))
}
}
match state.in_progress.next().await {
Some((num, result)) => match result {
Ok(TurnedPage {
items,
next_request,
}) => {
if next_request.is_none() {
state.last_page = Some(num);
}
return Ok(Some((items, state)));
}
// Don't return an error immediately, continue the loop to find the one for the
// first failed page instead, or to discard an error if it occured past the last existing page
Err(new_err) => state.update_err(num, new_err),
},
None => {
unreachable!(
"BUG(page-turner): We ensured that the unordered futures queue is not empty right above")
}
}
}
}
};
}
pub(crate) use pages_ahead_unordered_state_def;
pub(crate) use request_pages_ahead_unordered_decl;