rskit-stream 0.2.0-alpha.5

Foundational async stream toolkit: bounded fan-out broadcaster, sources, cancellable tasks, and futures::Stream extension operators
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

use futures_util::stream;
use tokio::sync::mpsc;

/// Create a stream from a `Vec<T>`.
pub fn from_slice<T: Send + 'static>(
    items: Vec<T>,
) -> impl futures::Stream<Item = T> + Send + 'static {
    stream::iter(items)
}

/// Create a lazy stream from an async generator function.
///
/// `f` is called repeatedly; it should return `Some(item)` while there are items,
/// and `None` to terminate.
pub fn from_fn<T, F, Fut>(mut f: F) -> impl futures::Stream<Item = T> + Send + 'static
where
    T: Send + 'static,
    F: FnMut() -> Fut + Send + 'static,
    Fut: std::future::Future<Output = Option<T>> + Send + 'static,
{
    stream::unfold((), move |()| {
        let fut = f();
        async move { fut.await.map(|v| (v, ())) }
    })
}

/// Wrap a `tokio::sync::mpsc::Receiver` as a `Stream`.
pub fn from_channel<T: Send + 'static>(
    rx: mpsc::Receiver<T>,
) -> impl futures::Stream<Item = T> + Send + 'static {
    tokio_stream::wrappers::ReceiverStream::new(rx)
}