sloray
cc6b82d7ec
- 固件：新增 link_control / stream_control，覆盖运行中 Wi-Fi 断线恢复、取帧节拍、 温控与质量/帧率减载；UDP 发送增加有界重试预算与帧期限，PSRAM JPEG 直发减少拷贝。 - 协议：重组器严格校验分片布局，统计未收齐、超时与完全缺失的帧，处理帧号回绕与设备重启。 - 桌面端：视频改 latest-value 通道并前置帧时间标记，拒绝积压和慢解码的过期画面；新增 延迟标记页与 --latency-report 校准，完善录制、服务端与退出流程。 - 文档与脚本：新增硬件、开发、延迟、功耗、实时性、稳定性文档和改动记录，重写根 README； 新增构建、性能、延迟与串口长测脚本，以及长测客户端脚本。 - 构建：固定 esp-1.97.0.0 工具链，分区表随构建复制，更新 sdkconfig 与忽略规则。
534 lines
20 KiB
Rust
534 lines
20 KiB
Rust
//! 界面与后端之间的视频通道：一个只监听回环地址的 WebSocket。
|
||
//!
|
||
//! # 为什么不是 Tauri 的 IPC
|
||
//!
|
||
//! 两路 22fps、每帧约 22 kB，也就是每秒约 1 MB 的裸 JPEG。走 IPC 的话它要被序列化成
|
||
//! JSON（base64 再涨 1/3），再在 JS 侧解回来——CPU 花在没人需要的编解码上，延迟也白
|
||
//! 涨一截。WebSocket 的二进制帧把原始 JPEG 字节直接交给前端，浏览器用原生解码器
|
||
//! （`createImageBitmap`）解，**后端连 JPEG 都不用解**。
|
||
//!
|
||
//! # 为什么要有 token
|
||
//!
|
||
//! 端口是随机分配的，但同机器上任何进程都能连回环地址。这条通道里有实时画面，给它一
|
||
//! 个每次启动随机生成的 token（前端从 `stream_info` 命令拿到），至少让"另一个本地程序
|
||
//! 顺手连上来看画面"变成一件需要先读到那个 token 的事。
|
||
//!
|
||
//! 说清楚它的边界：这**不是**防对手的措施——对手要看画面得先在自己的机器上跑这套客户
|
||
//! 端，那道闸门在 [`cam_core::receiver::Receiver`] 里。这里的 token 只防止本机上的顺手
|
||
//! 连接。
|
||
//!
|
||
//! # 二进制帧的布局
|
||
//!
|
||
//! ```text
|
||
//! 0 4 8 10 12 16
|
||
//! +--------------+--------------+------+------+--------------+---------+
|
||
//! | sender_id u32| frame_id u32 | w u16| h u16| jpeg_len u32 | JPEG... |
|
||
//! +--------------+--------------+------+------+--------------+---------+
|
||
//! ```
|
||
//!
|
||
//! 小端。16 字节头，前端切掉它就能拿到一段完整的 JPEG。`sender_id` 让前端知道这一帧是
|
||
//! 哪块板子的——用户端只画绑定的那一路，裁判端按槽位分配画两路。
|
||
|
||
use std::collections::HashMap;
|
||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||
use std::sync::Arc;
|
||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||
|
||
use futures_util::{SinkExt, StreamExt};
|
||
use serde::Serialize;
|
||
use tokio::net::{TcpListener, TcpStream};
|
||
use tokio::sync::{broadcast, watch};
|
||
use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
|
||
use tokio_tungstenite::tungstenite::http::StatusCode;
|
||
use tokio_tungstenite::tungstenite::{Bytes, Message};
|
||
|
||
use cam_core::snapshot::{HitDto, Snapshot};
|
||
|
||
/// 二进制帧头长度。
|
||
pub const FRAME_HEADER_LEN: usize = 16;
|
||
|
||
/// Ordered control-message capacity. JPEGs use a separate latest-value channel.
|
||
pub const CHANNEL_CAPACITY: usize = 256;
|
||
const LOCAL_VIDEO_MAX_AGE: Duration = Duration::from_millis(100);
|
||
|
||
/// 向界面推送的消息。
|
||
pub enum ServerMsg {
|
||
/// JSON 文本。
|
||
Text(String),
|
||
}
|
||
|
||
#[derive(Clone)]
|
||
pub struct VideoPacket {
|
||
frame_id: u32,
|
||
bytes: Bytes,
|
||
published_at: Instant,
|
||
published_unix_ms: u64,
|
||
}
|
||
|
||
pub type VideoFrames = HashMap<u32, VideoPacket>;
|
||
pub type VideoSender = watch::Sender<VideoFrames>;
|
||
|
||
/// Video uses latest-value semantics; control messages keep ordered delivery.
|
||
pub fn publish_video(tx: &VideoSender, frame: &protocol::Frame) {
|
||
let packet = VideoPacket {
|
||
frame_id: frame.frame_id,
|
||
bytes: encode_frame(frame).into(),
|
||
published_at: Instant::now(),
|
||
published_unix_ms: SystemTime::now()
|
||
.duration_since(UNIX_EPOCH)
|
||
.unwrap_or_default()
|
||
.as_millis() as u64,
|
||
};
|
||
tx.send_modify(|latest| {
|
||
latest.insert(frame.sender_id, packet);
|
||
});
|
||
}
|
||
|
||
/// JSON 文本消息的信封。
|
||
///
|
||
/// 用内部标签 `t` 而不是两个字段：前端一个 `switch (msg.t)` 就能分发，且不需要为每种
|
||
/// 消息各写一次 `JSON.parse` 的错误处理。
|
||
#[derive(Debug, Clone, Serialize)]
|
||
#[serde(tag = "t", rename_all = "camelCase")]
|
||
pub enum WireMsg {
|
||
/// 完整状态快照，5 Hz。
|
||
Snapshot(Box<Snapshot>),
|
||
/// 一次掉血，立即推。
|
||
Hit(HitDto),
|
||
/// 后端拦截了一次"关闭/退出"（用户端还没解锁）。前端据此弹出解锁对话框。
|
||
ConfirmQuit,
|
||
/// 一条说明（迁移、录制报错等），前端显示成一行提示。
|
||
///
|
||
/// 写成带字段的结构体而不是 `Note(String)`：serde 的**内部标签**（`#[serde(tag)]`）
|
||
/// 只支持载荷是 map 的变体，newtype 里放一个字符串会在运行时报错，而不是编译期。
|
||
Note { message: String },
|
||
/// Timestamp precedes its JPEG so a stalled WebView can reject queued data.
|
||
FrameTiming {
|
||
camera: u32,
|
||
#[serde(rename = "frameId")]
|
||
frame_id: u32,
|
||
#[serde(rename = "publishedUnixMs")]
|
||
published_unix_ms: u64,
|
||
},
|
||
}
|
||
|
||
/// 前端连接这条通道需要的信息。
|
||
#[derive(Debug, Clone, Serialize)]
|
||
#[serde(rename_all = "camelCase")]
|
||
pub struct StreamInfo {
|
||
/// `ws://127.0.0.1:<port>/?token=...`
|
||
pub url: String,
|
||
pub port: u16,
|
||
/// 当前连着的界面数。用于诊断"画面不动"到底是后端没推还是前端没连。
|
||
pub clients: usize,
|
||
}
|
||
|
||
/// 把一张帧编码成二进制消息。
|
||
pub fn encode_frame(frame: &protocol::Frame) -> Vec<u8> {
|
||
let mut out = Vec::with_capacity(FRAME_HEADER_LEN + frame.data.len());
|
||
|
||
out.extend_from_slice(&frame.sender_id.to_le_bytes());
|
||
out.extend_from_slice(&frame.frame_id.to_le_bytes());
|
||
out.extend_from_slice(&frame.width.to_le_bytes());
|
||
out.extend_from_slice(&frame.height.to_le_bytes());
|
||
out.extend_from_slice(&(frame.data.len() as u32).to_le_bytes());
|
||
out.extend_from_slice(&frame.data);
|
||
|
||
out
|
||
}
|
||
|
||
/// 从二进制消息里解出帧头。前后端对这段字节布局的约定由测试盯着。
|
||
#[cfg(test)]
|
||
pub fn decode_frame_header(message: &[u8]) -> Option<(u32, u32, u16, u16, usize)> {
|
||
if message.len() < FRAME_HEADER_LEN {
|
||
return None;
|
||
}
|
||
|
||
let sender_id = u32::from_le_bytes(message[0..4].try_into().ok()?);
|
||
let frame_id = u32::from_le_bytes(message[4..8].try_into().ok()?);
|
||
let width = u16::from_le_bytes(message[8..10].try_into().ok()?);
|
||
let height = u16::from_le_bytes(message[10..12].try_into().ok()?);
|
||
let len = u32::from_le_bytes(message[12..16].try_into().ok()?) as usize;
|
||
|
||
if message.len() < FRAME_HEADER_LEN + len {
|
||
return None;
|
||
}
|
||
|
||
Some((sender_id, frame_id, width, height, len))
|
||
}
|
||
|
||
/// 绑一个随机端口的回环 listener。
|
||
///
|
||
/// 随机端口而不是固定端口：同一台机器上同时开用户端和裁判端时不会撞车。
|
||
pub async fn bind() -> std::io::Result<(TcpListener, u16)> {
|
||
let listener = TcpListener::bind(("127.0.0.1", 0)).await?;
|
||
let port = listener.local_addr()?.port();
|
||
|
||
Ok((listener, port))
|
||
}
|
||
|
||
/// 接受循环。每个连接一个任务，活到连接断开。
|
||
pub async fn serve(
|
||
listener: TcpListener,
|
||
token: String,
|
||
tx: broadcast::Sender<Arc<ServerMsg>>,
|
||
video: VideoSender,
|
||
clients: Arc<AtomicUsize>,
|
||
) {
|
||
loop {
|
||
let (stream, peer) = match listener.accept().await {
|
||
Ok(pair) => pair,
|
||
Err(err) => {
|
||
log::warn!("stream: accept failed: {err}");
|
||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||
continue;
|
||
}
|
||
};
|
||
|
||
let tx = tx.clone();
|
||
let token = token.clone();
|
||
let clients = Arc::clone(&clients);
|
||
let video = video.subscribe();
|
||
|
||
tauri::async_runtime::spawn(async move {
|
||
if let Err(err) = handle_client(stream, peer, &token, tx, video, clients).await {
|
||
log::debug!("stream: client {peer} ended: {err}");
|
||
}
|
||
});
|
||
}
|
||
}
|
||
|
||
async fn handle_client(
|
||
stream: TcpStream,
|
||
peer: std::net::SocketAddr,
|
||
token: &str,
|
||
tx: broadcast::Sender<Arc<ServerMsg>>,
|
||
mut video: watch::Receiver<VideoFrames>,
|
||
clients: Arc<AtomicUsize>,
|
||
) -> anyhow::Result<()> {
|
||
stream.set_nodelay(true).ok();
|
||
|
||
let expected = token.to_string();
|
||
let callback = move |req: &Request, res: Response| -> Result<Response, ErrorResponse> {
|
||
let query = req.uri().query().unwrap_or("");
|
||
let presented = query
|
||
.split('&')
|
||
.filter_map(|pair| pair.split_once('='))
|
||
.find(|(key, _)| *key == "token")
|
||
.map(|(_, value)| value.to_string());
|
||
|
||
match presented {
|
||
Some(value) if constant_time_eq(&value, &expected) => Ok(res),
|
||
_ => {
|
||
log::warn!("stream: refusing {peer}: wrong or missing token");
|
||
let mut response = ErrorResponse::new(Some("bad token".into()));
|
||
*response.status_mut() = StatusCode::FORBIDDEN;
|
||
Err(response)
|
||
}
|
||
}
|
||
};
|
||
|
||
let ws = tokio_tungstenite::accept_hdr_async(stream, callback).await?;
|
||
let (mut sink, mut incoming) = ws.split();
|
||
|
||
let mut rx = tx.subscribe();
|
||
video.mark_changed();
|
||
let mut last_sent = HashMap::new();
|
||
let mut keepalive = tokio::time::interval(Duration::from_secs(20));
|
||
let count = clients.fetch_add(1, Ordering::Relaxed) + 1;
|
||
log::info!("stream: interface connected from {peer} ({count} connected)");
|
||
|
||
// 连上就先推一条提示，前端能据此在界面上显示"通道已接通"。
|
||
let hello = serde_json::to_string(&WireMsg::Note {
|
||
message: "video channel ready".into(),
|
||
})
|
||
.unwrap_or_else(|_| "{}".into());
|
||
let _ = sink.send(Message::text(hello)).await;
|
||
|
||
let result = 'messages: loop {
|
||
tokio::select! {
|
||
changed = video.changed() => {
|
||
if changed.is_err() { break Ok(()); }
|
||
// Release the watch read lock before awaiting socket writes.
|
||
let latest: Vec<_> = video.borrow_and_update().iter()
|
||
.map(|(&camera, packet)| (camera, packet.clone())).collect();
|
||
for (camera, packet) in latest {
|
||
if last_sent.get(&camera) == Some(&packet.frame_id) { continue; }
|
||
if packet.published_at.elapsed() >= LOCAL_VIDEO_MAX_AGE { continue; }
|
||
let timing = serde_json::to_string(&WireMsg::FrameTiming {
|
||
camera, frame_id: packet.frame_id, published_unix_ms: packet.published_unix_ms,
|
||
})?;
|
||
// Keep the 16-byte JPEG header compatible. This small text
|
||
// message carries host queue age through TCP/WebView queues.
|
||
let send = async {
|
||
sink.send(Message::text(timing)).await?;
|
||
sink.send(Message::Binary(packet.bytes)).await
|
||
};
|
||
match tokio::time::timeout(LOCAL_VIDEO_MAX_AGE, send).await {
|
||
Ok(Ok(())) => { last_sent.insert(camera, packet.frame_id); }
|
||
Ok(Err(err)) => break 'messages Err(anyhow::anyhow!("video send failed: {err}")),
|
||
Err(_) => break 'messages Err(anyhow::anyhow!("video client stalled")),
|
||
}
|
||
}
|
||
},
|
||
message = rx.recv() => match message {
|
||
Ok(msg) => {
|
||
let out = match &*msg {
|
||
ServerMsg::Text(text) => Message::text(text.clone()),
|
||
};
|
||
|
||
if let Err(err) = sink.send(out).await {
|
||
break Err(anyhow::anyhow!("send failed: {err}"));
|
||
}
|
||
}
|
||
// 接收慢，积压被挤掉了。对视频这是正常且期望的：跳过旧帧，继续送新的。
|
||
Err(broadcast::error::RecvError::Lagged(skipped)) => {
|
||
log::debug!("stream: {peer} lagged {skipped} messages; skipping ahead");
|
||
}
|
||
Err(broadcast::error::RecvError::Closed) => {
|
||
break Ok(());
|
||
}
|
||
},
|
||
|
||
_ = keepalive.tick() => {
|
||
// WKWebView 侧的连接可能在休眠后静默失效。ping 让这种情况变成一次明确的
|
||
// 发送失败，而不是一个永远不刷新的窗口。
|
||
if let Err(err) = sink.send(Message::Ping(Bytes::new())).await {
|
||
break Err(anyhow::anyhow!("keepalive failed: {err}"));
|
||
}
|
||
}
|
||
|
||
incoming_msg = incoming.next() => match incoming_msg {
|
||
None => break Ok(()),
|
||
Some(Err(err)) => break Err(anyhow::anyhow!("receive failed: {err}")),
|
||
// 这条通道是单向的：前端发来的东西（pong 除外）一律忽略。
|
||
Some(Ok(_)) => {}
|
||
},
|
||
}
|
||
};
|
||
|
||
let count = clients.fetch_sub(1, Ordering::Relaxed) - 1;
|
||
log::info!("stream: interface disconnected ({count} connected)");
|
||
|
||
result
|
||
}
|
||
|
||
/// 定长比较，避免 token 校验泄露时间信息。
|
||
fn constant_time_eq(a: &str, b: &str) -> bool {
|
||
let (a, b) = (a.as_bytes(), b.as_bytes());
|
||
|
||
if a.len() != b.len() {
|
||
return false;
|
||
}
|
||
|
||
let mut diff = 0u8;
|
||
for (x, y) in a.iter().zip(b) {
|
||
diff |= x ^ y;
|
||
}
|
||
|
||
diff == 0
|
||
}
|
||
|
||
/// 生成一个随机 token。
|
||
pub fn random_token() -> String {
|
||
use rand::RngCore;
|
||
|
||
let mut bytes = [0u8; 24];
|
||
rand::rng().fill_bytes(&mut bytes);
|
||
|
||
bytes.iter().map(|b| format!("{b:02x}")).collect()
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
fn frame(sender: u32, id: u32, len: usize) -> protocol::Frame {
|
||
protocol::Frame {
|
||
sender_id: sender,
|
||
frame_id: id,
|
||
width: 640,
|
||
height: 480,
|
||
data: vec![0xAB; len],
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn the_binary_header_roundtrips() {
|
||
let encoded = encode_frame(&frame(0x3D4D_8A9A, 4242, 22_000));
|
||
|
||
assert_eq!(encoded.len(), FRAME_HEADER_LEN + 22_000);
|
||
assert_eq!(
|
||
decode_frame_header(&encoded),
|
||
Some((0x3D4D_8A9A, 4242, 640, 480, 22_000))
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn a_truncated_message_is_refused_rather_than_read_past_the_end() {
|
||
let mut encoded = encode_frame(&frame(1, 1, 100));
|
||
encoded.truncate(FRAME_HEADER_LEN + 40);
|
||
|
||
// 头里声明 100 字节，实际只有 40：宁可返回 None，也不要让前端拿到一段不完整的
|
||
// JPEG（那会解码失败，或者更糟——解出一张半截的图）。
|
||
assert_eq!(decode_frame_header(&encoded), None);
|
||
assert_eq!(decode_frame_header(&[0u8; 4]), None);
|
||
}
|
||
|
||
#[test]
|
||
fn the_token_comparison_is_exact() {
|
||
assert!(constant_time_eq("abc", "abc"));
|
||
assert!(!constant_time_eq("abc", "abd"));
|
||
assert!(!constant_time_eq("abc", "abcd"));
|
||
assert!(!constant_time_eq("", "a"));
|
||
assert!(constant_time_eq("", ""));
|
||
}
|
||
|
||
#[test]
|
||
fn tokens_differ_between_runs() {
|
||
assert_ne!(random_token(), random_token());
|
||
assert_eq!(random_token().len(), 48);
|
||
}
|
||
|
||
#[test]
|
||
fn a_slow_video_consumer_keeps_only_the_latest_frame_of_each_camera() {
|
||
let (video, mut reader) = watch::channel(VideoFrames::new());
|
||
for id in 1..=1000 {
|
||
publish_video(&video, &frame(1, id, 100));
|
||
publish_video(&video, &frame(2, id + 10, 100));
|
||
}
|
||
let latest = reader.borrow_and_update();
|
||
assert_eq!(latest.len(), 2);
|
||
assert_eq!(latest[&1].frame_id, 1000);
|
||
assert_eq!(latest[&2].frame_id, 1010);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn a_reconnecting_interface_receives_cached_latest_frames_without_waiting_for_capture() {
|
||
let (listener, port) = bind().await.unwrap();
|
||
let (control, _) = broadcast::channel(CHANNEL_CAPACITY);
|
||
let (video, _) = watch::channel(VideoFrames::new());
|
||
for id in 1..=100 {
|
||
publish_video(&video, &frame(1, id, 100));
|
||
publish_video(&video, &frame(2, id + 10, 100));
|
||
}
|
||
let task = tokio::spawn(serve(
|
||
listener,
|
||
"test-token".into(),
|
||
control,
|
||
video,
|
||
Arc::new(AtomicUsize::new(0)),
|
||
));
|
||
let (mut client, _) =
|
||
tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}/?token=test-token"))
|
||
.await
|
||
.unwrap();
|
||
let mut received = Vec::new();
|
||
let mut timings = HashMap::new();
|
||
while received.len() < 2 {
|
||
let message = tokio::time::timeout(Duration::from_secs(2), client.next())
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.unwrap();
|
||
if let Message::Text(text) = &message {
|
||
let value: serde_json::Value = serde_json::from_str(text).unwrap();
|
||
if value["t"] == "frameTiming" {
|
||
assert!(value["publishedUnixMs"].as_u64().unwrap() > 0);
|
||
timings.insert(
|
||
value["camera"].as_u64().unwrap() as u32,
|
||
value["frameId"].as_u64().unwrap() as u32,
|
||
);
|
||
}
|
||
}
|
||
if let Message::Binary(bytes) = message {
|
||
let (camera, frame_id, _, _, _) = decode_frame_header(&bytes).unwrap();
|
||
assert_eq!(timings.get(&camera), Some(&frame_id));
|
||
received.push((camera, frame_id));
|
||
}
|
||
}
|
||
received.sort_unstable();
|
||
assert_eq!(received, vec![(1, 100), (2, 110)]);
|
||
client.close(None).await.unwrap();
|
||
task.abort();
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn reconnect_does_not_replay_expired_cached_video() {
|
||
let (listener, port) = bind().await.unwrap();
|
||
let (control, _) = broadcast::channel(CHANNEL_CAPACITY);
|
||
let (video, _) = watch::channel(VideoFrames::new());
|
||
publish_video(&video, &frame(1, 1, 100));
|
||
video.send_modify(|latest| {
|
||
latest.get_mut(&1).unwrap().published_at = Instant::now() - Duration::from_secs(1)
|
||
});
|
||
let task = tokio::spawn(serve(
|
||
listener,
|
||
"test-token".into(),
|
||
control,
|
||
video.clone(),
|
||
Arc::new(AtomicUsize::new(0)),
|
||
));
|
||
let (mut client, _) =
|
||
tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}/?token=test-token"))
|
||
.await
|
||
.unwrap();
|
||
let stale = tokio::time::timeout(Duration::from_millis(100), async {
|
||
while let Some(message) = client.next().await {
|
||
if matches!(message.unwrap(), Message::Binary(_)) {
|
||
return;
|
||
}
|
||
}
|
||
})
|
||
.await;
|
||
assert!(stale.is_err());
|
||
publish_video(&video, &frame(1, 2, 100));
|
||
let fresh = tokio::time::timeout(Duration::from_secs(2), async {
|
||
while let Some(message) = client.next().await {
|
||
if let Message::Binary(bytes) = message.unwrap() {
|
||
return decode_frame_header(&bytes).unwrap().1;
|
||
}
|
||
}
|
||
panic!("video connection closed");
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(fresh, 2);
|
||
client.close(None).await.unwrap();
|
||
task.abort();
|
||
}
|
||
|
||
#[test]
|
||
fn the_wire_envelope_carries_the_message_kind() {
|
||
let json = serde_json::to_string(&WireMsg::Hit(HitDto {
|
||
board: 1,
|
||
team: 0,
|
||
index: 2,
|
||
from: 100,
|
||
to: 97,
|
||
damage: 3,
|
||
at_ms: 1,
|
||
}))
|
||
.unwrap();
|
||
|
||
assert!(json.contains(r#""t":"hit""#), "got {json}");
|
||
assert!(json.contains(r#""damage":3"#), "got {json}");
|
||
}
|
||
|
||
#[test]
|
||
fn a_unit_variant_is_just_a_tag() {
|
||
let json = serde_json::to_string(&WireMsg::ConfirmQuit).unwrap();
|
||
assert_eq!(json, r#"{"t":"confirmQuit"}"#);
|
||
}
|
||
|
||
#[test]
|
||
fn a_note_is_tagged_too() {
|
||
let json = serde_json::to_string(&WireMsg::Note {
|
||
message: "hi".into(),
|
||
})
|
||
.unwrap();
|
||
|
||
assert_eq!(json, r#"{"t":"note","message":"hi"}"#);
|
||
}
|
||
}
|