use futures::channel::mpsc;
use futures::stream::{self, Stream, StreamExt};
use std::future::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)
}
pub fn try_channel<T, E, F>(
size: usize,
f: impl FnOnce(mpsc::Sender<T>) -> F,
) -> impl Stream<Item = Result<T, E>>
where
F: Future<Output = Result<(), E>>,
{
let (sender, receiver) = mpsc::channel(size);
let runner = stream::once(f(sender)).filter_map(|result| async {
match result {
Ok(()) => None,
Err(error) => Some(Err(error)),
}
});
stream::select(receiver.map(Ok), runner)
}