Horldsence/ESP32S3_vt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7b6593bb750386547e57e37fab98d8cee040635c
ESP32S3_vt/protocol/src/lib.rs
T
Horldsence 7be8d7d2ba protocol: 更正上一个提交的归因——坏帧不在这里 
上一个提交（1c65eaa）说"34 张 JPEG 不完整是重复分片凑数导致的"，并说修完从
34 降到 1。**这个结论是错的**，是我归因到了错误的地方。

实测复查：一到两个订阅者、20 秒 2000+ 个分片，重复投递数为 0（固定端口和
临时端口两种订阅方式都试过）。这条链路上根本不产生重复。

34 张坏帧的真实来源是探测脚本自己：scripts/udp_probe.py 的重组状态只按
sender_id 存，不按 frame_id 存。于是某帧丢了分片之后，下一帧的分片会接着往
同一个计数器里凑，凑够了就交出去一张混着两帧的 JPEG。

FrameAssembler 一直按 frame_id 隔离（换个 frame_id 就重置），所以它没这个
问题——脚本原先跟它不一致，而我又把脚本的输出当成了 crate 的行为。

本次改动：

  protocol/src/lib.rs
    位图去重保留，理由改为正确性而非正在发生的故障，并在注释里写明实测重复
    为 0、以及真正的坏帧来源。删掉一个死变量 `last`。行为不变。

  protocol/README.md
    "重复分片必须去重" 一节改写：说清防护的定位是"哪天真的重复了"。
    新增"按 frame_id 隔离，换帧即放弃"一节，讲清不隔离的具体后果。
    记下探到坏帧时先怀疑测量工具。

值得记的不是位图本身，是那个错误归因：34 张坏帧配上"帧率正常、无报错线索"
这个症状，太像一个真的丢包/重复 bug 了，于是一路写到了注释、文档和提交说明
里。以后探到链路异常，先确认测量工具和被测对象说的是同一件事。
2026-10-05 19:13:24 +08:00

