如何为 actix websocket 服务器配置有效负载限制

问题描述

我正在学习 Actix 并尝试创建一个 WebSocket 服务

代码片段：

启动服务器

pub async fn start(addr: &str) -> std::result::Result<(),IoError> {
    let connections = Connections::default().start();
    HttpServer::new(move || {
        App::new().service(
            web::resource("/ws/")
                .data(connections.clone())
                .route(web::get().to(ws_index)),)
    })
    .bind(addr)?
    .run()
    .await
}

处理程序

async fn ws_index(
    req: HttpRequest,stream: web::Payload,addr: web::Data<Addr<Connections>>,) -> Result<HttpResponse,Error> {
    let id = Uuid::new_v4().to_simple().to_string();
    let client = Connection::new(id,addr.get_ref().clone());
    let resp = ws::start(client,&req,stream);
    resp
}

流处理程序

impl StreamHandler<Result<ws::Message,ws::ProtocolError>> for Connection {
    fn handle(&mut self,msg: Result<ws::Message,ws::ProtocolError>,ctx: &mut Self::Context) {
        match msg {
            Ok(ws::Message::Ping(msg)) => ctx.pong(&msg),Ok(ws::Message::Pong(_)) => self.hb = Instant::Now(),Ok(ws::Message::Text(text)) => ctx.text(text),Ok(ws::Message::Close(reason)) => {
                ctx.stop();
                println!("Connection {} closed with reason {:?}",self.id,reason);
            }
            Err(e) => println!("Error: {}",e),_ => (),}
    }
}

现在 WebSocket 服务器正在运行，它可以接收文本并将其发送回去。但是如果我发送大文本，服务器会记录“错误：有效载荷达到大小限制。”。如何解决？

解决方法

您必须使用特定的编解码器创建 WebSocket，而不是使用 ws::start()，因为您可以在此处设置负载 https://docs.rs/actix-http/2.2.0/actix_http/ws/struct.Codec.html 的大小。

所以创建你的启动函数：

use actix_web::error::PayloadError;
use ws::{handshake,WebsocketContext};                        
use actix_http::ws::{Codec,Message,ProtocolError};
use bytes::Bytes;

fn start_with_codec<A,S>(actor: A,req: &HttpRequest,stream: S,codec: Codec) -> Result<HttpResponse,Error>
where
    A: Actor<Context = WebsocketContext<A>>
        + StreamHandler<Result<Message,ProtocolError>>,S: Stream<Item = Result<Bytes,PayloadError>> + 'static,{   
    let mut res = handshake(req)?;
    Ok(res.streaming(WebsocketContext::with_codec(actor,stream,codec)))
}

然后更改调用新函数并传递编解码器的代码：

async fn ws_index(
    req: HttpRequest,stream: web::Payload,addr: web::Data<Addr<Connections>>,) -> Result<HttpResponse,Error> {
    let id = Uuid::new_v4().to_simple().to_string();
    let client = Connection::new(id,addr.get_ref().clone());
    let resp = start_with_codec(client,&req,Codec::new().max_size(MAX_SIZE));
    resp
}

更新

在大负载上，消息可以通过 Message::Continuation 变体分解成几个较小的片段。

此变体包含具有以下变体的 Item 枚举：

pub enum Item {
    FirstText(Bytes),FirstBinary(Bytes),Continue(Bytes),Last(Bytes),}

要重组原始消息，您必须收集从 FirstText / FirstBinary 到 Last 的所有片段，或者确保按正确顺序将所有这些消息转发到将检索消息的目的地。

actix-web rust