rskit_stream/
source.rs

1use futures_util::stream;
2use tokio::sync::mpsc;
3
4/// Create a stream from a `Vec<T>`.
5pub fn from_slice<T: Send + 'static>(
6    items: Vec<T>,
7) -> impl futures::Stream<Item = T> + Send + 'static {
8    stream::iter(items)
9}
10
11/// Create a lazy stream from an async generator function.
12///
13/// `f` is called repeatedly; it should return `Some(item)` while there
14/// are items, and `None` to terminate.
15pub fn from_fn<T, F, Fut>(mut f: F) -> impl futures::Stream<Item = T> + Send + 'static
16where
17    T: Send + 'static,
18    F: FnMut() -> Fut + Send + 'static,
19    Fut: std::future::Future<Output = Option<T>> + Send + 'static,
20{
21    stream::unfold((), move |()| {
22        let fut = f();
23        async move { fut.await.map(|v| (v, ())) }
24    })
25}
26
27/// Wrap a `tokio::sync::mpsc::Receiver` as a `Stream`.
28pub fn from_channel<T: Send + 'static>(
29    rx: mpsc::Receiver<T>,
30) -> impl futures::Stream<Item = T> + Send + 'static {
31    tokio_stream::wrappers::ReceiverStream::new(rx)
32}
rskit_stream/source.rs

rskit_stream/
source.rs
