futuristic 0.5.0

Extensions to the futures crate
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
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

// Copyright (C) 2022 Stephane Raux. Distributed under the 0BSD license.

//! Tools for streams

use futures::Stream;

pub use zip_latest::ZipLatest;
pub use zip_latest_all::ZipLatestAll;
pub use zip_latest_with::ZipLatestWith;
pub use zip_latest_with_all::ZipLatestWithAll;

mod zip_latest;
mod zip_latest_all;
mod zip_latest_with;
mod zip_latest_with_all;

/// Extension trait for [`Stream`](futures::Stream).
pub trait StreamTools: Stream {
    /// Zips two streams using their latest values when one is not ready
    ///
    /// The zipped stream keeps the latest items produced by both streams. If one of the underlying
    /// streams is exhausted or not ready and the other stream yields a new item, it is combined
    /// with the latest item from the stream that did not yield anything new.
    ///
    /// The zipped stream ends when both underlying streams end, or if one of the streams ends
    /// without ever producing an item.
    ///
    /// Visually, this gives:
    /// ```text
    /// ---0-----------1-----------------2-------> self
    /// ------10-------11-------12---------------> other
    /// ------10-------12-------13-------14------> self.zip_latest_with(other, |a, b| a + b)
    /// ```
    fn zip_latest_with<S, F, T>(self, other: S, combine: F) -> ZipLatestWith<Self, S, F>
    where
        Self: Sized,
        S: Stream,
        F: FnMut(&Self::Item, &S::Item) -> T,
    {
        ZipLatestWith::new(self, other, combine)
    }

    /// Zips two streams using their latest values when one is not ready
    ///
    /// The zipped stream keeps a copy of the latest items produced by both streams. If one of the
    /// underlying streams is exhausted or not ready and the other stream yields a new item, it is
    /// returned alongside the latest item from the stream that did not yield anything new.
    ///
    /// The zipped stream ends when both underlying streams end, or if one of the streams ends
    /// without ever producing an item.
    ///
    /// Visually, this gives:
    /// ```text
    /// ---a-----------b-----------------c-------> self
    /// ------0--------1--------2----------------> other
    /// ------(a, 0)---(b, 1)---(b, 2)---(c, 2)--> self.zip_latest(other)
    /// ```
    fn zip_latest<S>(self, other: S) -> ZipLatest<Self, S>
    where
        Self: Sized,
        Self::Item: Clone,
        S: Stream,
        S::Item: Clone,
    {
        ZipLatest::new(self, other)
    }
}

impl<S: Stream> StreamTools for S {}

/// Zips multiple streams using their latest values for the ones that are not ready
///
/// The zipped stream keeps the latest items produced by all streams. If one of the underlying
/// streams is exhausted or not ready and at least one of the other streams yields a new item, it is
/// combined with the latest items from the streams that did not yield anything new.
///
/// The zipped stream ends when all underlying streams end, or if one of the streams ends
/// without ever producing an item.
///
/// Visually, this gives:
/// ```text
/// ---0-----------1-----------------2-------> a
/// ------10-------11-------12---------------> b
/// ------10-------12-------13-------14------> zip_latest_with_all([a, b], |a, b| a + b)
/// ```
pub fn zip_latest_with_all<I, F, T>(streams: I, combine: F) -> ZipLatestWithAll<I::Item, F>
where
    I: IntoIterator,
    I::Item: Stream + Unpin,
    F: FnMut(&[<I::Item as Stream>::Item]) -> T,
{
    ZipLatestWithAll::new(streams, combine)
}

/// Zips multiple streams using their latest values for the ones that are not ready
///
/// The zipped stream keeps a copy of the latest items produced by all streams. If one of the
/// underlying streams is exhausted or not ready and at least one of the other streams yields a new
/// item, it is returned alongside the latest items from the streams that did not yield anything
/// new.
///
/// The zipped stream ends when all underlying streams end, or if one of the streams ends
/// without ever producing an item.
///
/// Visually, this gives:
/// ```text
/// ---0--------------------1--------------------------> a
/// ------10----------------11-------------------------> b
/// ----------20--------------------------21-----------> c
/// ----------[0, 10, 20]---[1, 11, 20]---[1, 11, 21]--> zip_latest_all([a, b, c])
/// ```
pub fn zip_latest_all<I>(streams: I) -> ZipLatestAll<I::Item>
where
    I: IntoIterator,
    I::Item: Stream + Unpin,
    <I::Item as Stream>::Item: Clone,
{
    ZipLatestAll::new(streams)
}

#[cfg(test)]
mod test_util {
    use crate::future::yield_now;
    use futures::{Stream, StreamExt};

    pub fn yield_on_none<I, T>(items: I) -> impl Stream<Item = T>
    where
        I: IntoIterator<Item = Option<T>>,
    {
        futures::stream::iter(items).filter_map(|x| async move {
            if x.is_none() {
                yield_now().await;
            }
            x
        })
    }
}