如何检测Rocket_ws客户端WebSocket断开及实现主动周期性推送

1. 检测WebSocket连接关闭/客户端断开 #

在rocket_ws里，通过两种核心逻辑判断连接状态：

• 当stream.next().await返回None时，说明客户端主动断开连接，循环会自然退出，可在此后执行断开后的清理逻辑。
• 当stream.send()返回错误（比如BrokenPipe类异常），意味着连接已异常中断，需捕获该错误并处理。

示例代码：

#[get("/echo?channel")]
fn echo_channel(ws: ws::WebSocket) -> ws::Channel<'static> {
    use rocket::futures::{SinkExt, StreamExt};

    ws.channel(move |mut stream| Box::pin(async move {
        while let Some(message) = stream.next().await {
            match message {
                Ok(msg) => {
                    if let Err(e) = stream.send(msg).await {
                        eprintln!("连接断开：发送消息失败 - {}", e);
                        break;
                    }
                }
                Err(e) => {
                    eprintln!("连接异常：读取消息失败 - {}", e);
                    break;
                }
            }
        }

        eprintln!("客户端已断开连接");
        Ok(())
    }))
}

2. 周期性向客户端发送消息（无需读取客户端消息） #

结合tokio的interval实现定时发送，同时通过监听stream的关闭事件及时退出任务，避免无效发送。用select!宏同时处理定时事件和连接关闭事件即可。

示例代码：

#[get("/periodic?channel")]
fn periodic_send(ws: ws::WebSocket) -> ws::Channel<'static> {
    use rocket::futures::{SinkExt, StreamExt};
    use tokio::time::{interval, Duration};
    use rocket::serde::json::json;

    ws.channel(move |mut stream| Box::pin(async move {
        let mut interval = interval(Duration::from_secs(10));
        // 可选：先触发一次发送，避免等待10秒才发第一条消息
        interval.tick().await;

        loop {
            tokio::select! {
                _ = interval.tick() => {
                    // 构造示例消息，可替换为实际业务数据
                    let reading = json!({
                        "timestamp": chrono::Utc::now().to_rfc3339(),
                        "value": rand::random::<f64>() * 100.0
                    });
                    let msg = ws::Message::Text(reading.to_string());

                    if let Err(e) = stream.send(msg).await {
                        eprintln!("发送失败，连接已断开：{}", e);
                        break;
                    }
                }
                _ = stream.next() => {
                    eprintln!("客户端断开连接，停止定时发送");
                    break;
                }
            }
        }

        Ok(())
    }))
}

注：若项目未引入chrono和rand，需在Cargo.toml添加依赖，或替换为你实际需要的消息内容。

内容的提问来源于stack exchange，提问作者Uros Pocek