Merge pull request #2805 from iced-rs/feature/sipper-support

`sipper` support and some QoL

9 files changed, 168 insertions, 177 deletions

diff --git a/examples/download_progress/src/download.rs b/examples/download_progress/src/download.rs
index d63fb906..5b81f7a2 100644
--- a/examples/download_progress/src/download.rs
+++ b/examples/download_progress/src/download.rs
@@ -1,16 +1,14 @@
-use iced::futures::{SinkExt, Stream, StreamExt};
-use iced::stream::try_channel;
+use iced::futures::StreamExt;
+use iced::task::{sipper, Straw};
 
 use std::sync::Arc;
 
-pub fn download(
-    url: impl AsRef<str>,
-) -> impl Stream<Item = Result<Progress, Error>> {
-    try_channel(1, move |mut output| async move {
+pub fn download(url: impl AsRef<str>) -> impl Straw<(), Progress, Error> {
+    sipper(move |mut progress| async move {
         let response = reqwest::get(url.as_ref()).await?;
         let total = response.content_length().ok_or(Error::NoContentLength)?;
 
-        let _ = output.send(Progress::Downloading { percent: 0.0 }).await;
+        let _ = progress.send(Progress { percent: 0.0 }).await;
 
         let mut byte_stream = response.bytes_stream();
         let mut downloaded = 0;
@@ -19,23 +17,20 @@ pub fn download(
             let bytes = next_bytes?;
             downloaded += bytes.len();
 
-            let _ = output
-                .send(Progress::Downloading {
+            let _ = progress
+                .send(Progress {
                     percent: 100.0 * downloaded as f32 / total as f32,
                 })
                 .await;
         }
 
-        let _ = output.send(Progress::Finished).await;
-
         Ok(())
     })
 }
 
 #[derive(Debug, Clone)]
-pub enum Progress {
-    Downloading { percent: f32 },
-    Finished,
+pub struct Progress {
+    pub percent: f32,
 }
 
 #[derive(Debug, Clone)]
diff --git a/examples/download_progress/src/main.rs b/examples/download_progress/src/main.rs
index f4b07203..8082eccd 100644
--- a/examples/download_progress/src/main.rs
+++ b/examples/download_progress/src/main.rs
@@ -4,7 +4,7 @@ use download::download;
 
 use iced::task;
 use iced::widget::{button, center, column, progress_bar, text, Column};
-use iced::{Center, Element, Right, Task};
+use iced::{Center, Element, Function, Right, Task};
 
 pub fn main() -> iced::Result {
     iced::application(
@@ -25,7 +25,7 @@ struct Example {
 pub enum Message {
     Add,
     Download(usize),
-    DownloadProgressed(usize, Result<download::Progress, download::Error>),
+    DownloadUpdated(usize, Update),
 }
 
 impl Example {
@@ -52,15 +52,13 @@ impl Example {
 
                 let task = download.start();
 
-                task.map(move |progress| {
-                    Message::DownloadProgressed(index, progress)
-                })
+                task.map(Message::DownloadUpdated.with(index))
             }
-            Message::DownloadProgressed(id, progress) => {
+            Message::DownloadUpdated(id, update) => {
                 if let Some(download) =
                     self.downloads.iter_mut().find(|download| download.id == id)
                 {
-                    download.progress(progress);
+                    download.update(update);
                 }
 
                 Task::none()
@@ -95,6 +93,12 @@ struct Download {
     state: State,
 }
 
+#[derive(Debug, Clone)]
+pub enum Update {
+    Downloading(download::Progress),
+    Finished(Result<(), download::Error>),
+}
+
 #[derive(Debug)]
 enum State {
     Idle,
@@ -111,18 +115,20 @@ impl Download {
         }
     }
 
-    pub fn start(
-        &mut self,
-    ) -> Task<Result<download::Progress, download::Error>> {
+    pub fn start(&mut self) -> Task<Update> {
         match self.state {
             State::Idle { .. }
             | State::Finished { .. }
             | State::Errored { .. } => {
-                let (task, handle) = Task::stream(download(
-                    "https://huggingface.co/\
+                let (task, handle) = Task::sip(
+                    download(
+                        "https://huggingface.co/\
                         mattshumer/Reflection-Llama-3.1-70B/\
                         resolve/main/model-00001-of-00162.safetensors",
-                ))
+                    ),
+                    Update::Downloading,
+                    Update::Finished,
+                )
                 .abortable();
 
                 self.state = State::Downloading {
@@ -136,20 +142,18 @@ impl Download {
         }
     }
 
-    pub fn progress(
-        &mut self,
-        new_progress: Result<download::Progress, download::Error>,
-    ) {
+    pub fn update(&mut self, update: Update) {
         if let State::Downloading { progress, .. } = &mut self.state {
-            match new_progress {
-                Ok(download::Progress::Downloading { percent }) => {
-                    *progress = percent;
-                }
-                Ok(download::Progress::Finished) => {
-                    self.state = State::Finished;
+            match update {
+                Update::Downloading(new_progress) => {
+                    *progress = new_progress.percent;
                 }
-                Err(_error) => {
-                    self.state = State::Errored;
+                Update::Finished(result) => {
+                    self.state = if result.is_ok() {
+                        State::Finished
+                    } else {
+                        State::Errored
+                    };
                 }
             }
         }
diff --git a/examples/gallery/Cargo.toml b/examples/gallery/Cargo.toml
index c9dc1e9d..6e8aba06 100644
--- a/examples/gallery/Cargo.toml
+++ b/examples/gallery/Cargo.toml
@@ -17,6 +17,7 @@ serde.features = ["derive"]
 
 bytes.workspace = true
 image.workspace = true
+sipper.workspace = true
 tokio.workspace = true
 
 blurhash = "0.2.3"
diff --git a/examples/gallery/src/civitai.rs b/examples/gallery/src/civitai.rs
index 18d2a040..04589030 100644
--- a/examples/gallery/src/civitai.rs
+++ b/examples/gallery/src/civitai.rs
@@ -1,5 +1,6 @@
 use bytes::Bytes;
 use serde::Deserialize;
+use sipper::{sipper, Straw};
 use tokio::task;
 
 use std::fmt;
@@ -45,58 +46,72 @@ impl Image {
         self,
         width: u32,
         height: u32,
-    ) -> Result<Rgba, Error> {
+    ) -> Result<Blurhash, Error> {
         task::spawn_blocking(move || {
             let pixels = blurhash::decode(&self.hash, width, height, 1.0)?;
 
-            Ok::<_, Error>(Rgba {
-                width,
-                height,
-                pixels: Bytes::from(pixels),
+            Ok::<_, Error>(Blurhash {
+                rgba: Rgba {
+                    width,
+                    height,
+                    pixels: Bytes::from(pixels),
+                },
             })
         })
         .await?
     }
 
-    pub async fn download(self, size: Size) -> Result<Rgba, Error> {
-        let client = reqwest::Client::new();
-
-        let bytes = client
-            .get(match size {
-                Size::Original => self.url,
-                Size::Thumbnail { width } => self
-                    .url
-                    .split("/")
-                    .map(|part| {
-                        if part.starts_with("width=") {
-                            format!("width={}", width * 2) // High DPI
-                        } else {
-                            part.to_owned()
-                        }
-                    })
-                    .collect::<Vec<_>>()
-                    .join("/"),
+    pub fn download(self, size: Size) -> impl Straw<Rgba, Blurhash, Error> {
+        sipper(move |mut sender| async move {
+            let client = reqwest::Client::new();
+
+            if let Size::Thumbnail { width, height } = size {
+                let image = self.clone();
+
+                drop(task::spawn(async move {
+                    if let Ok(blurhash) = image.blurhash(width, height).await {
+                        sender.send(blurhash).await;
+                    }
+                }));
+            }
+
+            let bytes = client
+                .get(match size {
+                    Size::Original => self.url,
+                    Size::Thumbnail { width, .. } => self
+                        .url
+                        .split("/")
+                        .map(|part| {
+                            if part.starts_with("width=") {
+                                format!("width={}", width * 2) // High DPI
+                            } else {
+                                part.to_owned()
+                            }
+                        })
+                        .collect::<Vec<_>>()
+                        .join("/"),
+                })
+                .send()
+                .await?
+                .error_for_status()?
+                .bytes()
+                .await?;
+
+            let image = task::spawn_blocking(move || {
+                Ok::<_, Error>(
+                    image::ImageReader::new(io::Cursor::new(bytes))
+                        .with_guessed_format()?
+                        .decode()?
+                        .to_rgba8(),
+                )
             })
-            .send()
-            .await?
-            .error_for_status()?
-            .bytes()
-            .await?;
+            .await??;
 
-        let image = task::spawn_blocking(move || {
-            Ok::<_, Error>(
-                image::ImageReader::new(io::Cursor::new(bytes))
-                    .with_guessed_format()?
-                    .decode()?
-                    .to_rgba8(),
-            )
-        })
-        .await??;
-
-        Ok(Rgba {
-            width: image.width(),
-            height: image.height(),
-            pixels: Bytes::from(image.into_raw()),
+            Ok(Rgba {
+                width: image.width(),
+                height: image.height(),
+                pixels: Bytes::from(image.into_raw()),
+            })
         })
     }
 }
@@ -106,6 +121,11 @@ impl Image {
 )]
 pub struct Id(u32);
 
+#[derive(Debug, Clone)]
+pub struct Blurhash {
+    pub rgba: Rgba,
+}
+
 #[derive(Clone)]
 pub struct Rgba {
     pub width: u32,
@@ -125,7 +145,7 @@ impl fmt::Debug for Rgba {
 #[derive(Debug, Clone, Copy)]
 pub enum Size {
     Original,
-    Thumbnail { width: u32 },
+    Thumbnail { width: u32, height: u32 },
 }
 
 #[derive(Debug, Clone)]
diff --git a/examples/gallery/src/main.rs b/examples/gallery/src/main.rs
index ab22679d..abafaf2d 100644
--- a/examples/gallery/src/main.rs
+++ b/examples/gallery/src/main.rs
@@ -14,7 +14,8 @@ use iced::widget::{
 };
 use iced::window;
 use iced::{
-    color, Animation, ContentFit, Element, Fill, Subscription, Task, Theme,
+    color, Animation, ContentFit, Element, Fill, Function, Subscription, Task,
+    Theme,
 };
 
 use std::collections::HashMap;
@@ -40,7 +41,7 @@ enum Message {
     ImageDownloaded(Result<Rgba, Error>),
     ThumbnailDownloaded(Id, Result<Rgba, Error>),
     ThumbnailHovered(Id, bool),
-    BlurhashDecoded(Id, Result<Rgba, Error>),
+    BlurhashDecoded(Id, civitai::Blurhash),
     Open(Id),
     Close,
     Animate(Instant),
@@ -94,18 +95,14 @@ impl Gallery {
                     return Task::none();
                 };
 
-                Task::batch(vec![
-                    Task::perform(
-                        image.clone().blurhash(Preview::WIDTH, Preview::HEIGHT),
-                        move |result| Message::BlurhashDecoded(id, result),
-                    ),
-                    Task::perform(
-                        image.download(Size::Thumbnail {
-                            width: Preview::WIDTH,
-                        }),
-                        move |result| Message::ThumbnailDownloaded(id, result),
-                    ),
-                ])
+                Task::sip(
+                    image.download(Size::Thumbnail {
+                        width: Preview::WIDTH,
+                        height: Preview::HEIGHT,
+                    }),
+                    Message::BlurhashDecoded.with(id),
+                    Message::ThumbnailDownloaded.with(id),
+                )
             }
             Message::ImageDownloaded(Ok(rgba)) => {
                 self.viewer.show(rgba);
@@ -131,9 +128,11 @@ impl Gallery {
 
                 Task::none()
             }
-            Message::BlurhashDecoded(id, Ok(rgba)) => {
+            Message::BlurhashDecoded(id, blurhash) => {
                 if !self.previews.contains_key(&id) {
-                    let _ = self.previews.insert(id, Preview::loading(rgba));
+                    let _ = self
+                        .previews
+                        .insert(id, Preview::loading(blurhash.rgba));
                 }
 
                 Task::none()
@@ -167,8 +166,7 @@ impl Gallery {
             }
             Message::ImagesListed(Err(error))
             | Message::ImageDownloaded(Err(error))
-            | Message::ThumbnailDownloaded(_, Err(error))
-            | Message::BlurhashDecoded(_, Err(error)) => {
+            | Message::ThumbnailDownloaded(_, Err(error)) => {
                 dbg!(error);
 
                 Task::none()
diff --git a/examples/game_of_life/src/main.rs b/examples/game_of_life/src/main.rs
index dec3df7f..9516f832 100644
--- a/examples/game_of_life/src/main.rs
+++ b/examples/game_of_life/src/main.rs
@@ -9,7 +9,7 @@ use iced::time::{self, milliseconds};
 use iced::widget::{
     button, checkbox, column, container, pick_list, row, slider, text,
 };
-use iced::{Center, Element, Fill, Subscription, Task, Theme};
+use iced::{Center, Element, Fill, Function, Subscription, Task, Theme};
 
 pub fn main() -> iced::Result {
     tracing_subscriber::fmt::init();
@@ -37,7 +37,7 @@ struct GameOfLife {
 
 #[derive(Debug, Clone)]
 enum Message {
-    Grid(grid::Message, usize),
+    Grid(usize, grid::Message),
     Tick,
     TogglePlayback,
     ToggleGrid(bool),
@@ -61,7 +61,7 @@ impl GameOfLife {
 
     fn update(&mut self, message: Message) -> Task<Message> {
         match message {
-            Message::Grid(message, version) => {
+            Message::Grid(version, message) => {
                 if version == self.version {
                     self.grid.update(message);
                 }
@@ -78,9 +78,7 @@ impl GameOfLife {
 
                     let version = self.version;
 
-                    return Task::perform(task, move |message| {
-                        Message::Grid(message, version)
-                    });
+                    return Task::perform(task, Message::Grid.with(version));
                 }
             }
             Message::TogglePlayback => {
@@ -129,9 +127,7 @@ impl GameOfLife {
         );
 
         let content = column![
-            self.grid
-                .view()
-                .map(move |message| Message::Grid(message, version)),
+            self.grid.view().map(Message::Grid.with(version)),
             controls,
         ]
         .height(Fill);
diff --git a/examples/multi_window/src/main.rs b/examples/multi_window/src/main.rs
index f9021c8d..8cec9d4c 100644
--- a/examples/multi_window/src/main.rs
+++ b/examples/multi_window/src/main.rs
@@ -3,7 +3,9 @@ use iced::widget::{
     text_input,
 };
 use iced::window;
-use iced::{Center, Element, Fill, Subscription, Task, Theme, Vector};
+use iced::{
+    Center, Element, Fill, Function, Subscription, Task, Theme, Vector,
+};
 
 use std::collections::BTreeMap;
 
@@ -169,7 +171,7 @@ impl Window {
         let scale_input = column![
             text("Window scale factor:"),
             text_input("Window Scale", &self.scale_input)
-                .on_input(move |msg| { Message::ScaleInputChanged(id, msg) })
+                .on_input(Message::ScaleInputChanged.with(id))
                 .on_submit(Message::ScaleChanged(
                     id,
                     self.scale_input.to_string()
@@ -179,7 +181,7 @@ impl Window {
         let title_input = column![
             text("Window title:"),
             text_input("Window Title", &self.title)
-                .on_input(move |msg| { Message::TitleChanged(id, msg) })
+                .on_input(Message::TitleChanged.with(id))
                 .id(format!("input-{id}"))
         ];
 
diff --git a/examples/todos/src/main.rs b/examples/todos/src/main.rs
index 7faf742e..dfb73d96 100644
--- a/examples/todos/src/main.rs
+++ b/examples/todos/src/main.rs
@@ -4,7 +4,9 @@ use iced::widget::{
     scrollable, text, text_input, Text,
 };
 use iced::window;
-use iced::{Center, Element, Fill, Font, Subscription, Task as Command};
+use iced::{
+    Center, Element, Fill, Font, Function, Subscription, Task as Command,
+};
 
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
@@ -215,9 +217,8 @@ impl Todos {
                             .map(|(i, task)| {
                                 (
                                     task.id,
-                                    task.view(i).map(move |message| {
-                                        Message::TaskMessage(i, message)
-                                    }),
+                                    task.view(i)
+                                        .map(Message::TaskMessage.with(i)),
                                 )
                             }),
                     )
diff --git a/examples/websocket/src/echo.rs b/examples/websocket/src/echo.rs
index 14652936..149a260c 100644
--- a/examples/websocket/src/echo.rs
+++ b/examples/websocket/src/echo.rs
@@ -1,73 +1,59 @@
 pub mod server;
 
 use iced::futures;
-use iced::stream;
+use iced::task::{sipper, Never, Sipper};
 use iced::widget::text;
 
 use futures::channel::mpsc;
 use futures::sink::SinkExt;
-use futures::stream::{Stream, StreamExt};
+use futures::stream::StreamExt;
 
 use async_tungstenite::tungstenite;
 use std::fmt;
 
-pub fn connect() -> impl Stream<Item = Event> {
-    stream::channel(100, |mut output| async move {
-        let mut state = State::Disconnected;
-
+pub fn connect() -> impl Sipper<Never, Event> {
+    sipper(|mut output| async move {
         loop {
-            match &mut state {
-                State::Disconnected => {
-                    const ECHO_SERVER: &str = "ws://127.0.0.1:3030";
+            const ECHO_SERVER: &str = "ws://127.0.0.1:3030";
 
-                    match async_tungstenite::tokio::connect_async(ECHO_SERVER)
-                        .await
-                    {
-                        Ok((websocket, _)) => {
-                            let (sender, receiver) = mpsc::channel(100);
+            let (mut websocket, mut input) =
+                match async_tungstenite::tokio::connect_async(ECHO_SERVER).await
+                {
+                    Ok((websocket, _)) => {
+                        let (sender, receiver) = mpsc::channel(100);
 
-                            let _ = output
-                                .send(Event::Connected(Connection(sender)))
-                                .await;
+                        output.send(Event::Connected(Connection(sender))).await;
 
-                            state = State::Connected(websocket, receiver);
-                        }
-                        Err(_) => {
-                            tokio::time::sleep(
-                                tokio::time::Duration::from_secs(1),
-                            )
+                        (websocket.fuse(), receiver)
+                    }
+                    Err(_) => {
+                        tokio::time::sleep(tokio::time::Duration::from_secs(1))
                             .await;
 
-                            let _ = output.send(Event::Disconnected).await;
-                        }
+                        output.send(Event::Disconnected).await;
+                        continue;
                     }
-                }
-                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,
+                };
+
+            loop {
+                futures::select! {
+                    received = websocket.select_next_some() => {
+                        match received {
+                            Ok(tungstenite::Message::Text(message)) => {
+                                output.send(Event::MessageReceived(Message::User(message))).await;
+                            }
+                            Err(_) => {
+                                output.send(Event::Disconnected).await;
+                                break;
                             }
+                            Ok(_) => {},
                         }
+                    }
+                    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_err() {
-                                let _ = output.send(Event::Disconnected).await;
-
-                                state = State::Disconnected;
-                            }
+                        if result.is_err() {
+                            output.send(Event::Disconnected).await;
                         }
                     }
                 }
@@ -76,18 +62,6 @@ pub fn connect() -> impl Stream<Item = Event> {
     })
 }
 
-#[derive(Debug)]
-#[allow(clippy::large_enum_variant)]
-enum State {
-    Disconnected,
-    Connected(
-        async_tungstenite::WebSocketStream<
-            async_tungstenite::tokio::ConnectStream,
-        >,
-        mpsc::Receiver<Message>,
-    ),
-}
-
 #[derive(Debug, Clone)]
 pub enum Event {
     Connected(Connection),