rust - 如何使用 Rust 与 Actix Web 和 WebSockets 发送服务器事件?
问题描述
我正在将 Rust 与 Actix web 一起使用,并希望将 WebSockets 用于服务器事件,主要是发送到客户端的状态消息。我遵循了基本结构,
struct StatusWS {}
impl Actor for StatusWS {
type Context = ws::WebsocketContext<Self>;
fn started(&mut self, ctx: &mut Self::Context) {
// How to keep track of ctx???
}
}
当客户端连接时,started() 被调用,我可以使用 ctx.text() 等从那里发送消息。但是,我需要跟踪上下文以便稍后发送消息。我怎么做?它是作为参考传入的,所以借用检查器不会让我对它做太多事情。
解决方案
这对我来说是深水,但我认为你需要impl
Handler for StatusWS
。然后,当您向其发送消息时,您可以使用获得的Addr作为返回值。然后,您将使用可以访问ws 的方法,您可以发送您的 ws 消息。.start()
Actor
handle
Handler
ctx
我自己正在学习 Rust 和 Actix,所以这可能不是你实际做的方式。让我知道事情的后续。
[编辑] 我意识到我的答案有点黯淡,所以我举了一个可能对其他人有用的例子。
use actix::prelude::*;
use actix_web::{middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer};
use actix_web_actors::ws;
use std::time::Duration;
fn ws_index(
r: HttpRequest,
stream: web::Payload,
data: web::Data<Addr<ServerMonitor>>,
) -> Result<HttpResponse, Error> {
let (addr, res) = ws::start_with_addr(MyWebSocket {}, &r, stream)?;
data.get_ref().do_send(RegisterWSClient { addr: addr });
Ok(res)
}
struct MyWebSocket {}
impl Actor for MyWebSocket {
type Context = ws::WebsocketContext<Self>;
}
impl StreamHandler<ws::Message, ws::ProtocolError> for MyWebSocket {
fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
println!("WS: {:?}", msg);
match msg {
ws::Message::Ping(msg) => {
ctx.pong(&msg);
}
ws::Message::Pong(_) => {}
ws::Message::Text(text) => ctx.text(text),
ws::Message::Binary(bin) => ctx.binary(bin),
ws::Message::Close(_) => {
ctx.stop();
}
ws::Message::Nop => (),
}
}
}
impl Handler<ServerEvent> for MyWebSocket {
type Result = ();
fn handle(&mut self, msg: ServerEvent, ctx: &mut Self::Context) {
ctx.text(msg.event);
}
}
#[derive(Message)]
struct RegisterWSClient {
addr: Addr<MyWebSocket>,
}
#[derive(Message)]
struct ServerEvent {
event: String,
}
struct ServerMonitor {
listeners: Vec<Addr<MyWebSocket>>,
}
impl Actor for ServerMonitor {
type Context = Context<Self>;
fn started(&mut self, ctx: &mut Self::Context) {
ctx.run_interval(Duration::from_secs(5), |act, _| {
for l in &act.listeners {
l.do_send(ServerEvent{ event: String::from("Event:") });
}
});
}
}
impl Handler<RegisterWSClient> for ServerMonitor {
type Result = ();
fn handle(&mut self, msg: RegisterWSClient, _: &mut Context<Self>) {
self.listeners.push(msg.addr);
}
}
fn main() -> std::io::Result<()> {
std::env::set_var("RUST_LOG", "actix_server=info,actix_web=info");
env_logger::init();
let sys = actix_rt::System::new("example");
let srvmon = ServerMonitor { listeners: vec![] }.start();
HttpServer::new(move || {
App::new()
.data(srvmon.clone())
.wrap(middleware::Logger::default())
.service(web::resource("/ws/").route(web::get().to(ws_index)))
})
.bind("127.0.0.1:8080")?
.start();
sys.run()
}
推荐阅读
- javascript - Angular ngIf 可观察到的异步响应
- caching - 如何避免 Bazel 构建上的“没有空间”?
- wix - 安装期间运行 Wix 捆绑包安装程序失败
- python - Pyspark Dataframes:创建要在 python 中的聚类中使用的特征列
- unit-testing - 如何测试使用了一些经过充分测试但复杂的功能的功能?
- angular - Angular Universal 中的完全服务器端渲染
- three.js - 动画线框线
- sql - 将制表符分隔的文本文件加载到空白表中的 Derby SQL 命令是什么?
- image - 根据关联值将图像补丁保存到多个文件夹中
- r - 如何将平均函数转换为R中的字符串