绝尘
030ddb668a
Phase3 阶段2(D2 全19变体透传):trait TunnelClient 加 send_raw_event(Value) + WsTunnelClient 实现;OutFrame::Event 改持 String(send_event/send_raw_event 预序列化,run_loop 只写 socket DRY)。cargo check + 5 测试过。
548 lines
23 KiB
Rust
548 lines
23 KiB
Rust
//! df-tunnel 隧道客户端(Phase2 完整实现)
|
|
//!
|
|
//! 设计依据:设计文档「Layer1」—— 桌面端主动出站连云后端 `wss://host/ws/device`,
|
|
//! 穿 NAT 无需端口映射。tokio-tungstenite WS + Hello 握手 + 收发循环 + 指数退避重连。
|
|
//!
|
|
//! ## 协议对齐(df-relay Phase2 commit 2b8b30e)
|
|
//! - 桌面端连 `/ws/device`,首帧发 `Hello { kind:"device", device_id, token }`
|
|
//! - relay 双因子校验(kind == expected_device + token 匹配),失败发
|
|
//! `{"kind":"control","error":"..."}` + Close,成功则静默进入收发循环(无显式 ack)
|
|
//! - 握手成功信号:**未收到 error 帧 且 socket 保持打开**(connect 用 timeout 探测首帧)
|
|
//! - 入站消息:relay 把对端 Miniapp 的文本帧包成 BroadcastMessage 转发,桌面端收到的
|
|
//! 是完整 BroadcastMessage JSON,payload 是 miniapp 指令原样 JSON(Phase3 纯透传,不解析)
|
|
//! - 出站消息:桌面端发 TunnelEvent 的 JSON(relay 把它当 payload 包成 BroadcastMessage)
|
|
//!
|
|
//! ## AiChatEvent 透传原则
|
|
//! payload 用 serde_json::Value 透传,本 crate 不依赖 src-tauri/df-types(避跨 crate 强耦合)。
|
|
//! TunnelEvent 仅是高频子集,扩展变体不破坏向后兼容。
|
|
//!
|
|
//! ## 并发模型
|
|
//! 内部 spawn 收发循环 task,通过 mpsc 解耦 send_event(调用方)与 socket 写入。
|
|
//! 收到 command 方向 payload(`serde_json::Value`)走 on_command 回调(调用方在 connect 时注册),不阻塞收发循环。
|
|
//! Phase3 协议统一(D1=A):tunnel 纯透传,业务协议解析在 device 端桥接层(src-tauri)。
|
|
//! `&self` + `Arc<Mutex<...>>` 持连接状态,trait 方法全部 async + 不持 stream 借用。
|
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use async_trait::async_trait;
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use tokio::sync::{mpsc, Mutex};
|
|
use tokio::task::JoinHandle;
|
|
use tokio_tungstenite::tungstenite::protocol::CloseFrame;
|
|
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
|
|
|
use crate::error::{Result, TunnelError};
|
|
use crate::events::{Hello, HelloKind, RelayControlError, TunnelEvent};
|
|
|
|
/// 收到 miniapp→device 指令 payload 时的回调类型(Boxed Future,在收发循环 task 内 await)
|
|
///
|
|
/// Phase3 协议统一(D1=A):tunnel 纯透传,不再反序列化 TunnelCommand 强类型,
|
|
/// payload 以 `serde_json::Value` 上抛,业务协议解析在 device 端桥接层(src-tauri)。
|
|
/// 设计为 async 便于调用方执行 Tauri command 异步路由。回调内 panic 会中断收发循环,
|
|
/// 调用方应自行处理错误(回调返回值忽略,失败由调用方记录)。
|
|
pub type CommandHandler = Arc<dyn Fn(serde_json::Value) -> futures_util::future::BoxFuture<'static, ()> + Send + Sync>;
|
|
|
|
/// 隧道客户端抽象
|
|
///
|
|
/// 设计为 trait 而非具体结构,便于:
|
|
/// 1. 测试用 mock 实现(不依赖真实 WS)
|
|
/// 2. 未来替换底层实现(如换 quic / sse)不改调用方
|
|
///
|
|
/// 状态语义:
|
|
/// - `connect` 一次建立连接 + spawn 收发循环,重复 connect 视为重连(先 disconnect)
|
|
/// - `is_connected` 轻量查询,不发心跳
|
|
/// - `send_event` 非阻塞入队,实际 socket 写入由收发循环异步完成
|
|
/// - `disconnect` 优雅关闭(Close 帧 + 停 task + 清资源)
|
|
#[async_trait]
|
|
pub trait TunnelClient: Send + Sync {
|
|
/// 主动连接云后端 + 注册指令回调
|
|
///
|
|
/// `url` 形如 `wss://host/ws/device`(query 不带 token,token 走 Hello 帧)。
|
|
/// 成功后进入「已连接」状态并 spawn 收发循环。失败返回 TunnelError::Connect/Auth。
|
|
/// `on_command` 收到云后端转发的 command 方向 payload Value 时被调用(在收发 task 内异步执行)。
|
|
/// Phase3 纯透传(D1=A):payload 不在 tunnel 解析,由 device 端桥接层 match cmd 路由。
|
|
async fn connect(
|
|
&self,
|
|
url: &str,
|
|
device_id: &str,
|
|
token: &str,
|
|
on_command: CommandHandler,
|
|
) -> Result<()>;
|
|
|
|
/// 主动断开(优雅 Close 帧 + 释放 task / 通道)
|
|
async fn disconnect(&self) -> Result<()>;
|
|
|
|
/// 是否处于已连接状态(轻量查询,不阻塞,不发心跳)
|
|
fn is_connected(&self) -> bool;
|
|
|
|
/// 发送事件(桌面端 → 云后端 → 小程序)
|
|
///
|
|
/// 非阻塞入队,实际 socket 写入异步完成。连接断开时返回 NotConnected。
|
|
async fn send_event(&self, event: TunnelEvent) -> Result<()>;
|
|
|
|
/// 发送原始事件(桌面端 → 云后端 → 小程序),payload 为 AiChatEvent 序列化 JSON Value
|
|
///
|
|
/// Phase3 阶段2(D2 全 19 变体透传):EventBus subscriber 把 AiChatEvent Value 经此方法
|
|
/// 透传(不经 TunnelEvent 强类型子集)。与 [`send_event`](TunnelClient::send_event) 伴行,
|
|
/// 后者保留作高频子集快捷方式(D5 强类型保留)。非阻塞入队,连接断开返 NotConnected。
|
|
async fn send_raw_event(&self, payload: serde_json::Value) -> Result<()>;
|
|
}
|
|
|
|
/// 握手探测窗口(relay 校验通过后静默进入收发循环,无 ack 帧;
|
|
/// 此窗口内若收到 error 帧或连接 Close,判定握手失败)
|
|
const HANDSHAKE_PROBE: Duration = Duration::from_secs(3);
|
|
|
|
/// 心跳间隔(应用层 Ping,补协议层 keepalive,防 NAT 连接表超时回收)
|
|
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(25);
|
|
|
|
/// 重连初始退避
|
|
const RECONNECT_BASE: Duration = Duration::from_secs(1);
|
|
/// 重连退避上限
|
|
const RECONNECT_MAX: Duration = Duration::from_secs(30);
|
|
|
|
/// 内部连接状态(disconnect 时 drop 即清资源)
|
|
struct ConnState {
|
|
/// 发送队列:send_event 投递 → 收发循环消费写 socket
|
|
tx: mpsc::UnboundedSender<OutFrame>,
|
|
/// 收发循环 task handle(disconnect 时 abort)
|
|
task: JoinHandle<()>,
|
|
}
|
|
|
|
/// 收发循环出站帧(SendLoop 内部消费)
|
|
enum OutFrame {
|
|
/// 业务事件(已序列化 JSON 字符串,run_loop 直接写 socket)
|
|
///
|
|
/// send_event(TunnelEvent) 与 send_raw_event(Value) 各自预序列化为 String 入队,
|
|
/// 统一经此变体写 socket(DRY:序列化在调用侧,run_loop 只负责写)。
|
|
Event(String),
|
|
/// 优雅关闭
|
|
Close,
|
|
}
|
|
|
|
/// 默认 WS 隧道客户端(完整实现)
|
|
pub struct WsTunnelClient {
|
|
/// 目标服务器 URL(connect 后保存,重连用;字段供未来 auto-reconnect 读)
|
|
/// 注:当前 connect 一次性建立,未实现自动后台重连(手动 connect 即重连),
|
|
/// 该字段保留以备 Phase3「断线后台自动重连」消费。
|
|
server_url: Mutex<Option<String>>,
|
|
device_id: Mutex<Option<String>>,
|
|
token: Mutex<Option<String>>,
|
|
/// 当前活跃连接状态(None = 未连接)
|
|
conn: Mutex<Option<ConnState>>,
|
|
/// 连接状态标志(与收发循环 task 共享同一 Arc,is_connected 无锁查询;
|
|
/// task 退出时置 false,connect 时置 true)
|
|
connected: Arc<AtomicBool>,
|
|
}
|
|
|
|
impl WsTunnelClient {
|
|
/// 创建未连接的客户端实例
|
|
pub fn new() -> Self {
|
|
Self {
|
|
server_url: Mutex::new(None),
|
|
device_id: Mutex::new(None),
|
|
token: Mutex::new(None),
|
|
conn: Mutex::new(None),
|
|
connected: Arc::new(AtomicBool::new(false)),
|
|
}
|
|
}
|
|
|
|
/// 当前已连接时取一份重连参数(未来自动重连用)
|
|
pub async fn reconnect_params(&self) -> Option<(String, String, String)> {
|
|
let url = self.server_url.lock().await.clone()?;
|
|
let device_id = self.device_id.lock().await.clone()?;
|
|
let token = self.token.lock().await.clone()?;
|
|
Some((url, device_id, token))
|
|
}
|
|
}
|
|
|
|
impl Default for WsTunnelClient {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl TunnelClient for WsTunnelClient {
|
|
async fn connect(
|
|
&self,
|
|
url: &str,
|
|
device_id: &str,
|
|
token: &str,
|
|
on_command: CommandHandler,
|
|
) -> Result<()> {
|
|
// 已连接则先清(支持重复 connect 视作重连)
|
|
if self.connected.load(Ordering::Relaxed) {
|
|
self.cleanup_conn().await;
|
|
}
|
|
|
|
// 1. 建立 WS 连接
|
|
let (ws_stream, _resp) = tokio_tungstenite::connect_async(url)
|
|
.await
|
|
.map_err(|e| TunnelError::Connect(format!("WS 连接失败 {url}: {e}")))?;
|
|
tracing::info!(%url, "WS 连接已建立,开始 Hello 握手");
|
|
|
|
let (mut sink, mut stream) = ws_stream.split();
|
|
|
|
// 2. 发 Hello 握手帧
|
|
let hello = Hello {
|
|
kind: HelloKind::Device,
|
|
device_id: device_id.to_string(),
|
|
token: token.to_string(),
|
|
};
|
|
let hello_json = serde_json::to_string(&hello)?;
|
|
sink.send(WsMessage::Text(hello_json)).await.map_err(|e| {
|
|
TunnelError::Connect(format!("发送 Hello 失败: {e}"))
|
|
})?;
|
|
|
|
// 3. 探测握手结果:HANDSHAKE_PROBE 窗口内
|
|
// - 收到 error 帧 → relay 拒绝(返回 Auth 错误,附原因)
|
|
// - 收到 Close / EOF → 连接被关(返回 Connect 错误)
|
|
// - 收到首条正常消息 / 超时无负帧 → 视为握手成功(relay 静默进入收发循环)
|
|
// - Ping/Pong 协议帧:忽略继续等
|
|
let mut first_frame_pending: Option<WsMessage> = None;
|
|
let probe = tokio::time::timeout(HANDSHAKE_PROBE, stream.next()).await;
|
|
match probe {
|
|
Err(_) => {
|
|
// 超时无负帧:relay 静默通过(常见路径)
|
|
tracing::debug!("握手探测窗口无负帧,视为成功");
|
|
}
|
|
Ok(None) => {
|
|
return Err(TunnelError::Connect(
|
|
"握手阶段连接关闭(relay 未响应)".into(),
|
|
));
|
|
}
|
|
Ok(Some(Err(e))) => {
|
|
return Err(TunnelError::Connect(format!("握手 recv 错误: {e}")));
|
|
}
|
|
Ok(Some(Ok(msg))) => {
|
|
first_frame_pending = Some(msg);
|
|
}
|
|
}
|
|
|
|
// 校验首帧是否为 relay 拒绝帧
|
|
if let Some(msg) = &first_frame_pending {
|
|
if let Some(err) = parse_relay_error(msg) {
|
|
return Err(TunnelError::Auth(format!(
|
|
"relay 拒绝握手: {}",
|
|
err.error
|
|
)));
|
|
}
|
|
if matches!(msg, WsMessage::Close(_)) {
|
|
return Err(TunnelError::Connect("握手被 relay Close".into()));
|
|
}
|
|
}
|
|
|
|
// 4. 握手成功:保存参数 + 建发送队列 + spawn 收发循环
|
|
*self.server_url.lock().await = Some(url.to_string());
|
|
*self.device_id.lock().await = Some(device_id.to_string());
|
|
*self.token.lock().await = Some(token.to_string());
|
|
|
|
let (tx, rx) = mpsc::unbounded_channel::<OutFrame>();
|
|
// 复用 self.connected 的 Arc,task 退出时置 false 与 self 同步
|
|
self.connected.store(true, Ordering::Relaxed);
|
|
let connected_for_task = self.connected.clone();
|
|
|
|
let task = tokio::spawn(run_loop(
|
|
sink,
|
|
stream,
|
|
first_frame_pending,
|
|
rx,
|
|
on_command,
|
|
connected_for_task,
|
|
));
|
|
|
|
*self.conn.lock().await = Some(ConnState { tx, task });
|
|
tracing::info!(%url, device_id, "WsTunnelClient 连接就绪,收发循环已启动");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn disconnect(&self) -> Result<()> {
|
|
self.cleanup_conn().await;
|
|
tracing::info!("WsTunnelClient 已断开");
|
|
Ok(())
|
|
}
|
|
|
|
fn is_connected(&self) -> bool {
|
|
self.connected.load(Ordering::Relaxed)
|
|
}
|
|
|
|
async fn send_event(&self, event: TunnelEvent) -> Result<()> {
|
|
// 序列化提前(OutFrame::Event 持 String),失败返 Serde 错给调用方
|
|
let json = serde_json::to_string(&event)?;
|
|
let conn = self.conn.lock().await;
|
|
let Some(state) = conn.as_ref() else {
|
|
return Err(TunnelError::NotConnected);
|
|
};
|
|
state
|
|
.tx
|
|
.send(OutFrame::Event(json))
|
|
.map_err(|_| TunnelError::NotConnected)?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn send_raw_event(&self, payload: serde_json::Value) -> Result<()> {
|
|
// AiChatEvent Value 序列化为 JSON 字符串入队(与 send_event 共用 OutFrame::Event(String))
|
|
let json = serde_json::to_string(&payload)?;
|
|
let conn = self.conn.lock().await;
|
|
let Some(state) = conn.as_ref() else {
|
|
return Err(TunnelError::NotConnected);
|
|
};
|
|
state
|
|
.tx
|
|
.send(OutFrame::Event(json))
|
|
.map_err(|_| TunnelError::NotConnected)?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
impl WsTunnelClient {
|
|
/// 清理当前连接(Close 帧 + abort task + 置标志)
|
|
async fn cleanup_conn(&self) {
|
|
let mut conn = self.conn.lock().await;
|
|
if let Some(state) = conn.take() {
|
|
// 通知收发循环优雅关闭(忽略发送失败:循环已退出)
|
|
let _ = state.tx.send(OutFrame::Close);
|
|
// 给 task 一点时间自然退出(发完 Close 帧);超时则 abort 兜底
|
|
let mut task = state.task;
|
|
tokio::select! {
|
|
_ = &mut task => {}
|
|
_ = tokio::time::sleep(Duration::from_millis(500)) => {
|
|
task.abort();
|
|
}
|
|
}
|
|
}
|
|
*self.server_url.lock().await = None;
|
|
*self.device_id.lock().await = None;
|
|
*self.token.lock().await = None;
|
|
self.connected.store(false, Ordering::Relaxed);
|
|
}
|
|
}
|
|
|
|
/// 收发循环主体(spawn 后独立运行)
|
|
///
|
|
/// 职责:
|
|
/// 1. socket 入帧 → 解析 BroadcastMessage → 取 payload Value 纯透传 → on_command 回调(协议解析在桥接层)
|
|
/// 2. mpsc 出帧 → socket 写入(Event 序列化 / Ping 心跳 / Close 关闭)
|
|
/// 3. 心跳定时器:每 HEARTBEAT_INTERVAL 发一次 Ping
|
|
/// 4. 任一端断开 → 退出 task,置 connected=false
|
|
///
|
|
/// 注:本任务范围不含「自动后台重连」(disconnect 即退出)。Phase3 可在外层包一层
|
|
/// supervisor task,检测 task 退出后用指数退避重新 connect(参数从 reconnect_params 取)。
|
|
async fn run_loop(
|
|
mut sink: futures_util::stream::SplitSink<
|
|
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
|
|
WsMessage,
|
|
>,
|
|
mut stream: futures_util::stream::SplitStream<
|
|
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
|
|
>,
|
|
first_frame: Option<WsMessage>,
|
|
mut rx: mpsc::UnboundedReceiver<OutFrame>,
|
|
on_command: CommandHandler,
|
|
connected: Arc<AtomicBool>,
|
|
) {
|
|
// 心跳 tick
|
|
let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
|
|
heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
|
// 首次 tick 立即触发会干扰,跳过
|
|
heartbeat.tick().await;
|
|
|
|
// 处理握手期已读但未消费的首帧(可能是首条业务消息)
|
|
if let Some(msg) = first_frame {
|
|
if !handle_inbound(&msg, &on_command).await {
|
|
tracing::info!("首帧处理后连接应关闭,退出收发循环");
|
|
connected.store(false, Ordering::Relaxed);
|
|
let _ = sink.close().await;
|
|
return;
|
|
}
|
|
}
|
|
|
|
loop {
|
|
tokio::select! {
|
|
// socket 入帧
|
|
maybe_msg = stream.next() => {
|
|
match maybe_msg {
|
|
Some(Ok(msg)) => {
|
|
if !handle_inbound(&msg, &on_command).await {
|
|
tracing::info!("对端关闭连接,收发循环退出");
|
|
break;
|
|
}
|
|
}
|
|
Some(Err(e)) => {
|
|
tracing::warn!(error = %e, "socket recv 错误,收发循环退出");
|
|
break;
|
|
}
|
|
None => {
|
|
tracing::info!("socket 流结束,收发循环退出");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
// 出站队列
|
|
maybe_out = rx.recv() => {
|
|
match maybe_out {
|
|
// OutFrame::Event 持已序列化 JSON 字符串(send_event/send_raw_event 预序列化)
|
|
Some(OutFrame::Event(json)) => {
|
|
if let Err(e) = sink.send(WsMessage::Text(json)).await {
|
|
tracing::warn!(error = %e, "socket 写入失败,收发循环退出");
|
|
break;
|
|
}
|
|
}
|
|
Some(OutFrame::Close) => {
|
|
tracing::debug!("收到 disconnect 指令,发 Close 帧优雅关闭");
|
|
let _ = sink.send(WsMessage::Close(Some(CloseFrame {
|
|
code: tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Normal,
|
|
reason: "client_disconnect".into(),
|
|
}))).await;
|
|
break;
|
|
}
|
|
None => {
|
|
tracing::debug!("发送队列关闭(客户端 drop),退出收发循环");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
// 心跳
|
|
_ = heartbeat.tick() => {
|
|
if let Err(e) = sink.send(WsMessage::Ping(Vec::new())).await {
|
|
tracing::warn!(error = %e, "心跳发送失败,收发循环退出");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
connected.store(false, Ordering::Relaxed);
|
|
// 确保 sink 关闭(若上面 break 未发 Close,这里兜底)
|
|
let _ = sink.close().await;
|
|
}
|
|
|
|
/// 处理入站帧,返回 false 表示连接应关闭(Close/EOF/错误)
|
|
async fn handle_inbound(msg: &WsMessage, on_command: &CommandHandler) -> bool {
|
|
match msg {
|
|
WsMessage::Text(text) => {
|
|
// Phase3 纯透传(D1=A):relay 转发完整 BroadcastMessage,
|
|
// payload 是 miniapp 指令原样 JSON。tunnel 只提取 payload Value 上抛,
|
|
// 不解析业务协议(协议解析在 device 端桥接层 src-tauri)。
|
|
if let Some(payload) = parse_payload_from_broadcast(text) {
|
|
// 回调内执行指令路由(桥接层 match cmd → Tauri command);回调失败不影响收发循环
|
|
let fut = on_command(payload);
|
|
fut.await;
|
|
} else {
|
|
// 非 Command 消息(Control/Event 回环/未知):忽略,不中断
|
|
tracing::debug!(text = %text.chars().take(200).collect::<String>(), "收到非 Command 入站消息,忽略");
|
|
}
|
|
true
|
|
}
|
|
WsMessage::Binary(_) => {
|
|
tracing::debug!("收到二进制帧,忽略");
|
|
true
|
|
}
|
|
WsMessage::Ping(_) => {
|
|
// tungstenite 协议层自动应答 Pong,这里不处理
|
|
true
|
|
}
|
|
WsMessage::Pong(_) => true,
|
|
WsMessage::Close(_) => false,
|
|
WsMessage::Frame(_) => true, // 原始帧,上层已聚合,忽略
|
|
}
|
|
}
|
|
|
|
/// 从 BroadcastMessage JSON 提取 command 方向的 payload Value(纯透传,不解析业务协议)
|
|
///
|
|
/// Phase3 协议统一(D1=A):df-relay 入站包成 `BroadcastMessage { ..., payload: Value }`,
|
|
/// tunnel 只提取 payload 字段以 Value 上抛 on_command,不反序列化为 TunnelCommand 强类型
|
|
/// (业务协议解析在 device 端桥接层)。仅 kind=="command" 时提取(Event/Control 不提,避免回环噪音)。
|
|
fn parse_payload_from_broadcast(raw: &str) -> Option<serde_json::Value> {
|
|
// 仅取 payload 字段,避整结构强类型耦合(device_id/source 等字段本客户端不关心)
|
|
#[derive(serde::Deserialize)]
|
|
struct BroadcastLike {
|
|
#[serde(default)]
|
|
payload: Option<serde_json::Value>,
|
|
#[serde(default)]
|
|
kind: Option<String>,
|
|
}
|
|
let parsed: BroadcastLike = serde_json::from_str(raw).ok()?;
|
|
// 仅 kind == "command" 时才提 payload(Event/Control 不提)
|
|
if parsed.kind.as_deref() != Some("command") {
|
|
return None;
|
|
}
|
|
parsed.payload
|
|
}
|
|
|
|
/// 判断是否为 relay 控制面错误帧
|
|
fn parse_relay_error(msg: &WsMessage) -> Option<RelayControlError> {
|
|
if let WsMessage::Text(t) = msg {
|
|
serde_json::from_str::<RelayControlError>(t).ok()
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// 计算指数退避延迟(供未来 supervisor 重连使用,本任务未直接调用,但属完整重连语义的一部分)
|
|
#[allow(dead_code)]
|
|
fn backoff_delay(attempt: u32) -> Duration {
|
|
// 1s, 2s, 4s, 8s, 16s, 30s, 30s, ...
|
|
let secs = RECONNECT_BASE
|
|
.as_secs()
|
|
.saturating_mul(1u64 << attempt.min(5));
|
|
Duration::from_secs(secs.min(RECONNECT_MAX.as_secs()))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn backoff_caps_at_max() {
|
|
assert_eq!(backoff_delay(0), RECONNECT_BASE);
|
|
assert_eq!(backoff_delay(1), Duration::from_secs(2));
|
|
assert_eq!(backoff_delay(5), Duration::from_secs(32).min(RECONNECT_MAX));
|
|
assert_eq!(backoff_delay(10), RECONNECT_MAX);
|
|
}
|
|
|
|
#[test]
|
|
fn parse_payload_from_broadcast_extracts_value() {
|
|
// 模拟 relay 转发的 BroadcastMessage(miniapp 发的 MiniCommand {cmd, args} 在 payload)
|
|
let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"cmd":"send_message","args":{"message":"hi","conversation_id":"c1"}},"ts":1700000000000}"#;
|
|
let payload = parse_payload_from_broadcast(raw).expect("应提取出 payload Value");
|
|
// 纯透传:payload 原样上抛,字段保持 miniapp 端 {cmd, args} 结构(未做协议转换)
|
|
assert_eq!(payload["cmd"], "send_message");
|
|
assert_eq!(payload["args"]["message"], "hi");
|
|
assert_eq!(payload["args"]["conversation_id"], "c1");
|
|
}
|
|
|
|
#[test]
|
|
fn parse_payload_ignores_event_kind() {
|
|
// Event 方向(device→miniapp)不提取,避免回环噪音
|
|
let raw = r#"{"device_id":"dev-1","kind":"event","source":{"0":1},"from":"device","payload":{"type":"text_delta","conversation_id":"c1","delta":"x"},"ts":1}"#;
|
|
assert!(parse_payload_from_broadcast(raw).is_none());
|
|
}
|
|
|
|
/// D5 弱校验:payload Value 仍可反序列化为 TunnelCommand(保留强类型作桥接层可选校验)
|
|
#[test]
|
|
fn payload_compatible_with_tunnel_command() {
|
|
use crate::events::TunnelCommand;
|
|
let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"kind":"send","conv_id":"c1","content":"hi"},"ts":1}"#;
|
|
let payload = parse_payload_from_broadcast(raw).expect("应提取 payload");
|
|
let cmd: TunnelCommand = serde_json::from_value(payload).expect("payload 应兼容 TunnelCommand");
|
|
match cmd {
|
|
TunnelCommand::Send { conv_id, content } => {
|
|
assert_eq!(conv_id, "c1");
|
|
assert_eq!(content, "hi");
|
|
}
|
|
other => panic!("预期 Send,实际 {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn parse_relay_error_detects_auth_failure() {
|
|
let msg = WsMessage::Text(r#"{"kind":"control","error":"auth_failed"}"#.into());
|
|
let err = parse_relay_error(&msg).expect("应解析出错误帧");
|
|
assert_eq!(err.error, "auth_failed");
|
|
}
|
|
}
|