blob: 06f9230d88c80992fd4474036510f1bdfbf202f3 (
plain) (
blame)
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
|
//! Create asynchronous streams of data.
use futures::channel::mpsc;
use futures::stream::{self, Stream, StreamExt};
use std::future::Future;
/// Creates a new [`Stream`] that produces the items sent from a [`Future`]
/// to the [`mpsc::Sender`] provided to the closure.
///
/// This is a more ergonomic [`stream::unfold`], which allows you to go
/// from the "world of futures" to the "world of streams" by simply looping
/// and publishing to an async channel from inside a [`Future`].
pub fn channel<T, F>(
size: usize,
f: impl FnOnce(mpsc::Sender<T>) -> F,
) -> impl Stream<Item = T>
where
F: Future<Output = ()>,
{
let (sender, receiver) = mpsc::channel(size);
let runner = stream::once(f(sender)).filter_map(|_| async { None });
stream::select(receiver, runner)
}
|
