Implement `subscription::channel` and simplify `unfold`

1 files changed, 52 insertions, 48 deletions

diff --git a/examples/websocket/src/echo.rs b/examples/websocket/src/echo.rs
index e74768a6..4fabb660 100644
--- a/examples/websocket/src/echo.rs
+++ b/examples/websocket/src/echo.rs
@@ -13,63 +13,67 @@ use std::fmt;
 pub fn connect() -> Subscription<Event> {
     struct Connect;
 
-    subscription::unfold(
+    subscription::channel(
         std::any::TypeId::of::<Connect>(),
-        State::Disconnected,
-        |state| async move {
-            match state {
-                State::Disconnected => {
-                    const ECHO_SERVER: &str = "ws://localhost:3030";
-
-                    match async_tungstenite::tokio::connect_async(ECHO_SERVER)
+        100,
+        |mut output| async move {
+            let mut state = State::Disconnected;
+
+            loop {
+                match &mut state {
+                    State::Disconnected => {
+                        const ECHO_SERVER: &str = "ws://localhost:3030";
+
+                        match async_tungstenite::tokio::connect_async(
+                            ECHO_SERVER,
+                        )
                         .await
-                    {
-                        Ok((websocket, _)) => {
-                            let (sender, receiver) = mpsc::channel(100);
-
-                            (
-                                Some(Event::Connected(Connection(sender))),
-                                State::Connected(websocket, receiver),
-                            )
-                        }
-                        Err(_) => {
-                            tokio::time::sleep(
-                                tokio::time::Duration::from_secs(1),
-                            )
-                            .await;
+                        {
+                            Ok((websocket, _)) => {
+                                let (sender, receiver) = mpsc::channel(100);
+
+                                let _ = output
+                                    .send(Event::Connected(Connection(sender)))
+                                    .await;
 
-                            (Some(Event::Disconnected), State::Disconnected)
+                                state = State::Connected(websocket, receiver);
+                            }
+                            Err(_) => {
+                                tokio::time::sleep(
+                                    tokio::time::Duration::from_secs(1),
+                                )
+                                .await;
+
+                                let _ = output.send(Event::Disconnected).await;
+                            }
                         }
                     }
-                }
-                State::Connected(mut websocket, mut input) => {
-                    let mut fused_websocket = websocket.by_ref().fuse();
-
-                    futures::select! {
-                        received = fused_websocket.select_next_some() => {
-                            match received {
-                                Ok(tungstenite::Message::Text(message)) => {
-                                    (
-                                        Some(Event::MessageReceived(Message::User(message))),
-                                        State::Connected(websocket, input)
-                                    )
-                                }
-                                Ok(_) => {
-                                    (None, State::Connected(websocket, input))
-                                }
-                                Err(_) => {
-                                    (Some(Event::Disconnected), State::Disconnected)
+                    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,
                                 }
                             }
-                        }
 
-                        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_ok() {
+                                    let _ = output.send(Event::Disconnected).await;
 
-                            if result.is_ok() {
-                                (None, State::Connected(websocket, input))
-                            } else {
-                                (Some(Event::Disconnected), State::Disconnected)
+                                    state = State::Disconnected;
+                                }
                             }
                         }
                     }