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
use crate::prelude::*;
use crate::producer::Status;
/// A wrapper for a [`Producer`] which tracks the [`Status`] of the wrapped producer, and the number of [`items`](Producer::Item) produced since the wrapper was created.
///
/// Use the [`into_inner`](Stats::into_inner) method to consume the wrapper and access the wrapped consumer.
///
/// Created via the [`ProducerExt::to_stats`] method.
///
/// <br/>Counterpart: the [`consumer::Stats`] type.
pub struct Stats<C> {
inner: C,
count: usize,
status: Status,
}
impl<P> Stats<P> {
pub(crate) fn new(inner: P) -> Self {
Stats {
inner,
count: 0,
status: Status::Processing,
}
}
/// Returns the number of [`items`](Producer::Item) which have been produced since since this wrapper was created.
pub fn count(&self) -> usize {
self.count
}
/// Returns the [`Status`] of the [`Producer`].
pub fn status(&self) -> Status {
self.status
}
/// Consumes this wrapper to return the inner [`Producer`].
pub fn into_inner(self) -> P {
self.inner
}
}
impl<P> Producer for Stats<P>
where
P: Producer,
{
type Item = P::Item;
type Final = P::Final;
type Error = P::Error;
async fn produce(&mut self) -> Result<Either<Self::Item, Self::Final>, Self::Error> {
match self.inner.produce().await {
Ok(Left(item)) => {
self.count = self
.count
.checked_add(1)
.expect("producer stats cannot count more than usize::MAX items");
Ok(Left(item))
}
Ok(Right(fin)) => {
self.status = Status::Finalised;
Ok(Right(fin))
}
Err(err) => {
self.status = Status::Errored;
Err(err)
}
}
}
async fn slurp(&mut self) -> Result<(), Self::Error> {
let result = self.inner.slurp().await;
if result.is_err() {
self.status = Status::Errored;
}
result
}
}
impl<P> BulkProducer for Stats<P>
where
P: BulkProducer,
{
async fn expose_items_gracefully<F, R>(
&mut self,
f: F,
) -> Result<Either<R, (F, Self::Final)>, (F, Self::Error)>
where
F: AsyncFnOnce(&[Self::Item]) -> (usize, R),
{
let mut f = Some(f);
match self
.inner
.expose_items_gracefully(async |items| {
let f = f.take().expect("constructed as a Some variant");
let (produced, result) = f(items).await;
self.count = self
.count
.checked_add(produced)
.expect("producer stats cannot count more than usize::MAX items");
(produced, result)
})
.await
{
Ok(Left(result)) => Ok(Left(result)),
Ok(Right((_, fin))) => {
self.status = Status::Finalised;
Ok(Right((
f.take().expect(
"provided closure must not be called when the final item is encountered",
),
fin,
)))
}
Err((_, err)) => {
self.status = Status::Errored;
Err((
f.take()
.expect("provided closure must not be called when an error occurs"),
err,
))
}
}
}
}