946 lines
32 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! 线路格式：图传视频 + 设备发现 + 装甲板预留接口。
//!
//! 三端共用这一个 crate：两块 ESP32 图传板、队伍客户端、裁判终端。
//!
//! # 拓扑
//!
//! ```text
//! ┌──────────────────────────┐
//! │ 裁判终端（PC，仅显示） │
//! │ 队伍一 + 队伍二 + 血量 │
//! └───────────┬──────────────┘
//! │ 同一个局域网，广播可用
//! ┌──────────────────┼──────────────────┐
//! │ │ │
//! ┌───────┴───────┐ ┌───────┴───────┐ ┌───────┴───────┐
//! │ 图传板 A │ │ 图传板 B │ │ 装甲板 ×4 │
//! │ (队伍一) │ │ (队伍二) │ │ (预留，未实现) │
//! └───────────────┘ └───────────────┘ └───────────────┘
//! ```
//!
//! # 为什么视频走 UDP 而不是 TCP
//!
//! 实测：传感器产出 22 fps，TCP MJPEG 只送出 1.9 fps，且只占用了链路能力的 1%。
//! 链路是空的，帧却堵在 TCP 的队头阻塞后面等一个实时画面等不起的重传计时器。
//! UDP 没有重传与确认，缺了分片的帧整个丢弃、下一帧顶上。
//!
//! # 分片
//!
//! 每个视频数据报 = [`HEADER_LEN`] 字节头 + 至多 [`MAX_FRAGMENT_PAYLOAD`]
//! 字节 JPEG 数据。头里的 `sender_id` 让接收端能在同一端口上区分多块板子，
//! 这是裁判终端能同时看两路的前提。
//!
//! ```text
//! 0 1 2 3
//! 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
//! +-------------------------------+-------+-----------------------+
//! | magic (u32) | type | version (u16) |
//! +-------------------------------+-------+-----------------------+
//! | sender_id (u32) | frame_id (u32) |
//! +-------------------------------+-------+-----------------------+
//! | frag_index (u16) | frag_count (u16) |
//! +-------------------------------+-------+-----------------------+
//! | total_len (u32) | width (u16) | height (u16) |
//! +-------------------------------+---------------+---------------+
//! | payload (0..=1400) |
//! +---------------------------------------------------------------+
//! ```
use std::net::{Ipv4Addr, SocketAddrV4};
// ---------------------------------------------------------------------------
// 端口
// ---------------------------------------------------------------------------
/// 视频与设备发现共用端口。
///
/// 发现和控制消息体积小、频率低，与视频混在一个端口上可以让板子只开一个 socket，
/// 简化固件侧。视频与控制的数据报头不同（`msg_type`），不会混淆。
pub const DEFAULT_PORT: u16 = 5600;
/// 装甲板预留端口。
///
/// 现在没有实现，裁判终端只是监听它并把血量显示为占位符。板子接上以后往这个端口
/// 发消息即可，不需要改动线路格式或界面结构。
pub const ARMOR_PORT: u16 = 5601;
// ---------------------------------------------------------------------------
// 帧头
// ---------------------------------------------------------------------------
/// 线路格式版本。
///
/// 字段增删时递增。不匹配的对端会拒绝处理，而不是按错误布局解析出垃圾数据——
/// 那会表现为随机崩溃或画面错乱，比拒绝连接难查得多。
pub const PROTOCOL_VERSION: u16 = 1;
/// 标识本协议。
///
/// 取值使得它不可能出现在 JPEG 数据开头，所以一个同端口上的无关 UDP 包会被挡掉。
pub const MAGIC: u32 = 0x4D56_4331; // "MVC1"
/// 帧头长度（字节）。
///
/// `magic`(4) + `msg_type`(1) + `version`(2) + `sender_id`(4) + `frame_id`(4) +
/// `frag_index`(2) + `frag_count`(2) + `total_len`(4) + `width`(2) + `height`(2)
/// = **27**。
///
/// 27 不如 4 字节对齐，但这里刻意不去凑：头只占每个 1427 字节数据报的 1.9%，而
/// 为对齐在固件和客户端两侧各写一段手写的偏移常量，是引入静默错位的更划算的代价。
pub const HEADER_LEN: usize = 27;
/// 单个数据报的最大 JPEG 载荷。
///
/// 1400 + 27 = 1427 字节，低于典型以太网的 1500 MTU。这一点比看起来重要：超 MTU
/// 的数据报要么被 AP 丢弃，要么被 IP 分片，两种情况的丢包都记在应用账上而不是网络账
/// 上——后者本来更容易排查。
pub const MAX_FRAGMENT_PAYLOAD: usize = 1400;
/// 数据报类型。
///
/// 决定同一端口上的字节如何解释。把控制和视频放在一个端口的前提就是这个字段
/// 一定在最前面，且值互不相同。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum MsgType {
/// 一帧 JPEG 的第 N 个分片。
Video = 1,
/// 客户端广播："我是这个角色，请给我画面。"
Subscribe = 2,
/// 板子回复："我在线，我是这个 id。"
Hello = 3,
/// 心跳，用于发现与保活。
Heartbeat = 4,
/// 装甲板上报血量（预留，当前无实现）。
ArmorHealth = 5,
}
impl MsgType {
/// 从线上的字节还原。
pub fn from_u8(v: u8) -> Option<Self> {
Some(match v {
1 => MsgType::Video,
2 => MsgType::Subscribe,
3 => MsgType::Hello,
4 => MsgType::Heartbeat,
5 => MsgType::ArmorHealth,
_ => return None,
})
}
}
/// 客户端在订阅里声明的角色。
///
/// 板子据此决定要不要给这个客户端推流。裁判终端用
/// [`SubscribeRole::Referee`](SubscribeRole::Referee) 表示"我要两路都看"，但它仍然
/// 是靠收到两块板子的 Hello 后分别订阅来完成——角色只是省掉了手工筛选。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum SubscribeRole {
/// 队伍客户端，只要一台板子的画面。
Team = 1,
/// 裁判终端，要全部板子的画面。
Referee = 2,
}
impl SubscribeRole {
pub fn from_u8(v: u8) -> Option<Self> {
match v {
1 => Some(SubscribeRole::Team),
2 => Some(SubscribeRole::Referee),
_ => None,
}
}
}
/// 数据报头，20 字节，小端。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C)]
pub struct Header {
pub msg_type: MsgType,
pub version: u16,
/// 发送方标识。
///
/// 板子用 MAC 派生（见 [`id_from_mac`]），这样重启后 id 不变，客户端保存的绑定
/// 不会失效。裁判终端用固定的 [`REFEREE_ID`]。
pub sender_id: u32,
/// 帧序号，按发送方独立计数，回绕无害。
pub frame_id: u32,
pub frag_index: u16,
pub frag_count: u16,
/// 整张 JPEG 的长度。
pub total_len: u32,
pub width: u16,
pub height: u16,
}
impl Default for Header {
fn default() -> Self {
Self {
msg_type: MsgType::Video,
version: PROTOCOL_VERSION,
sender_id: 0,
frame_id: 0,
frag_index: 0,
frag_count: 0,
total_len: 0,
width: 0,
height: 0,
}
}
}
/// 解析帧头并校验魔数与版本。
///
/// 任何不满足的输入都返回 `None`，调用方据此当作"不是给我的"跳过，而不是当作
/// 错误——同一端口上可能出现板子的存活探测包，也可能日后出现其他协议。
pub fn decode_header(bytes: &[u8]) -> Option<Header> {
if bytes.len() < HEADER_LEN {
return None;
}
let magic = u32::from_le_bytes(bytes[0..4].try_into().ok()?);
if magic != MAGIC {
return None;
}
let msg_type = MsgType::from_u8(bytes[4])?;
let version = u16::from_le_bytes(bytes[5..7].try_into().ok()?);
if version != PROTOCOL_VERSION {
return None;
}
Some(Header {
msg_type,
version,
sender_id: u32::from_le_bytes(bytes[7..11].try_into().ok()?),
frame_id: u32::from_le_bytes(bytes[11..15].try_into().ok()?),
frag_index: u16::from_le_bytes(bytes[15..17].try_into().ok()?),
frag_count: u16::from_le_bytes(bytes[17..19].try_into().ok()?),
total_len: u32::from_le_bytes(bytes[19..23].try_into().ok()?),
width: u16::from_le_bytes(bytes[23..25].try_into().ok()?),
height: u16::from_le_bytes(bytes[25..27].try_into().ok()?),
})
}
/// 把帧头写进 `out`（追加，不清空）。
pub fn encode_header_into(header: &Header, out: &mut Vec<u8>) {
out.extend_from_slice(&MAGIC.to_le_bytes());
out.push(header.msg_type as u8);
out.extend_from_slice(&header.version.to_le_bytes());
out.extend_from_slice(&header.sender_id.to_le_bytes());
out.extend_from_slice(&header.frame_id.to_le_bytes());
out.extend_from_slice(&header.frag_index.to_le_bytes());
out.extend_from_slice(&header.frag_count.to_le_bytes());
out.extend_from_slice(&header.total_len.to_le_bytes());
out.extend_from_slice(&header.width.to_le_bytes());
out.extend_from_slice(&header.height.to_le_bytes());
debug_assert_eq!(out.len(), HEADER_LEN);
}
/// 裁判终端的固定标识。
///
/// 固定值而非 MAC 派生：裁判终端不是网络意义上的"设备集合"的一员，它只有一个，
/// 用常量比让它参与 MAC 派生更简单也更清楚。
pub const REFEREE_ID: u32 = 0;
/// 从 MAC 地址派生发送方标识。
///
/// 用固定长度输入的 FNV-1a：ESP32 没有随机数源可依赖，而 id 必须在重启后保持不变，
/// 否则客户端保存的绑定每次重启都会失效。
///
/// 返回 0 的输入（组播地址）被避开，因为 0 留给裁判终端。
pub fn id_from_mac(mac: &[u8; 6]) -> u32 {
const FNV_OFFSET: u32 = 0x811c9dc5;
const FNV_PRIME: u32 = 0x01000193;
let mut hash = FNV_OFFSET;
for byte in mac {
hash ^= *byte as u32;
hash = hash.wrapping_mul(FNV_PRIME);
}
// 避开 0，它是裁判终端的保留值。
if hash == 0 {
1
} else {
hash
}
}
/// 视频分片：一个数据报。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Fragment {
pub header: Header,
pub payload: Vec<u8>,
}
impl Fragment {
/// 追加写入 `out`，不分配。
///
/// 发送端对每帧的每个分片复用同一个缓冲：在 22fps、每帧 8–10 个分片下，这是
/// 每秒约 250 次分配与不分配的区别——在这块同时要抢内部 RAM 的芯片上很实在。
pub fn encode_into(&self, out: &mut Vec<u8>) {
encode_header_into(&self.header, out);
out.extend_from_slice(&self.payload);
}
pub fn encode(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(HEADER_LEN + self.payload.len());
self.encode_into(&mut out);
out
}
/// 解析一个数据报。
///
/// 头之后的所有字节都算载荷。校验不一致的分片在这里就被挡住，因此重组器不必
/// 再操心。
pub fn decode(bytes: &[u8]) -> Option<Self> {
let header = decode_header(bytes)?;
if header.msg_type != MsgType::Video {
return None;
}
// 一帧至少一个分片，且分片下标必须落在范围内。
if header.frag_count == 0 || header.frag_index >= header.frag_count {
return None;
}
let payload = &bytes[HEADER_LEN..];
// JPEG 不可能比承载它的分片还短。检查很便宜，却能在造成一次大额分配之前
// 挡掉损坏的头。
if payload.len() as u64 > header.total_len as u64 {
return None;
}
Some(Fragment {
header,
payload: payload.to_vec(),
})
}
}
/// 一张完整重组好的 JPEG。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Frame {
/// 来源板子的标识。
pub sender_id: u32,
pub frame_id: u32,
pub width: u16,
pub height: u16,
pub data: Vec<u8>,
}
/// 把一张 JPEG 切成数据报。
///
/// 分片数由载荷大小推出，因此调用方不可能产出与自身矛盾的集合。
pub fn fragment_frame(
sender_id: u32,
frame_id: u32,
jpeg: &[u8],
width: u16,
height: u16,
) -> Vec<Fragment> {
let count = jpeg.len().div_ceil(MAX_FRAGMENT_PAYLOAD).max(1) as u16;
let total_len = jpeg.len() as u32;
(0..count)
.map(|i| {
let start = i as usize * MAX_FRAGMENT_PAYLOAD;
let end = (start + MAX_FRAGMENT_PAYLOAD).min(jpeg.len());
Fragment {
header: Header {
msg_type: MsgType::Video,
version: PROTOCOL_VERSION,
sender_id,
frame_id,
frag_index: i,
frag_count: count,
total_len,
width,
height,
},
payload: jpeg[start..end].to_vec(),
}
})
.collect()
}
/// 客户端的订阅请求。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Subscribe {
pub role: SubscribeRole,
pub client_id: u32,
}
impl Subscribe {
pub fn encode(&self) -> Vec<u8> {
let header = Header {
msg_type: MsgType::Subscribe,
version: PROTOCOL_VERSION,
sender_id: self.client_id,
..Default::default()
};
let mut out = Vec::with_capacity(HEADER_LEN + 1);
encode_header_into(&header, &mut out);
out.push(self.role as u8);
out
}
pub fn decode(bytes: &[u8]) -> Option<Self> {
let header = decode_header(bytes)?;
if header.msg_type != MsgType::Subscribe || bytes.len() < HEADER_LEN + 1 {
return None;
}
Some(Subscribe {
role: SubscribeRole::from_u8(bytes[HEADER_LEN])?,
client_id: header.sender_id,
})
}
}
/// 板子的在线通告与心跳。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Hello {
pub sender_id: u32,
/// 固件版本，便于现场排查烧录了哪一版。
pub firmware: u16,
}
impl Hello {
pub fn encode(&self, msg_type: MsgType) -> Vec<u8> {
let header = Header {
msg_type,
version: PROTOCOL_VERSION,
sender_id: self.sender_id,
..Default::default()
};
let mut out = Vec::with_capacity(HEADER_LEN + 2);
encode_header_into(&header, &mut out);
out.extend_from_slice(&self.firmware.to_le_bytes());
out
}
pub fn decode(bytes: &[u8]) -> Option<Self> {
let header = decode_header(bytes)?;
if !matches!(header.msg_type, MsgType::Hello | MsgType::Heartbeat) {
return None;
}
// 固件版本是可选的：早期或精简的通告可能不带这个字段。
let firmware = if bytes.len() >= HEADER_LEN + 2 {
u16::from_le_bytes(bytes[HEADER_LEN..HEADER_LEN + 2].try_into().ok()?)
} else {
0
};
Some(Hello {
sender_id: header.sender_id,
firmware,
})
}
}
// ---------------------------------------------------------------------------
// 装甲板（预留）
// ---------------------------------------------------------------------------
/// 一块装甲板的上报。
///
/// 现在没有实现也没有数据源，但格式在这里定下来，并且 [`MsgType::ArmorHealth`]
/// 与 [`ARMOR_PORT`] 已经就位：板子接上以后向 `ARMOR_PORT` 发这个结构，裁判终端
/// 监听即可，界面与其它代码都不用改。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ArmorHealth {
/// 哪一边：0 = 队伍一，1 = 队伍二。
pub team: u8,
/// 机器人在该队伍的装甲板序号，0..3。
pub index: u8,
/// 当前血量。
pub hp: u16,
}
impl ArmorHealth {
pub fn encode(&self) -> Vec<u8> {
let header = Header {
msg_type: MsgType::ArmorHealth,
version: PROTOCOL_VERSION,
..Default::default()
};
let mut out = Vec::with_capacity(HEADER_LEN + 4);
encode_header_into(&header, &mut out);
out.push(self.team);
out.push(self.index);
out.extend_from_slice(&self.hp.to_le_bytes());
out
}
pub fn decode(bytes: &[u8]) -> Option<Self> {
let header = decode_header(bytes)?;
if header.msg_type != MsgType::ArmorHealth || bytes.len() < HEADER_LEN + 4 {
return None;
}
Some(ArmorHealth {
team: bytes[HEADER_LEN],
index: bytes[HEADER_LEN + 1],
hp: u16::from_le_bytes(bytes[HEADER_LEN + 2..HEADER_LEN + 4].try_into().ok()?),
})
}
}
/// 视频端口地址。
pub fn video_endpoint(ip: Ipv4Addr) -> SocketAddrV4 {
SocketAddrV4::new(ip, DEFAULT_PORT)
}
/// 255.255.255.255，用于广播发现。
pub const BROADCAST_ADDR: Ipv4Addr = Ipv4Addr::new(255, 255, 255, 255);
// ---------------------------------------------------------------------------
// 重组
// ---------------------------------------------------------------------------
/// 帧被丢弃的原因。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DropReason {
/// 新的帧开始了，而这一张还没收齐。
Superseded,
/// 发送方沉默过久。
TimedOut,
}
/// 从乱序、有丢失的 UDP 数据报里重组帧。
///
/// 策略就是全部要点：**一张帧只有所有分片都到齐才交付，且最新的帧永远优先。**
///
/// UDP 不保证跨数据报的顺序与到达，所以试图"补洞"的接收端做的功比"等下一张完整
/// 帧"更多——而且补的过程中会显示撕裂的画面。
///
/// 超时堵住另一种失败：发送方在帧中途死掉时，半截缓冲会永远留在这里。
pub struct FrameAssembler {
sender_id: u32,
current_id: Option<u32>,
expected: usize,
buffer: Vec<u8>,
width: u16,
height: u16,
since: Option<std::time::Instant>,
timeout: std::time::Duration,
/// 已收到的分片数（不含重复）。用来统计网络丢了多少。
seen: usize,
/// 每个分片是否已收到。
///
/// 存在的原因是 `seen` 会被重复包灌水：UDP 会重复投递，一个丢失的分片可以
/// 被另一个包的副本凑数，于是"收够了"却留着空洞，交出去一张尾部是 0 的坏图。
have: Vec<bool>,
}
impl FrameAssembler {
/// 新建一个重组器，`timeout` 后放弃未收齐的帧。
pub fn new(timeout: std::time::Duration) -> Self {
Self {
sender_id: 0,
current_id: None,
expected: 0,
buffer: Vec::new(),
width: 0,
height: 0,
since: None,
timeout,
seen: 0,
have: Vec::new(),
}
}
/// 这个重组器正在处理哪个发送方。
pub fn sender_id(&self) -> u32 {
self.sender_id
}
/// 收到一个分片。
///
/// 这张分片补齐一张帧时返回 [`Frame`]。
///
/// `Err(Superseded)` 表示这个分片属于一张**已被放弃**的帧。多路复用时（裁判
/// 终端同时收两块板子）必须每个发送方一个重组器——混在一起会让两块板子的分片
/// 互相顶掉。这里不做检查，是调用方的责任。
pub fn push(&mut self, frag: Fragment) -> Result<Frame, DropReason> {
let sender_id = frag.header.sender_id;
let frame_id = frag.header.frame_id;
// 换了发送方：上一张属于别人的帧就此作废。
if self.sender_id != 0 && self.sender_id != sender_id {
self.reset();
}
self.sender_id = sender_id;
let superseded = match self.current_id {
None => false,
Some(id) => id != frame_id,
};
if superseded {
self.reset();
}
match self.current_id {
None => {
self.current_id = Some(frame_id);
self.expected = frag.header.frag_count as usize;
self.buffer.clear();
// 按 total_len 预留，最后拼接时不再重新分配。
self.buffer.reserve(frag.header.total_len as usize);
self.width = frag.header.width;
self.height = frag.header.height;
self.since = Some(std::time::Instant::now());
self.seen = 0;
self.have.clear();
self.have.resize(self.expected, false);
}
Some(id) if id == frame_id => {}
_ => unreachable!("current_id was just reconciled with frame_id"),
}
let index = frag.header.frag_index as usize;
// 重复分片直接跳过，且**不**增加计数。
//
// 用 `seen` 一个计数判断收齐是不够健壮的：一个丢失的分片若能被某个分片的
// 副本凑数，计数就达标了，而缓冲区里留着空洞，交出去一张尾部是 0 的 JPEG。
// 位图让"收到过"这件事是显式的，不用靠计数的算术去推断。
//
// 要说明的是：在本项目实测的链路上（2.4GHz AP，一到两个订阅者，20 秒 2000+
// 个分片）**一次重复投递都没出现过**，所以这层防护是为了正确性而不是为了
// 修一个正在发生的故障。之前观察到的 34 张坏 JPEG 另有原因：探测脚本自己
// 的重组状态只按 sender_id 存、不按 frame_id 存，于是丢片之后下一帧的分片
// 会接着凑上一帧的计数。那个脚本已修，见 scripts/udp_probe.py。
if self.have[index] {
return Err(DropReason::Superseded);
}
self.have[index] = true;
let start = index * MAX_FRAGMENT_PAYLOAD;
let end = start + frag.payload.len();
if end > self.buffer.len() {
self.buffer.resize(end, 0);
}
self.buffer[start..end].copy_from_slice(&frag.payload);
self.seen += 1;
// 计数达标还不够：最后一个分片必须真的到过，否则中间可能缺一块。
// 显式检查尾块，比依赖计数的算术更直接，也更容易看出意图。
let tail_index =
(frag.header.total_len as usize).saturating_sub(1) / MAX_FRAGMENT_PAYLOAD;
if self.seen >= self.expected && self.have[tail_index] {
let total = frag.header.total_len as usize;
self.buffer.truncate(total.min(self.buffer.len()));
let frame = Frame {
sender_id,
frame_id,
width: self.width,
height: self.height,
data: std::mem::take(&mut self.buffer),
};
self.reset();
return Ok(frame);
}
Err(DropReason::Superseded)
}
/// 若未收齐的帧等待过久则丢弃它。
///
/// 需要由定时器调用。否则发送方在首末分片之间死掉时，重组器会一直留着那张帧。
pub fn expire(&mut self) -> Option<DropReason> {
match self.since {
Some(started) if started.elapsed() > self.timeout => {
self.reset();
Some(DropReason::TimedOut)
}
_ => None,
}
}
/// 是否有一张帧正在收。
pub fn is_in_progress(&self) -> bool {
self.current_id.is_some()
}
fn reset(&mut self) {
self.current_id = None;
self.expected = 0;
self.buffer = Vec::new();
self.buffer.shrink_to_fit();
self.since = None;
self.seen = 0;
self.have = Vec::new();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn header_roundtrips() {
let header = Header {
msg_type: MsgType::Video,
version: PROTOCOL_VERSION,
sender_id: 0xDEAD_BEEF,
frame_id: 12345,
frag_index: 3,
frag_count: 9,
total_len: 20_000,
width: 640,
height: 480,
};
let mut buf = Vec::new();
encode_header_into(&header, &mut buf);
assert_eq!(buf.len(), HEADER_LEN);
assert_eq!(decode_header(&buf), Some(header));
}
#[test]
fn fragment_roundtrips_and_is_multiple_sources() {
let jpeg: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
let frags = fragment_frame(0xAAAA_BBBB, 7, &jpeg, 640, 480);
assert_eq!(frags.len(), 4);
for f in &frags {
assert_eq!(f.header.sender_id, 0xAAAA_BBBB);
let decoded = Fragment::decode(&f.encode()).expect("decodes");
assert_eq!(&decoded, f);
}
}
#[test]
fn reassembles_and_separates_senders() {
let jpeg: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
// 模拟裁判终端：两个发送方各自的重组器。
let mut a = FrameAssembler::new(std::time::Duration::from_secs(2));
let mut b = FrameAssembler::new(std::time::Duration::from_secs(2));
let mut frame_a = None;
let mut frame_b = None;
for f in fragment_frame(1, 1, &jpeg, 640, 480) {
if let Ok(frame) = a.push(f) {
frame_a = Some(frame);
}
}
for f in fragment_frame(2, 1, &jpeg, 320, 240) {
if let Ok(frame) = b.push(f) {
frame_b = Some(frame);
}
}
let fa = frame_a.expect("sender 1 completed");
let fb = frame_b.expect("sender 2 completed");
assert_eq!(fa.sender_id, 1);
assert_eq!(fb.sender_id, 2);
assert_eq!(fa.data, jpeg);
assert_eq!(fb.data, jpeg);
assert_eq!((fa.width, fa.height), (640, 480));
assert_eq!((fb.width, fb.height), (320, 240));
}
#[test]
fn a_gap_means_no_frame_until_the_next_one() {
let jpeg = vec![0xAAu8; 5000];
let frags = fragment_frame(1, 1, &jpeg, 640, 480);
let mut asm = FrameAssembler::new(std::time::Duration::from_secs(2));
// 丢掉第 2 个分片。
for f in frags.iter().filter(|f| f.header.frag_index != 2) {
assert!(asm.push(f.clone()).is_err());
}
assert!(asm.is_in_progress(), "still waiting on the missing fragment");
// 新的帧取代没收齐的那张，而不是去补洞。
let next = fragment_frame(1, 2, &jpeg, 640, 480);
assert!(asm.push(next[0].clone()).is_err());
let mut got = None;
for f in next.iter().skip(1) {
if let Ok(frame) = asm.push(f.clone()) {
got = Some(frame);
}
}
assert_eq!(got.expect("second frame completes").frame_id, 2);
}
#[test]
fn expires_a_stalled_frame() {
let jpeg = vec![0x11u8; 3000];
let frags = fragment_frame(9, 1, &jpeg, 640, 480);
let mut asm = FrameAssembler::new(std::time::Duration::from_millis(1));
let _ = asm.push(frags[0].clone());
assert!(asm.is_in_progress());
std::thread::sleep(std::time::Duration::from_millis(5));
assert_eq!(asm.expire(), Some(DropReason::TimedOut));
assert!(!asm.is_in_progress());
}
#[test]
fn rejects_foreign_and_malformed_datagrams() {
assert!(decode_header(b"").is_none());
assert!(decode_header(&[0u8; 8]).is_none());
// 长度够但魔数不对。
assert!(decode_header(&[0u8; HEADER_LEN]).is_none());
// 正确的魔数但版本不对：宁可拒绝，也不能按错误布局解析。
let header = Header {
sender_id: 1,
frag_count: 1,
total_len: 10,
..Default::default()
};
let mut buf = Vec::new();
encode_header_into(&header, &mut buf);
buf[5] = 99; // version 低字节
assert!(decode_header(&buf).is_none(), "wrong version must be refused");
let jpeg = vec![0u8; 100];
// frag_index >= frag_count 自相矛盾。
let mut bad = fragment_frame(1, 1, &jpeg, 64, 48).remove(0).encode();
bad[15..17].copy_from_slice(&9u16.to_le_bytes());
assert!(Fragment::decode(&bad).is_none());
// 载荷比声明的总长还大：损坏。
let mut over = fragment_frame(1, 1, &jpeg, 64, 48).remove(0).encode();
over[19..23].copy_from_slice(&4u32.to_le_bytes());
assert!(Fragment::decode(&over).is_none());
}
#[test]
fn subscribe_and_hello_roundtrip() {
let sub = Subscribe {
role: SubscribeRole::Referee,
client_id: 0x1234_5678,
};
assert_eq!(Subscribe::decode(&sub.encode()), Some(sub));
let hello = Hello {
sender_id: 0x8765_4321,
firmware: 1,
};
assert_eq!(Hello::decode(&hello.encode(MsgType::Hello)), Some(hello));
assert_eq!(
Hello::decode(&hello.encode(MsgType::Heartbeat)),
Some(hello)
);
}
#[test]
fn armor_health_roundtrips() {
// 现在没有实现，但格式必须先定下来并且可往返，否则接上板子时才发现布局
// 不兼容就晚了。
let hp = ArmorHealth {
team: 1,
index: 2,
hp: 420,
};
let encoded = hp.encode();
assert_eq!(ArmorHealth::decode(&encoded), Some(hp));
}
#[test]
fn ids_are_stable_and_never_zero() {
let mac = [0x90, 0x70, 0x69, 0xF8, 0xF2, 0x14];
let a = id_from_mac(&mac);
let b = id_from_mac(&mac);
assert_eq!(a, b, "the same MAC must always give the same id");
assert_ne!(a, 0, "0 is reserved for the referee terminal");
// 不同 MAC 不应撞车。
assert_ne!(a, id_from_mac(&[1, 2, 3, 4, 5, 6]));
}
#[test]
fn a_duplicate_cannot_complete_a_frame_with_a_hole() {
// UDP 会重复投递。一个丢失的分片若能被另一片的副本凑数，接收端就会以为
// 收齐了，交出一张中间是 0 的坏 JPEG——而且帧率正常、没有任何报错线索。
let jpeg: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
let frags = fragment_frame(1, 1, &jpeg, 640, 480);
let mut asm = FrameAssembler::new(std::time::Duration::from_secs(2));
// 丢掉第 2 片，然后把第 0 片连送三次来灌 seen。
for f in frags.iter().filter(|f| f.header.frag_index != 2) {
let _ = asm.push(f.clone());
}
for _ in 0..3 {
assert!(
asm.push(frags[0].clone()).is_err(),
"a duplicate must not be allowed to complete the frame"
);
}
assert!(asm.is_in_progress(), "the hole must keep it incomplete");
// 补上真正的第 2 片，才在这一片上收齐——所以结果要从这一次的返回值拿。
let completed = asm.push(frags[2].clone());
assert_eq!(
completed.expect("completes once the hole is filled").data,
jpeg,
);
}
#[test]
fn a_gap_that_is_never_filled_never_yields_a_frame() {
let jpeg = vec![0xAAu8; 5000];
let frags = fragment_frame(1, 1, &jpeg, 640, 480);
let mut asm = FrameAssembler::new(std::time::Duration::from_secs(2));
// 反复送同一片，永远凑不齐——接收端必须一直等着，而不是交出半张图。
for _ in 0..50 {
assert!(asm.push(frags[0].clone()).is_err());
}
assert!(asm.is_in_progress(), "50 duplicates must not complete a frame");
assert_eq!(asm.sender_id(), 1, "and it must still be the same camera");
}
#[test]
fn empty_jpeg_still_produces_one_fragment() {
let frags = fragment_frame(1, 1, &[], 64, 48);
assert_eq!(frags.len(), 1);
assert!(frags[0].payload.is_empty());
}
#[test]
fn datagrams_stay_under_a_typical_mtu() {
let jpeg = vec![0u8; 100_000];
for f in fragment_frame(1, 1, &jpeg, 800, 600) {
assert!(f.encode().len() <= HEADER_LEN + MAX_FRAGMENT_PAYLOAD);
assert!(f.encode().len() < 1500, "must fit an Ethernet MTU");
}
}
}
Reference in New Issue View Git Blame Copy Permalink