refactor: use ractor

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.10", default-features = false }
 tokio = { workspace = true, features = ["full"] }
 tokio-stream.workspace = true

websockets/echo/src/main.rs

@@ -4,18 +4,30 @@
 
 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::{AsMessage, MyWebSocket};
 
 async fn index() -> impl Responder {
     NamedFile::open_async("./static/index.html").await.unwrap()
 }
 
 /// 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(AsMessage::Ws(msg)).unwrap();
+        }
+    });
+
+    Ok(res)
 }
 
 // the actor-based WebSocket examples REQUIRE `actix_web::main` for actor support

websockets/echo/src/server.rs

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