diff --git a/futures/src/stream.rs b/futures/src/stream.rs index 06f9230d88..af2f8c990c 100644 --- a/futures/src/stream.rs +++ b/futures/src/stream.rs @@ -23,3 +23,24 @@ where 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( + size: usize, + f: impl FnOnce(mpsc::Sender) -> F, +) -> impl Stream> +where + F: Future>, +{ + 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) +}