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

`sipper` support and some QoL

17 files changed, 272 insertions, 190 deletions

diff --git a/Cargo.lock b/Cargo.lock
index c5e9443f..5c015e2a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1875,6 +1875,7 @@ dependencies = [
  "image",
  "reqwest",
  "serde",
+ "sipper",
  "tokio",
 ]
 
@@ -2592,6 +2593,7 @@ dependencies = [
  "iced_core",
  "iced_futures",
  "raw-window-handle 0.6.2",
+ "sipper",
  "thiserror 1.0.69",
 ]
 
@@ -5228,6 +5230,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "56199f7ddabf13fe5074ce809e7d3f42b42ae711800501b5b16ea82ad029c39d"
 
 [[package]]
+name = "sipper"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1bccb4192828b3d9a08e0b5a73f17795080dfb278b50190216e3ae2132cf4f95"
+dependencies = [
+ "futures",
+ "pin-project-lite",
+]
+
+[[package]]
 name = "skrifa"
 version = "0.22.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index c95fbd1e..abdb2f80 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -172,6 +172,7 @@ raw-window-handle = "0.6"
 resvg = "0.42"
 rustc-hash = "2.0"
 sha2 = "0.10"
+sipper = "0.1"
 smol = "1.0"
 smol_str = "0.2"
 softbuffer = "0.4"
