Merge pull request #912 from actix/ractor

Ractor WebSockets

.vscode/settings.json

@@ -20,6 +20,7 @@
         "pemfile",
         "prost",
         "protobuf",
+        "ractor",
         "reqwest",
         "rustls",
         "rustup",

Cargo.lock

@@ -1003,7 +1003,7 @@ dependencies = [
  "proc-macro-crate 1.3.1",
  "proc-macro2",
  "quote",
- "strum",
+ "strum 0.25.0",
  "syn 2.0.93",
  "thiserror 1.0.69",
 ]
@@ -1807,6 +1807,29 @@ dependencies = [
  "generic-array",
 ]
 
+[[package]]
+name = "bon"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "97493a391b4b18ee918675fb8663e53646fd09321c58b46afa04e8ce2499c869"
+dependencies = [
+ "bon-macros",
+ "rustversion",
+]
+
+[[package]]
+name = "bon-macros"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a2af3eac944c12cdf4423eab70d310da0a8e5851a18ffb192c0a5e3f7ae1663"
+dependencies = [
+ "darling",
+ "ident_case",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.93",
+]
+
 [[package]]
 name = "borsh"
 version = "1.5.3"
@@ -2534,6 +2557,20 @@ dependencies = [
  "parking_lot_core",
 ]
 
+[[package]]
+name = "dashmap"
+version = "6.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf"
+dependencies = [
+ "cfg-if",
+ "crossbeam-utils",
+ "hashbrown 0.14.5",
+ "lock_api",
+ "once_cell",
+ "parking_lot_core",
+]
+
 [[package]]
 name = "data-encoding"
 version = "2.6.0"
