Implement `subscription::channel` and simplify `unfold`

3 files changed, 122 insertions, 86 deletions

diff --git a/examples/download_progress/src/download.rs b/examples/download_progress/src/download.rs
index 39dd843f..cd7647e8 100644
--- a/examples/download_progress/src/download.rs
+++ b/examples/download_progress/src/download.rs
@@ -18,10 +18,7 @@ pub struct Download<I> {
     url: String,
 }
 
-async fn download<I: Copy>(
-    id: I,
-    state: State,
-) -> (Option<(I, Progress)>, State) {
+async fn download<I: Copy>(id: I, state: State) -> ((I, Progress), State) {
     match state {
         State::Ready(url) => {
             let response = reqwest::get(&url).await;
@@ -30,7 +27,7 @@ async fn download<I: Copy>(
                 Ok(response) => {
                     if let Some(total) = response.content_length() {
                         (
-                            Some((id, Progress::Started)),
+                            (id, Progress::Started),
                             State::Downloading {
                                 response,
                                 total,
@@ -38,10 +35,10 @@ async fn download<I: Copy>(
                             },
                         )
                     } else {
-                        (Some((id, Progress::Errored)), State::Finished)
+                        ((id, Progress::Errored), State::Finished)
                     }
                 }
-                Err(_) => (Some((id, Progress::Errored)), State::Finished),
+                Err(_) => ((id, Progress::Errored), State::Finished),
             }
         }
         State::Downloading {
@@ -55,7 +52,7 @@ async fn download<I: Copy>(
                 let percentage = (downloaded as f32 / total as f32) * 100.0;
 
                 (
-                    Some((id, Progress::Advanced(percentage))),
+                    (id, Progress::Advanced(percentage)),
                     State::Downloading {
                         response,
                         total,
@@ -63,8 +60,8 @@ async fn download<I: Copy>(
                     },
                 )
             }
-            Ok(None) => (Some((id, Progress::Finished)), State::Finished),
-            Err(_) => (Some((id, Progress::Errored)), State::Finished),
+            Ok(None) => ((id, Progress::Finished), State::Finished),
+            Err(_) => ((id, Progress::Errored), State::Finished),
         },
         State::Finished => {
             // We do not let the stream die, as it would start a
diff --git a/examples/websocket/src/echo.rs b/examples/websocket/src/echo.rs
index e74768a6..4fabb660 100644
--- a/examples/websocket/src/echo.rs
+++ b/examples/websocket/src/echo.rs
@@ -13,63 +13,67 @@ use std::fmt;
 pub fn connect() -> Subscription<Event> {
     struct Connect;
 
-    subscription::unfold(
+    subscription::channel(
         std::any::TypeId::of::<Connect>(),
-        State::Disconnected,
-        |state| async move {
-            match state {
-                State::Disconnected => {
-                    const ECHO_SERVER: &str = "ws://localhost:3030";
-
-                    match async_tungstenite::tokio::connect_async(ECHO_SERVER)
+        100,
+        |mut output| async move {
+            let mut state = State::Disconnected;
+
+            loop {
+                match &mut state {
+                    State::Disconnected => {
+                        const ECHO_SERVER: &str = "ws://localhost:3030";
+
+                        match async_tungstenite::tokio::connect_async(
+                            ECHO_SERVER,
+                        )
                         .await
-                    {
-                        Ok((websocket, _)) => {
-                            let (sender, receiver) = mpsc::channel(100);
-
-                            (
-                                Some(Event::Connected(Connection(sender))),
-                                State::Connected(websocket, receiver),
-                            )
-                        }
-                        Err(_) => {
-                            tokio::time::sleep(
-                                tokio::time::Duration::from_secs(1),
-                            )
-                            .await;
+                        {
+                            Ok((websocket, _)) => {
+                                let (sender, receiver) = mpsc::channel(100);
+
+                                let _ = output
+                                    .send(Event::Connected(Connection(sender)))
+                                    .await;
 
-                            (Some(Event::Disconnected), State::Disconnected)
+                                state = State::Connected(websocket, receiver);
+                            }
+                            Err(_) => {
+                                tokio::time::sleep(
+                                    tokio::time::Duration::from_secs(1),
+                                )
+                                .await;
+
+                                let _ = output.send(Event::Disconnected).await;
+                            }
                         }
                     }
-                }
-                State::Connected(mut websocket, mut input) => {
-                    let mut fused_websocket = websocket.by_ref().fuse();
-
-                    futures::select! {
-                        received = fused_websocket.select_next_some() => {
-                            match received {
-                                Ok(tungstenite::Message::Text(message)) => {
-                                    (
-                                        Some(Event::MessageReceived(Message::User(message))),
-                                        State::Connected(websocket, input)
-                                    )
-                                }
-                                Ok(_) => {
-                                    (None, State::Connected(websocket, input))
-                                }
-                                Err(_) => {
-                                    (Some(Event::Disconnected), State::Disconnected)
+                    State::Connected(websocket, input) => {
+                        let mut fused_websocket = websocket.by_ref().fuse();
+
+                        futures::select! {
+                            received = fused_websocket.select_next_some() => {
+                                match received {
+                                    Ok(tungstenite::Message::Text(message)) => {
+                                       let _ = output.send(Event::MessageReceived(Message::User(message))).await;
+                                    }
+                                    Err(_) => {
+                                        let _ = output.send(Event::Disconnected).await;
+
+                                        state = State::Disconnected;
+                                    }
+                                    Ok(_) => continue,
                                 }
                             }
-                        }
 
-                        message = input.select_next_some() => {
-                            let result = websocket.send(tungstenite::Message::Text(message.to_string())).await;
+                            message = input.select_next_some() => {
+                                let result = websocket.send(tungstenite::Message::Text(message.to_string())).await;
+
+                                if !result.is_ok() {
+                                    let _ = output.send(Event::Disconnected).await;
 
-                            if result.is_ok() {
-                                (None, State::Connected(websocket, input))
-                            } else {
-                                (Some(Event::Disconnected), State::Disconnected)
+                                    state = State::Disconnected;
+                                }
                             }
                         }
                     }
diff --git a/native/src/subscription.rs b/native/src/subscription.rs
index 16e78e82..0ff5e320 100644
--- a/native/src/subscription.rs
+++ b/native/src/subscription.rs
@@ -3,6 +3,8 @@ use crate::event::{self, Event};
 use crate::window;
 use crate::Hasher;
 
+use iced_futures::futures::channel::mpsc;
+use iced_futures::futures::never::Never;
 use iced_futures::futures::{self, Future, Stream};
 use iced_futures::{BoxStream, MaybeSend};
 
@@ -133,6 +135,27 @@ where
 /// [`Stream`] that will call the provided closure to produce every `Message`.
 ///
 /// The `id` will be used to uniquely identify the [`Subscription`].
+pub fn unfold<I, T, Fut, Message>(
+    id: I,
+    initial: T,
+    mut f: impl FnMut(T) -> Fut + MaybeSend + Sync + 'static,
+) -> Subscription<Message>
+where
+    I: Hash + 'static,
+    T: MaybeSend + 'static,
+    Fut: Future<Output = (Message, T)> + MaybeSend + 'static,
+    Message: 'static + MaybeSend,
+{
+    use futures::future::FutureExt;
+
+    run_with_id(
+        id,
+        futures::stream::unfold(initial, move |state| f(state).map(Some)),
+    )
+}
+
+/// Creates a [`Subscription`] that publishes the events sent from a [`Future`]
+/// to an [`mpsc::Sender`] with the given bounds.
 ///
 /// # Creating an asynchronous worker with bidirectional communication
 /// You can leverage this helper to create a [`Subscription`] that spawns
@@ -145,6 +168,7 @@ where
 /// ```
 /// use iced_native::subscription::{self, Subscription};
 /// use iced_native::futures::channel::mpsc;
+/// use iced_native::futures::sink::SinkExt;
 ///
 /// pub enum Event {
 ///     Ready(mpsc::Sender<Input>),
@@ -165,27 +189,35 @@ where
 /// fn some_worker() -> Subscription<Event> {
 ///     struct SomeWorker;
 ///
-///     subscription::unfold(std::any::TypeId::of::<SomeWorker>(), State::Starting, |state| async move {
-///         match state {
-///             State::Starting => {
-///                 // Create channel
-///                 let (sender, receiver) = mpsc::channel(100);
+///     subscription::channel(std::any::TypeId::of::<SomeWorker>(), 100, |mut output| async move {
+///         let mut state = State::Starting;
 ///
-///                 (Some(Event::Ready(sender)), State::Ready(receiver))
-///             }
-///             State::Ready(mut receiver) => {
-///                 use iced_native::futures::StreamExt;
+///         loop {
+///             match &mut state {
+///                 State::Starting => {
+///                     // Create channel
+///                     let (sender, receiver) = mpsc::channel(100);
+///
+///                     // Send the sender back to the application
+///                     output.send(Event::Ready(sender)).await;
+///
+///                     // We are ready to receive messages
+///                     state = State::Ready(receiver);
+///                 }
+///                 State::Ready(receiver) => {
+///                     use iced_native::futures::StreamExt;
 ///
-///                 // Read next input sent from `Application`
-///                 let input = receiver.select_next_some().await;
+///                     // Read next input sent from `Application`
+///                     let input = receiver.select_next_some().await;
 ///
-///                 match input {
-///                     Input::DoSomeWork => {
-///                         // Do some async work...
+///                     match input {
+///                         Input::DoSomeWork => {
+///                             // Do some async work...
 ///
-///                         // Finally, we can optionally return a message to tell the
-///                         // `Application` the work is done
-///                         (Some(Event::WorkFinished), State::Ready(receiver))
+///                             // Finally, we can optionally produce a message to tell the
+///                             // `Application` the work is done
+///                             output.send(Event::WorkFinished).await;
+///                         }
 ///                     }
 ///                 }
 ///             }
@@ -198,25 +230,28 @@ where
 /// connection open.
 ///
 /// [`websocket`]: https://github.com/iced-rs/iced/tree/0.8/examples/websocket
-pub fn unfold<I, T, Fut, Message>(
+pub fn channel<I, Fut, Message>(
     id: I,
-    initial: T,
-    mut f: impl FnMut(T) -> Fut + MaybeSend + Sync + 'static,
+    size: usize,
+    f: impl Fn(mpsc::Sender<Message>) -> Fut + MaybeSend + Sync + 'static,
 ) -> Subscription<Message>
 where
     I: Hash + 'static,
-    T: MaybeSend + 'static,
-    Fut: Future<Output = (Option<Message>, T)> + MaybeSend + 'static,
+    Fut: Future<Output = Never> + MaybeSend + 'static,
     Message: 'static + MaybeSend,
 {
-    use futures::future::{self, FutureExt};
-    use futures::stream::StreamExt;
+    use futures::stream::{self, StreamExt};
 
-    run_with_id(
+    Subscription::from_recipe(Runner {
         id,
-        futures::stream::unfold(initial, move |state| f(state).map(Some))
-            .filter_map(future::ready),
-    )
+        spawn: move |_| {
+            let (sender, receiver) = mpsc::channel(size);
+
+            let runner = stream::once(f(sender)).map(|_| unreachable!());
+
+            stream::select(receiver, runner)
+        },
+    })
 }
 
 struct Runner<I, F, S, Message>