improve actorless echo example

Cargo.lock

@@ -2446,21 +2446,6 @@ version = "0.3.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc4045962a5a5e935ee2fdedaa4e08284547402885ab326734432bed5d12966b"
 
-[[package]]
-name = "futures-lite"
-version = "1.12.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7694489acd39452c77daa48516b894c153f192c3578d5a839b62c58099fcbf48"
-dependencies = [
- "fastrand",
- "futures-core",
- "futures-io",
- "memchr",
- "parking",
- "pin-project-lite 0.2.9",
- "waker-fn",
-]
-
 [[package]]
 name = "futures-macro"
 version = "0.3.21"
@@ -4000,12 +3985,6 @@ version = "6.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "21326818e99cfe6ce1e524c2a805c189a99b5ae555a35d19f9a284b427d86afa"
 
-[[package]]
-name = "parking"
-version = "2.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "427c3892f9e783d91cc128285287e70a59e206ca452770ece88a76f7a3eddd72"
-
 [[package]]
 name = "parking_lot"
 version = "0.11.2"
@@ -6870,12 +6849,6 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6a02e4885ed3bc0f2de90ea6dd45ebcbb66dacffe03547fadbb0eeae2770887d"
 
-[[package]]
-name = "waker-fn"
-version = "1.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d5b2c62b4012a3e1eca5a7e077d13b3bf498c4073e33ccd58626607748ceeca"
-
 [[package]]
 name = "walkdir"
 version = "2.3.2"
