Rewrite websocket example using sipper

examples/websocket/src/echo.rs

@@ -1,73 +1,63 @@
 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> {
+pub fn connect() -> impl Sipper<Never, Event> {
-    stream::channel(100, |mut output| async move {
+    sipper(|mut output| async move {
-        let mut state = State::Disconnected;
-
         loop {
-            match &mut state {
+            const ECHO_SERVER: &str = "ws://127.0.0.1:3030";
-                State::Disconnected => {
-                    const ECHO_SERVER: &str = "ws://127.0.0.1:3030";
 
-                    match async_tungstenite::tokio::connect_async(ECHO_SERVER)
+            let (mut websocket, mut input) =
-                        .await
+                match async_tungstenite::tokio::connect_async(ECHO_SERVER).await
-                    {
+                {
-                        Ok((websocket, _)) => {
+                    Ok((websocket, _)) => {
-                            let (sender, receiver) = mpsc::channel(100);
+                        let (sender, receiver) = mpsc::channel(100);
 
-                            let _ = output
+                        let _ = output
-                                .send(Event::Connected(Connection(sender)))
+                            .send(Event::Connected(Connection(sender)))
-                                .await;
-
-                            state = State::Connected(websocket, receiver);
-                        }
-                        Err(_) => {
-                            tokio::time::sleep(
-                                tokio::time::Duration::from_secs(1),
-                            )
                             .await;
 
-                            let _ = output.send(Event::Disconnected).await;
+                        (websocket.fuse(), receiver)
+                    }
+                    Err(_) => {
+                        tokio::time::sleep(tokio::time::Duration::from_secs(1))
+                            .await;
+
+                        let _ = output.send(Event::Disconnected).await;
+
+                        continue;
+                    }
+                };
+
+            loop {
+                futures::select! {
+                    received = 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;
+                                break;
+                            }
+                            Ok(_) => continue,
                         }
                     }
-                }
-                State::Connected(websocket, input) => {
-                    let mut fused_websocket = websocket.by_ref().fuse();
 
-                    futures::select! {
+                    message = input.select_next_some() => {
-                        received = fused_websocket.select_next_some() => {
+                        let result = websocket.send(tungstenite::Message::Text(message.to_string())).await;
-                            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;
+                        if result.is_err() {
-                                }
+                            let _ = output.send(Event::Disconnected).await;
-                                Ok(_) => continue,
-                            }
-                        }
-
-                        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;
-                            }
                         }
                     }
                 }
@@ -76,18 +66,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),

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 {
+            stream::unfold(0, move |i| async move {
-                loop {
+                if i > 0 {
-                    let _ = output.send(f().await).await;
-
                     tokio::time::sleep(interval).await;
                 }
+
+                Some((f().await, i + 1))
             })
         })
     }

futures/src/lib.rs

@@ -15,7 +15,6 @@ pub mod backend;
 pub mod event;
 pub mod executor;
 pub mod keyboard;
-pub mod stream;
 pub mod subscription;
 
 pub use executor::Executor;

src/lib.rs

@@ -478,7 +478,6 @@ use iced_winit::core;
 use iced_winit::runtime;
 
 pub use iced_futures::futures;
-pub use iced_futures::stream;
 
 #[cfg(feature = "highlighter")]
 pub use iced_highlighter as highlighter;