2 files changed, 61 insertions, 70 deletions

diff --git a/examples/websocket/src/echo.rs b/examples/websocket/src/echo.rs
index cd32cb66..14652936 100644
--- a/examples/websocket/src/echo.rs
+++ b/examples/websocket/src/echo.rs
@@ -1,87 +1,79 @@
 pub mod server;
 
 use iced::futures;
-use iced::subscription::{self, Subscription};
+use iced::stream;
 use iced::widget::text;
 
 use futures::channel::mpsc;
 use futures::sink::SinkExt;
-use futures::stream::StreamExt;
+use futures::stream::{Stream, StreamExt};
 
 use async_tungstenite::tungstenite;
 use std::fmt;
 
-pub fn connect() -> Subscription<Event> {
-    struct Connect;
+pub fn connect() -> impl Stream<Item = Event> {
+    stream::channel(100, |mut output| async move {
+        let mut state = State::Disconnected;
 
-    subscription::channel(
-        std::any::TypeId::of::<Connect>(),
-        100,
-        |mut output| async move {
-            let mut state = State::Disconnected;
+        loop {
+            match &mut state {
+                State::Disconnected => {
+                    const ECHO_SERVER: &str = "ws://127.0.0.1:3030";
 
-            loop {
-                match &mut state {
-                    State::Disconnected => {
-                        const ECHO_SERVER: &str = "ws://127.0.0.1:3030";
-
-                        match async_tungstenite::tokio::connect_async(
-                            ECHO_SERVER,
-                        )
+                    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;
+                    {
+                        Ok((websocket, _)) => {
+                            let (sender, receiver) = mpsc::channel(100);
 
-                                state = State::Connected(websocket, receiver);
-                            }
-                            Err(_) => {
-                                tokio::time::sleep(
-                                    tokio::time::Duration::from_secs(1),
-                                )
+                            let _ = output
+                                .send(Event::Connected(Connection(sender)))
                                 .await;
 
-                                let _ = output.send(Event::Disconnected).await;
-                            }
+                            state = State::Connected(websocket, receiver);
+                        }
+                        Err(_) => {
+                            tokio::time::sleep(
+                                tokio::time::Duration::from_secs(1),
+                            )
+                            .await;
+
+                            let _ = output.send(Event::Disconnected).await;
                         }
                     }
-                    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,
+                }
+                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;
                                 }
-                            }
-
-                            message = input.select_next_some() => {
-                                let result = websocket.send(tungstenite::Message::Text(message.to_string())).await;
-
-                                if result.is_err() {
+                                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;
+
+                            if result.is_err() {
+                                let _ = output.send(Event::Disconnected).await;
+
+                                state = State::Disconnected;
                             }
                         }
                     }
                 }
             }
-        },
-    )
+        }
+    })
 }
 
 #[derive(Debug)]
diff --git a/examples/websocket/src/main.rs b/examples/websocket/src/main.rs
index 8422ce16..d8246436 100644
--- a/examples/websocket/src/main.rs
+++ b/examples/websocket/src/main.rs
@@ -9,12 +9,10 @@ use once_cell::sync::Lazy;
 
 pub fn main() -> iced::Result {
     iced::application("WebSocket - Iced", WebSocket::update, WebSocket::view)
-        .load(WebSocket::load)
         .subscription(WebSocket::subscription)
-        .run()
+        .run_with(WebSocket::new)
 }
 
-#[derive(Default)]
 struct WebSocket {
     messages: Vec<echo::Message>,
     new_message: String,
@@ -30,11 +28,18 @@ enum Message {
 }
 
 impl WebSocket {
-    fn load() -> Task<Message> {
-        Task::batch([
-            Task::perform(echo::server::run(), |_| Message::Server),
-            widget::focus_next(),
-        ])
+    fn new() -> (Self, Task<Message>) {
+        (
+            Self {
+                messages: Vec::new(),
+                new_message: String::new(),
+                state: State::Disconnected,
+            },
+            Task::batch([
+                Task::perform(echo::server::run(), |_| Message::Server),
+                widget::focus_next(),
+            ]),
+        )
     }
 
     fn update(&mut self, message: Message) -> Task<Message> {
@@ -83,7 +88,7 @@ impl WebSocket {
     }
 
     fn subscription(&self) -> Subscription<Message> {
-        echo::connect().map(Message::Echo)
+        Subscription::run(echo::connect).map(Message::Echo)
     }
 
     fn view(&self) -> Element<Message> {
@@ -140,10 +145,4 @@ enum State {
     Connected(echo::Connection),
 }
 
-impl Default for State {
-    fn default() -> Self {
-        Self::Disconnected
-    }
-}
-
 static MESSAGE_LOG: Lazy<scrollable::Id> = Lazy::new(scrollable::Id::unique);