新增: df-tunnel Phase2 WS隧道客户端(出站穿NAT+Hello握手+收发循环)
- WsTunnelClient: connect_async连df-relay /ws/device + Hello(HelloKind::Device) - 握手探测(HANDSHAKE_PROBE 3s):无负帧=relay静默通过 / error帧=Auth失败 / Close=Connect失败 - 收发循环: socket入帧→parse_command_from_broadcast提payload(kind=command)→on_command回调; mpsc出帧→socket写; heartbeat 25s Ping - cleanup_conn: Close帧+500ms等task+abort兜底; connected用Arc<AtomicBool>共享 - backoff_delay(1→2→4→8→16→30s cap)预留Phase3 supervisor自动重连 - 协议对齐 df-relay(Hello字段/方向Event↔Command/payload透传),cargo check零warning+4测试通过
This commit is contained in:
@@ -11,6 +11,44 @@
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// 握手身份宣告帧(对齐 df-relay::relay::Hello)
|
||||
///
|
||||
/// 桌面端 WS 连 `/ws/device` 后,首条消息发此 JSON,relay 校验 kind 与 token。
|
||||
/// 字段名/serde rename 与 df-relay 严格一致(`#[serde(rename_all="snake_case")]`),
|
||||
/// 否则 relay 反序列化失败 → 握手被拒。
|
||||
///
|
||||
/// 注意:本 crate 不依赖 df-relay(避免 lib→server 反向依赖),自行镜像协议结构。
|
||||
/// 协议字段如有变更须同步 df-relay::relay::Hello 与本类型。
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Hello {
|
||||
/// 客户端类型("device" / "miniapp"),桌面端恒为 Device
|
||||
pub kind: HelloKind,
|
||||
/// 配对绑定的设备 ID
|
||||
pub device_id: String,
|
||||
/// 配对 token(env DF_RELAY_TOKEN 同源)
|
||||
pub token: String,
|
||||
}
|
||||
|
||||
/// Hello.kind 传输表示(serde 字符串,与 df-relay::ClientKindWire 一致)
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum HelloKind {
|
||||
Device,
|
||||
Miniapp,
|
||||
}
|
||||
|
||||
/// relay 控制面错误帧(握手失败时 relay 发送)
|
||||
///
|
||||
/// relay 在校验失败/连接异常时发 `{"kind":"control","error":"<reason>"}` 文本帧。
|
||||
/// 桌面端据此识别握手失败原因(不依赖 error 帧亦可:连接被 Close 即视为失败)。
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RelayControlError {
|
||||
/// 固定 "control"(用于与正常 BroadcastMessage 区分,relay 错误帧简化为两字段)
|
||||
pub kind: String,
|
||||
/// 错误原因(handshake_failed / kind_mismatch / auth_failed)
|
||||
pub error: String,
|
||||
}
|
||||
|
||||
/// 隧道透传的事件类型(AiChatEvent 简化子集)
|
||||
///
|
||||
/// 变体语义对齐桌面端 ai-chat-event:
|
||||
|
||||
@@ -22,5 +22,5 @@ pub mod tunnel;
|
||||
|
||||
// 顶层再导出常用项,简化调用方 use
|
||||
pub use error::{Result, TunnelError};
|
||||
pub use events::{TunnelCommand, TunnelEvent, TunnelMessage};
|
||||
pub use events::{Hello, HelloKind, RelayControlError, TunnelCommand, TunnelEvent, TunnelMessage};
|
||||
pub use tunnel::{TunnelClient, WsTunnelClient};
|
||||
|
||||
@@ -1,35 +1,73 @@
|
||||
//! df-tunnel 隧道客户端
|
||||
//! df-tunnel 隧道客户端(Phase2 完整实现)
|
||||
//!
|
||||
//! 设计依据:设计文档「Layer1」—— 桌面端主动出站连云后端 `wss://your-server/ws/device`,
|
||||
//! 穿 NAT 无需端口映射。本模块定义 `TunnelClient` trait + tokio-tungstenite WS 连接骨架。
|
||||
//! 设计依据:设计文档「Layer1」—— 桌面端主动出站连云后端 `wss://host/ws/device`,
|
||||
//! 穿 NAT 无需端口映射。tokio-tungstenite WS + Hello 握手 + 收发循环 + 指数退避重连。
|
||||
//!
|
||||
//! 脚手架范围:trait 定义 + 连接骨架(占位实现),Phase2 填:
|
||||
//! - 实际 WS 握手 + 鉴权(token / device_id)
|
||||
//! - 断线指数退避重连
|
||||
//! - 事件桥接(app.emit 同步 clone 推隧道)
|
||||
//! - 指令路由(收到 TunnelCommand 调 Tauri command)
|
||||
//! ## 协议对齐(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 才是真正的 TunnelCommand
|
||||
//! - 出站消息:桌面端发 TunnelEvent 的 JSON(relay 把它当 payload 包成 BroadcastMessage)
|
||||
//!
|
||||
//! ## AiChatEvent 透传原则
|
||||
//! payload 用 serde_json::Value 透传,本 crate 不依赖 src-tauri/df-types(避跨 crate 强耦合)。
|
||||
//! TunnelEvent 仅是高频子集,扩展变体不破坏向后兼容。
|
||||
//!
|
||||
//! ## 并发模型
|
||||
//! 内部 spawn 收发循环 task,通过 mpsc 解耦 send_event(调用方)与 socket 写入。
|
||||
//! 收到 TunnelCommand 走 on_command 回调(调用方在 connect 时注册),不阻塞收发循环。
|
||||
//! `&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;
|
||||
use crate::events::{TunnelCommand, TunnelEvent};
|
||||
use crate::error::{Result, TunnelError};
|
||||
use crate::events::{Hello, HelloKind, RelayControlError, TunnelCommand, TunnelEvent};
|
||||
|
||||
/// 收到 TunnelCommand 时的回调类型(Boxed Future,在收发循环 task 内 await)
|
||||
///
|
||||
/// 设计为 async 便于调用方执行 Tauri command 异步路由。回调内 panic 会中断收发循环,
|
||||
/// 调用方应自行处理错误(回调返回值忽略,失败由调用方记录)。
|
||||
pub type CommandHandler = Arc<dyn Fn(TunnelCommand) -> futures_util::future::BoxFuture<'static, ()> + Send + Sync>;
|
||||
|
||||
/// 隧道客户端抽象
|
||||
///
|
||||
/// 设计为 trait 而非具体结构,便于:
|
||||
/// 1. 测试用 mock 实现(不依赖真实 WS)
|
||||
/// 2. 未来替换底层实现(如换 sse / quic)不改调用方
|
||||
/// 2. 未来替换底层实现(如换 quic / sse)不改调用方
|
||||
///
|
||||
/// 所有方法 async + 返回 Result<T>,失败显式上抛不吞错。
|
||||
/// 状态语义:
|
||||
/// - `connect` 一次建立连接 + spawn 收发循环,重复 connect 视为重连(先 disconnect)
|
||||
/// - `is_connected` 轻量查询,不发心跳
|
||||
/// - `send_event` 非阻塞入队,实际 socket 写入由收发循环异步完成
|
||||
/// - `disconnect` 优雅关闭(Close 帧 + 停 task + 清资源)
|
||||
#[async_trait]
|
||||
pub trait TunnelClient: Send + Sync {
|
||||
/// 主动连接云后端(出站长连接,穿 NAT)
|
||||
/// 主动连接云后端 + 注册指令回调
|
||||
///
|
||||
/// `url` 形如 `wss://your-server/ws/device?token=xxx`。
|
||||
/// 成功后进入「已连接」状态,失败返回 TunnelError::Connect。
|
||||
async fn connect(&self, url: &str) -> Result<()>;
|
||||
/// `url` 形如 `wss://host/ws/device`(query 不带 token,token 走 Hello 帧)。
|
||||
/// 成功后进入「已连接」状态并 spawn 收发循环。失败返回 TunnelError::Connect/Auth。
|
||||
/// `on_command` 收到云后端转发的 TunnelCommand 时被调用(在收发 task 内异步执行)。
|
||||
async fn connect(
|
||||
&self,
|
||||
url: &str,
|
||||
device_id: &str,
|
||||
token: &str,
|
||||
on_command: CommandHandler,
|
||||
) -> Result<()>;
|
||||
|
||||
/// 主动断开(优雅关闭 WS,释放底层 TCP)
|
||||
/// 主动断开(优雅 Close 帧 + 释放 task / 通道)
|
||||
async fn disconnect(&self) -> Result<()>;
|
||||
|
||||
/// 是否处于已连接状态(轻量查询,不阻塞,不发心跳)
|
||||
@@ -37,39 +75,72 @@ pub trait TunnelClient: Send + Sync {
|
||||
|
||||
/// 发送事件(桌面端 → 云后端 → 小程序)
|
||||
///
|
||||
/// 设计为按事件粒度发送,不聚合;调用方在 app.emit 的同时 clone 推隧道。
|
||||
/// 非阻塞入队,实际 socket 写入异步完成。连接断开时返回 NotConnected。
|
||||
async fn send_event(&self, event: TunnelEvent) -> Result<()>;
|
||||
|
||||
/// 接收指令(云后端 → 桌面端)
|
||||
///
|
||||
/// 阻塞等待下一条指令。返回 None 表示连接已关闭(EOF)。
|
||||
async fn recv_command(&self) -> Result<Option<TunnelCommand>>;
|
||||
}
|
||||
|
||||
/// 默认 WS 隧道客户端骨架(tokio-tungstenite)
|
||||
///
|
||||
/// 脚手架阶段:字段占位,方法返回 `TunnelError::Other("未实现")`。
|
||||
/// Phase2 替换为完整实现:
|
||||
/// - `WebSocketStream<MaybeTlsStream<TcpStream>>` 持有连接
|
||||
/// - `mpsc` 通道解耦 send/recv 任务
|
||||
/// - 鉴权握手 + 心跳 + 重连
|
||||
/// 握手探测窗口(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 {
|
||||
/// 业务事件(序列化后写 socket)
|
||||
Event(TunnelEvent),
|
||||
/// 优雅关闭
|
||||
Close,
|
||||
}
|
||||
|
||||
/// 默认 WS 隧道客户端(完整实现)
|
||||
pub struct WsTunnelClient {
|
||||
/// 目标服务器 URL(连接后保存,重连用)
|
||||
#[allow(dead_code)]
|
||||
server_url: Option<String>,
|
||||
/// 连接状态标志(轻量 is_connected 查询用)
|
||||
#[allow(dead_code)]
|
||||
connected: bool,
|
||||
/// 目标服务器 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: None,
|
||||
connected: false,
|
||||
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 {
|
||||
@@ -80,35 +151,364 @@ impl Default for WsTunnelClient {
|
||||
|
||||
#[async_trait]
|
||||
impl TunnelClient for WsTunnelClient {
|
||||
async fn connect(&self, url: &str) -> Result<()> {
|
||||
// 脚手架占位:Phase2 实现 tokio_tungstenite::connect_async + 鉴权握手。
|
||||
// 此处仅记录意图,返回未实现错误(不静默假成功,避免掩盖调用方逻辑)。
|
||||
tracing::warn!(url = %url, "WsTunnelClient.connect 尚未实现(Phase2 填充)");
|
||||
Err(crate::error::TunnelError::Other(anyhow::anyhow!(
|
||||
"WsTunnelClient.connect 尚未实现(Phase2)"
|
||||
)))
|
||||
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<()> {
|
||||
tracing::warn!("WsTunnelClient.disconnect 尚未实现(Phase2 填充)");
|
||||
Err(crate::error::TunnelError::Other(anyhow::anyhow!(
|
||||
"WsTunnelClient.disconnect 尚未实现(Phase2)"
|
||||
)))
|
||||
self.cleanup_conn().await;
|
||||
tracing::info!("WsTunnelClient 已断开");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn is_connected(&self) -> bool {
|
||||
self.connected
|
||||
self.connected.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
async fn send_event(&self, _event: TunnelEvent) -> Result<()> {
|
||||
Err(crate::error::TunnelError::Other(anyhow::anyhow!(
|
||||
"WsTunnelClient.send_event 尚未实现(Phase2)"
|
||||
)))
|
||||
}
|
||||
|
||||
async fn recv_command(&self) -> Result<Option<TunnelCommand>> {
|
||||
Err(crate::error::TunnelError::Other(anyhow::anyhow!(
|
||||
"WsTunnelClient.recv_command 尚未实现(Phase2)"
|
||||
)))
|
||||
async fn send_event(&self, event: TunnelEvent) -> Result<()> {
|
||||
let conn = self.conn.lock().await;
|
||||
let Some(state) = conn.as_ref() else {
|
||||
return Err(TunnelError::NotConnected);
|
||||
};
|
||||
state
|
||||
.tx
|
||||
.send(OutFrame::Event(event))
|
||||
.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 反序列化 TunnelCommand → 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 {
|
||||
Some(OutFrame::Event(ev)) => {
|
||||
let json = match serde_json::to_string(&ev) {
|
||||
Ok(j) => j,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "事件序列化失败,跳过");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
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) => {
|
||||
// relay 转发的是完整 BroadcastMessage JSON
|
||||
// (含 device_id/kind/source/from/payload/ts,payload 才是 TunnelCommand)
|
||||
// 先尝试按 BroadcastMessage 镜像解析取 payload;失败则尝试直接解析为 TunnelCommand
|
||||
// (兼容 relay 未来直发 payload 的变更)
|
||||
let cmd_opt = parse_command_from_broadcast(text)
|
||||
.or_else(|| serde_json::from_str::<TunnelCommand>(text).ok());
|
||||
|
||||
if let Some(cmd) = cmd_opt {
|
||||
// 回调内执行指令路由(Tauri command);回调失败不影响收发循环
|
||||
let fut = on_command(cmd);
|
||||
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 提取 TunnelCommand
|
||||
///
|
||||
/// df-relay 入站包成 `BroadcastMessage { ..., payload: <原文本解析为 Value> }`,
|
||||
/// 故 payload 字段即 TunnelCommand 序列化后的 JSON 对象(tag="kind")。
|
||||
/// 此处只镜像解析 payload 字段(避免依赖 df-relay crate),payload 用 Value 透传不耦合。
|
||||
fn parse_command_from_broadcast(raw: &str) -> Option<TunnelCommand> {
|
||||
// 仅取 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" 时才尝试提 TunnelCommand(Event/Control 不提)
|
||||
if parsed.kind.as_deref() != Some("command") {
|
||||
return None;
|
||||
}
|
||||
let payload = parsed.payload?;
|
||||
serde_json::from_value::<TunnelCommand>(payload).ok()
|
||||
}
|
||||
|
||||
/// 判断是否为 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_command_from_broadcast_extracts_payload() {
|
||||
// 模拟 relay 转发的 BroadcastMessage
|
||||
let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"kind":"send","conv_id":"c1","content":"hi"},"ts":1700000000000}"#;
|
||||
let cmd = parse_command_from_broadcast(raw).expect("应提取出 TunnelCommand");
|
||||
match cmd {
|
||||
TunnelCommand::Send { conv_id, content } => {
|
||||
assert_eq!(conv_id, "c1");
|
||||
assert_eq!(content, "hi");
|
||||
}
|
||||
other => panic!("预期 Send,实际 {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_command_ignores_event_kind() {
|
||||
let raw = r#"{"device_id":"dev-1","kind":"event","source":{"0":1},"from":"device","payload":{"kind":"text_delta","conv_id":"c1","delta":"x"},"ts":1}"#;
|
||||
assert!(parse_command_from_broadcast(raw).is_none());
|
||||
}
|
||||
|
||||
#[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");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user