流管理

StreamManager 是 Monibuca V6 的流管理核心，负责流的创建、发布、订阅和销毁的全生命周期管理。

流生命周期

flowchart LR
  C["Create<br/>创建"] --> P["Publish<br/>推流"]
  P --> S["Subscribe<br/>订阅"]
  S --> D["Destroy<br/>销毁"]
  P -.-> InitTracks["Init Tracks<br/>Set Codec<br/>Write Frames"]
  S -.-> ReadFrames["Read Frames<br/>Wait / Unsubscribe"]
  D -.-> Dispose["Dispose<br/>Clean Up<br/>Remove from Registry"]

详细状态流转

stateDiagram-v2
  [*] --> Init
  Init --> TrackAdded: init_video_track / init_audio_track
  TrackAdded --> Subscribed: add_subscriber
  Subscribed --> WaitingSubscriber: 所有订阅者离开
  WaitingSubscriber --> Subscribed: 新订阅者加入
  WaitingSubscriber --> WaitingReconnect: 推流断开且启用断连续推
  Subscribed --> WaitingReconnect: 推流断开且启用断连续推
  WaitingReconnect --> Subscribed: 重新推流
  WaitingReconnect --> Disposed: 超时
  Disposed --> [*]

  note right of Subscribed
    特殊状态 Paused：暂停数据超时检测
    （VOD / 回放场景）
  end note

StreamManager 职责

StreamManager 是流管理的统一入口，由三个子组件组成：

flowchart TB
  SM["StreamManager"]
  SM --> Reg["StreamRegistry<br/>DashMap 高并发存储"]
  SM --> Life["StreamLifecycle<br/>创建 / 销毁 / 断连续推"]
  SM --> WQ["WaitingQueue<br/>订阅者等待尚未发布的流"]

核心 API

impl StreamManager {
    // 创建流（推流端调用）
    fn create_publisher(stream_path) -> Publisher

    // 获取流（订阅端调用）
    fn get_publisher(stream_path) -> Option<Publisher>

    // 带 Dispatcher 的创建
    fn create_publisher_with_dispatcher(stream_path)
        -> (Publisher, Dispatcher)

    // 事件订阅
    fn subscribe_events() -> Receiver<StreamEvent>

    // 流信息查询
    fn stream_exists(stream_path) -> bool
    fn stream_count() -> usize
    fn list_streams() -> Vec<StreamInfo>
}

Publisher（推流者）

Publisher 负责管理一路流的推流端：

pub struct Publisher {
    stream_path: String,       // 流路径（如 "live/camera1"）
    plugin_name: String,       // 创建此流的插件名
    stream_type: String,       // 流类型（如 "rtmp", "rtsp"）
    video_track: Option<Arc<VideoTrack>>,  // 视频轨道
    audio_track: Option<Arc<AudioTrack>>,  // 音频轨道
    subscribers: DashMap<u64, SubscriberInfo>,  // 订阅者列表
    frame_notify: watch::Sender<u64>,  // 帧通知 channel
    config: PublisherConfig,   // 配置
    task: Option<Arc<Task>>,   // 关联的任务（层级取消）
}

推流配置

PublisherConfig {
    video_buffer_capacity: 1024,       // 视频缓冲区大小
    audio_buffer_capacity: 64,         // 音频缓冲区大小
    default_buffer_time: 2s,           // 默认缓冲时间
    max_buffer_time: 10s,              // 最大缓冲时间
    wait_timeout: 30s,                 // 无订阅者等待超时
    max_fps: 0,                        // 最大帧率（0=不限）
    data_timeout: 60s,                 // 数据超时（0=不超时）
    continue_push_timeout: 30s,        // 断连续推超时
}

推流流程

// 1. 创建 Publisher
let publisher = stream_manager.create_publisher("live/stream1")?;

// 2. 初始化轨道
let mut pub_guard = publisher.write();
pub_guard.init_video_track();
pub_guard.init_audio_track();

