ufotofu 0.12.2

Abstractions for lazily consuming and producing sequences
Documentation 
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

use crate::errors::ProcessWhileError;
use crate::prelude::*;

/// A wrapper for a [`consumer`](Consumer) which continues to consume items only while those items meet a given predicate.
///
/// Use the [`into_inner`](`ConsumeWhile::into_inner`) method or [`AsRef<C>`](AsRef) impl to access the wrapped consumer.
///
/// Created via the [`ConsumerExt::to_consume_while`] method.
///
/// <br/>Counterpart: the [producer::ProduceWhile] type.
pub struct ConsumeWhile<C, Pred> {
    inner: C,
    predicate: Pred,
}

impl<C, Pred> ConsumeWhile<C, Pred> {
    pub(crate) fn new(consumer: C, predicate: Pred) -> ConsumeWhile<C, Pred>
    where
        C: Consumer,
        Pred: AsyncFnMut(&C::Item) -> bool,
    {
        ConsumeWhile {
            inner: consumer,
            predicate,
        }
    }

    /// Retrieves the wrapped [`Consumer`], dropping the provided predicate.
    pub fn into_inner(self) -> C {
        self.inner
    }
}

impl<C, Pred> AsRef<C> for ConsumeWhile<C, Pred> {
    fn as_ref(&self) -> &C {
        &self.inner
    }
}

impl<C, Pred> Consumer for ConsumeWhile<C, Pred>
where
    C: Consumer,
    Pred: AsyncFnMut(&C::Item) -> bool,
{
    type Item = C::Item;

    type Final = C::Final;

    type Error = ProcessWhileError<C::Item, C::Error>;

    async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
        match val {
            Left(item) => {
                if (self.predicate)(&item).await {
                    self.inner
                        .consume(Left(item))
                        .await
                        .map_err(ProcessWhileError::Inner)
                } else {
                    Err(ProcessWhileError::PredicateFailed(item))
                }
            }
            Right(fin) => self
                .inner
                .consume(Right(fin))
                .await
                .map_err(ProcessWhileError::Inner),
        }
    }

    async fn flush(&mut self) -> Result<(), Self::Error> {
        self.inner.flush().await.map_err(ProcessWhileError::Inner)
    }
}