@@ -7099,7 +7072,6 @@ dependencies = [
  "actix-ws",
  "awc",
  "env_logger 0.9.0",
- "futures-lite",
  "futures-util",
  "log",
  "tokio 1.19.2",

websockets/chat-actorless/Cargo.toml

@@ -11,7 +11,7 @@ awc = "3"
 
 derive_more = "0.99.5"
 env_logger = "0.9"
-futures-util = { version = "0.3.7", default-features = false, features = ["std", "sink"] }
+futures-util = { version = "0.3.17", default-features = false, features = ["std", "sink"] }
 log = "0.4"
 rand = "0.8"
-tokio = { version = "1.13.1", features = ["full"] }
+tokio = { version = "1.13.1", features = ["rt", "macros", "sync", "time"] }

websockets/chat-actorless/src/main.rs

@@ -42,7 +42,8 @@ async fn chat_ws(
     Ok(res)
 }
 
-#[actix_web::main]
+// note that the `actix` based WebSocket handling would NOT work under `tokio::main`
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> std::io::Result<()> {
     env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
 

websockets/chat-actorless/src/server.rs

@@ -9,7 +9,7 @@ use std::{
 };
 
 use rand::{thread_rng, Rng as _};
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::mpsc;
 
 use crate::{Command, ConnId, Msg, RoomId};
 

websockets/echo-actorless/Cargo.toml

@@ -10,7 +10,6 @@ actix-ws = "0.2.5"
 awc = "3"
 
 env_logger = "0.9"
-futures-util = { version = "0.3.7", default-features = false, features = ["std", "sink"] }
+futures-util = { version = "0.3.17", default-features = false, features = ["std"] }
-futures-lite = "1.3"
 log = "0.4"
-tokio = { version = "1.13.1", features = ["full"] }
+tokio = { version = "1.13.1", features = ["rt", "macros", "sync", "time"] }

websockets/echo-actorless/src/handler.rs

@@ -1,14 +1,18 @@
 use std::time::{Duration, Instant};
 
-use actix_web::rt;
 use actix_ws::Message;
-use futures_lite::future;
+use futures_util::{
-use futures_util::{future::Either, stream::StreamExt as _, FutureExt as _};
+    future::{self, Either},
+    StreamExt as _,
+};
+use tokio::{pin, time::interval};
 
-/// How often heartbeat pings are sent
+/// How often heartbeat pings are sent.
+///
+/// Should be half (or less) of the acceptable client timeout.
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
 
-/// How long before lack of client response causes a timeout
+/// How long before lack of client response causes a timeout.
 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
 
 /// Echo text & binary messages received from the client, respond to ping messages, and monitor
@@ -20,19 +24,34 @@ pub async fn echo_heartbeat_ws(
     log::info!("connected");
 
     let mut last_heartbeat = Instant::now();
-    let mut interval = rt::time::interval(HEARTBEAT_INTERVAL);
+    let mut interval = interval(HEARTBEAT_INTERVAL);
 
-    loop {
+    let reason = loop {
-        match future::or(
+        // create "next client timeout check" future
-            msg_stream.next().map(Either::Left),
+        let tick = interval.tick();
-            interval.tick().map(Either::Right),
+        // required for select()
-        )
+        pin!(tick);
-        .await
+
-        {
+        // waits for either `msg_stream` to receive a message from the client or the heartbeat
-            Either::Left(Some(Ok(msg))) => {
+        // interval timer to tick, yielding the value of whichever one is ready first
+        match future::select(msg_stream.next(), tick).await {
+            // received message from WebSocket client
+            Either::Left((Some(Ok(msg)), _)) => {
                 log::debug!("msg: {msg:?}");
 
                 match msg {
+                    Message::Text(text) => {
+                        session.text(text).await.unwrap();
+                    }
+
+                    Message::Binary(bin) => {
+                        session.binary(bin).await.unwrap();
+                    }
+
+                    Message::Close(reason) => {
+                        break reason;
+                    }
+
                     Message::Ping(bytes) => {
                         last_heartbeat = Instant::now();
                         let _ = session.pong(&bytes).await;
@@ -42,46 +61,44 @@ pub async fn echo_heartbeat_ws(
                         last_heartbeat = Instant::now();
                     }
 
-                    Message::Text(text) => {
+                    Message::Continuation(_) => {
-                        session.text(text).await.unwrap();
+                        log::warn!("no support for continuation frames");
                     }
 
-                    Message::Binary(bin) => {
+                    // no-op; ignore
-                        session.binary(bin).await.unwrap();
+                    Message::Nop => {}
+                };
             }
 
-                    Message::Close(reason) => {
+            // client WebSocket stream error
-                        let _ = session.close(reason).await;
+            Either::Left((Some(Err(err)), _)) => {
-                        break;
+                log::error!("{}", err);
             }
 
-                    _ => {
+            // client WebSocket stream ended
-                        let _ = session.close(None).await;
+            Either::Left((None, _)) => break None,
-                        break;
-                    }
-                }
-            }
 
-            Either::Left(_) => {}
+            // heartbeat interval ticked
-
+            Either::Right((_inst, _)) => {
-            Either::Right(_) => {
                 // if no heartbeat ping/pong received recently, close the connection
                 if Instant::now().duration_since(last_heartbeat) > CLIENT_TIMEOUT {
                     log::info!(
                         "client has not sent heartbeat in over {CLIENT_TIMEOUT:?}; disconnecting"
                     );
-                    let _ = session.close(None).await;
+
-                    break;
+                    break None;
                 }
 
                 // send heartbeat ping
                 let _ = session.ping(b"").await;
-
+            }
-                // reset interval duration
-                interval.reset();
         }
     };
-    }
+
+    // attempt to close connection gracefully
+    let _ = session.close(reason).await;
+
+    log::info!("disconnected");
 }
 
 /// Echo text & binary messages received from the client and respond to ping messages.
@@ -89,19 +106,17 @@ pub async fn echo_heartbeat_ws(
 /// This example is just for demonstration of simplicity. In reality, you likely want to include
 /// some handling of heartbeats for connection health tracking to free up server resources when
 /// connections die or network issues arise.
+///
+/// See [`echo_heartbeat_ws`] for a more realistic implementation.
 pub async fn echo_ws(mut session: actix_ws::Session, mut msg_stream: actix_ws::MessageStream) {
     log::info!("connected");
 
-    while let Some(Ok(msg)) = msg_stream.next().await {
+    let close_reason = loop {
+        match msg_stream.next().await {
+            Some(Ok(msg)) => {
                 log::debug!("msg: {msg:?}");
 
                 match msg {
-            Message::Ping(bytes) => {
-                let _ = session.pong(&bytes).await;
-            }
-
-            Message::Pong(_) => {}
-
                     Message::Text(text) => {
                         session.text(text).await.unwrap();
                     }
@@ -111,14 +126,31 @@ pub async fn echo_ws(mut session: actix_ws::Session, mut msg_stream: actix_ws::M
                     }
 
                     Message::Close(reason) => {
-                let _ = session.close(reason).await;
+                        break reason;
-                break;
                     }
 
-            _ => {
+                    Message::Ping(bytes) => {
-                let _ = session.close(None).await;
+                        let _ = session.pong(&bytes).await;
-                break;
                     }
+
+                    Message::Pong(_) => {}
+
+                    Message::Continuation(_) => {
+                        log::warn!("no support for continuation frames");
                     }
+
+                    // no-op; ignore
+                    Message::Nop => {}
+                };
             }
+
+            // error or end of stream
+            _ => break None,
+        }
+    };
+
+    // attempt to close connection gracefully
+    let _ = session.close(close_reason).await;
+
+    log::info!("disconnected");
 }

websockets/echo-actorless/src/main.rs

@@ -37,7 +37,8 @@ async fn echo_ws(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse,
     Ok(res)
 }
 
-#[actix_web::main]
+// note that the `actix` based WebSocket handling would NOT work under `tokio::main`
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> std::io::Result<()> {
     env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));