// 3. 设置编码器信息
pub_guard.set_video_codec(VideoCodec::new(H264, 1920, 1080));
pub_guard.set_audio_codec(AudioCodec::new(AAC, 44100, 2));

// 4. 写入帧数据（在推流循环中）
pub_guard.write_video(IDR, timestamp, dts, data)?;
pub_guard.write_audio(timestamp, data)?;

Subscriber（订阅者）

Subscriber 从 Publisher 的 RingBuffer 中消费帧数据：

三种订阅模式

模式	说明
RealTime	从最新的 IDR 帧开始播放（默认）
Buffer	从指定缓冲时间前的 IDR 帧开始
WaitKeyframe	等待下一个关键帧

订阅配置

SubscriberConfig {
    mode: SubscribeMode::RealTime,
    buffer_time: Duration::from_secs(2),
    receive_video: true,
    receive_audio: true,
    keyframe_timeout: Some(Duration::from_secs(30)),
}

订阅流程

// 1. 创建 Subscriber
let mut subscriber = Subscriber::new("live/stream1");

// 2. 订阅 Publisher
let publisher = stream_manager.get_publisher("live/stream1")?;
subscriber.subscribe(&publisher)?;

// 3. 读取帧数据（在播放循环中）
loop {
    subscriber.wait_for_frames().await?;

    while let Ok(Some(frame)) = subscriber.read_video() {
        // 处理视频帧
    }

    while let Some(frame) = subscriber.read_audio() {
        // 处理音频帧
    }
}

// 4. 取消订阅
subscriber.unsubscribe(&publisher);

自动 Opus→AAC 转码

当源流音频为 Opus 编码（如 WebRTC 推流）而订阅者需要 AAC（如 RTMP/FLV 播放）时，Subscriber 会自动进行转码：

flowchart LR
  WebRTC["WebRTC 推流 Opus"] --> Pub["Publisher"] --> Sub["Subscriber"]
  Sub -->|默认| AAC["AAC 自动转码"] --> RTMP["RTMP 拉流"]
  Sub -->|skip_opus_transcode=true| Direct["WebRTC 订阅者直接消费 Opus"]

音视频轨道管理

VideoTrack

flowchart TB
  VT["VideoTrack"]
  VT --> VRB["RingBuffer 1024 slots"]
  VT --> VC["VideoCodec H.264/H.265/AV1"]
  VT --> VSeq["SequenceGenerator"]
  VT --> VStats["Stats BPS/FPS"]
  VT --> VIDR["IDR Tracking"]

VideoTrack 支持两种帧写入格式：

• AVCC 格式: 来自 RTMP/FLV（长度前缀 NAL 单元）
• Raw NAL 格式: 来自 RTSP/RTP（裸 NAL 单元列表）

AudioTrack

flowchart TB
  AT["AudioTrack"]
  AT --> ARB["RingBuffer 64 slots"]
  AT --> AC["AudioCodec AAC/Opus/G.711/…"]
  AT --> ASeq["SequenceGenerator"]
  AT --> AStats["Stats BPS"]

序列头（Sequence Header）

Publisher 维护视频和音频的序列头：

• 视频序列头: AVC/HEVC Decoder Configuration Record
• 音频序列头: AAC AudioSpecificConfig

新订阅者连接时，Dispatcher 会先发送序列头，确保解码器正确初始化。

Pull/Push 代理

Pull Proxy（拉流代理）

从远程源拉流到本地发布：

flowchart LR
  Remote["远程 RTMP 服务器"] -->|拉流| Pub["Monibuca Publisher"] --> Local["本地订阅者"]

支持的拉流协议：RTMP、RTSP、SRT、HLS、HTTP-FLV、MP4

Push Proxy（推流代理）

将本地流推送到远程服务器：

flowchart LR
  Local["本地 Publisher"] -->|推流| Remote["远程 RTMP / RTSP / SRT 服务器"]

代理配置支持自动重试、重连间隔和最大重试次数。