@@ -3522,7 +3559,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "68a7f542ee6b35af73b06abc0dad1c1bae89964e4e253bc4b587b91c9637867b"
 dependencies = [
  "cfg-if",
- "dashmap",
+ "dashmap 5.5.3",
  "futures",
  "futures-timer",
  "no-std-compat",
@@ -6292,6 +6329,21 @@ dependencies = [
  "r2d2",
 ]
 
+[[package]]
+name = "ractor"
+version = "0.15.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc03eb43cae6ef79bae037e58e645664a480a1b17232e5a9cc77b37a99aa7b1f"
+dependencies = [
+ "bon",
+ "dashmap 6.1.0",
+ "futures",
+ "once_cell",
+ "strum 0.26.3",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "radium"
 version = "0.7.0"
@@ -7724,7 +7776,16 @@ version = "0.25.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "290d54ea6f91c969195bdbcd7442c8c2a2ba87da8bf60a7ee86a235d4bc1e125"
 dependencies = [
- "strum_macros",
+ "strum_macros 0.25.3",
+]
+
+[[package]]
+name = "strum"
+version = "0.26.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06"
+dependencies = [
+ "strum_macros 0.26.4",
 ]
 
 [[package]]
@@ -7740,6 +7801,19 @@ dependencies = [
  "syn 2.0.93",
 ]
 
+[[package]]
+name = "strum_macros"
+version = "0.26.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be"
+dependencies = [
+ "heck 0.5.0",
+ "proc-macro2",
+ "quote",
+ "rustversion",
+ "syn 2.0.93",
+]
+
 [[package]]
 name = "subprocess"
 version = "0.2.9"
@@ -9168,14 +9242,14 @@ dependencies = [
 name = "websocket"
 version = "1.0.0"
 dependencies = [
- "actix",
  "actix-files",
  "actix-web",
- "actix-web-actors",
+ "actix-ws",
  "awc",
  "env_logger",
  "futures-util",
  "log",
+ "ractor",
  "tokio",
  "tokio-stream",
 ]

justfile

@@ -7,6 +7,11 @@ fmt:
     cargo +nightly fmt
     fd --type=file --hidden --extension=yml --extension=md --extension=js --exec-batch npx -y prettier --write
 
+# Check project.
+[group("lint")]
+check: clippy
+    fd --type=file --hidden --extension=yml --extension=md --extension=js --exec-batch npx -y prettier --check
+
 # Run Clippy over workspace.
 [group("lint")]
 clippy:

websockets/echo/Cargo.toml

@@ -12,14 +12,14 @@ name = "websocket-client"
 path = "src/client.rs"
 
 [dependencies]
-actix.workspace = true
 actix-files.workspace = true
 actix-web.workspace = true
-actix-web-actors.workspace = true
+actix-ws.workspace = true
 awc.workspace = true
 
 env_logger.workspace = true
 futures-util = { workspace = true, features = ["sink"] }
 log.workspace = true
+ractor = { version = "0.15", default-features = false }
 tokio = { workspace = true, features = ["full"] }
 tokio-stream.workspace = true

websockets/echo/src/main.rs

@@ -4,10 +4,10 @@
 
 use actix_files::NamedFile;
 use actix_web::{middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer, Responder};
-use actix_web_actors::ws;
+use ractor::Actor;
 
 mod server;
-use self::server::MyWebSocket;
+use self::server::{MyWebSocket, WsMessage};
 
 async fn index() -> impl Responder {
     NamedFile::open_async("./static/index.html").await.unwrap()
@@ -15,7 +15,19 @@ async fn index() -> impl Responder {
 
 /// WebSocket handshake and start `MyWebSocket` actor.
 async fn echo_ws(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
-    ws::start(MyWebSocket::new(), &req, stream)
+    let (res, session, stream) = actix_ws::handle(&req, stream)?;
+
+    let (actor, _handle) = Actor::spawn(None, MyWebSocket, session).await.unwrap();
+
+    actix_web::rt::spawn(async move {
+        let mut stream = stream.aggregate_continuations();
+
+        while let Some(Ok(msg)) = stream.recv().await {
+            actor.send_message(WsMessage::Ws(msg)).unwrap();
+        }
+    });
+
+    Ok(res)
 }
 
 // the actor-based WebSocket examples REQUIRE `actix_web::main` for actor support

websockets/echo/src/server.rs

@@ -1,7 +1,7 @@
 use std::time::{Duration, Instant};
 
-use actix::prelude::*;
+use actix_ws::AggregatedMessage;
-use actix_web_actors::ws;
+use ractor::{ActorProcessingErr, ActorRef};
 
 /// How often heartbeat pings are sent
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
@@ -9,70 +9,104 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
 /// How long before lack of client response causes a timeout
 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
 
+#[derive(Debug)]
+pub(crate) enum WsMessage {
+    Ws(actix_ws::AggregatedMessage),
+    Hb,
+}
+
 /// websocket connection is long running connection, it easier
 /// to handle with an actor
-pub struct MyWebSocket {
+pub(crate) struct MyWebSocket;
-    /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
-    /// otherwise we drop connection.
-    hb: Instant,
-}
 
 impl MyWebSocket {
-    pub fn new() -> Self {
+    async fn handle_hb(
-        Self { hb: Instant::now() }
+        &self,
+        state: &mut (Instant, actix_ws::Session),
+        myself: &ActorRef<WsMessage>,
+    ) -> Result<(), ActorProcessingErr> {
+        if Instant::now().duration_since(state.0) > CLIENT_TIMEOUT {
+            // heartbeat timed out
+            println!("Websocket Client heartbeat failed, disconnecting!");
+
+            let _ = state.1.clone().close(None).await;
+            myself.stop(None);
+
+            // don't try to send a ping
+        } else {
+            state.1.ping(b"").await?;
+        };
+
+        Ok(())
     }
 
-    /// helper method that sends ping to client every 5 seconds (HEARTBEAT_INTERVAL).
+    async fn handle_ws_msg(
-    ///
+        &self,
-    /// also this method checks heartbeats from client
+        msg: AggregatedMessage,
-    fn hb(&self, ctx: &mut <Self as Actor>::Context) {
+        state: &mut (Instant, actix_ws::Session),
-        ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
+        myself: ActorRef<WsMessage>,
-            // check client heartbeats
+    ) -> Result<(), ActorProcessingErr> {
-            if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
-                // heartbeat timed out
-                println!("Websocket Client heartbeat failed, disconnecting!");
-
-                // stop actor
-                ctx.stop();
-
-                // don't try to send a ping
-                return;
-            }
-
-            ctx.ping(b"");
-        });
-    }
-}
-
-impl Actor for MyWebSocket {
-    type Context = ws::WebsocketContext<Self>;
-
-    /// Method is called on actor start. We start the heartbeat process here.
-    fn started(&mut self, ctx: &mut Self::Context) {
-        self.hb(ctx);
-    }
-}
-
-/// Handler for `ws::Message`
-impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWebSocket {
-    fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
-        // process websocket messages
         println!("WS: {msg:?}");
+
         match msg {
-            Ok(ws::Message::Ping(msg)) => {
+            AggregatedMessage::Ping(msg) => {
-                self.hb = Instant::now();
+                state.0 = Instant::now();
-                ctx.pong(&msg);
+                state.1.pong(&msg).await?;
             }
-            Ok(ws::Message::Pong(_)) => {
+
-                self.hb = Instant::now();
+            AggregatedMessage::Pong(_) => {
+                state.0 = Instant::now();
             }
-            Ok(ws::Message::Text(text)) => ctx.text(text),
+
-            Ok(ws::Message::Binary(bin)) => ctx.binary(bin),
+            AggregatedMessage::Text(text) => {
-            Ok(ws::Message::Close(reason)) => {
+                state.1.text(text).await?;
-                ctx.close(reason);
-                ctx.stop();
             }
-            _ => ctx.stop(),
+
-        }
+            AggregatedMessage::Binary(bin) => {
+                state.1.binary(bin).await?;
+            }
+
+            AggregatedMessage::Close(reason) => {
+                let _ = state.1.clone().close(reason).await;
+                myself.stop(None);
+            }
+        };
+
+        Ok(())
+    }
+}
+
+impl ractor::Actor for MyWebSocket {
+    type Msg = WsMessage;
+    type State = (Instant, actix_ws::Session);
+    type Arguments = actix_ws::Session;
+
+    async fn pre_start(
+        &self,
+        myself: ActorRef<Self::Msg>,
+        session: Self::Arguments,
+    ) -> Result<Self::State, ActorProcessingErr> {
+        myself.send_interval(HEARTBEAT_INTERVAL, || WsMessage::Hb);
+
+        Ok((Instant::now(), session))
+    }
+
+    async fn handle(
+        &self,
+        myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            WsMessage::Hb => {
+                self.handle_hb(state, &myself).await?;
+            }
+
+            WsMessage::Ws(msg) => {
+                self.handle_ws_msg(msg, state, myself).await?;
+            }
+        }
+
+        Ok(())
     }
 }