#input-stream #async-stream #output-stream #latest #combine #tuple #values

combine-latest

Combines two streams into a new stream which yields tuples with the latest values from each input stream

4 stable releases

1.1.2 Aug 13, 2023
1.1.1 Feb 24, 2023
1.0.0 Feb 22, 2023

#11 in #latest

Download history 10/week @ 2024-06-22 2/week @ 2024-06-29 7/week @ 2024-07-06 25/week @ 2024-07-13 14/week @ 2024-07-20 8/week @ 2024-07-27 42/week @ 2024-08-03 22/week @ 2024-08-10 16/week @ 2024-08-17 32/week @ 2024-08-24 87/week @ 2024-08-31 66/week @ 2024-09-07 67/week @ 2024-09-14 96/week @ 2024-09-21 1/week @ 2024-09-28 15/week @ 2024-10-05

185 downloads per month

MIT license

8KB
71 lines

combine-latest

Combines two streams into a new stream which yields tuples with the latest values from each input stream. Inspired by Rx's combineLatest.

use async_stream::stream;
use combine_latest::combine_latest;

fn combine_weather_data_streams(
    temperature: impl Stream<i32>,
    weather_notes: impl Stream<String>,
) -> Stream<String> {
    stream! {
        for await (t, n) in combine_latest(temperature, weather_notes) {
            yield format!("Temperature {t}°, note: {n}");
        }
    }
}

combine_latest won't yield its first tuple until both input streams have yielded a value. If you need to get items as soon as the first is available, there is a combine_latest_opt function that yields (Option<T1>, Option<T2>) tuples.

As values come in over time on the temperature and weather_notes streams, combine_latest and combine_latest_opt will yield values like so:

time temperature weather_notes combine_latest combine_latest_opt
0 25 (Some(25), None)
1 26 (Some(26), None)
2 Low visibility (26, "Low visibility") (Some(26), Some("Low visibility"))
3 Foggy (26, "Foggy") (Some(26), Some("Foggy"))
4 25 (25, "Foggy") (Some(25), Some("Foggy"))

Since the same input value might be returned several times in the output stream, the items that the input streams yield must implement Clone.

For types that don't implement Clone, it's possible to use the function map_latest:

fn combine_weather_data_streams(
    temperature: impl Stream<NonCloneTemperature>,
    weather_notes: impl Stream<String>,
) -> Stream<String> {
    stream! {
        for await output in map_latest(
            temperature,
            weather_notes,
            |t, n| format!("Temperature {t}°, note: {n}"),
        ) {
            yield output;
        }
    }
}

To add to you project: cargo add combine-latest

Dependencies

~560–760KB
~14K SLoC