Rewrite `websocket` example using `sipper`

1 files changed, 38 insertions, 60 deletions

diff --git a/examples/websocket/src/echo.rs b/examples/websocket/src/echo.rs
index 14652936..6fa4c54a 100644
--- a/examples/websocket/src/echo.rs
+++ b/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> {
-    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;
+                        let _ = 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;
-                        }
+                        let _ = 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)) => {
+                               let _ = output.send(Event::MessageReceived(Message::User(message))).await;
+                            }
+                            Err(_) => {
+                                let _ = output.send(Event::Disconnected).await;
+                                break;
                             }
+                            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_err() {
-                                let _ = output.send(Event::Disconnected).await;
-
-                                state = State::Disconnected;
-                            }
+                        if result.is_err() {
+                            let _ = output.send(Event::Disconnected).await;
                         }
                     }
                 }
@@ -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),