combine-latest 1.0.0

Combines two streams into a new stream which yields tuples with the latest values from each input stream
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

use std::future;

use either::Either;
use futures_core::Stream;
use futures_util::{stream::select, StreamExt};

/// Combines two streams into a new stream that always contains the latest items from both streams
/// as a tuple. This stream won't yield a tuple until both input streams have yielded at least one
/// item each.
pub fn combine_latest<T1: Clone, T2: Clone>(
    s1: impl Stream<Item = T1>,
    s2: impl Stream<Item = T2>,
) -> impl Stream<Item = (T1, T2)> {
    combine_latest_optional(s1, s2).filter_map(|(v1, v2)| future::ready(v1.zip(v2)))
}

/// Combines two streams into a new stream, yielding tuples of `(Option<T1>, Option<T2>)`. The
/// stream starts yielding tuples as soon as one of the input streams yields an item, and the one
/// that has not yet yielded has a corresponding `None` in it's field of the tuple.
pub fn combine_latest_optional<T1: Clone, T2: Clone>(
    s1: impl Stream<Item = T1>,
    s2: impl Stream<Item = T2>,
) -> impl Stream<Item = (Option<T1>, Option<T2>)> {
    let mut current1 = None;
    let mut current2 = None;
    select(s1.map(Either::Left), s2.map(Either::Right)).map(move |either| {
        match either {
            Either::Left(l) => current1 = Some(l),
            Either::Right(r) => current2 = Some(r),
        };
        (current1.clone(), current2.clone())
    })
}