Simplify `run` and `unfold` helpers to build a `Subscription`

1 files changed, 52 insertions, 70 deletions

diff --git a/examples/download_progress/src/download.rs b/examples/download_progress/src/download.rs
index 20682e7a..7db1206b 100644
--- a/examples/download_progress/src/download.rs
+++ b/examples/download_progress/src/download.rs
@@ -1,22 +1,15 @@
-use futures::Stream;
-use iced_futures::futures;
 use iced_native::subscription;
 
 use std::hash::Hash;
 
 // Just a little utility function
-pub fn file<I: 'static + Hash + Copy + Send, T: ToString>(
+pub fn file<I: 'static + Hash + Copy + Send + Sync, T: ToString>(
     id: I,
     url: T,
 ) -> iced::Subscription<(I, Progress)> {
-    subscription::run(
-        id,
-        Download {
-            id,
-            url: url.to_string(),
-        },
-        download,
-    )
+    subscription::unfold(id, State::Ready(url.to_string()), move |state| {
+        download(id, state)
+    })
 }
 
 #[derive(Debug, Hash, Clone)]
@@ -25,74 +18,63 @@ pub struct Download<I> {
     url: String,
 }
 
-fn download<I: Copy>(
-    download: Download<I>,
-) -> impl Stream<Item = (I, Progress)> {
-    let id = download.id;
-
-    futures::stream::unfold(
-        State::Ready(download.url),
-        move |state| async move {
-            match state {
-                State::Ready(url) => {
-                    let response = reqwest::get(&url).await;
-
-                    match response {
-                        Ok(response) => {
-                            if let Some(total) = response.content_length() {
-                                Some((
-                                    (id, Progress::Started),
-                                    State::Downloading {
-                                        response,
-                                        total,
-                                        downloaded: 0,
-                                    },
-                                ))
-                            } else {
-                                Some(((id, Progress::Errored), State::Finished))
-                            }
-                        }
-                        Err(_) => {
-                            Some(((id, Progress::Errored), State::Finished))
-                        }
-                    }
-                }
-                State::Downloading {
-                    mut response,
-                    total,
-                    downloaded,
-                } => match response.chunk().await {
-                    Ok(Some(chunk)) => {
-                        let downloaded = downloaded + chunk.len() as u64;
-
-                        let percentage =
-                            (downloaded as f32 / total as f32) * 100.0;
+async fn download<I: Copy>(
+    id: I,
+    state: State,
+) -> (Option<(I, Progress)>, State) {
+    match state {
+        State::Ready(url) => {
+            let response = reqwest::get(&url).await;
 
-                        Some((
-                            (id, Progress::Advanced(percentage)),
+            match response {
+                Ok(response) => {
+                    if let Some(total) = response.content_length() {
+                        (
+                            Some((id, Progress::Started)),
                             State::Downloading {
                                 response,
                                 total,
-                                downloaded,
+                                downloaded: 0,
                             },
-                        ))
+                        )
+                    } else {
+                        (Some((id, Progress::Errored)), State::Finished)
                     }
-                    Ok(None) => {
-                        Some(((id, Progress::Finished), State::Finished))
-                    }
-                    Err(_) => Some(((id, Progress::Errored), State::Finished)),
-                },
-                State::Finished => {
-                    // We do not let the stream die, as it would start a
-                    // new download repeatedly if the user is not careful
-                    // in case of errors.
-                    let _: () = iced::futures::future::pending().await;
-
-                    None
                 }
+                Err(_) => (Some((id, Progress::Errored)), State::Finished),
+            }
+        }
+        State::Downloading {
+            mut response,
+            total,
+            downloaded,
+        } => match response.chunk().await {
+            Ok(Some(chunk)) => {
+                let downloaded = downloaded + chunk.len() as u64;
+
+                let percentage = (downloaded as f32 / total as f32) * 100.0;
+
+                (
+                    Some((id, Progress::Advanced(percentage))),
+                    State::Downloading {
+                        response,
+                        total,
+                        downloaded,
+                    },
+                )
             }
+            Ok(None) => (Some((id, Progress::Finished)), State::Finished),
+            Err(_) => (Some((id, Progress::Errored)), State::Finished),
         },
-    )
+        State::Finished => {
+            // We do not let the stream die, as it would start a
+            // new download repeatedly if the user is not careful
+            // in case of errors.
+            let _: () = iced::futures::future::pending().await;
+
+            unreachable!()
+        }
+    }
 }
 
 #[derive(Debug, Clone)]