录制

录制模块将流数据写入文件：

flowchart TB
  Pub["Publisher"] --> RecSub["Subscriber 录制"] --> FW["FileWriter"]
  FW --> FLV["FLV"] & MP4["MP4"] & HLS["HLS"]
  MP4 --> Frag["fMP4 / Raw"]

支持的录制格式：

格式	说明
FLV	FLV 文件录制
MP4	标准 MP4 文件
fMP4	碎片化 MP4（Fragmented MP4）
HLS	HLS TS 分片
Raw	原始帧数据

录制支持 Normal（可自动删除）、High（受保护）和 Event（事件触发）三种优先级。

转换器（Transformer）

Transformer 订阅源流，处理后发布为新流：

flowchart TB
  Src["源流 live/camera1"] --> X["Transformer<br/>订阅 → 处理 → 发布"] --> Dst["目标流 live/camera1_720p"]

应用场景：

• 视频转码（分辨率/编码格式转换）
• 添加水印/叠加层
• SEI 数据注入
• 格式转换

等待队列（WaitQueue）

当订阅者请求一个尚未发布的流时，不会立即失败，而是进入等待队列：

flowchart TB
  Req["订阅者请求 live/camera1"]
  Req --> Exists{流已存在?}
  Exists -->|是| Now["立即订阅"]
  Exists -->|否| WQ["加入 WaitQueue"]
  WQ --> Wait["等待流发布…"]
  Wait --> Pub{流已发布?}
  Pub -->|是| Auto["自动订阅"]
  Pub -->|超时| Err["返回错误"]

WaitQueue 使流的发布和订阅解耦 — 订阅者可以先于推流端启动。这在以下场景特别有用：

• 按需拉流: 订阅触发 Pull Proxy 从远程源拉流
• 设备重连: 设备断线后订阅者等待重连
• 直播预约: 观众提前进入直播间等待开播

断连续推

当推流端断开连接时，Monibuca 不会立即销毁流，而是进入 WaitingReconnect 状态：

flowchart TB
  Disc["推流端断开"] --> WR["Publisher → WaitingReconnect"]
  WR --> Choice{在 continue_push_timeout 内重连?}
  Choice -->|是| TakeOver["take_over 接管<br/>继承时间戳 · 转移序列头<br/>恢复订阅者 · 无感知切换"]
  Choice -->|否| Disp["Publisher → Disposed<br/>通知所有订阅者 EOS"]

关键设计: 断连续推保证了观看端在推流短暂中断时不需要重新连接，提供了无缝的观看体验。

默认断连续推超时为 30 秒，可通过 continue_push_timeout 配置。设置为 0 禁用此功能。

数据超时检测

Publisher 会检测推流端是否长时间未发送数据：

// 配置数据超时（默认 60 秒）
data_timeout: Duration::from_secs(60)

• 每次写入帧时更新 last_data_time
• 超时后自动 Dispose Publisher
• 在 Paused 状态下超时检测被禁用（用于 VOD/回放场景）
• 设置为 0 禁用超时检测

事件系统

StreamManager 通过 broadcast channel 发布流事件：

let mut events = stream_manager.subscribe_events();

loop {
    match events.recv().await {
        Ok(StreamEvent::Created { path, plugin }) => {
            println!("流创建: {} (插件: {})", path, plugin);
        }
        Ok(StreamEvent::Disposed { path, reason }) => {
            println!("流销毁: {} ({})", path, reason);
        }
        Ok(StreamEvent::SubscriberAdded { stream, subscriber_id, protocol }) => {
            println!("新订阅: {} #{} ({})", stream, subscriber_id, protocol);
        }
        Ok(StreamEvent::SubscriberRemoved { stream, subscriber_id }) => {
            println!("订阅离开: {} #{}", stream, subscriber_id);
        }
        _ => {}
    }
}

插件可以监听流事件来触发相应的业务逻辑（如按需录制、自动转码等）。