diff --git a/Cargo.lock b/Cargo.lock index f569eb8..e0cd637 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1008,7 +1008,7 @@ dependencies = [ "proc-macro-crate 1.3.1", "proc-macro2", "quote", - "strum", + "strum 0.25.0", "syn 2.0.77", "thiserror", ] @@ -6070,6 +6070,20 @@ dependencies = [ "r2d2", ] +[[package]] +name = "ractor" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11e9a53fcabb680bb70a3ce495e744676ee7e1800a188e01a68d18916a1b48e1" +dependencies = [ + "dashmap", + "futures", + "once_cell", + "strum 0.26.3", + "tokio", + "tracing", +] + [[package]] name = "radium" version = "0.7.0" @@ -7484,7 +7498,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]] @@ -7500,6 +7523,19 @@ dependencies = [ "syn 2.0.77", ] +[[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.77", +] + [[package]] name = "subprocess" version = "0.2.9" @@ -8884,14 +8920,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", ] diff --git a/websockets/echo/Cargo.toml b/websockets/echo/Cargo.toml index e649145..a8aa22c 100644 --- a/websockets/echo/Cargo.toml +++ b/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 diff --git a/websockets/echo/src/main.rs b/websockets/echo/src/main.rs index 7a6d656..a4817b0 100644 --- a/websockets/echo/src/main.rs +++ b/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::{AsMessage, MyWebSocket}; 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 { - 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 diff --git a/websockets/echo/src/server.rs b/websockets/echo/src/server.rs index e262eb7..1f4f695 100644 --- a/websockets/echo/src/server.rs +++ b/websockets/echo/src/server.rs @@ -1,78 +1,111 @@ +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); /// How long before lack of client response causes a timeout const CLIENT_TIMEOUT: Duration = Duration::from_secs(10); +#[derive(Debug)] +pub(crate) enum AsMessage { + Ws(actix_ws::AggregatedMessage), + Hb, +} + /// 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; impl MyWebSocket { - pub fn new() -> Self { - Self { hb: Instant::now() } + async fn handle_hb( + &self, + state: &mut (Instant, actix_ws::Session), + myself: &ActorRef, + ) -> 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). - /// - /// also this method checks heartbeats from client - fn hb(&self, ctx: &mut ::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; - } - - ctx.ping(b""); - }); - } -} - -impl Actor for MyWebSocket { - type Context = ws::WebsocketContext; - - /// 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> for MyWebSocket { - fn handle(&mut self, msg: Result, ctx: &mut Self::Context) { - // process websocket messages + async fn handle_ws_msg( + &self, + msg: AggregatedMessage, + state: &mut (Instant, actix_ws::Session), + myself: ActorRef, + ) -> Result<(), ActorProcessingErr> { println!("WS: {msg:?}"); + match msg { - Ok(ws::Message::Ping(msg)) => { - self.hb = Instant::now(); - ctx.pong(&msg); + AggregatedMessage::Ping(msg) => { + state.0 = Instant::now(); + 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), - Ok(ws::Message::Close(reason)) => { - ctx.close(reason); - ctx.stop(); + + AggregatedMessage::Text(text) => { + state.1.text(text).await?; } - _ => 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 = AsMessage; + type State = (Instant, actix_ws::Session); + type Arguments = actix_ws::Session; + + async fn pre_start( + &self, + myself: ActorRef, + session: Self::Arguments, + ) -> Result { + myself.send_interval(HEARTBEAT_INTERVAL, || AsMessage::Hb); + + Ok((Instant::now(), session)) + } + + async fn handle( + &self, + myself: ActorRef, + message: Self::Msg, + state: &mut Self::State, + ) -> Result<(), ActorProcessingErr> { + match message { + AsMessage::Hb => { + self.handle_hb(state, &myself).await?; + } + + AsMessage::Ws(msg) => { + self.handle_ws_msg(msg, state, myself).await?; + } + } + + Ok(()) } }