docs(webrtc): 中文注释 str0m WebRTC 信令服务器

This commit is contained in:
dailz
2026-06-22 17:51:40 +08:00
parent d94431bd1e
commit 2f8197210e
+277
View File
@@ -1,3 +1,47 @@
//! # WebRTC 传输模块 — str0m Sans-IO 信令服务器与媒体出口
//!
//! ## 模块定位
//! 将 H.264 编码帧通过 WebRTC 推送到浏览器(替代文件输出)。仅在 `--port > 0` 时启用;
//! `--port 0`(默认)走纯文件输出路径,本模块不会被实例化(见 `main.rs` 入口判断)。
//!
//! ## str0m 是 Sans-IO WebRTC 库
//! 类比 Go 的 `net/http`,但 Sans-IO 哲学不同:
//! - **没有 background goroutine**str0m 不创建任何线程,所有进度都靠外部 poll 推动
//! - **手动驱动 3 步循环**(见 `poll_and_feed`/`feed_network`/`poll_rtc`):
//! 1. 读 UDP 包 → `Rtc::handle_input(Input::Receive(...))` 喂给 str0m
//! 2. 调 `Rtc::poll_output()` 拿 `Output::Transmit` 包 → 写回 UDP socket
//! 3. 定时喂 `Input::Timeout(Instant::now())` 推动内部时钟
//! - **同步而非 async**str0m 不是 async/await 库(与 `tokio::net::TcpListener` 等
//! 异步运行时无关);本文件用 `std::net::TcpListener` + `UdpSocket`(手动
//! `set_nonblocking(true)`),完全同步代码;上层 `main.rs` 在 mio 事件循环里
//! 周期性调 `poll_and_feed()` 推动 RTC 状态机
//! - **Go 等价物**`github.com/pion/webrtc`Go 主流 WebRTC 库)也是同步 + 手动驱动,
//! 但 str0m 把 Sans-IO 推得更彻底——连 UDP socket 都不持有,所有 I/O 都由调用方管理
//!
//! ## 内嵌 HTTP 信令服务器
//! 本模块自带一个极简 HTTP 服务器(`std::net::TcpListener`,非 tokio/axum),3 个端点:
//! - `GET /` → 返回 `HTML_PAGE`(自带 SDP 协商 + `<video>` 播放 + 实时 stats 的测试页)
//! - `POST /sdp`Content-Type: application/json)→ 接收浏览器 `RTCPeerConnection`
//! localDescriptionOffer SDP),交给 `Rtc::sdp_api().accept_offer()` 生成 Answer
//! 返回 JSON body 给浏览器 `setRemoteDescription`
//! - `GET /sdp`(无 JSON Content-Type)→ 与 `GET /` 同(兼容旧路径)
//!
//! ICE candidate 通过 SDP offer/answer 完成:浏览器等 `iceGatheringState == 'complete'`
//! 才 POST(见 `HTML_PAGE` 的 `onicegatheringstatechange`),所以 candidate 已全在
//! SDP 里,本服务端无需单独的 ICE endpointtrickle ICE 关闭)。
//!
//! ## 关键不变量
//! - **单连接**`WebRtcState::inner: Option<WebRtcInner>` 只持有 1 个 peer;新连接
//! POST 进来时,旧 `inner` 被 drop(旧 `Rtc` 析构,UDP socket 关闭)
//! - **非阻塞 IO**:所有 socket `set_nonblocking(true)``WouldBlock` 是常态而非错误
//! - **BWE 启动**`RtcConfig::enable_bwe(Some(Bitrate::mbps(5)))` 启用带宽估计,
//! 用于动态分辨率切换(见 `state_portal.rs::select_resolution`
//!
//! ## 引用
//! - `Cargo.toml`: `str0m = "0.20"`
//! - git `727893f`: bitrate 修复(BWE 与 VBV 协同)
//! - issue #23: PLI 节流(`FORCED_KEYFRAME_MIN_INTERVAL`
// WebRTC 传输模块 — 使用 str0m (Sans-IO) 将 H.264 编码帧推送到浏览器 // WebRTC 传输模块 — 使用 str0m (Sans-IO) 将 H.264 编码帧推送到浏览器
use std::io::{Read, Write}; use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, UdpSocket}; use std::net::{SocketAddr, TcpListener, UdpSocket};
@@ -192,28 +236,60 @@ connect();
// ── WebRTC 状态 ─────────────────────────────────────────────────────────── // ── WebRTC 状态 ───────────────────────────────────────────────────────────
// 对外门面:持有 HTTP 信令监听器 + 当前唯一的 peer 连接(`inner`)。
// 类比 Go 的 `*http.Server`,但 Sans-IO:所有推进都靠调用方主动 poll。
pub struct WebRtcState { pub struct WebRtcState {
// HTTP 信令监听器(`POST /sdp` 协商;`GET /` 测试页面)。`set_nonblocking(true)`
// 由上层 mio 事件循环可读时调 `handle_signaling()` 接受连接。
signal_listener: TcpListener, signal_listener: TcpListener,
// 当前 peer。`None` = 尚无连接 / 上次连接已断开。新 `POST /sdp` 会整体替换此字段,
// 旧 `Rtc` 实例被 dropUDP socket 随之关闭)。
inner: Option<WebRtcInner>, inner: Option<WebRtcInner>,
// 上层期望的帧率(来自 CLI `--fps`),用于初始化 `WebRtcInner`。
fps: u32, fps: u32,
} }
// 单个 WebRTC peer 的全部状态:str0m `Rtc` 实例 + 它专用的 UDP socket +
// 编解码参数协商结果 + 关键帧请求/BWE 估计的运行时缓存。
//
// 字段访问路径(每帧一次,由 `main.rs` 的事件循环驱动):
// 1. `feed_network()` 把 UDP 入包喂给 `Rtc::handle_input`
// 2. `poll_rtc()` 取出 `Rtc::poll_output` 的 `Transmit` 包写回 UDP,并处理 `Event`
// 3. `write_h264_frame()` 把编码后的 H.264 NALU 通过 `Rtc::writer(mid).write(...)` 发出
struct WebRtcInner { struct WebRtcInner {
// str0m `Rtc`:一个完整的 WebRTC peer connectionICE / DTLS / SRTP / RTP / RTCP)。
// Sans-IO:不持有任何 socket 或线程,只持有协议状态机。
rtc: Rtc, rtc: Rtc,
// 本 peer 专用的 UDP socket(每连接一个,避免与不存在的其他 peer 串扰)。
socket: UdpSocket, socket: UdpSocket,
// 该 socket 绑定的本地地址(带随机端口),用作 `Candidate::host` 的发地址。
udp_addr: SocketAddr, udp_addr: SocketAddr,
// 视频 Media IDSDP 协商后从 `Event::MediaAdded` 捕获)。`None` = 尚未协商到。
video_mid: Option<Mid>, video_mid: Option<Mid>,
// H.264 payload type(从 `Rtc::writer(mid).payload_params()` 扫描得到)。
video_pt: Option<Pt>, video_pt: Option<Pt>,
// ICE+DTLS 是否已完成(`Event::Connected`)。未连接时 `write_h264_frame` 静默丢弃。
connected: bool, connected: bool,
// 等待下一个 IDR 关键帧(连接建立/分辨率切换时置 true,写帧时若非 IDR 则丢帧)。
need_keyframe: bool, need_keyframe: bool,
// 通知上游编码器下一次输出 IDR`state.rs::State::take_force_keyframe` 拉取)。
force_keyframe_to_encode: bool, force_keyframe_to_encode: bool,
// 最近一次强制关键帧时刻,用于 `FORCED_KEYFRAME_MIN_INTERVAL` 节流(防 PLI 风暴)。
last_forced_keyframe_at: Option<Instant>, last_forced_keyframe_at: Option<Instant>,
// 最近一次 BWE 估计(来自 `Event::EgressBitrateEstimate`),用于上层动态分辨率选择。
current_bwe_estimate: Option<Bitrate>, current_bwe_estimate: Option<Bitrate>,
// 最近一次写入的 RTP 时间戳(90kHz),仅用于日志 trace,不参与协议正确性。
rtp_clock: u32, rtp_clock: u32,
// UDP 接收缓冲(重复利用以避免每包分配;65535 = max UDP payload)。
buf: Vec<u8>, buf: Vec<u8>,
} }
impl WebRtcState { impl WebRtcState {
// 构造函数:绑定 HTTP 信令 TCP 监听器并设为非阻塞。`port` 来自 CLI `--port`
// `fps` 来自 CLI `--fps`,仅在 `--port > 0` 时被 `main.rs` 调用。
//
// 注意:本函数只创建信令监听器,**不**创建 UDP socket 或 `Rtc` 实例——
// 那些在第一次 `POST /sdp` 时由 `WebRtcInner::new` 按需创建。
pub fn new(port: u16, fps: u32) -> Result<Self> { pub fn new(port: u16, fps: u32) -> Result<Self> {
let signal_listener = TcpListener::bind(format!("0.0.0.0:{port}"))?; let signal_listener = TcpListener::bind(format!("0.0.0.0:{port}"))?;
signal_listener.set_nonblocking(true)?; signal_listener.set_nonblocking(true)?;
@@ -226,18 +302,36 @@ impl WebRtcState {
}) })
} }
// 处理所有待接受的 HTTP 信令连接。上层 mio 循环在 `signal_listener` 可读时调用。
//
// 返回 `Ok(true)` 表示至少处理了一个请求(用于上层日志/计数)。
// 单次调用 drain 当前 accept 队列里所有连接(`Err(WouldBlock)` 时退出循环)。
//
// 路由:
// - `GET /` 或 `GET /sdp`(非 JSON)→ 返回 `HTML_PAGE`
// - `POST /sdp` → 解析 body,构造新 `WebRtcInner` 并替换 `self.inner`
// - 其他路径 → 404
pub fn handle_signaling(&mut self) -> Result<bool> { pub fn handle_signaling(&mut self) -> Result<bool> {
let mut handled = false; let mut handled = false;
loop { loop {
// `TcpListener::accept()` 类比 Go `ln.Accept()`;非阻塞模式下队列为空返回
// `WouldBlock`,是 drain 完成的信号而非错误(类比 Go `accept` + nonblocking + EAGAIN)。
let (mut stream, _addr) = match self.signal_listener.accept() { let (mut stream, _addr) = match self.signal_listener.accept() {
Ok(s) => s, Ok(s) => s,
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break, Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
// `bail!` 是 anyhow 提供的宏,等价于 `return Err(anyhow::anyhow!(...))`
// 类比 Go `return fmt.Errorf("TCP accept error: %w", err)`。
Err(e) => bail!("TCP accept error: {e}"), Err(e) => bail!("TCP accept error: {e}"),
}; };
handled = true; handled = true;
// 设为非阻塞——类比 Go `syscall.SetNonblock(fd, true)`。后续 `stream.read`
// 在没数据时返回 `WouldBlock`(用 `continue` 跳过本连接)。
stream.set_nonblocking(true)?; stream.set_nonblocking(true)?;
// 64KB 一次性读完:HTTP/1.0 客户端默认 `Connection: close`,浏览器 POST 整个
// SDP offer 不会超过 64KB。`vec![0u8; N]` 类比 Go `make([]byte, N)`。
let mut req = vec![0u8; 65536]; let mut req = vec![0u8; 65536];
// `stream.read(&mut req)` 类比 Go `conn.Read(buf)`——`Read` trait 即 Go `io.Reader`。
let n = match stream.read(&mut req) { let n = match stream.read(&mut req) {
Ok(n) => n, Ok(n) => n,
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => continue, Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => continue,
@@ -246,6 +340,8 @@ impl WebRtcState {
continue; continue;
} }
}; };
// `String::from_utf8_lossy` 把字节转成字符串,无效 UTF-8 替换为 U+FFFDHTTP 头都是 ASCII)。
// 类比 Go `string(buf[:n])`Go 字符串可包含任意字节,但后续 `starts_with` 也只看 ASCII)。
let req_str = String::from_utf8_lossy(&req[..n]); let req_str = String::from_utf8_lossy(&req[..n]);
if req_str.starts_with("GET / ") if req_str.starts_with("GET / ")
@@ -270,12 +366,19 @@ impl WebRtcState {
continue; continue;
} }
// `and_then`Result 链式组合,类比 Go `if err != nil { return err }` 后继续。
// `new_inner.handle_sdp_offer(...)?``?` 操作符传播 `Result::Err`
// 类比 Go `result, err := ...; if err != nil { return err }` 的简写。
match WebRtcInner::new(self.fps).and_then(|mut new_inner| { match WebRtcInner::new(self.fps).and_then(|mut new_inner| {
let answer_json = new_inner.handle_sdp_offer(body.as_bytes())?; let answer_json = new_inner.handle_sdp_offer(body.as_bytes())?;
Ok((new_inner, answer_json)) Ok((new_inner, answer_json))
}) { }) {
Ok((new_inner, answer_json)) => { Ok((new_inner, answer_json)) => {
// `Option::is_some()` = Rust 检查 `Option` 是否为 `Some(_)`
// 类比 Go `if p != nil`。这里用于日志区分"替换"vs"首次"。
let replacing = self.inner.is_some(); let replacing = self.inner.is_some();
// 整体替换 `self.inner`:旧 `Rtc` 实例 dropUDP socket 关闭,
// peer 连接断开)。这是单连接不变量的核心实现。
self.inner = Some(new_inner); self.inner = Some(new_inner);
if replacing { if replacing {
tracing::info!("Replaced WebRTC connection (old dropped)"); tracing::info!("Replaced WebRTC connection (old dropped)");
@@ -311,6 +414,10 @@ impl WebRtcState {
Ok(handled) Ok(handled)
} }
// 推动 str0m `Rtc` 状态机:取出 `poll_output` 的 `Transmit` 包写回 UDP,处理 `Event`。
// 返回 `Ok(())`;若 `poll_rtc` 上报 peer 已断开,则清空 `self.inner`。
//
// 类比 Go pion/webrtc:没有 `go func()` 自动循环,必须由 main 线程显式调用。
pub fn poll_rtc(&mut self) -> Result<()> { pub fn poll_rtc(&mut self) -> Result<()> {
if let Some(inner) = self.inner.as_mut() { if let Some(inner) = self.inner.as_mut() {
if inner.poll_rtc()? { if inner.poll_rtc()? {
@@ -321,6 +428,8 @@ impl WebRtcState {
Ok(()) Ok(())
} }
// 从 UDP socket 读所有待处理包喂给 `Rtc::handle_input`。`WouldBlock` 退出循环。
// Go 类比:`for { n, _ := conn.ReadFrom(buf); if errors.Is(err, EAGAIN) { break } }`。
pub fn feed_network(&mut self) -> Result<()> { pub fn feed_network(&mut self) -> Result<()> {
if let Some(inner) = self.inner.as_mut() { if let Some(inner) = self.inner.as_mut() {
inner.feed_network()?; inner.feed_network()?;
@@ -328,12 +437,20 @@ impl WebRtcState {
Ok(()) Ok(())
} }
// `poll_rtc` → `feed_network` → `poll_rtc` 三明治。中间多一次 poll 是因为
// `feed_network` 喂的入包可能触发 str0m 产生新的 `Transmit`(如 RTCP ACK),
// 这些出包必须在同一轮循环里写回 UDP,避免延迟一帧。
pub fn poll_and_feed(&mut self) -> Result<()> { pub fn poll_and_feed(&mut self) -> Result<()> {
self.poll_rtc()?; self.poll_rtc()?;
self.feed_network()?; self.feed_network()?;
self.poll_rtc() self.poll_rtc()
} }
// 把一帧 H.264 NALU(已 annex-B 转码)写入 str0m `Rtc`,通过 RTP 发给 peer。
// `pts_ticks` = 90kHz 时钟下的 PTS(编码器 time_base = 1/90000,等同 RTP 时间戳)。
//
// 返回 `Ok(())`;若 `WebRtcInner::write_h264_frame` 上报 peer 断开,则清空 `self.inner`。
// 未连接 / 未协商到 mid/pt / 等待 IDR 时静默丢帧(`Ok(false)`)。
pub fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64) -> Result<()> { pub fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64) -> Result<()> {
let should_destroy = if let Some(inner) = self.inner.as_mut() { let should_destroy = if let Some(inner) = self.inner.as_mut() {
inner.write_h264_frame(data, pts_ticks)? inner.write_h264_frame(data, pts_ticks)?
@@ -347,10 +464,15 @@ impl WebRtcState {
Ok(()) Ok(())
} }
// 是否有已连接的 peer。`Option::is_some_and` = Rust 短路求值,类比 Go
// `if p != nil && p.connected { ... }`。
pub fn is_connected(&self) -> bool { pub fn is_connected(&self) -> bool {
self.inner.as_ref().is_some_and(WebRtcInner::is_connected) self.inner.as_ref().is_some_and(WebRtcInner::is_connected)
} }
// 上层(`state_portal.rs::select_resolution`)查询最近一次 BWE 估计(bps)。
// `None` = 尚未收到 `Event::EgressBitrateEstimate``Some(bps)` = str0m 推断的可用带宽。
// 上层据此切换分辨率 tier(防止过载导致卡顿)。
/// Returns the latest bandwidth estimation estimate in bits per second, if available. /// Returns the latest bandwidth estimation estimate in bits per second, if available.
pub fn get_bwe_estimate(&self) -> Option<u64> { pub fn get_bwe_estimate(&self) -> Option<u64> {
self.inner self.inner
@@ -358,6 +480,9 @@ impl WebRtcState {
.and_then(|inner| inner.current_bwe_estimate.map(|b| b.as_u64())) .and_then(|inner| inner.current_bwe_estimate.map(|b| b.as_u64()))
} }
// 内部触发:连接刚建立或分辨率刚变化,需要立刻 IDR 以让对端解码器重置。
// 不受 `FORCED_KEYFRAME_MIN_INTERVAL` 节流(本函数总是 honor),但会刷新
// `last_forced_keyframe_at`,使紧接着的 1 秒内 viewer PLI 被丢弃。
/// Internal keyframe request (connect, resolution change). Always honored, /// Internal keyframe request (connect, resolution change). Always honored,
/// but updates last_forced_keyframe_at so a subsequent viewer PLI in the next /// but updates last_forced_keyframe_at so a subsequent viewer PLI in the next
/// second is throttled. /// second is throttled.
@@ -367,6 +492,9 @@ impl WebRtcState {
} }
} }
// 外部触发:viewer 通过 RTCP PLI/FIR 主动请求关键帧(`Event::KeyframeRequest`)。
// 受 `FORCED_KEYFRAME_MIN_INTERVAL` 节流(1 秒),防止恶意/频繁 PLI 触发 IDR 风暴
// 撑爆上行带宽。See issue #23。
/// External keyframe request from viewer (PLI/FIR via str0m /// External keyframe request from viewer (PLI/FIR via str0m
/// `Event::KeyframeRequest`). Rate-limited to FORCED_KEYFRAME_MIN_INTERVAL /// `Event::KeyframeRequest`). Rate-limited to FORCED_KEYFRAME_MIN_INTERVAL
/// to prevent PLI storms from swamping the network with IDR bursts. /// to prevent PLI storms from swamping the network with IDR bursts.
@@ -378,6 +506,8 @@ impl WebRtcState {
} }
} }
// 上层拉取"是否需要下一帧为 IDR"。返回 `true` 仅一次(取后自动复位),
// 类比 Go `atomic.SwapInt32(&flag, 0)`。编码线程据此在下一帧 `force_idr=1`。
pub fn take_force_keyframe(&mut self) -> bool { pub fn take_force_keyframe(&mut self) -> bool {
if let Some(inner) = self.inner.as_mut() { if let Some(inner) = self.inner.as_mut() {
let v = inner.force_keyframe_to_encode; let v = inner.force_keyframe_to_encode;
@@ -390,19 +520,46 @@ impl WebRtcState {
} }
impl WebRtcInner { impl WebRtcInner {
// 构造一个全新的 WebRTC peer:创建 str0m `Rtc` 实例 + UDP socket + 候选地址。
// 在 `handle_signaling` 接到 `POST /sdp` 时被调用——也就是说**每来一个 SDP offer
// 都新建一个 peer**,旧 `Rtc` 实例随之 dropUDP socket 关闭,连接断开)。
//
// 步骤:
// 1. `RtcConfig::new().enable_bwe(...).build(...)`str0m 构造器链式 Builder 模式,
// 类比 Go `webrtc.NewAPI(webrtc.WithSettingEngine(...))`;启用 BWE5 Mbps 初始)
// 2. `UdpSocket::bind("0.0.0.0:0")`OS 随机分配端口,类比 Go `net.ListenUDP("udp", nil)`
// 3. `unsafe { libc::setsockopt(SO_SNDBUF) }`:扩大 UDP 发送缓冲到 2MB(默认 ~208KB
// 在 IDR 突发下会 EAGAIN 丢包);英文 SAFETY 注释见下方
// 4. `Candidate::host(addr, "udp")`:构造 host ICE candidate(局域网用),
// `Rtc::add_local_candidate` 注册到 str0m
fn new(fps: u32) -> Result<Self> { fn new(fps: u32) -> Result<Self> {
// `let _ = fps;` 显式标记 fps 暂未使用(保留接口给未来 fps-based pacing)。
// 类比 Go `_ = fps`。
let _ = fps; let _ = fps;
// str0m `Rtc` 构造:Builder 模式 + 链式 setter。
// - `RtcConfig::new()`:空配置
// - `.enable_bwe(Some(Bitrate::mbps(5)))`:启用 bandwidth estimation,初始估 5 Mbps
// - `.build(Instant::now())`:传入当前时刻作为 Rtc 内部时钟起点
// 类比 Go pion/webrtc`webrtc.NewAPI(webrtc.WithSettingEngine(...))`
let mut rtc = RtcConfig::new() let mut rtc = RtcConfig::new()
.enable_bwe(Some(Bitrate::mbps(5))) .enable_bwe(Some(Bitrate::mbps(5)))
.build(Instant::now()); .build(Instant::now());
// `UdpSocket::bind("0.0.0.0:0")`OS 随机分配端口(每 peer 独享一个 socket)。
// 类比 Go `net.ListenUDP("udp", &net.UDPAddr{Port: 0})`。
let socket = UdpSocket::bind("0.0.0.0:0")?; let socket = UdpSocket::bind("0.0.0.0:0")?;
socket.set_nonblocking(true)?; socket.set_nonblocking(true)?;
// 中文概述:调大 UDP 发送缓冲到 2MB(默认 ~208KB),原因详见下方英文注释。
// 然后用 `getsockopt` 读取内核实际分配的大小(Linux 可能受 `wmem_max` 截断,且
// 通常会翻倍)。Go 等价:`net.ListenConfig{Control: ...}`。
// Increase UDP send buffer to absorb IDR frame bursts (256KB IDR → ~145 RTP // Increase UDP send buffer to absorb IDR frame bursts (256KB IDR → ~145 RTP
// packets in a single poll_rtc loop). Default Linux wmem is ~208KB which // packets in a single poll_rtc loop). Default Linux wmem is ~208KB which
// causes EAGAIN on large keyframes. 2MB comfortably buffers several IDRs. // causes EAGAIN on large keyframes. 2MB comfortably buffers several IDRs.
const SND_BUF_REQ: usize = 2 * 1024 * 1024; const SND_BUF_REQ: usize = 2 * 1024 * 1024;
// 中文概述:调用 `setsockopt(SO_SNDBUF)` 调大 UDP 发送缓冲,然后用
// `getsockopt` 读取内核实际分配的大小(Linux 可能受 `wmem_max` 截断,且通常会
// 翻倍)。FFI 安全性论证见下方英文 `// SAFETY:` 块。
// SAFETY: fd is a valid UDP socket; setsockopt/getsockopt with SOL_SOCKET + // SAFETY: fd is a valid UDP socket; setsockopt/getsockopt with SOL_SOCKET +
// SO_SNDBUF are safe on Linux. We check the return value and log the actual // SO_SNDBUF are safe on Linux. We check the return value and log the actual
// kernel-assigned buffer (Linux may cap at wmem_max and/or double the value). // kernel-assigned buffer (Linux may cap at wmem_max and/or double the value).
@@ -443,13 +600,22 @@ impl WebRtcInner {
let local_addr = socket.local_addr()?; let local_addr = socket.local_addr()?;
// `local_ip().unwrap_or_else(closure)``Option<T>::unwrap_or_else` 类比 Go
// `if ip == "" { ip = "127.0.0.1" }`——`Option::None` 时执行闭包取兜底值。
let lan_ip = local_ip().unwrap_or_else(|| { let lan_ip = local_ip().unwrap_or_else(|| {
tracing::debug!("Failed to detect LAN IP, falling back to 127.0.0.1"); tracing::debug!("Failed to detect LAN IP, falling back to 127.0.0.1");
"127.0.0.1".to_string() "127.0.0.1".to_string()
}); });
// `format!("{lan_ip}:{}", port)`Rust 格式化宏,类比 Go `fmt.Sprintf("%s:%d", ...)`.
// `.parse::<SocketAddr>()`:字符串解析为 `SocketAddr``?` 自动传播 `AddrParseError`。
let candidate_addr: SocketAddr = format!("{lan_ip}:{}", local_addr.port()).parse()?; let candidate_addr: SocketAddr = format!("{lan_ip}:{}", local_addr.port()).parse()?;
// `Candidate::host(addr, "udp")`:构造 host ICE candidate(局域网用,无 STUN/TURN)。
// `.map_err(|e| anyhow::anyhow!(...))?`:把 str0m 自定义错误转成 `anyhow::Error`
// 并传播,类比 Go `if err != nil { return fmt.Errorf("candidate: %w", err) }`。
let candidate = Candidate::host(candidate_addr, "udp") let candidate = Candidate::host(candidate_addr, "udp")
.map_err(|e| anyhow::anyhow!("candidate: {e}"))?; .map_err(|e| anyhow::anyhow!("candidate: {e}"))?;
// `Rtc::add_local_candidate`:把 candidate 注册到 str0m,之后 SDP 协商时它会被
// 包含进 answer 的 `a=candidate:` 行。
rtc.add_local_candidate(candidate); rtc.add_local_candidate(candidate);
tracing::info!("WebRTC UDP: {candidate_addr} (bound 0.0.0.0)"); tracing::info!("WebRTC UDP: {candidate_addr} (bound 0.0.0.0)");
@@ -469,10 +635,27 @@ impl WebRtcInner {
}) })
} }
// SDP offer/answer 交换:解析浏览器 POST 来的 SDP offer JSON → 喂给 str0m 协商 →
// 返回 answer JSON。
//
// 关键步骤:
// 1. `serde_json::from_slice`:反序列化 SDP offer(类比 Go `json.Unmarshal`
// 2. `self.rtc.sdp_api().accept_offer(offer)`str0m 内部协商出 answer
// 副作用是设置 `Event::MediaAdded` 等待异步触发
// 3. `self.need_keyframe = true; self.force_keyframe_to_encode = true;`
// 协商完成后立即请求 IDR,让对端尽快解码首帧
// 4. `discover_video_params()`:扫描 str0m writer 找到 H.264 payload type
// 5. `serde_json::to_vec`:序列化 answer(类比 Go `json.Marshal`
fn handle_sdp_offer(&mut self, body: &[u8]) -> Result<String> { fn handle_sdp_offer(&mut self, body: &[u8]) -> Result<String> {
// `serde_json::from_slice::<SdpOffer>(body)`:把浏览器 POST 的 JSON 反序列化成
// str0m 的 `SdpOffer` 类型,类比 Go `json.Unmarshal(body, &offer)`。
// `.map_err(...)?`:把 serde 错误包装成 anyhow 错误并传播。
let offer: SdpOffer = let offer: SdpOffer =
serde_json::from_slice(body).map_err(|e| anyhow::anyhow!("parse SDP offer: {e}"))?; serde_json::from_slice(body).map_err(|e| anyhow::anyhow!("parse SDP offer: {e}"))?;
// `Rtc::sdp_api().accept_offer(offer)`str0m SDP 协商核心入口——
// 解析 offer 中的 m= 行、codec 列表、ICE candidate,构造对应的 answer。
// 副作用:触发后续 `Event::MediaAdded`(异步,要等 poll_rtc 才发)。
let answer = self let answer = self
.rtc .rtc
.sdp_api() .sdp_api()
@@ -491,6 +674,13 @@ impl WebRtcInner {
String::from_utf8(answer_json).map_err(|e| anyhow::anyhow!("answer utf8: {e}")) String::from_utf8(answer_json).map_err(|e| anyhow::anyhow!("answer utf8: {e}"))
} }
// 扫描 str0m 内部协商出的 codec 列表,找到 H.264 payload type`Pt`)。
// 在 SDP 协商后、`Event::MediaAdded` 后、`Event::Connected` 后各调用一次
// (三处调用是因为 str0m 的 codec 信息可能在不同时机可用——多保险)。
//
// 副作用:调用 `direct_api().stream_tx_by_mid(mid, None).set_unpaced(true)`
// 关闭 str0m 的 LeakyBucketPacer(默认每包加 ~100ms pacing 延迟,与我们的 VBV
// 8 Mbps 上限冲突;关掉后由编码器侧 VBV 做速率控制)。
fn discover_video_params(&mut self) { fn discover_video_params(&mut self) {
let mid = match self.video_mid { let mid = match self.video_mid {
Some(m) => m, Some(m) => m,
@@ -503,12 +693,18 @@ impl WebRtcInner {
// Disable str0m's LeakyBucketPacer for this video stream. Default pacing // Disable str0m's LeakyBucketPacer for this video stream. Default pacing
// adds ~100ms send latency per large IDR; our 8Mbps cap + VBV already // adds ~100ms send latency per large IDR; our 8Mbps cap + VBV already
// provide rate control. BWE stays enabled for adaptation feedback. // provide rate control. BWE stays enabled for adaptation feedback.
// `direct_api()` 返回 str0m 内部 API(不公开稳定接口),`stream_tx_by_mid(mid, None)`
// 取得该 mid 的发送流控制器;`set_unpaced(true)` 关闭 pacing。
if let Some(stream_tx) = self.rtc.direct_api().stream_tx_by_mid(mid, None) { if let Some(stream_tx) = self.rtc.direct_api().stream_tx_by_mid(mid, None) {
stream_tx.set_unpaced(true); stream_tx.set_unpaced(true);
} }
// `Rtc::writer(mid)` 返回媒体写入器,`payload_params()` 列出协商出的所有 codec。
// 我们扫描找 H.264`Codec::H264`)的 payload type,存入 `video_pt` 供后续 `write_h264_frame` 使用。
if let Some(writer) = self.rtc.writer(mid) { if let Some(writer) = self.rtc.writer(mid) {
for pp in writer.payload_params() { for pp in writer.payload_params() {
tracing::debug!("Codec: pt={:?} spec={:?}", pp.pt(), pp.spec()); tracing::debug!("Codec: pt={:?} spec={:?}", pp.pt(), pp.spec());
// `pp.spec().codec.is_video()`:先确认是视频 codec
// `pp.spec().codec == Codec::H264`:再确认是 H.264(非 VP8/VP9/AV1)。
if pp.spec().codec.is_video() && pp.spec().codec == Codec::H264 { if pp.spec().codec.is_video() && pp.spec().codec == Codec::H264 {
self.video_pt = Some(pp.pt()); self.video_pt = Some(pp.pt());
tracing::info!("H.264 payload type: {:?}", pp.pt()); tracing::info!("H.264 payload type: {:?}", pp.pt());
@@ -521,6 +717,8 @@ impl WebRtcInner {
} }
} }
// 内部不节流版本:直接置位 `need_keyframe` + `force_keyframe_to_encode`
// 并刷新 `last_forced_keyframe_at`(防紧接着 1 秒内的 viewer PLI 重复触发 IDR)。
/// Unthrottled keyframe trigger. Always sets the keyframe flags and refreshes /// Unthrottled keyframe trigger. Always sets the keyframe flags and refreshes
/// `last_forced_keyframe_at` so a follow-up viewer PLI within the next /// `last_forced_keyframe_at` so a follow-up viewer PLI within the next
/// `FORCED_KEYFRAME_MIN_INTERVAL` is dropped. /// `FORCED_KEYFRAME_MIN_INTERVAL` is dropped.
@@ -530,6 +728,8 @@ impl WebRtcInner {
self.last_forced_keyframe_at = Some(Instant::now()); self.last_forced_keyframe_at = Some(Instant::now());
} }
// 节流版本:仅在距离 `last_forced_keyframe_at` 已过 `FORCED_KEYFRAME_MIN_INTERVAL`
//1 秒)时才 honor,否则记 warn 日志并丢弃。对应 `Event::KeyframeRequest`PLI/FIR)。
/// Throttled keyframe trigger used for viewer-originated PLI/FIR requests. /// Throttled keyframe trigger used for viewer-originated PLI/FIR requests.
/// Honored only if enough time has elapsed since the last forced keyframe. /// Honored only if enough time has elapsed since the last forced keyframe.
fn request_keyframe_from_viewer(&mut self) { fn request_keyframe_from_viewer(&mut self) {
@@ -550,11 +750,24 @@ impl WebRtcInner {
} }
} }
// Sans-IO 推进主循环(出方向):取出 str0m 待发的 `Output::Transmit` 包写回 UDP
// 处理 `Output::Event`Connected/Disconnected/MediaAdded/KeyframeRequest/BWE 等)。
// 返回 `Ok(true)` 表示 peer 已断开(调用方应 drop `WebRtcInner`)。
//
// `Output::Timeout` 表示 str0m 需要在未来某时刻被再次唤醒——本实现简单 `break`,
// 依赖上层 mio 循环的 1ms tick 重新进入;更高性能的做法是读取 `_t` 安排 timer。
fn poll_rtc(&mut self) -> Result<bool> { fn poll_rtc(&mut self) -> Result<bool> {
loop { loop {
// `Rtc::poll_output()`str0m 主推进入口,返回 `Output` 枚举(Transmit/Event/Timeout
// 或 `Err`。Sans-IO 设计:调用方必须循环 poll 直到拿到 `Timeout`(表示 str0m
// 当前没活干了,等下一次外部输入)。
match self.rtc.poll_output() { match self.rtc.poll_output() {
// `Output::Transmit`str0m 想发的网络包(RTP/RTCP/DTLS/STUN)。
// 我们写回 UDP socket——这就是 Sans-IO 的"输出"侧。
Ok(Output::Transmit(t)) => { Ok(Output::Transmit(t)) => {
tracing::trace!("TX {} bytes -> {}", t.contents.len(), t.destination); tracing::trace!("TX {} bytes -> {}", t.contents.len(), t.destination);
// `UdpSocket::send_to` 类比 Go `conn.WriteToUDP(b, addr)`。
// `WouldBlock` = 内核发送缓冲满(罕见,因为我们在 new() 里调大了)。
if let Err(e) = self.socket.send_to(&t.contents, t.destination) { if let Err(e) = self.socket.send_to(&t.contents, t.destination) {
if e.kind() == std::io::ErrorKind::WouldBlock { if e.kind() == std::io::ErrorKind::WouldBlock {
tracing::debug!( tracing::debug!(
@@ -566,20 +779,28 @@ impl WebRtcInner {
} }
} }
} }
// `Output::Event`str0m 内部状态变化通知(ICE 连接、媒体添加、keyframe 请求等)。
// `Event` 是 enum,下方 `match &e` 对每种 variant 分发处理。
Ok(Output::Event(e)) => { Ok(Output::Event(e)) => {
tracing::debug!("RTC event: {e:?}"); tracing::debug!("RTC event: {e:?}");
match &e { match &e {
// `Event::Connected`ICE+DTLS 握手完成,可以发 RTP 了。
// 立即触发 IDR 请求(让对端解码器拿到关键帧尽快起播)+ 重新扫 codec 参数。
Event::Connected => { Event::Connected => {
tracing::info!("WebRTC connected!"); tracing::info!("WebRTC connected!");
self.connected = true; self.connected = true;
self.set_need_keyframe(); self.set_need_keyframe();
self.discover_video_params(); self.discover_video_params();
} }
// `Event::IceConnectionStateChange`ICE 状态变化。
// `Disconnected` 视为连接已死,向上层返回 `Ok(true)` 触发 drop。
Event::IceConnectionStateChange(IceConnectionState::Disconnected) => { Event::IceConnectionStateChange(IceConnectionState::Disconnected) => {
tracing::warn!("WebRTC disconnected"); tracing::warn!("WebRTC disconnected");
self.connected = false; self.connected = false;
return Ok(true); return Ok(true);
} }
// `Event::MediaAdded`SDP 协商后有新 m= 行就绪。
// 捕获视频 mid(只取第一个 sending direction 的视频流)。
Event::MediaAdded(ma) => { Event::MediaAdded(ma) => {
tracing::info!("Media added: mid={} kind={:?}", ma.mid, ma.kind); tracing::info!("Media added: mid={} kind={:?}", ma.mid, ma.kind);
if ma.kind == MediaKind::Video { if ma.kind == MediaKind::Video {
@@ -592,10 +813,15 @@ impl WebRtcInner {
} }
} }
} }
// `Event::KeyframeRequest`:对端发来 PLI/FIR,请求 IDR。
// 转发到节流版本 `request_keyframe_from_viewer`(防止 PLI 风暴)。
Event::KeyframeRequest(_) => { Event::KeyframeRequest(_) => {
tracing::info!("received keyframe request from viewer"); tracing::info!("received keyframe request from viewer");
self.request_keyframe_from_viewer(); self.request_keyframe_from_viewer();
} }
// `Event::EgressBitrateEstimate`BWE 推断的可用上行带宽。
// `BweKind::Twcc`Transport-CC,新标准)或 `BweKind::Remb`(老标准)。
// 提取数值存入 `current_bwe_estimate`,供 `state_portal.rs::select_resolution` 使用。
Event::EgressBitrateEstimate(est) => { Event::EgressBitrateEstimate(est) => {
let bitrate = match est { let bitrate = match est {
BweKind::Twcc(b) => *b, BweKind::Twcc(b) => *b,
@@ -613,6 +839,8 @@ impl WebRtcInner {
} }
} }
} }
// `Output::Timeout`str0m 内部定时器到期点。本实现忽略 `_t`(即下次唤醒时刻),
// 简单 `break`——上层 mio 循环 1ms tick 会很快再次调用 `poll_rtc`。
Ok(Output::Timeout(_t)) => break, Ok(Output::Timeout(_t)) => break,
Err(e) => { Err(e) => {
tracing::error!("rtc.poll_output error: {e}"); tracing::error!("rtc.poll_output error: {e}");
@@ -624,15 +852,27 @@ impl WebRtcInner {
Ok(false) Ok(false)
} }
// Sans-IO 推进主循环(入方向):从 UDP socket 读所有待处理包,封装为
// `Input::Receive` 喂给 str0m;最后喂一次 `Input::Timeout(now)` 推动内部时钟。
// 类比 Go pion/webrtc:手动调用 `peerConnection.Receive(rtpPacket)` 而不是
// 起 goroutine 监听 UDP。
fn feed_network(&mut self) -> Result<()> { fn feed_network(&mut self) -> Result<()> {
let mut recv_count = 0u32; let mut recv_count = 0u32;
loop { loop {
// `UdpSocket::recv_from(&mut self.buf)`:类比 Go `conn.ReadFrom(buf)`。
// 返回 `(n_bytes_read, source_addr)`。`WouldBlock`/`Interrupted` 是常态,
// 前者 break 出循环,后者重试(类比 Go EINTR 处理)。
match self.socket.recv_from(&mut self.buf) { match self.socket.recv_from(&mut self.buf) {
Ok((n, source)) => { Ok((n, source)) => {
recv_count += 1; recv_count += 1;
if recv_count <= 5 { if recv_count <= 5 {
tracing::trace!("UDP recv {} bytes from {}", n, source); tracing::trace!("UDP recv {} bytes from {}", n, source);
} }
// 构造 `Input::Receive`str0m 的"入包"事件。
// `Receive { proto, source, destination, contents }` 完整描述一个网络包:
// - `proto: Protocol::Udp`str0m 也支持 TCP,但 WebRTC 主流用 UDP
// - `source` / `destination`ICE candidate 端点
// - `contents``self.buf[..n]` 转 `Box<[u8]>``.try_into()` 因为 slice→Box 长度可能变化)
let input = Input::Receive( let input = Input::Receive(
Instant::now(), Instant::now(),
Receive { Receive {
@@ -644,6 +884,8 @@ impl WebRtcInner {
.map_err(|e| anyhow::anyhow!("receive contents: {e}"))?, .map_err(|e| anyhow::anyhow!("receive contents: {e}"))?,
}, },
); );
// `Rtc::handle_input(input)`:把入包喂给 str0m 解析(ICE/DTLS/SRTP/RTP/RTCP)。
// 这是 Sans-IO 的"输入"侧——str0m 不主动读 socket,全靠调用方喂。
self.rtc.handle_input(input).map_err(|e| { self.rtc.handle_input(input).map_err(|e| {
anyhow::anyhow!("handle_input({n} bytes from {source}): {e}") anyhow::anyhow!("handle_input({n} bytes from {source}): {e}")
})?; })?;
@@ -654,6 +896,8 @@ impl WebRtcInner {
} }
} }
// 喂一次 `Input::Timeout(now)`:让 str0m 推进内部定时器(重传、keepalive、BWE 周期等)。
// 即使没有任何入包,也必须定期调用,否则 str0m 内部超时不会触发。
self.rtc self.rtc
.handle_input(Input::Timeout(Instant::now())) .handle_input(Input::Timeout(Instant::now()))
.map_err(|e| anyhow::anyhow!("handle timeout: {e}"))?; .map_err(|e| anyhow::anyhow!("handle timeout: {e}"))?;
@@ -661,6 +905,17 @@ impl WebRtcInner {
Ok(()) Ok(())
} }
// 把一帧 H.264 NALUannex-B 格式,含 0x000001 起始码)写入 str0m,转 RTP 发出。
//
// 5 步:
// 1. 检查 `connected`、`video_mid`、`video_pt`,未就绪则 `Ok(false)` 静默丢帧
// 2. 若 `need_keyframe`,校验此帧必须是 IDRNAL type=5),否则丢帧等下一帧
// 3. PTS 90kHz 时钟 → RTP 时间戳(直接复用,因编码器 time_base = 1/90000
// 4. `Rtc::writer(mid).write(pt, now, rtp_time, data)`str0m 内部分包(>MTU 切片)
// 并加密 SRTP,产生 `Output::Transmit` 包
// 5. 立即 `poll_rtc()` 把 Transmit 包写回 UDP(同步发出,避免延迟)
//
// 返回 `Ok(true)` = peer 断开,调用方应 drop 本 `WebRtcInner`。
fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64) -> Result<bool> { fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64) -> Result<bool> {
if !self.connected { if !self.connected {
return Ok(false); return Ok(false);
@@ -696,10 +951,16 @@ impl WebRtcInner {
self.need_keyframe = false; self.need_keyframe = false;
} }
// PTS 90kHz → RTP 时间戳。`rtp_timestamp_from_pts_ticks` 把 i64 clamp 到 u64
//(见该函数文档)。`Frequency::NINETY_KHZ` 是视频 RTP 的标准时钟频率。
let rtp_timestamp = rtp_timestamp_from_pts_ticks(pts_ticks); let rtp_timestamp = rtp_timestamp_from_pts_ticks(pts_ticks);
self.rtp_clock = rtp_timestamp as u32; self.rtp_clock = rtp_timestamp as u32;
// `MediaTime::new(rtp_timestamp, Frequency::NINETY_KHZ)`:构造 str0m 媒体时间戳,
// 用于 RTP 头部 + jitter buffer 同步。
let rtp_time = MediaTime::new(rtp_timestamp, Frequency::NINETY_KHZ); let rtp_time = MediaTime::new(rtp_timestamp, Frequency::NINETY_KHZ);
// `Rtc::writer(mid)`:取得 mid 对应的媒体写入器(之前在 `discover_video_params` 用过)。
// None 表示 mid 还没就绪(罕见,已在前面的 video_mid 检查里处理)。
let writer = match self.rtc.writer(mid) { let writer = match self.rtc.writer(mid) {
Some(w) => w, Some(w) => w,
None => { None => {
@@ -714,6 +975,9 @@ impl WebRtcInner {
pt, pt,
self.rtp_clock self.rtp_clock
); );
// `writer.write(pt, Instant::now(), rtp_time, data)`:媒体写入入口。
// str0m 内部完成 (a) H.264 RTP 分包(FU-A for >MTU),(b) SRTP 加密,
// (c) 产生 `Output::Transmit` 包供 `poll_rtc` 取出。
writer writer
.write(pt, Instant::now(), rtp_time, data) .write(pt, Instant::now(), rtp_time, data)
.map_err(|e| anyhow::anyhow!("writer.write: {e}"))?; .map_err(|e| anyhow::anyhow!("writer.write: {e}"))?;
@@ -723,11 +987,15 @@ impl WebRtcInner {
Ok(should_destroy) Ok(should_destroy)
} }
// 简单 getter,对应 `Event::Connected` / `Event::IceConnectionStateChange(Disconnected)`。
fn is_connected(&self) -> bool { fn is_connected(&self) -> bool {
self.connected self.connected
} }
} }
// PTS→RTP 时间戳换算:编码器侧 time_base 已是 1/90000(与 RTP 视频时钟一致),
// 因此 1:1 直接复用,无需 fps-based 换算(旧版本曾用 `90000 / fps` 误导致时间戳错乱)。
// 返回 `u64` 喂 `MediaTime::new` 避免 u32 在 13.25 小时后过早回绕;str0m 内部处理 RTP u32 回绕。
/// Convert PTS in 90kHz media-clock ticks to RTP MediaTime ticks (u64). /// Convert PTS in 90kHz media-clock ticks to RTP MediaTime ticks (u64).
/// ///
/// With WebRTC encoder time_base = 1/90000, pts_ticks ARE RTP timestamps. /// With WebRTC encoder time_base = 1/90000, pts_ticks ARE RTP timestamps.
@@ -748,6 +1016,9 @@ fn extract_body(req: &str) -> &str {
} }
} }
// 探测本机 LAN IP(用于 ICE host candidate)。Go 等价:`net.Dial("udp", "1.1.1.1:80")`
// 后读 `LocalAddr()`——`connect` 不会发包,只设置路由表,从而选出默认网关对应的网卡 IP。
// `127.x` / `0.0.0.0` 视为无 LAN IP,由调用方 fallback 到 127.0.0.1loopback 调试用)。
fn local_ip() -> Option<String> { fn local_ip() -> Option<String> {
std::net::UdpSocket::bind("0.0.0.0:0").ok().and_then(|s| { std::net::UdpSocket::bind("0.0.0.0:0").ok().and_then(|s| {
s.connect("1.1.1.1:80").ok()?; s.connect("1.1.1.1:80").ok()?;
@@ -761,6 +1032,12 @@ fn local_ip() -> Option<String> {
}) })
} }
// 检测 H.264 NALU 流中是否含 IDR sliceNAL type=5)。两种起始码:
// - 4 字节 `00 00 00 01`AVCC boundary,主流)
// - 3 字节 `00 00 01` Annex-B inline,少见)
// NAL header 低 5 位 = type5 = IDR slice。SPS=7、PPS=8、SEI=6 等不算 IDR。
//
// 用于 `need_keyframe` 时丢非 IDR 帧——Go 等价:`bytes.Index(data, []byte{0,0,0,1})` 循环。
fn is_idr_nalu(data: &[u8]) -> bool { fn is_idr_nalu(data: &[u8]) -> bool {
let mut i = 0; let mut i = 0;
while i < data.len() { while i < data.len() {