Update to ewebsock 0.4.0 (#3729)

This unblocks us from updating `tunstenite` to fix RUSTSEC-2023-0065

This also removes a few dependencies, thanks to:
* rerun-io/ewebsock#24

I've tested that it still works.

### Checklist
* [x] I have read and agree to [Contributor
Guide](https://github.com/rerun-io/rerun/blob/main/CONTRIBUTING.md) and
the [Code of
Conduct](https://github.com/rerun-io/rerun/blob/main/CODE_OF_CONDUCT.md)
* [x] I've included a screenshot or gif (if applicable)
* [x] I have tested [demo.rerun.io](https://demo.rerun.io/pr/3729) (if
applicable)
* [x] The PR title and labels are set such as to maximize their
usefulness for the next release's CHANGELOG

- [PR Build Summary](https://build.rerun.io/pr/3729)
- [Docs
preview](https://rerun.io/preview/1378634af462a895fea11b44a2dd08e766ed7176/docs)
<!--DOCS-PREVIEW-->
- [Examples
preview](https://rerun.io/preview/1378634af462a895fea11b44a2dd08e766ed7176/examples)
<!--EXAMPLES-PREVIEW-->
- [Recent benchmark results](https://ref.rerun.io/dev/bench/)
- [Wasm size tracking](https://ref.rerun.io/dev/sizes/)

Cargo.toml

@@ -115,7 +115,7 @@ emath = "0.23.0"
 enumset = "1.0.12"
 env_logger = "0.10"
 epaint = "0.23.0"
-ewebsock = "0.2"
+ewebsock = "0.4.0"
 futures-channel = "0.3"
 futures-util = { version = "0.3", default-features = false }
 getrandom = "0.2"

crates/re_data_source/src/web_sockets.rs

@@ -35,7 +35,6 @@ pub fn connect_to_ws_url(
         }
     };
 
-    let connection = re_ws_comms::Connection::viewer_to_server(url.to_owned(), callback)?;
-    std::mem::drop(connection); // Never close the connection. TODO(emilk): is this wise?
+    re_ws_comms::viewer_to_server(url.to_owned(), callback)?;
     Ok(rx)
 }

crates/re_sdk_comms/Cargo.toml

@@ -35,4 +35,4 @@ crossbeam.workspace = true
 document-features.workspace = true
 rand = { workspace = true, features = ["std", "std_rng", "small_rng"] }
 thiserror.workspace = true
-tokio.workspace = true
+tokio = { workspace = true, features = ["io-util", "net"] }

crates/re_sdk_comms/src/server.rs

@@ -1,10 +1,10 @@
 use std::{io::ErrorKind, time::Instant};
 
 use rand::{Rng as _, SeedableRng};
+use tokio::net::{TcpListener, TcpStream};
 
 use re_log_types::{LogMsg, TimePoint, TimeType, TimelineName};
 use re_smart_channel::{Receiver, Sender};
-use tokio::net::{TcpListener, TcpStream};
 
 #[derive(thiserror::Error, Debug)]
 pub enum ServerError {

crates/re_ws_comms/src/client.rs

@@ -1,60 +1,51 @@
 use std::ops::ControlFlow;
 
-use ewebsock::{WsEvent, WsMessage, WsSender};
+use ewebsock::{WsEvent, WsMessage};
 
 // TODO(jleibs): use thiserror
 pub type Result<T> = anyhow::Result<T>;
 
-/// Represents a connection to the server.
-/// Disconnects on drop.
-#[must_use]
-pub struct Connection(WsSender);
-
-impl Connection {
-    /// Connect viewer to server
-    pub fn viewer_to_server(
-        url: String,
-        on_binary_msg: impl Fn(Vec<u8>) -> ControlFlow<()> + Send + 'static,
-    ) -> Result<Self> {
-        re_log::info!("Connecting to {url:?}…");
-        let sender = ewebsock::ws_connect(
-            url,
-            Box::new(move |event: WsEvent| match event {
-                WsEvent::Opened => {
-                    re_log::info!("Connection established");
+/// Connect viewer to server
+pub fn viewer_to_server(
+    url: String,
+    on_binary_msg: impl Fn(Vec<u8>) -> ControlFlow<()> + Send + 'static,
+) -> Result<()> {
+    re_log::info!("Connecting to {url:?}…");
+    ewebsock::ws_receive(
+        url,
+        Box::new(move |event: WsEvent| match event {
+            WsEvent::Opened => {
+                re_log::info!("Connection established");
+                ControlFlow::Continue(())
+            }
+            WsEvent::Message(message) => match message {
+                WsMessage::Binary(binary) => on_binary_msg(binary),
+                WsMessage::Text(text) => {
+                    re_log::warn!("Unexpected text message: {text:?}");
+                    ControlFlow::Continue(())
+                }
+                WsMessage::Unknown(text) => {
+                    re_log::warn!("Unknown message: {text:?}");
                     ControlFlow::Continue(())
                 }
-                WsEvent::Message(message) => match message {
-                    WsMessage::Binary(binary) => on_binary_msg(binary),
-                    WsMessage::Text(text) => {
-                        re_log::warn!("Unexpected text message: {text:?}");
-                        ControlFlow::Continue(())
-                    }
-                    WsMessage::Unknown(text) => {
-                        re_log::warn!("Unknown message: {text:?}");
-                        ControlFlow::Continue(())
-                    }
-                    WsMessage::Ping(_data) => {
-                        re_log::warn!("Unexpected PING");
-                        ControlFlow::Continue(())
-                    }
-                    WsMessage::Pong(_data) => {
-                        re_log::warn!("Unexpected PONG");
-                        ControlFlow::Continue(())
-                    }
-                },
-                WsEvent::Error(error) => {
-                    re_log::error!("Connection error: {error}");
-                    ControlFlow::Break(())
+                WsMessage::Ping(_data) => {
+                    re_log::warn!("Unexpected PING");
+                    ControlFlow::Continue(())
                 }
-                WsEvent::Closed => {
-                    re_log::info!("Connection to server closed.");
-                    ControlFlow::Break(())
+                WsMessage::Pong(_data) => {
+                    re_log::warn!("Unexpected PONG");
+                    ControlFlow::Continue(())
                 }
-            }),
-        )
-        .map_err(|err| anyhow::format_err!("ewebsock: {err}"))?;
-
-        Ok(Self(sender))
-    }
+            },
+            WsEvent::Error(error) => {
+                re_log::error!("Connection error: {error}");
+                ControlFlow::Break(())
+            }
+            WsEvent::Closed => {
+                re_log::info!("Connection to server closed.");
+                ControlFlow::Break(())
+            }
+        }),
+    )
+    .map_err(|err| anyhow::format_err!("ewebsock: {err}"))
 }

crates/re_ws_comms/src/lib.rs

@@ -9,7 +9,7 @@ mod client;
 use std::{fmt::Display, str::FromStr};
 
 #[cfg(feature = "client")]
-pub use client::Connection;
+pub use client::viewer_to_server;
 
 #[cfg(feature = "server")]
 mod server;

crates/re_ws_comms/src/server.rs

@@ -70,6 +70,7 @@ impl RerunServer {
             let (tcp_stream, _) = tokio::select! {
                 res = self.listener.accept() => res?,
                 _ = shutdown_rx.recv() => {
+                    re_log::debug!("Shutting down WebSocket server");
                     return Ok(());
                 }
             };