Introduce `stream::try_channel` helper

1 files changed, 21 insertions, 0 deletions

diff --git a/futures/src/stream.rs b/futures/src/stream.rs
index 06f9230d..af2f8c99 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<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)
+}