diff --git a/core/src/element.rs b/core/src/element.rs
index ede9e16c..b7d51aeb 100644
--- a/core/src/element.rs
+++ b/core/src/element.rs
@@ -93,6 +93,7 @@ impl<'a, Message, Theme, Renderer> Element<'a, Message, Theme, Renderer> {
     ///
     /// ```no_run
     /// # mod iced {
+    /// #     pub use iced_core::Function;
     /// #     pub type Element<'a, Message> = iced_core::Element<'a, Message, iced_core::Theme, ()>;
     /// #
     /// #     pub mod widget {
@@ -119,7 +120,7 @@ impl<'a, Message, Theme, Renderer> Element<'a, Message, Theme, Renderer> {
     /// use counter::Counter;
     ///
     /// use iced::widget::row;
-    /// use iced::Element;
+    /// use iced::{Element, Function};
     ///
     /// struct ManyCounters {
     ///     counters: Vec<Counter>,
@@ -142,7 +143,7 @@ impl<'a, Message, Theme, Renderer> Element<'a, Message, Theme, Renderer> {
     ///                     // Here we turn our `Element<counter::Message>` into
     ///                     // an `Element<Message>` by combining the `index` and the
     ///                     // message of the `element`.
-    ///                     counter.map(move |message| Message::Counter(index, message))
+    ///                     counter.map(Message::Counter.with(index))
     ///                 }),
     ///         )
     ///         .into()
diff --git a/core/src/lib.rs b/core/src/lib.rs
index d5c221ac..03cc0632 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -93,3 +93,60 @@ pub use smol_str::SmolStr;
 pub fn never<T>(never: std::convert::Infallible) -> T {
     match never {}
 }
+
+/// A trait extension for binary functions (`Fn(A, B) -> O`).
+///
+/// It enables you to use a bunch of nifty functional programming paradigms
+/// that work well with iced.
+pub trait Function<A, B, O> {
+    /// Applies the given first argument to a binary function and returns
+    /// a new function that takes the other argument.
+    ///
+    /// This lets you partially "apply" a function—equivalent to currying,
+    /// but it only works with binary functions. If you want to apply an
+    /// arbitrary number of arguments, create a little struct for them.
+    ///
+    /// # When is this useful?
+    /// Sometimes you will want to identify the source or target
+    /// of some message in your user interface. This can be achieved through
+    /// normal means by defining a closure and moving the identifier
+    /// inside:
+    ///
+    /// ```rust
+    /// # let element: Option<()> = Some(());
+    /// # enum Message { ButtonPressed(u32, ()) }
+    /// let id = 123;
+    ///
+    /// # let _ = {
+    /// element.map(move |result| Message::ButtonPressed(id, result))
+    /// # };
+    /// ```
+    ///
+    /// That's quite a mouthful. [`with`](Self::with) lets you write:
+    ///
+    /// ```rust
+    /// # use iced_core::Function;
+    /// # let element: Option<()> = Some(());
+    /// # enum Message { ButtonPressed(u32, ()) }
+    /// let id = 123;
+    ///
+    /// # let _ = {
+    /// element.map(Message::ButtonPressed.with(id))
+    /// # };
+    /// ```
+    ///
+    /// Effectively creating the same closure that partially applies
+    /// the `id` to the message—but much more concise!
+    fn with(self, prefix: A) -> impl Fn(B) -> O;
+}
+
+impl<F, A, B, O> Function<A, B, O> for F
+where
+    F: Fn(A, B) -> O,
+    Self: Sized,
+    A: Copy,
+{
+    fn with(self, prefix: A) -> impl Fn(B) -> O {
+        move |result| self(prefix, result)
+    }
+}
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),
diff --git a/futures/src/backend/native/tokio.rs b/futures/src/backend/native/tokio.rs
index e0be83a6..c38ef566 100644
--- a/futures/src/backend/native/tokio.rs
+++ b/futures/src/backend/native/tokio.rs
@@ -23,11 +23,10 @@ impl crate::Executor for Executor {
 pub mod time {
     //! Listen and react to time.
     use crate::core::time::{Duration, Instant};
-    use crate::stream;
     use crate::subscription::Subscription;
     use crate::MaybeSend;
 
-    use futures::SinkExt;
+    use futures::stream;
     use std::future::Future;
 
     /// Returns a [`Subscription`] that produces messages at a set interval.
@@ -66,12 +65,12 @@ pub mod time {
             let f = *f;
             let interval = *interval;
 
-            stream::channel(1, move |mut output| async move {
-                loop {
-                    let _ = output.send(f().await).await;
-
+            stream::unfold(0, move |i| async move {
+                if i > 0 {
                     tokio::time::sleep(interval).await;
                 }
+
+                Some((f().await, i + 1))
             })
         })
     }
diff --git a/runtime/Cargo.toml b/runtime/Cargo.toml
index 703c3ed9..fc212ef8 100644
--- a/runtime/Cargo.toml
+++ b/runtime/Cargo.toml
@@ -23,5 +23,6 @@ iced_core.workspace = true
 iced_futures.workspace = true
 iced_futures.features = ["thread-pool"]
 
-thiserror.workspace = true
 raw-window-handle.workspace = true
+sipper.workspace = true
+thiserror.workspace = true
diff --git a/runtime/src/task.rs b/runtime/src/task.rs
index 22cfb63e..022483f7 100644
--- a/runtime/src/task.rs
+++ b/runtime/src/task.rs
@@ -3,7 +3,6 @@ use crate::core::widget;
 use crate::futures::futures::channel::mpsc;
 use crate::futures::futures::channel::oneshot;
 use crate::futures::futures::future::{self, FutureExt};
-use crate::futures::futures::never::Never;
 use crate::futures::futures::stream::{self, Stream, StreamExt};
 use crate::futures::{boxed_stream, BoxStream, MaybeSend};
 use crate::Action;
@@ -11,6 +10,9 @@ use crate::Action;
 use std::future::Future;
 use std::sync::Arc;
 
+#[doc(no_inline)]
+pub use sipper::{sipper, stream, Never, Sender, Sipper, Straw};
+
 /// A set of concurrent actions to be performed by the iced runtime.
 ///
 /// A [`Task`] _may_ produce a bunch of values of type `T`.
@@ -57,6 +59,22 @@ impl<T> Task<T> {
         Self::stream(stream.map(f))
     }
 
+    /// Creates a [`Task`] that runs the given [`Sipper`] to completion, mapping
+    /// progress with the first closure and the output with the second one.
+    pub fn sip<S>(
+        sipper: S,
+        on_progress: impl FnMut(S::Progress) -> T + MaybeSend + 'static,
+        on_output: impl FnOnce(<S as Future>::Output) -> T + MaybeSend + 'static,
+    ) -> Self
+    where
+        S: sipper::Core + MaybeSend + 'static,
+        T: MaybeSend + 'static,
+    {
+        Self::stream(stream(sipper::sipper(move |sender| async move {
+            on_output(sipper.with(on_progress).run(sender).await)
+        })))
+    }
+
     /// Combines the given tasks and produces a single [`Task`] that will run all of them
     /// in parallel.
     pub fn batch(tasks: impl IntoIterator<Item = Self>) -> Self
diff --git a/src/lib.rs b/src/lib.rs
index 849b51e9..441826d6 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -506,8 +506,8 @@ pub use crate::core::padding;
 pub use crate::core::theme;
 pub use crate::core::{
     never, Alignment, Animation, Background, Border, Color, ContentFit,
-    Degrees, Gradient, Length, Padding, Pixels, Point, Radians, Rectangle,
-    Rotation, Settings, Shadow, Size, Theme, Transformation, Vector,
+    Degrees, Function, Gradient, Length, Padding, Pixels, Point, Radians,
+    Rectangle, Rotation, Settings, Shadow, Size, Theme, Transformation, Vector,
 };
 pub use crate::runtime::exit;
 pub use iced_futures::Subscription;
@@ -519,7 +519,9 @@ pub use Length::{Fill, FillPortion, Shrink};
 
 pub mod task {
     //! Create runtime tasks.
-    pub use crate::runtime::task::{Handle, Task};
+    pub use crate::runtime::task::{
+        sipper, stream, Handle, Never, Sipper, Straw, Task,
+    };
 }
 
 pub mod clipboard {