使用 Hyperlane 实现 WebSocket广播

使用 Hyperlane 实现 WebSocket广播

hyperlane 框架原生支持 WebSocket 协议，开发者无需关心协议升级过程，即可通过统一接口处理 WebSocket 请求。本文将介绍如何使用 hyperlane 实现服务端的单点发送与广播发送功能，以及如何配套实现一个简单的 WebSocket 客户端。

框架特性提示

hyperlane 框架支持 WebSocket 协议，服务端自动处理协议升级，支持请求中间件、路由处理、响应中间件等特性。

需要注意的是，WebSocket 响应应使用 send_response_body 方法进行发送；使用 send_response 将导致客户端解析失败，因为该方法不会将响应封装成符合 WebSocket 协议格式。

服务端：单点发送示例

该示例中，服务端仅将客户端发送的数据原样返回，实现 WebSocket 回显（echo）功能。

pub async fn handle(ctx: Context) {
    let request_body: Vec<u8> = ctx.get_request_body().await;
    let _ = ctx.send_response_body(request_body).await;
}

服务端：广播发送示例

在广播模式下，多个客户端连接共享同一个消息通道，任一客户端发送的消息会被广播至所有连接的客户端。

注意事项

• 广播功能使用 tokio::broadcast 实现。
• 使用 tokio::select 同时监听客户端发来的新消息与广播通道中的新数据。
• 如果未开启 enable_inner_websocket_handle，客户端需连接后主动发送任意消息（哪怕是空消息）才能接收广播。
• 开启该配置后，连接即代表已准备好接收广播。

示例代码

static BROADCAST_CHANNEL: OnceLock<Broadcast<ResponseBody>> = OnceLock::new();

fn ensure_broadcast_channel() -> Broadcast<ResponseBody> {
    BROADCAST_CHANNEL
        .get_or_init(|| Broadcast::default())
        .clone()
}

pub async fn handle(ctx: Context) {
    if ctx.get_stream().await.is_none() {
        ctx.aborted().await;
        return;
    }
    let broadcast: Broadcast<ResponseBody> = ensure_broadcast_channel();
    let mut receiver: BroadcastReceiver<Vec<u8>> = broadcast.subscribe();
    loop {
        tokio::select! {
            request_res = ctx.websocket_request_from_stream(10000) => {
                if request_res.is_err() {
                    break;
                }
                let request = request_res.unwrap_or_default();
                let body: RequestBody = request.get_body().clone();
                if broadcast.send(body).is_err() {
                    break;
                }
            },
            msg_res = receiver.recv() => {
                if let Ok(msg) = msg_res {
                    if ctx.send_response_body(msg).await.is_err() || ctx.flush().await.is_err() {
                        break;
                    }
                }
           }
        }
    }
}

客户端代码示例（JavaScript）

以下是一个基于浏览器环境的 WebSocket 客户端示例，每秒发送当前时间到服务器，并输出收到的广播消息。

const ws = new WebSocket('ws://localhost:60000/websocket');

ws.onopen = () => {
  console.log('WebSocket opened');
  setInterval(() => {
    ws.send(`Now time: ${new Date().toISOString()}`);
  }, 1000);
};

ws.onmessage = (event) => {
  console.log('Receive: ', event.data);
};

ws.onerror = (error) => {
  console.error('WebSocket error: ', error);
};

ws.onclose = () => {
  console.log('WebSocket closed');
};

总结

通过 hyperlane，你可以轻松构建基于 WebSocket 的实时服务，无需手动处理握手与协议细节。借助统一的 send_response_body 接口，你可以使用与 HTTP 一致的方式处理 WebSocket 消息，极大简化开发体验。