Introduce stream::try_channel helper
This commit is contained in:
parent
70f44a6e26
commit
4ce2a207a6
1 changed files with 21 additions and 0 deletions
|
|
@ -23,3 +23,24 @@ where
|
||||||
|
|
||||||
stream::select(receiver, runner)
|
stream::select(receiver, runner)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Creates a new [`Stream`] that produces the items sent from a [`Future`]
|
||||||
|
/// that can fail to the [`mpsc::Sender`] provided to the closure.
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue