moirai-iter 0.7.0

Parallel and async iterator combinators for Moirai concurrency library
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
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

use std::ops::ControlFlow;

use super::ParallelIterator;

mod private {
    pub trait Sealed {}

    impl<T> Sealed for Option<T> {}
    impl<T, E> Sealed for Result<T, E> {}
}

/// Sealed fallible stream item contract for `try_reduce_with`.
pub trait TryStreamItem: private::Sealed + Send {
    /// The success value carried by a `Continue` branch.
    type Output: Send;

    /// Split the item into its success value or a terminating `Break` carrying the failure item.
    fn branch(self) -> ControlFlow<Self, Self::Output>
    where
        Self: Sized;

    /// Reconstruct the item from a success value.
    fn from_output(output: Self::Output) -> Self;
}

impl<T> TryStreamItem for Option<T>
where
    T: Send,
{
    type Output = T;

    fn branch(self) -> ControlFlow<Self, Self::Output> {
        match self {
            Some(value) => ControlFlow::Continue(value),
            None => ControlFlow::Break(None),
        }
    }

    fn from_output(output: Self::Output) -> Self {
        Some(output)
    }
}

impl<T, E> TryStreamItem for Result<T, E>
where
    T: Send,
    E: Send,
{
    type Output = T;

    fn branch(self) -> ControlFlow<Self, Self::Output> {
        match self {
            Ok(value) => ControlFlow::Continue(value),
            Err(error) => ControlFlow::Break(Err(error)),
        }
    }

    fn from_output(output: Self::Output) -> Self {
        Ok(output)
    }
}

pub(in crate::parallel) fn try_reduce_with<I, F>(iterator: I, reduce_fn: F) -> Option<I::Item>
where
    I: ParallelIterator,
    I::Item: TryStreamItem,
    F: Fn(<I::Item as TryStreamItem>::Output, <I::Item as TryStreamItem>::Output) -> I::Item
        + Send
        + Sync
        + Clone,
{
    try_reduce_with_items(iterator.seq_items(), reduce_fn)
}

pub(in crate::parallel) fn try_reduce_with_items<Items, Item, F>(
    items: Items,
    reduce_fn: F,
) -> Option<Item>
where
    Items: IntoIterator<Item = Item>,
    Item: TryStreamItem,
    F: Fn(<Item as TryStreamItem>::Output, <Item as TryStreamItem>::Output) -> Item
        + Send
        + Sync
        + Clone,
{
    let mut items = items.into_iter();
    let first = items.next()?;
    let mut accumulator = match first.branch() {
        ControlFlow::Continue(value) => value,
        ControlFlow::Break(residual) => return Some(residual),
    };

    for item in items {
        let value = match item.branch() {
            ControlFlow::Continue(value) => value,
            ControlFlow::Break(residual) => return Some(residual),
        };
        accumulator = match reduce_fn(accumulator, value).branch() {
            ControlFlow::Continue(value) => value,
            ControlFlow::Break(residual) => return Some(residual),
        };
    }

    Some(Item::from_output(accumulator))
}