1 files changed, 26 insertions, 0 deletions
diff --git a/futures/src/stream.rs b/futures/src/stream.rs
new file mode 100644
index 00000000..2ec505f1
--- /dev/null
+++ b/futures/src/stream.rs
@@ -0,0 +1,26 @@
+//! Create asynchronous streams of data.
+use futures::channel::mpsc;
+use futures::never::Never;
+use futures::stream::{self, Stream, StreamExt};
+
+use std::future::Future;
+
+/// Creates a new [`Stream`] that produces the items sent from a [`Future`]
+/// to the [`mpsc::Sender`] provided to the closure.
+///
+/// This is a more ergonomic [`stream::unfold`], which allows you to go
+/// from the "world of futures" to the "world of streams" by simply looping
+/// and publishing to an async channel from inside a [`Future`].
+pub fn channel<T, F>(
+    size: usize,
+    f: impl FnOnce(mpsc::Sender<T>) -> F,
+) -> impl Stream<Item = T>
+where
+    F: Future<Output = Never>,
+{
+    let (sender, receiver) = mpsc::channel(size);
+
+    let runner = stream::once(f(sender)).map(|_| unreachable!());
+
+    stream::select(receiver, runner)
+}