当前位置：首页 > article >正文

使用 Hyperlane 实现 WebSocket广播

article 2025/6/30 12:33:44

使用 `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');
};